From b889584906771c8da742559f2c528a82201766bb Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 14 Oct 2026 13:46:51 -0400 Subject: [PATCH] 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 --- tests/integration/test_anomalies.py | 68 ++++++++++++++++------------- 1 file changed, 37 insertions(+), 31 deletions(-) diff --git a/tests/integration/test_anomalies.py b/tests/integration/test_anomalies.py index 7d3a815..86d092c 100644 --- a/tests/integration/test_anomalies.py +++ b/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")