fix(rustdesk): punch-hole version gate via deployed_identity (loud on unparsable, skip <1.1.16)

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:
+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))."""
+    """The running app service's image tag (e.g. rustdesk/rustdesk-server:1.1.16 -> (1,1,16)),
-    out = subprocess.run(
+    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
-            "docker",
+    never silently run (vacuous-green) or silently skip (the first enrollment run shipped a
-            "service",
+    None-returning hand-rolled inspect and observed pre-fix answering on a 1.1.12 head)."""
-            "inspect",
+    from harness import lifecycle  # noqa: PLC0415  (lazy: keeps import cost off collection)
-            f"{domain.replace('.', '_')}_app",
+
-            "--format",
+    image = (lifecycle.deployed_identity(domain).get("image") or "").strip()
-            "{{.Spec.TaskTemplate.ContainerSpec.Image}}",
+    tag = image.split("@", 1)[0].rsplit(":", 1)[-1]
-        ],
-        capture_output=True,
-        text=True,
-        timeout=60,
-    ).stdout.strip()
-    tag = out.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"