test: add migration + translation cache tests (briefs_db) and segment-map unit tests (translate)

23 tests total: language migration, save/get/invalidate translations,
_collect_segments coverage (prose/list/nested/exclusions/empty), _set_at_path
edge cases, and language_directive drift guard.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/test_briefs_db.py

@@ -13,7 +13,9 @@ def test_save_and_get_round_trip():
     brief = {"ticker": "AAPL", "company_name": "Apple Inc.", "what_changed": ["Revenue up 5%"]}
     save_brief("AAPL", brief)
     result = get_brief("AAPL")
-    assert result == brief
+    # get_brief injects a default "language" key; strip it for the equality check
+    result_without_lang = {k: v for k, v in result.items() if k != "language"}
+    assert result_without_lang == brief
 
 
 def test_get_brief_returns_none_for_unknown_ticker():
@@ -25,14 +27,15 @@ def test_save_brief_overwrites_on_same_ticker():
     from storage.briefs_db import save_brief, get_brief
     save_brief("AAPL", {"v": 1})
     save_brief("AAPL", {"v": 2})
-    assert get_brief("AAPL") == {"v": 2}
+    result = get_brief("AAPL")
+    assert result["v"] == 2
 
 
 def test_save_brief_normalizes_ticker_to_uppercase():
     from storage.briefs_db import save_brief, get_brief
     save_brief("aapl", {"company_name": "Apple"})
-    assert get_brief("AAPL") == {"company_name": "Apple"}
-    assert get_brief("aapl") == {"company_name": "Apple"}
+    assert get_brief("AAPL")["company_name"] == "Apple"
+    assert get_brief("aapl")["company_name"] == "Apple"
 
 
 def test_list_tickers_returns_all_saved():
@@ -53,4 +56,55 @@ def test_init_db_is_idempotent():
     init_db()
     init_db()
     save_brief("AAPL", {"x": 1})
-    assert get_brief("AAPL") == {"x": 1}
+    assert get_brief("AAPL")["x"] == 1
+
+
+def test_migration_adds_language_column():
+    """init_db() migrates an old briefs table that lacks the language column."""
+    import sqlite3
+    import storage.briefs_db as m
+    # Create old-schema row manually (no language column)
+    with sqlite3.connect(m.DB_PATH) as conn:
+        conn.execute("DROP TABLE IF EXISTS briefs")
+        conn.execute("""
+            CREATE TABLE briefs (
+                ticker TEXT PRIMARY KEY,
+                filing_date TEXT,
+                brief_json TEXT NOT NULL,
+                saved_at TEXT NOT NULL
+            )
+        """)
+        conn.execute("INSERT INTO briefs VALUES ('AAPL', '2025-01-01', '{\"ticker\": \"AAPL\"}', '2025-01-01T00:00:00+00:00')")
+    m.init_db()
+    with sqlite3.connect(m.DB_PATH) as conn:
+        cols = {r[1] for r in conn.execute("PRAGMA table_info(briefs)")}
+        assert "language" in cols
+        row = conn.execute("SELECT language FROM briefs WHERE ticker = 'AAPL'").fetchone()
+        assert row[0] == "English"
+
+
+def test_save_brief_persists_language():
+    from storage.briefs_db import save_brief, get_brief
+    save_brief("AAPL", {"ticker": "AAPL", "language": "French", "what_matters_most": "Bonjour"})
+    result = get_brief("AAPL")
+    assert result["language"] == "French"
+
+
+def test_save_and_get_translation_round_trip():
+    from storage.briefs_db import save_brief, save_translation, get_translation
+    save_brief("AAPL", {"ticker": "AAPL", "language": "English"})
+    translated = {"ticker": "AAPL", "language": "French", "what_matters_most": "Bonjour"}
+    save_translation("AAPL", "French", translated)
+    result = get_translation("AAPL", "French")
+    assert result == translated
+    assert get_translation("AAPL", "German") is None
+
+
+def test_save_brief_invalidates_translations():
+    from storage.briefs_db import save_brief, save_translation, get_translation
+    save_brief("AAPL", {"ticker": "AAPL", "language": "English"})
+    save_translation("AAPL", "French", {"ticker": "AAPL", "language": "French"})
+    assert get_translation("AAPL", "French") is not None
+    # Re-save brief → should delete cached translations
+    save_brief("AAPL", {"ticker": "AAPL", "language": "English", "what_matters_most": "Updated"})
+    assert get_translation("AAPL", "French") is None

tests/test_translate.py

@@ -0,0 +1,113 @@
+# tests/test_translate.py
+"""Tests for agent/translate.py — pure functions only, no LLM calls."""
+import copy
+import pytest
+from agent.translate import _collect_segments, _set_at_path
+from agent.prompts import PROSE_FIELDS, PROSE_LIST_FIELDS
+
+
+# ── _collect_segments ──────────────────────────────────────────────────────────
+
+def test_collect_segments_prose_field():
+    brief = {"what_matters_most": "Revenue grew strongly."}
+    segs = _collect_segments(brief)
+    assert len(segs) == 1
+    path, text = segs[0]
+    assert path == ("what_matters_most",)
+    assert text == "Revenue grew strongly."
+
+
+def test_collect_segments_excludes_evidence_snippet():
+    brief = {"evidence_snippet": "verbatim quote here", "headline": "Q3 beat"}
+    segs = _collect_segments(brief)
+    paths = [p for p, _ in segs]
+    assert ("evidence_snippet",) not in paths
+    assert ("headline",) in paths
+
+
+def test_collect_segments_skips_quarter_deltas_subtree():
+    brief = {
+        "what_matters_most": "Growth accelerated.",
+        "quarter_deltas": {"headline": "Should NOT be collected", "text": "Also skip"},
+    }
+    segs = _collect_segments(brief)
+    texts = [t for _, t in segs]
+    assert "Growth accelerated." in texts
+    assert "Should NOT be collected" not in texts
+    assert "Also skip" not in texts
+
+
+def test_collect_segments_prose_list_field():
+    brief = {"what_to_watch": ["Watch margins closely.", "Monitor capex spend."]}
+    segs = _collect_segments(brief)
+    assert len(segs) == 2
+    paths = [p for p, _ in segs]
+    assert ("what_to_watch", 0) in paths
+    assert ("what_to_watch", 1) in paths
+
+
+def test_collect_segments_nested_dict():
+    brief = {
+        "bull_points": [
+            {"text": "Revenue beat expectations.", "source": "10-K", "reliability": "HIGH"}
+        ]
+    }
+    segs = _collect_segments(brief)
+    paths = [p for p, _ in segs]
+    assert ("bull_points", 0, "text") in paths
+    # source and reliability are enum fields, should not be collected
+    assert ("bull_points", 0, "source") not in paths
+    assert ("bull_points", 0, "reliability") not in paths
+
+
+def test_collect_segments_skips_empty_strings():
+    brief = {"what_matters_most": "", "headline": "Has content"}
+    segs = _collect_segments(brief)
+    paths = [p for p, _ in segs]
+    assert ("what_matters_most",) not in paths
+    assert ("headline",) in paths
+
+
+def test_collect_segments_nested_key_quote():
+    brief = {"key_quote": {"text": "This is the key quote.", "source": "transcript"}}
+    segs = _collect_segments(brief)
+    paths = [p for p, _ in segs]
+    assert ("key_quote", "text") in paths
+    assert ("key_quote", "source") not in paths
+
+
+# ── _set_at_path ──────────────────────────────────────────────────────────────
+
+def test_set_at_path_top_level():
+    obj = {"what_matters_most": "English text"}
+    _set_at_path(obj, ("what_matters_most",), "Texte français")
+    assert obj["what_matters_most"] == "Texte français"
+
+
+def test_set_at_path_nested():
+    obj = {"bull_points": [{"text": "English", "reliability": "HIGH"}]}
+    _set_at_path(obj, ("bull_points", 0, "text"), "Français")
+    assert obj["bull_points"][0]["text"] == "Français"
+    assert obj["bull_points"][0]["reliability"] == "HIGH"  # untouched
+
+
+def test_set_at_path_missing_key_is_silent():
+    obj = {"bull_points": []}
+    _set_at_path(obj, ("bull_points", 0, "text"), "Français")
+    # Should not raise; list is empty so index 0 doesn't exist
+
+
+def test_set_at_path_empty_path_is_silent():
+    obj = {"x": 1}
+    _set_at_path(obj, (), "value")
+    assert obj == {"x": 1}  # unchanged
+
+
+# ── drift guard ──────────────────────────────────────────────────────────────
+
+def test_prose_fields_in_language_directive():
+    """Every name in PROSE_FIELDS | PROSE_LIST_FIELDS appears in language_directive output."""
+    from agent.prompts import language_directive
+    directive = language_directive("French")
+    for field in PROSE_FIELDS | PROSE_LIST_FIELDS:
+        assert field in directive, f"Field '{field}' missing from language_directive"