""" Story 9.05 — Test Suite ======================== NightlyEpochRunner: Full Orchestration + Epoch Log File under test: core/epoch/nightly_epoch_runner.py BB Tests (5): BB1: Full run with 3 conversations → epoch log written with status="success" BB2: distill() failure → epoch log written with status="failed" + error key BB3: duration_s > 0 in log BB4: conversations_processed count matches aggregate_week output BB5: axioms_written count matches write_axioms return value WB Tests (6): WB1: All 3 stages called in order (aggregate_week → distill → write_axioms) WB2: Timing uses time.monotonic() (not datetime diff) — mock time.monotonic WB3: Exception caught at top level, status="failed" logged WB4: EPOCH_LOG_PATH file created with os.makedirs for parent dirs WB5: Epoch log is JSONL (append mode, not overwrite) WB6: aggregate_week failure → status="failed", conversations_processed=0 Regression: R1: aggregate_week() still works (Story 9.02) R2: distill() still works (Story 9.03) R3: write_axioms() still works (Story 9.04) All tests use MagicMock / AsyncMock — zero live API calls. All file I/O uses tmp_path pytest fixture to redirect EPOCH_LOG_PATH so the real observability log is never touched. """ from __future__ import annotations import asyncio import json import os from pathlib import Path from unittest.mock import AsyncMock, MagicMock, call, patch import pytest import sys sys.path.insert(0, "/mnt/e/genesis-system") from core.epoch.nightly_epoch_runner import ( EPOCH_LOG_PATH, NightlyEpochRunner, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _arun(coro): """Run a coroutine synchronously.""" return asyncio.run(coro) def _make_axioms(n: int) -> list[dict]: """Build *n* valid axiom dicts.""" categories = ["preference", "fact", "strategy", "directive"] return [ { "id": f"epoch_2026_02_25_{i:03d}", "content": f"Axiom content number {i}", "category": categories[i % len(categories)], "confidence": round(0.7 + (i % 3) * 0.1, 1), } for i in range(n) ] def _make_rows(n: int) -> list[tuple]: """Build *n* fake DB rows matching AGGREGATION_QUERY column order.""" return [ ( f"conv-{i:04d}", f"2026-02-{i + 1:02d}", f"transcript text {i}", {"entity": i}, [f"decision-{i}"], [f"action-{i}"], [f"fact-{i}"], [f"directive-{i}"], ) for i in range(n) ] def _make_pg_conn(rows: list[tuple]) -> MagicMock: """Return a mock Postgres connection that returns *rows* on fetchall.""" cursor_mock = MagicMock() cursor_mock.fetchall.return_value = rows conn_mock = MagicMock() conn_mock.cursor.return_value = cursor_mock return conn_mock def _make_gemini_client( axioms: list[dict] | None = None, week_summary: str = "A great week.", ) -> MagicMock: """ Return a MagicMock Gemini client whose generate() returns a JSON string matching the distillation response shape, and whose embed() returns a fake vector. """ ax = axioms if axioms is not None else _make_axioms(3) response = json.dumps({"axioms": ax, "week_summary": week_summary}) client = MagicMock() client.generate = AsyncMock(return_value=response) client.embed = AsyncMock(return_value=[0.1] * 768) return client def _make_qdrant_client() -> MagicMock: """Return a MagicMock Qdrant client.""" client = MagicMock() client.upsert = MagicMock(return_value=None) return client def _patch_epoch_log(tmp_path: Path): """ Return the epoch log path in tmp_path and a patch context manager that redirects EPOCH_LOG_PATH to it. """ log_file = str(tmp_path / "epoch_log.jsonl") return log_file, patch( "core.epoch.nightly_epoch_runner.EPOCH_LOG_PATH", log_file, ) def _patch_all_output_paths(tmp_path: Path): """ Redirect KG_AXIOM_PATH, WEEKLY_SUMMARY_PATH, and EPOCH_LOG_PATH to tmp_path so the real files are never touched. """ axiom_file = str(tmp_path / "axioms.jsonl") summary_file = str(tmp_path / "summaries.jsonl") log_file = str(tmp_path / "epoch_log.jsonl") return ( axiom_file, summary_file, log_file, patch("core.epoch.nightly_epoch_runner.KG_AXIOM_PATH", axiom_file), patch("core.epoch.nightly_epoch_runner.WEEKLY_SUMMARY_PATH", summary_file), patch("core.epoch.nightly_epoch_runner.EPOCH_LOG_PATH", log_file), ) # --------------------------------------------------------------------------- # BB1 — Full run with 3 conversations → epoch log written with status="success" # --------------------------------------------------------------------------- class TestBB1_FullRunSuccessEpochLog: """BB1: A complete successful run must write an epoch log with status='success'.""" def test_epoch_log_file_created(self, tmp_path): """BB1: epoch_log.jsonl must be created after a successful run.""" rows = _make_rows(3) pg = _make_pg_conn(rows) gemini = _make_gemini_client(axioms=_make_axioms(3), week_summary="Great week.") qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) _arun(runner.run_epoch()) assert Path(log_file).exists(), ( "epoch_log.jsonl must be created after run_epoch()" ) def test_epoch_log_status_success(self, tmp_path): """BB1: Epoch log entry must have status='success' when all stages succeed.""" rows = _make_rows(3) pg = _make_pg_conn(rows) gemini = _make_gemini_client(axioms=_make_axioms(3), week_summary="Great week.") qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) result = _arun(runner.run_epoch()) assert result["status"] == "success", ( f"Expected status='success', got {result['status']!r}" ) def test_epoch_log_has_required_keys(self, tmp_path): """BB1: Epoch log dict must contain all 6 required keys.""" rows = _make_rows(3) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) result = _arun(runner.run_epoch()) required_keys = { "date", "conversations_processed", "axioms_written", "week_summary", "duration_s", "status", } missing = required_keys - set(result.keys()) assert not missing, ( f"Epoch log missing required keys: {missing}" ) def test_epoch_log_written_as_valid_json_line(self, tmp_path): """BB1: The line written to epoch_log.jsonl must be valid JSON.""" rows = _make_rows(2) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) _arun(runner.run_epoch()) content = Path(log_file).read_text().strip() assert content, "epoch_log.jsonl must not be empty after run_epoch()" parsed = json.loads(content.splitlines()[0]) assert isinstance(parsed, dict), ( f"Epoch log line is not a JSON object: {content!r}" ) def test_epoch_log_no_error_key_on_success(self, tmp_path): """BB1: The 'error' key must NOT appear in the log when status='success'.""" rows = _make_rows(1) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) result = _arun(runner.run_epoch()) assert "error" not in result, ( f"'error' key must not appear on success, got: {result}" ) # --------------------------------------------------------------------------- # BB2 — distill() failure → epoch log written with status="failed" + error key # --------------------------------------------------------------------------- class TestBB2_DistillFailureEpochLogFailed: """BB2: If distill() raises, epoch log must record status='failed' + error. Note: distill() internally catches Gemini API errors and returns safe fallbacks. To test run_epoch()'s failure path we patch distill() on the runner instance directly so it raises — bypassing distill()'s own internal error handling. """ def test_distill_failure_status_is_failed(self, tmp_path): """BB2: distill() raising → epoch log status='failed'.""" rows = _make_rows(2) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) # Patch distill() directly to raise — bypasses distill's own catch async def failing_distill(conversations): raise RuntimeError("Gemini timeout") runner.distill = failing_distill result = _arun(runner.run_epoch()) assert result["status"] == "failed", ( f"Expected status='failed' after distill() error, got {result['status']!r}" ) def test_distill_failure_error_key_present(self, tmp_path): """BB2: distill() raising → epoch log contains 'error' key with message.""" rows = _make_rows(2) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) async def failing_distill(conversations): raise RuntimeError("Gemini timeout") runner.distill = failing_distill result = _arun(runner.run_epoch()) assert "error" in result, ( f"Expected 'error' key in epoch log on failure, got: {result}" ) assert "Gemini timeout" in result["error"], ( f"Expected error message to contain 'Gemini timeout', got: {result['error']!r}" ) def test_distill_failure_epoch_log_still_written(self, tmp_path): """BB2: Even on distill() failure, epoch_log.jsonl must be written.""" rows = _make_rows(1) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) async def failing_distill(conversations): raise ValueError("bad response") runner.distill = failing_distill _arun(runner.run_epoch()) assert Path(log_file).exists(), ( "epoch_log.jsonl must still be written even after distill() failure" ) def test_run_epoch_does_not_raise_on_stage_failure(self, tmp_path): """BB2: run_epoch() must NOT propagate stage exceptions to the caller.""" rows = _make_rows(2) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) async def failing_distill(conversations): raise RuntimeError("boom") runner.distill = failing_distill try: result = _arun(runner.run_epoch()) except Exception as exc: pytest.fail( f"run_epoch() must not propagate stage exceptions, " f"but raised: {exc}" ) # --------------------------------------------------------------------------- # BB3 — duration_s > 0 in log # --------------------------------------------------------------------------- class TestBB3_DurationPositive: """BB3: duration_s in the epoch log must be a positive float.""" def test_duration_s_is_positive(self, tmp_path): """BB3: duration_s > 0 for any non-trivial run.""" rows = _make_rows(2) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) result = _arun(runner.run_epoch()) assert result["duration_s"] >= 0, ( f"duration_s must be >= 0, got {result['duration_s']}" ) assert isinstance(result["duration_s"], float), ( f"duration_s must be a float, got {type(result['duration_s'])}" ) def test_duration_s_is_a_float(self, tmp_path): """BB3: duration_s must be a float (from round(x, 3)).""" rows = _make_rows(1) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) result = _arun(runner.run_epoch()) assert isinstance(result["duration_s"], float), ( f"duration_s must be a float, got {type(result['duration_s'])}" ) # --------------------------------------------------------------------------- # BB4 — conversations_processed count matches aggregate_week output # --------------------------------------------------------------------------- class TestBB4_ConversationsProcessedCount: """BB4: conversations_processed must equal the number of rows returned.""" def test_conversations_processed_matches_row_count(self, tmp_path): """BB4: 5 DB rows → conversations_processed=5.""" rows = _make_rows(5) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) result = _arun(runner.run_epoch()) assert result["conversations_processed"] == 5, ( f"Expected conversations_processed=5, got {result['conversations_processed']}" ) def test_conversations_processed_zero_when_no_rows(self, tmp_path): """BB4: 0 DB rows → conversations_processed=0.""" pg = _make_pg_conn(rows=[]) gemini = _make_gemini_client(axioms=[], week_summary="No conversations this week") qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) result = _arun(runner.run_epoch()) assert result["conversations_processed"] == 0, ( f"Expected conversations_processed=0, got {result['conversations_processed']}" ) # --------------------------------------------------------------------------- # BB5 — axioms_written count matches write_axioms return value # --------------------------------------------------------------------------- class TestBB5_AxiomsWrittenCount: """BB5: axioms_written must equal the count returned by write_axioms().""" def test_axioms_written_matches_distillation_output(self, tmp_path): """BB5: distill() returns 4 axioms → axioms_written=4.""" rows = _make_rows(3) pg = _make_pg_conn(rows) axioms = _make_axioms(4) gemini = _make_gemini_client(axioms=axioms, week_summary="Four axiom week.") qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) result = _arun(runner.run_epoch()) assert result["axioms_written"] == 4, ( f"Expected axioms_written=4, got {result['axioms_written']}" ) def test_axioms_written_zero_when_no_axioms(self, tmp_path): """BB5: distill() returns 0 axioms → axioms_written=0.""" pg = _make_pg_conn(rows=[]) gemini = _make_gemini_client(axioms=[], week_summary="Quiet week.") qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) result = _arun(runner.run_epoch()) assert result["axioms_written"] == 0, ( f"Expected axioms_written=0, got {result['axioms_written']}" ) # --------------------------------------------------------------------------- # WB1 — All 3 stages called in order: aggregate_week → distill → write_axioms # --------------------------------------------------------------------------- class TestWB1_StagesCalledInOrder: """WB1: run_epoch() must call the 3 pipeline stages in the correct order.""" def test_all_three_stages_are_called(self, tmp_path): """WB1: aggregate_week, distill, and write_axioms must all be called.""" rows = _make_rows(2) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) # Wrap the methods to track call order call_order: list[str] = [] original_aggregate = runner.aggregate_week original_distill = runner.distill original_write = runner.write_axioms async def tracked_aggregate(): call_order.append("aggregate_week") return await original_aggregate() async def tracked_distill(conversations): call_order.append("distill") return await original_distill(conversations) async def tracked_write(axioms, week_summary): call_order.append("write_axioms") return await original_write(axioms, week_summary) runner.aggregate_week = tracked_aggregate runner.distill = tracked_distill runner.write_axioms = tracked_write _arun(runner.run_epoch()) assert call_order == ["aggregate_week", "distill", "write_axioms"], ( f"Expected stages in order [aggregate_week, distill, write_axioms], " f"got {call_order}" ) def test_distill_receives_aggregate_output(self, tmp_path): """WB1: distill() must be called with the list returned by aggregate_week().""" rows = _make_rows(3) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) captured_distill_input: list[list] = [] with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) original_distill = runner.distill async def capturing_distill(conversations): captured_distill_input.append(conversations) return await original_distill(conversations) runner.distill = capturing_distill _arun(runner.run_epoch()) assert len(captured_distill_input) == 1, ( "distill() must be called exactly once" ) assert len(captured_distill_input[0]) == 3, ( f"distill() must receive 3 conversations, got {len(captured_distill_input[0])}" ) # --------------------------------------------------------------------------- # WB2 — Timing uses time.monotonic() — mock to verify it is called # --------------------------------------------------------------------------- class TestWB2_TimingUsesMonotonic: """WB2: run_epoch() must measure duration with time.monotonic(), not datetime. Implementation note: time.monotonic() is called exactly TWICE in run_epoch (once at start, once after the try/except to compute duration). We mock it with a callable counter so that extra calls after the second one just keep returning the last value — avoiding StopIteration-in-coroutine errors that occur when a side_effect list is exhausted inside an async frame. """ def test_time_monotonic_called_for_start_and_end(self, tmp_path): """WB2: time.monotonic() must be called at least twice (start + end).""" rows = _make_rows(1) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) call_count = [0] times = [1000.0, 1000.5] def counting_monotonic(): idx = call_count[0] call_count[0] += 1 return times[min(idx, len(times) - 1)] with p1, p2, p3: with patch("core.epoch.nightly_epoch_runner.time.monotonic", side_effect=counting_monotonic) as mock_mono: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) result = _arun(runner.run_epoch()) # time.monotonic() must have been called at least twice assert call_count[0] >= 2, ( f"time.monotonic() must be called >=2 times (start + end), " f"got {call_count[0]}" ) def test_duration_computed_from_monotonic_values(self, tmp_path): """WB2: duration_s is computed from two time.monotonic() calls. asyncio internals also call time.monotonic() before run_epoch's body begins. We use a monotonically-increasing counter (step=0.1s per call) so that however many pre-amble calls happen, the two calls INSIDE run_epoch (start + end) are always separated by exactly 0.1 s, and duration_s must be > 0. """ rows = _make_rows(1) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) # Each successive call returns a value 0.1 s higher. call_count = [0] def incrementing_monotonic(): value = 100.0 + call_count[0] * 0.1 call_count[0] += 1 return value with p1, p2, p3: with patch("core.epoch.nightly_epoch_runner.time.monotonic", side_effect=incrementing_monotonic): runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) result = _arun(runner.run_epoch()) # duration_s must be positive — the two run_epoch calls are 0.1 s apart assert result["duration_s"] > 0, ( f"Expected duration_s > 0 when monotonic increases per call, " f"got {result['duration_s']}" ) assert result["duration_s"] == pytest.approx( round(result["duration_s"], 3), abs=1e-9 ), ( "duration_s must be rounded to 3 decimal places" ) # --------------------------------------------------------------------------- # WB3 — Exception caught at top level, status="failed" logged # --------------------------------------------------------------------------- class TestWB3_ExceptionCaughtAtTopLevel: """WB3: Any stage exception must be caught, logged, and reflected in status.""" def test_write_axioms_failure_status_failed(self, tmp_path): """WB3: write_axioms() raising → status='failed'.""" rows = _make_rows(2) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) # Force write_axioms to raise async def failing_write(axioms, week_summary): raise IOError("disk full") runner.write_axioms = failing_write result = _arun(runner.run_epoch()) assert result["status"] == "failed", ( f"Expected status='failed' after write_axioms() failure, " f"got {result['status']!r}" ) assert "disk full" in result.get("error", ""), ( f"Expected 'disk full' in error key, got {result.get('error')!r}" ) def test_error_key_contains_exception_message(self, tmp_path): """WB3: The 'error' key must contain str(exception). We patch distill() directly on the runner so the exception reaches run_epoch()'s top-level except block (distill() normally catches its own internal errors and returns safe fallbacks). """ rows = _make_rows(1) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) async def failing_distill(conversations): raise ConnectionError("network unreachable") runner.distill = failing_distill result = _arun(runner.run_epoch()) assert result.get("error") == "network unreachable", ( f"Expected error='network unreachable', got {result.get('error')!r}" ) # --------------------------------------------------------------------------- # WB4 — EPOCH_LOG_PATH file created with os.makedirs for parent dirs # --------------------------------------------------------------------------- class TestWB4_EpochLogDirCreated: """WB4: run_epoch() must create EPOCH_LOG_PATH parent dirs via os.makedirs.""" def test_deeply_nested_epoch_log_path_created(self, tmp_path): """WB4: Deeply nested EPOCH_LOG_PATH created without manual mkdir.""" deep_log = str(tmp_path / "a" / "b" / "c" / "epoch_log.jsonl") axiom_file = str(tmp_path / "axioms.jsonl") summary_file = str(tmp_path / "summaries.jsonl") rows = _make_rows(1) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() with ( patch("core.epoch.nightly_epoch_runner.KG_AXIOM_PATH", axiom_file), patch("core.epoch.nightly_epoch_runner.WEEKLY_SUMMARY_PATH", summary_file), patch("core.epoch.nightly_epoch_runner.EPOCH_LOG_PATH", deep_log), ): runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) try: _arun(runner.run_epoch()) except FileNotFoundError as exc: pytest.fail( f"run_epoch() must create parent dirs for EPOCH_LOG_PATH, " f"but raised FileNotFoundError: {exc}" ) assert Path(deep_log).exists(), ( "Epoch log file must exist at deeply nested path" ) def test_makedirs_called_with_exist_ok_for_epoch_log(self, tmp_path): """WB4: os.makedirs for EPOCH_LOG_PATH must use exist_ok=True.""" rows = _make_rows(1) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) makedirs_calls: list[tuple] = [] original_makedirs = os.makedirs def tracking_makedirs(path, **kwargs): makedirs_calls.append((path, kwargs)) return original_makedirs(path, **kwargs) with p1, p2, p3: with patch("core.epoch.nightly_epoch_runner.os.makedirs", side_effect=tracking_makedirs): runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) _arun(runner.run_epoch()) # At least one call must be for the epoch log path epoch_log_dir = str(tmp_path) # parent of epoch_log.jsonl in tmp_path epoch_calls = [c for c in makedirs_calls if "epoch_log" in str(c[0]) or str(c[0]) == epoch_log_dir] # Verify exist_ok=True on at least one epoch-related makedirs call for _, kwargs in makedirs_calls: assert kwargs.get("exist_ok") is True, ( f"os.makedirs must be called with exist_ok=True, got kwargs={kwargs}" ) # --------------------------------------------------------------------------- # WB5 — Epoch log is JSONL (append mode, not overwrite) # --------------------------------------------------------------------------- class TestWB5_EpochLogIsJsonlAppend: """WB5: run_epoch() must append to epoch_log.jsonl, never overwrite.""" def test_second_run_appends_to_epoch_log(self, tmp_path): """WB5: Running run_epoch() twice → epoch_log.jsonl has 2 lines.""" rows = _make_rows(2) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) _arun(runner.run_epoch()) # Reset mock cursor for second run pg2 = _make_pg_conn(_make_rows(1)) runner.pg = pg2 _arun(runner.run_epoch()) lines = Path(log_file).read_text().strip().splitlines() assert len(lines) == 2, ( f"Expected 2 epoch log lines after 2 runs, got {len(lines)}: {lines}" ) def test_pre_existing_epoch_log_preserved(self, tmp_path): """WB5: Pre-existing content in epoch_log.jsonl must NOT be erased.""" log_path = tmp_path / "epoch_log.jsonl" axiom_file = str(tmp_path / "axioms.jsonl") summary_file = str(tmp_path / "summaries.jsonl") # Write a pre-existing entry pre_entry = json.dumps({"date": "2026-02-18", "status": "success", "conversations_processed": 10, "axioms_written": 5, "week_summary": "Old week.", "duration_s": 1.234}) log_path.write_text(pre_entry + "\n") rows = _make_rows(1) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() with ( patch("core.epoch.nightly_epoch_runner.KG_AXIOM_PATH", axiom_file), patch("core.epoch.nightly_epoch_runner.WEEKLY_SUMMARY_PATH", summary_file), patch("core.epoch.nightly_epoch_runner.EPOCH_LOG_PATH", str(log_path)), ): runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) _arun(runner.run_epoch()) lines = log_path.read_text().strip().splitlines() assert len(lines) == 2, ( f"Expected 2 lines (1 pre-existing + 1 new), got {len(lines)}" ) first_entry = json.loads(lines[0]) assert first_entry["date"] == "2026-02-18", ( "Pre-existing epoch log entry must not be erased" ) def test_each_epoch_log_line_is_valid_json(self, tmp_path): """WB5: Every line in epoch_log.jsonl must be a valid JSON object.""" rows = _make_rows(3) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) _arun(runner.run_epoch()) _arun(runner.run_epoch()) for i, line in enumerate(Path(log_file).read_text().strip().splitlines()): parsed = json.loads(line) assert isinstance(parsed, dict), ( f"Line {i} in epoch_log.jsonl is not a JSON object: {line!r}" ) # --------------------------------------------------------------------------- # WB6 — aggregate_week failure → status="failed", conversations_processed=0 # --------------------------------------------------------------------------- class TestWB6_AggregateWeekFailure: """WB6: aggregate_week() raising must produce status='failed', processed=0.""" def test_aggregate_failure_status_failed(self, tmp_path): """WB6: If aggregate_week() raises, status must be 'failed'.""" rows = _make_rows(2) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) async def failing_aggregate(): raise RuntimeError("DB connection lost") runner.aggregate_week = failing_aggregate result = _arun(runner.run_epoch()) assert result["status"] == "failed", ( f"Expected status='failed' after aggregate_week() failure, " f"got {result['status']!r}" ) def test_aggregate_failure_conversations_processed_zero(self, tmp_path): """WB6: aggregate_week() failure → conversations_processed=0 in log.""" rows = _make_rows(2) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) async def failing_aggregate(): raise RuntimeError("DB connection lost") runner.aggregate_week = failing_aggregate result = _arun(runner.run_epoch()) assert result["conversations_processed"] == 0, ( f"Expected conversations_processed=0 on aggregate failure, " f"got {result['conversations_processed']}" ) def test_aggregate_failure_epoch_log_still_written(self, tmp_path): """WB6: Even on aggregate_week() failure, epoch_log.jsonl must be written.""" rows = _make_rows(1) pg = _make_pg_conn(rows) gemini = _make_gemini_client() qdrant = _make_qdrant_client() _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: runner = NightlyEpochRunner( pg_conn=pg, gemini_client=gemini, qdrant_client=qdrant ) async def failing_aggregate(): raise RuntimeError("total failure") runner.aggregate_week = failing_aggregate _arun(runner.run_epoch()) assert Path(log_file).exists(), ( "epoch_log.jsonl must be written even when aggregate_week() fails" ) # --------------------------------------------------------------------------- # Regression — Stories 9.02, 9.03, 9.04 still work # --------------------------------------------------------------------------- class TestRegression_PriorStoriesUnchanged: """Regression: Stories 9.02, 9.03, and 9.04 must pass after 9.05 changes.""" def test_r1_aggregate_week_no_pg_returns_empty_list(self): """R1 (9.02): aggregate_week() with pg=None returns [].""" runner = NightlyEpochRunner() result = _arun(runner.aggregate_week()) assert result == [], ( f"R1 regression: aggregate_week() should return [] with pg=None, got {result!r}" ) def test_r2_distill_empty_conversations(self): """R2 (9.03): distill([]) returns safe empty result.""" runner = NightlyEpochRunner() result = _arun(runner.distill([])) assert result == { "axioms": [], "week_summary": "No conversations this week", }, ( f"R2 regression: distill([]) returned unexpected value: {result!r}" ) def test_r3_write_axioms_empty_returns_zero(self, tmp_path): """R3 (9.04): write_axioms([], ...) returns 0.""" _, _, log_file, p1, p2, p3 = _patch_all_output_paths(tmp_path) with p1, p2, p3: with patch("core.epoch.nightly_epoch_runner.KG_AXIOM_PATH", str(tmp_path / "axioms.jsonl")): with patch("core.epoch.nightly_epoch_runner.WEEKLY_SUMMARY_PATH", str(tmp_path / "summaries.jsonl")): runner = NightlyEpochRunner() result = _arun(runner.write_axioms([], "Empty.")) assert result == 0, ( f"R3 regression: write_axioms([]) should return 0, got {result!r}" ) def test_r3_write_axioms_three_axioms_returns_three(self, tmp_path): """R3 (9.04): write_axioms() with 3 axioms returns 3 (no Qdrant).""" axioms = _make_axioms(3) axiom_file = str(tmp_path / "axioms.jsonl") summary_file = str(tmp_path / "summaries.jsonl") with ( patch("core.epoch.nightly_epoch_runner.KG_AXIOM_PATH", axiom_file), patch("core.epoch.nightly_epoch_runner.WEEKLY_SUMMARY_PATH", summary_file), ): runner = NightlyEpochRunner(gemini_client=None, qdrant_client=None) result = _arun(runner.write_axioms(axioms, "Three axiom week.")) assert result == 3, ( f"R3 regression: write_axioms() with 3 axioms should return 3, got {result!r}" ) def test_module_constant_epoch_log_path_present(self): """Regression: EPOCH_LOG_PATH constant must exist and be a string.""" from core.epoch.nightly_epoch_runner import EPOCH_LOG_PATH assert isinstance(EPOCH_LOG_PATH, str), ( f"EPOCH_LOG_PATH must be a string, got {type(EPOCH_LOG_PATH)}" ) assert "epoch_log.jsonl" in EPOCH_LOG_PATH, ( f"EPOCH_LOG_PATH must contain 'epoch_log.jsonl', got {EPOCH_LOG_PATH!r}" ) def test_module_import_time_available(self): """Regression: 'time' module must be importable from the module.""" import core.epoch.nightly_epoch_runner as mod assert hasattr(mod, "time") or True # time is used internally via import # Verify time.monotonic is accessible in the module's namespace import inspect src = inspect.getsource(mod) assert "time.monotonic" in src, ( "time.monotonic() must be used in nightly_epoch_runner.py" ) def test_run_epoch_method_exists(self): """Regression: NightlyEpochRunner must have a run_epoch method.""" runner = NightlyEpochRunner() assert hasattr(runner, "run_epoch"), ( "NightlyEpochRunner must have a run_epoch method" ) import asyncio assert asyncio.iscoroutinefunction(runner.run_epoch), ( "run_epoch must be an async method" ) # --------------------------------------------------------------------------- # Run summary # --------------------------------------------------------------------------- if __name__ == "__main__": import sys as _sys result = pytest.main([__file__, "-v", "--tb=short"]) _sys.exit(result)