fix(tests): align stale tests with current behavior

- swarm/test_swarm_api, swarm/test_heartbeat: replace deprecated
  asyncio.get_event_loop().run_until_complete() with asyncio.run();
  the former raises in 3.11 once another test has set+closed a loop on
  the main thread.
- prober/test_prober_bus, prober/test_prober_worker: extend tcp_fingerprint
  mocks with tos/dscp/ecn/server_isn so the worker doesn't KeyError into
  the prober_error branch.
- services/test_service_isolation: collector now retries on event-stream
  errors instead of exiting; assert it stays running and cancel cleanly.
- live/test_imap_live, live/test_pop3_live: log format emits
  outcome="failure", not "failed".
- live/test_service_isolation_live: is_service_container accepts label
  OR state-name; rewrite the empty-state test against a synthetic
  unlabeled container instead of the host's real fleet.

tests/live/test_imap_live.py

@@ -60,7 +60,7 @@ class TestIMAPLive:
             pass
         lines += drain()
         matched = assert_rfc5424(lines, service="imap", event_type="auth")
-        assert "failed" in matched, f"Expected auth failure in log. Got:\n{matched!r}"
+        assert "failure" in matched, f"Expected auth failure in log. Got:\n{matched!r}"
 
     def test_select_inbox_after_login(self, live_service):
         port, drain = live_service("imap")

tests/live/test_pop3_live.py

@@ -55,4 +55,4 @@ class TestPOP3Live:
             pass
         lines += drain()
         matched = assert_rfc5424(lines, service="pop3", event_type="auth")
-        assert "failed" in matched, f"Expected auth failure in log. Got:\n{matched!r}"
+        assert "failure" in matched, f"Expected auth failure in log. Got:\n{matched!r}"

tests/live/test_service_isolation_live.py

@@ -137,29 +137,31 @@ class TestCollectorLiveIsolation:
     """Real collector behaviour against the actual Docker daemon."""
 
     async def test_collector_finds_no_deckies_without_state(self, tmp_path):
-        """With no deckies in state, collector's container scan finds nothing.
+        """With no deckies in state and no DECNET labels, the scan rejects
+        every container.
 
-        We avoid calling the full worker because client.events() blocks
-        the thread indefinitely — instead we test the scan logic directly
-        against the real Docker daemon.
+        is_service_container has two acceptance paths:
+          1. label-based (decnet.fleet.service / decnet.topology.service)
+          2. name match against decnet-state.json
+
+        With state empty AND labels absent, both paths must reject.  We
+        feed synthetic container objects (no real Docker call) so the
+        result is independent of whatever fleet may already be running on
+        the host — which would otherwise satisfy path (1).
         """
-        import docker
         import decnet.config as cfg
+        from unittest.mock import MagicMock
 
         original_state = cfg.STATE_FILE
         try:
             cfg.STATE_FILE = tmp_path / "empty-state.json"
 
-            # Real Docker client, real container list — but no state means
-            # is_service_container rejects everything.
-            client = docker.from_env()
-            matched = [c for c in client.containers.list() if is_service_container(c)]
-            client.close()
+            unlabeled = MagicMock()
+            unlabeled.name = "some-random-container"
+            unlabeled.attrs = {"Config": {"Labels": {}}}
+            unlabeled.labels = {}
 
-            assert matched == [], (
-                f"Expected no matching containers without state, got: "
-                f"{[c.name for c in matched]}"
-            )
+            assert is_service_container(unlabeled) is False
         finally:
             cfg.STATE_FILE = original_state
 

tests/prober/test_prober_bus.py

@@ -114,6 +114,10 @@ def test_tcpfp_phase_invokes_publish_fn_on_success(monkeypatch, tmp_path: Path)
             "sack_ok": True,
             "timestamp": True,
             "options_order": "mss,sack,ts,nop,wscale",
+            "tos": 0,
+            "dscp": 0,
+            "ecn": 0,
+            "server_isn": 0,
         },
     )
 

tests/prober/test_prober_worker.py

@@ -342,6 +342,7 @@ class TestProbeCycleTCPFP:
             "ttl": 64, "window_size": 65535, "df_bit": 1,
             "mss": 1460, "window_scale": 7, "sack_ok": 1,
             "timestamp": 1, "options_order": "M,N,W,N,N,T,S,E",
+            "tos": 0, "dscp": 0, "ecn": 0, "server_isn": 0,
         }
         log_path = tmp_path / "decnet.log"
         json_path = tmp_path / "decnet.json"
@@ -368,6 +369,7 @@ class TestProbeCycleTCPFP:
             "ttl": 128, "window_size": 8192, "df_bit": 1,
             "mss": 1460, "window_scale": 8, "sack_ok": 1,
             "timestamp": 0, "options_order": "M,N,W,N,N,S",
+            "tos": 0, "dscp": 0, "ecn": 0, "server_isn": 0,
         }
         log_path = tmp_path / "decnet.log"
         json_path = tmp_path / "decnet.json"

tests/services/test_service_isolation.py

@@ -16,6 +16,7 @@ worker degrades gracefully while unrelated workers remain unaffected.
 """
 
 import asyncio
+import contextlib
 import json
 import os
 import time
@@ -68,8 +69,12 @@ class TestCollectorIsolation:
             with patch("decnet.config.load_state", return_value=None):
                 task = asyncio.create_task(log_collector_worker("/tmp/decnet-test-collector.log"))
                 await asyncio.sleep(0.1)
-                assert task.done()
-                assert task.exception() is None
+                # Collector now retries on event-stream errors instead of
+                # exiting; it should still be running (i.e. surviving) here.
+                assert not task.done()
+                task.cancel()
+                with contextlib.suppress(asyncio.CancelledError):
+                    await task
 
     def test_collector_container_filter_with_unknown_containers(self):
         """is_service_container must reject containers not in state."""

tests/swarm/test_heartbeat.py

@@ -167,7 +167,7 @@ def test_heartbeat_happy_path_primary_extraction(
         assert s["archetype"] == "generic"
         assert s["service_config"] == {"ssh": {"port": 22}}
 
-    asyncio.get_event_loop().run_until_complete(_verify())
+    asyncio.run(_verify())
 
 
 def test_heartbeat_fallback_extraction_path_also_accepted(
@@ -241,7 +241,7 @@ def test_heartbeat_decommissioned_host_returns_404(
         ok = await repo.delete_swarm_host(host["host_uuid"])
         assert ok
 
-    asyncio.get_event_loop().run_until_complete(_delete())
+    asyncio.run(_delete())
 
     _pin_fingerprint(monkeypatch, fp)
     resp = client.post(
@@ -272,7 +272,7 @@ def test_heartbeat_deployed_false_bumps_host_but_writes_no_shards(
         shards = await repo.list_decky_shards(host["host_uuid"])
         assert shards == []
 
-    asyncio.get_event_loop().run_until_complete(_verify())
+    asyncio.run(_verify())
 
 
 def test_heartbeat_decky_missing_from_runtime_is_degraded(
@@ -297,4 +297,4 @@ def test_heartbeat_decky_missing_from_runtime_is_degraded(
         assert by["decky-01"]["state"] == "running"
         assert by["decky-02"]["state"] == "degraded"
 
-    asyncio.get_event_loop().run_until_complete(_verify())
+    asyncio.run(_verify())

tests/swarm/test_swarm_api.py

@@ -464,7 +464,7 @@ def test_list_deckies_joins_host_identity(client: TestClient, repo) -> None:
             "services": ["smb", "ssh"], "state": "failed", "last_error": "boom",
         })
 
-    asyncio.get_event_loop().run_until_complete(_seed())
+    asyncio.run(_seed())
 
     rows = client.get("/swarm/deckies").json()
     assert len(rows) == 2