feat: add Sascha direct media tunnel support (tests/test_app.py)

tests/test_app.py

@@ -43,7 +43,7 @@ def test_health_exposes_current_version():
     with TestClient(app.app) as client:
         response = client.get("/health")
     assert response.status_code == 200
-    assert response.json()["version"] == app.VERSION == "2.3.6"
+    assert response.json()["version"] == app.VERSION == "2.3.7"
 
 
 def test_media_handoff_proxies_strict_category_contract(monkeypatch):
@@ -392,6 +392,9 @@ def test_capabilities_is_live_machine_readable_safety_map():
     assert removal["mode"] == "mutation"
     assert removal["dry_run"] is True
     assert removal["critical"] is True
+    tunnel = by_operation[("POST", "/network/media-tunnel/sascha")]
+    assert tunnel["dry_run"] is True
+    assert tunnel["critical"] is True
     assert by_operation[("DELETE", "/vm/destroy/{vmid}")]["dry_run"] is True
     assert all(item["path"] != "/{service}/{path}" for item in payload["operations"])
     assert payload["model_contract"]["instruction"].startswith("Prefer read_only")
@@ -1274,3 +1277,70 @@ def test_speedtest_deploy_requires_strong_secrets_and_uses_full_git_app(monkeypa
     assert deploy[1]["compose.yaml"] == "content:compose.yaml"
     assert deploy[2] == "correct-horse-battery-staple"
     assert deploy[3] == "streamscope-session-secret-with-entropy"
+
+
+def test_sascha_media_edge_audit_uses_fixed_hetzner_host(monkeypatch):
+    payload = {
+        "hostname": "pfannkuchen",
+        "caddy": {"container_running": True, "protocols": ["h1", "h2", "h3"], "upstreams": ["10.6.1.103:8096"]},
+        "network": {"wg_media": False, "udp_51821": False},
+    }
+    calls = []
+    monkeypatch.setattr(app, "_find_inventory_host", lambda name: {"name": name, "user": "root", "ip": "46.225.230.72"})
+    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=30: (calls.append((host, command, timeout)) or (0, json.dumps(payload), "")))
+    with TestClient(app.app) as client:
+        response = client.get("/media/edge/sascha", headers={"Authorization": "Bearer test-token"})
+    assert response.status_code == 200
+    assert response.json()["caddy"]["upstreams"] == ["10.6.1.103:8096"]
+    assert calls[0][0] == "root@46.225.230.72"
+    assert "PrivateKey" not in response.text
+
+
+def test_sascha_media_tunnel_defaults_to_side_effect_free_dry_run(monkeypatch):
+    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("dry-run must not SSH")))
+    with TestClient(app.app) as client:
+        response = client.post(
+            "/network/media-tunnel/sascha",
+            headers={"Authorization": "Bearer test-token"},
+            json={},
+        )
+    assert response.status_code == 200
+    body = response.json()
+    assert body["status"] == "would_deploy"
+    assert body["vps_address"] == "10.11.13.1/32"
+    assert body["emby_address"] == "10.11.13.3/32"
+    assert body["listen_port"] == 51821
+
+
+def test_sascha_media_tunnel_requires_explicit_confirmation(monkeypatch):
+    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("unconfirmed request must not SSH")))
+    with TestClient(app.app) as client:
+        response = client.post(
+            "/network/media-tunnel/sascha",
+            headers={"Authorization": "Bearer test-token"},
+            json={"dry_run": False},
+        )
+    assert response.status_code == 400
+
+
+def test_sascha_media_tunnel_apply_returns_redacted_result(monkeypatch):
+    result = {
+        "status": "deployed",
+        "interface": "wg-media",
+        "vps_address": "10.11.13.1/32",
+        "emby_address": "10.11.13.3/32",
+        "listen_port": 51821,
+        "handshake": True,
+        "ping_vps_to_emby": True,
+        "ping_emby_to_vps": True,
+    }
+    monkeypatch.setattr(app, "_deploy_sascha_media_tunnel", lambda: result)
+    with TestClient(app.app) as client:
+        response = client.post(
+            "/network/media-tunnel/sascha",
+            headers={"Authorization": "Bearer test-token"},
+            json={"dry_run": False, "confirmation": "DEPLOY_DIRECT_SASCHA_MEDIA_TUNNEL"},
+        )
+    assert response.status_code == 200
+    assert response.json()["handshake"] is True
+    assert "private" not in response.text.lower()