diff --git a/tests/rustdesk-server/custom/test_punch_hole_no_amplification.py b/tests/rustdesk-server/custom/test_punch_hole_no_amplification.py index 19ce8af..77d4eee 100644 --- a/tests/rustdesk-server/custom/test_punch_hole_no_amplification.py +++ b/tests/rustdesk-server/custom/test_punch_hole_no_amplification.py @@ -36,31 +36,28 @@ import _rustdesk_proto as proto # noqa: E402 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "runner")) -def _deployed_app_version(domain: str) -> tuple[int, int, int] | None: - """The running app service's image tag (e.g. rustdesk/rustdesk-server:1.1.16 -> (1,1,16)).""" - out = subprocess.run( - [ - "docker", - "service", - "inspect", - f"{domain.replace('.', '_')}_app", - "--format", - "{{.Spec.TaskTemplate.ContainerSpec.Image}}", - ], - capture_output=True, - text=True, - timeout=60, - ).stdout.strip() - tag = out.rsplit(":", 1)[-1] +def _deployed_app_version(domain: str) -> tuple[int, int, int]: + """The running app service's image tag (e.g. rustdesk/rustdesk-server:1.1.16 -> (1,1,16)), + via the harness's first-class deployed_identity (labels + image in one inspect). Fails + LOUDLY on an unparsable image rather than returning None — a gate that cannot decide must + never silently run (vacuous-green) or silently skip (the first enrollment run shipped a + None-returning hand-rolled inspect and observed pre-fix answering on a 1.1.12 head).""" + from harness import lifecycle # noqa: PLC0415 (lazy: keeps import cost off collection) + + image = (lifecycle.deployed_identity(domain).get("image") or "").strip() + tag = image.split("@", 1)[0].rsplit(":", 1)[-1] try: - return tuple(int(x) for x in tag.split(".")[:3]) + return tuple(int(x) for x in tag.split(".")[:3]) # type: ignore[return-value] except ValueError: - return None + pytest.fail( + f"cannot determine deployed rustdesk-server version from image {image!r} — fix " + "the read before trusting this test's verdict" + ) def test_unauthenticated_punch_hole_is_not_answered(live_app): version = _deployed_app_version(live_app) - if version is not None and version < (1, 1, 16): + if version < (1, 1, 16): pytest.skip( f"amplification fix landed upstream in 1.1.16 (running {version}) — pre-fix " "answering is expected behavior, covered by the upgrade tier's base deploy"