test(anomalies): each test runs in one event loop (#5740)

The app's pooled connections belong to the loop that opened them, so a
second asyncio.run in one test failed with "attached to a different loop"
(run 9116). Tests that do several passes now await them in one coroutine.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/integration/test_anomalies.py

@@ -61,7 +61,7 @@ async def _report(s, src: str, res: str, until: datetime, value: float):
                                      value=value, recorded_at=until))
 
 
-def _run(app, src: str, res: str, steps, *, seed: bool = True):
+async def _steps(app, src: str, res: str, steps, *, seed: bool = True):
     """Seed the normal (unless `seed` is false), then for each (at, value)
     report the window before `at` and run one detection pass at `at`. Returns
     the key's anomalies, oldest first, with their spans."""
@@ -69,25 +69,29 @@ def _run(app, src: str, res: str, steps, *, seed: bool = True):
     from steward.core.anomalies import detect
     from steward.models.norms import MetricAnomaly, MetricAnomalySpan
 
-    async def _go():
-        async with app.db_sessionmaker() as s:
-            if seed:
-                async with s.begin():
-                    s.add_all(_norms(src, res))
-            for at, value in steps:
-                async with s.begin():
-                    await _report(s, src, res, at, value)
-                async with s.begin():
-                    await detect(s, at)
-            rows = list((await s.execute(
-                select(MetricAnomaly).where(MetricAnomaly.resource_name == res)
-                .order_by(MetricAnomaly.started_at))).scalars())
-            spans = {a.id: list((await s.execute(
-                select(MetricAnomalySpan).where(MetricAnomalySpan.anomaly_id == a.id)
-                .order_by(MetricAnomalySpan.started_at))).scalars()) for a in rows}
-            return rows, spans
+    async with app.db_sessionmaker() as s:
+        if seed:
+            async with s.begin():
+                s.add_all(_norms(src, res))
+        for at, value in steps:
+            async with s.begin():
+                await _report(s, src, res, at, value)
+            async with s.begin():
+                await detect(s, at)
+        rows = list((await s.execute(
+            select(MetricAnomaly).where(MetricAnomaly.resource_name == res)
+            .order_by(MetricAnomaly.started_at))).scalars())
+        spans = {a.id: list((await s.execute(
+            select(MetricAnomalySpan).where(MetricAnomalySpan.anomaly_id == a.id)
+            .order_by(MetricAnomalySpan.started_at))).scalars()) for a in rows}
+        return rows, spans
 
-    return asyncio.run(_go())
+
+def _run(app, src: str, res: str, steps):
+    """_steps for a test that needs nothing else. A test doing more runs it
+    all in one asyncio.run: the app's pooled connections belong to the loop
+    that opened them, so a second loop can't use them."""
+    return asyncio.run(_steps(app, src, res, steps))
 
 
 def _res(tag: str) -> str:
@@ -140,10 +144,10 @@ def test_an_ended_anomaly_lapses_after_the_window_and_an_ongoing_one_never(app,
     from steward.models.norms import MetricAnomaly
     ended, going = _res("ended"), _res("going")
     end = NOW + timedelta(minutes=30)
-    _run(app, src, ended, [(NOW, 60.0), (end, 10.0)])
-    _run(app, src, going, [(NOW + timedelta(days=2), 60.0)])
 
     async def _go():
+        await _steps(app, src, ended, [(NOW, 60.0), (end, 10.0)])
+        await _steps(app, src, going, [(NOW + timedelta(days=2), 60.0)])
         async with app.db_sessionmaker() as s:
             async with s.begin():
                 # The ongoing one keeps reporting high, so it stays open.
@@ -171,15 +175,15 @@ def test_an_ended_anomaly_lapses_after_the_window_and_an_ongoing_one_never(app,
 def test_a_repeat_after_review_is_a_new_row(app, src):
     from steward.core.anomalies import REAL, TO_REVIEW, mark
     res = _res("again")
-    rows, _ = _run(app, src, res, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)])
 
-    async def _mark():
+    async def _go():
+        rows, _ = await _steps(app, src, res, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)])
         async with app.db_sessionmaker() as s:
             async with s.begin():
                 await mark(s, rows[0].id, REAL, "tester", "disk filled")
+        return await _steps(app, src, res, [(NOW + timedelta(hours=5), 60.0)], seed=False)
 
-    asyncio.run(_mark())
-    rows, _ = _run(app, src, res, [(NOW + timedelta(hours=5), 60.0)], seed=False)
+    rows, _ = asyncio.run(_go())
     assert [(a.review, a.occurrences) for a in rows] == [(REAL, 1), (TO_REVIEW, 1)]
     assert rows[0].note == "disk filled" and rows[0].reviewed_by == "tester"
 
@@ -187,11 +191,13 @@ def test_a_repeat_after_review_is_a_new_row(app, src):
 def test_to_review_is_listed_first(app, src):
     from steward.core.anomalies import NORMAL, TO_REVIEW, list_anomalies, mark
     older, newer = _res("older"), _res("newer")
-    old_rows, _ = _run(app, src, older, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)])
-    new_rows, _ = _run(app, src, newer, [(NOW + timedelta(hours=2), 60.0),
-                                         (NOW + timedelta(hours=2, minutes=30), 10.0)])
+    seen = {}
 
     async def _go():
+        old_rows, _ = await _steps(app, src, older, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)])
+        new_rows, _ = await _steps(app, src, newer, [(NOW + timedelta(hours=2), 60.0),
+                                                     (NOW + timedelta(hours=2, minutes=30), 10.0)])
+        seen["older"] = old_rows[0].review
         async with app.db_sessionmaker() as s:
             async with s.begin():
                 await mark(s, new_rows[0].id, NORMAL, "tester")
@@ -199,7 +205,7 @@ def test_to_review_is_listed_first(app, src):
             return [a.resource_name for a in got if a.resource_name in (older, newer)]
 
     assert asyncio.run(_go()) == [older, newer]
-    assert old_rows[0].review == TO_REVIEW
+    assert seen["older"] == TO_REVIEW
 
 
 @pytest.mark.parametrize("review,learned_hold", [
@@ -250,11 +256,11 @@ def test_cleanup_prunes_reviewed_anomalies_and_never_one_to_review(app, src):
     from steward.core.anomalies import REAL, mark, prune
     from steward.models.norms import MetricAnomaly
     kept, gone = _res("kept"), _res("gone")
-    _run(app, src, kept, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)])
     later = NOW + timedelta(days=1)
-    rows, _ = _run(app, src, gone, [(later, 60.0), (later + timedelta(minutes=30), 10.0)])
 
     async def _go():
+        await _steps(app, src, kept, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)])
+        rows, _ = await _steps(app, src, gone, [(later, 60.0), (later + timedelta(minutes=30), 10.0)])
         async with app.db_sessionmaker() as s:
             async with s.begin():
                 await mark(s, rows[0].id, REAL, "tester")