from __future__ import annotations import json import tempfile import unittest from datetime import date, datetime, timedelta, timezone from pathlib import Path from unittest.mock import patch import brief import judgement import register import rules from collect import ( Collection, _BodyStructureParser, _build_search_criteria, _load_accounts, _parts_from_structure, ) class RuleTests(unittest.TestCase): def message(self, subject: str, body: str = "") -> dict[str, object]: return { "message_id": subject, "from": "A Person ", "subject": subject, "body": body, "attachments": [], "date": datetime.now(timezone.utc).isoformat(), } def test_clear_near_date_is_act(self) -> None: due = datetime.now(timezone.utc).date() + timedelta(days=10) finding = rules.classify(self.message(f"Respond by {due.isoformat()}")) self.assertEqual("ACT", finding["bucket"]) self.assertEqual(due.isoformat(), finding["normalized_date"]) def test_ambiguous_numeric_date_is_review(self) -> None: finding = rules.classify(self.message("Appointment on 08/09")) self.assertEqual("REVIEW", finding["bucket"]) self.assertIsNone(finding["normalized_date"]) self.assertEqual("08/09", finding["original_date_text"]) def test_keyword_reasons_accumulate_and_act_wins(self) -> None: finding = rules.classify( self.message("Statement", "Failed payment. Is this a new device?") ) self.assertEqual("ACT", finding["bucket"]) self.assertGreaterEqual(len(finding["reasons"]), 3) def test_duplicate_finding_is_not_appended(self) -> None: with tempfile.TemporaryDirectory() as directory: target = Path(directory) message = self.message("A receipt") with patch.object(rules, "FINDINGS_DIR", target): first = rules.write_findings([message]) second = rules.write_findings([message]) self.assertEqual(1, len(first)) self.assertEqual([], second) lines = next(target.glob("*.jsonl")).read_text(encoding="utf-8").splitlines() self.assertEqual(1, len(lines)) def test_same_message_id_in_two_accounts_is_kept_twice(self) -> None: with tempfile.TemporaryDirectory() as directory: target = Path(directory) first = self.message("Shared message") first["account"] = "rocky" second = dict(first) second["account"] = "signup" with patch.object(rules, "FINDINGS_DIR", target): findings = rules.write_findings([first, second]) self.assertEqual(2, len(findings)) class MimeTests(unittest.TestCase): def test_attachment_is_listed_but_not_selected_as_text(self) -> None: raw = ( b'(("TEXT" "PLAIN" ("CHARSET" "UTF-8") NIL NIL "7BIT" 20 2)' b'("APPLICATION" "PDF" ("NAME" "visa.pdf") NIL NIL "BASE64" 100 ' b'NIL ("ATTACHMENT" ("FILENAME" "visa.pdf"))) "MIXED")' ) structure = _BodyStructureParser(raw).parse() text, attachments = _parts_from_structure(structure) self.assertEqual([("1", "plain", "UTF-8")], text) self.assertEqual(["visa.pdf"], attachments) class AccountAndSearchTests(unittest.TestCase): def test_seeded_accounts_and_signup_password_name(self) -> None: accounts = _load_accounts() self.assertEqual(["rocky", "rateright", "signup"], [item.alias for item in accounts]) self.assertEqual("PCOS_IMAP_APP_PASSWORD", accounts[-1].password_env) def test_multiple_search_terms_are_ored_with_since(self) -> None: criteria = _build_search_criteria( ["visa", "home affairs", 'say "hello"'], date(2020, 1, 2) ) self.assertEqual( 'SINCE 02-Jan-2020 OR TEXT "visa" ' 'OR TEXT "home affairs" TEXT "say \\"hello\\""', criteria, ) class RegisterAndBriefTests(unittest.TestCase): def test_deadline_append_only_and_surfaces(self) -> None: due = datetime.now(timezone.utc).date() + timedelta(days=20) finding = { "bucket": "REVIEW", "normalized_date": due.isoformat(), "subject": "Registration renewal", "sender": "Authority ", "reasons": ["future date"], } with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "deadlines.jsonl" with patch.object(register, "REGISTER_PATH", path): register.append_deadlines([finding]) register.append_deadlines([finding]) self.assertEqual(2, len(path.read_text(encoding="utf-8").splitlines())) with patch.object(brief, "REGISTER_PATH", path): upcoming = brief._upcoming_deadlines() self.assertEqual(1, len(upcoming)) self.assertEqual(due.isoformat(), upcoming[0]["date"]) def test_brief_never_exceeds_twenty_lines(self) -> None: collection = Collection([], 2, False, datetime.now(timezone.utc).isoformat()) findings = [] for number in range(10): findings.append( { "bucket": "ACT", "normalized_date": None, "subject": f"Action {number}", "sender": "Sender ", "message_date": datetime.now(timezone.utc).isoformat(), } ) with tempfile.TemporaryDirectory() as directory: with patch.object(brief, "REGISTER_PATH", Path(directory) / "missing"): text = brief.render(collection, findings, 3) self.assertLessEqual(len(text.splitlines()), 20) self.assertIn("work emails excluded", text) class NoiseControlTests(unittest.TestCase): """The 2026-07-29 rebuild. Each test is a measured false positive, named.""" def message(self, subject: str, body: str = "", sender: str = "A Person "): return { "message_id": subject + body[:20] + sender, "account": "rocky", "from": sender, "subject": subject, "body": body, "attachments": [], "date": datetime.now(timezone.utc).isoformat(), } def test_fine_does_not_match_define(self) -> None: # 92 of 'fine's 140 corpus hits were 'define' or 'refined'. finding = rules.classify(self.message("Newsletter", "We define our terms clearly.")) self.assertNotIn("legal: fine", finding["reasons"]) def test_visa_does_not_match_the_card_brand_in_a_footer(self) -> None: body = "Thanks for your order.\n" + ("filler. " * 80) + "We accept Visa and Mastercard." finding = rules.classify(self.message("Your order", body)) self.assertNotIn("legal: visa", finding["reasons"]) self.assertEqual("NOTE", finding["bucket"]) def test_keyword_past_the_scope_is_not_seen(self) -> None: # ANZ statement footers put 'password' at 710-1394 and 'notice' at 1091-7905. body = ("x" * (rules.SCOPE_CHARS + 50)) + " change your password here" finding = rules.classify(self.message("A newsletter", body)) self.assertNotIn("security: password", finding["reasons"]) def test_keyword_inside_the_scope_is_seen(self) -> None: finding = rules.classify(self.message("Hello", "Your payment has failed.")) self.assertEqual("ACT", finding["bucket"]) def test_written_date_without_a_year_never_becomes_a_date(self) -> None: # This invented 32 phantom obligations in July 2027 alone. # Subject deliberately carries no ACT keyword — an earlier draft used # "Invoice", which is itself an ACT term, and the test failed for the # wrong reason. finding = rules.classify(self.message("Monthly summary", "Issued 21 July for the period.")) self.assertIsNone(finding["normalized_date"]) self.assertEqual("REVIEW", finding["bucket"]) self.assertTrue(any("ambiguous" in r for r in finding["reasons"])) def test_written_date_with_a_year_is_kept(self) -> None: finding = rules.classify( self.message("Skills assessment", "He turns 45 on the 23 September 2026.") ) self.assertEqual("2026-09-23", finding["normalized_date"]) def test_the_visa_date_as_it_actually_appears(self) -> None: """Verbatim from the migration agent's 26/05 email, at its real offset.""" body = ( "Michael requires a skills assessment for the occupation of Carpenter " "for a permanent 186 Direct Entry application.\n\n" "He turns 45 on the 23/09/2026. We would require the assessment to be " "finalised by then in order to submit a valid application." ) finding = rules.classify( self.message("Skills Assessment - Carpenter", body, "Sibeal Ni Mhaille ") ) self.assertEqual("2026-09-23", finding["normalized_date"]) self.assertEqual("ACT", finding["bucket"]) def test_critical_sender_catches_what_no_keyword_can(self) -> None: # The real subject. No ACT keyword appears anywhere in it. finding = rules.classify( self.message("VETASSESS Application - P26MC85241", "Please log in.", "TradeAssess ") ) self.assertEqual("ACT", finding["bucket"]) self.assertIn("critical sender", finding["reasons"]) def test_critical_domain_does_not_drag_in_its_newsletter(self) -> None: self.assertTrue(rules._is_critical_sender("someone@ato.gov.au")) self.assertFalse(rules._is_critical_sender("x@news.ato.gov.au")) def test_findings_dedup_across_different_day_files(self) -> None: """The actual 8,758-for-4,351 bug: dedup that only looked at one day file.""" with tempfile.TemporaryDirectory() as directory: target = Path(directory) old = self.message("Old news") old["date"] = "2026-01-15T09:00:00+00:00" with patch.object(rules, "FINDINGS_DIR", target): first = rules.write_findings([old]) second = rules.write_findings([old]) self.assertEqual(1, len(first)) self.assertEqual([], second) files = sorted(p.name for p in target.glob("*.jsonl")) self.assertEqual(["2026-01-15.jsonl"], files) def test_findings_are_filed_under_the_message_day_not_today(self) -> None: with tempfile.TemporaryDirectory() as directory: target = Path(directory) old = self.message("Historical") old["date"] = "2026-03-02T11:00:00+00:00" with patch.object(rules, "FINDINGS_DIR", target): rules.write_findings([old]) self.assertTrue((target / "2026-03-02.jsonl").exists()) class HorizonTests(unittest.TestCase): def test_sixty_day_horizon_surfaces_the_visa_thirty_could_not(self) -> None: due = datetime.now(timezone.utc).date() + timedelta(days=56) record = { "account": "rocky", "date": due.isoformat(), "what": "date", "source_subject": "Skills Assessment - Carpenter", "source_sender": "Sibeal Ni Mhaille ", "first_seen": "2026-05-26T00:00:00+00:00", "last_seen": "2026-07-29T00:00:00+00:00", } with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "deadlines.jsonl" path.write_text(json.dumps(record) + "\n", encoding="utf-8") with patch.object(brief, "REGISTER_PATH", path): upcoming = brief._upcoming_deadlines() self.assertEqual(1, len(upcoming)) self.assertEqual(due.isoformat(), upcoming[0]["date"]) def test_same_deadline_on_two_accounts_shows_once(self) -> None: due = datetime.now(timezone.utc).date() + timedelta(days=56) base = { "date": due.isoformat(), "what": "date", "source_subject": "Skills Assessment - Carpenter", "source_sender": "Sibeal Ni Mhaille ", "first_seen": "2026-05-26T00:00:00+00:00", "last_seen": "2026-07-29T00:00:00+00:00", } with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "deadlines.jsonl" path.write_text( json.dumps({**base, "account": "rocky"}) + "\n" + json.dumps({**base, "account": "signup"}) + "\n", encoding="utf-8", ) with patch.object(brief, "REGISTER_PATH", path): upcoming = brief._upcoming_deadlines() self.assertEqual(1, len(upcoming)) class JudgementCannotSuppressTests(unittest.TestCase): """Stage 3's one guarantee: judgement may ADD and may never take away.""" def _collection(self): return Collection([], 0, False, datetime.now(timezone.utc).isoformat()) def _act_finding(self): return { "account": "rocky", "bucket": "ACT", "normalized_date": None, "subject": "VETASSESS Application - P26MC85241", "sender": "TradeAssess ", "message_date": datetime.now(timezone.utc).isoformat(), } def _render_with_judgement(self, payload, directory): target = Path(directory) day = datetime.now().astimezone().date().isoformat() if payload is not None: (target / f"{day}.json").write_text( json.dumps(payload), encoding="utf-8" ) with patch.object(judgement, "JUDGEMENT_DIR", target), \ patch.object(brief, "REGISTER_PATH", target / "missing-register"): return brief.render(self._collection(), [self._act_finding()], 0) def test_no_judgement_file_changes_nothing(self) -> None: with tempfile.TemporaryDirectory() as directory: text = self._render_with_judgement(None, directory) self.assertIn("VETASSESS Application", text) self.assertNotIn("WORTH A THOUGHT", text) def test_judgement_is_appended_and_the_act_survives(self) -> None: payload = { "written_at_utc": datetime.now(timezone.utc).isoformat(), "lines": ["VETASSESS has not moved in 20 days and the 186 closes 23 Sep."], } with tempfile.TemporaryDirectory() as directory: text = self._render_with_judgement(payload, directory) self.assertIn("VETASSESS Application", text) # the rule's finding, intact self.assertIn("WORTH A THOUGHT", text) self.assertIn("has not moved in 20 days", text) # and it sits BELOW the rules, never above them self.assertLess(text.index("VETASSESS Application"), text.index("WORTH A THOUGHT")) def test_malformed_judgement_is_ignored_not_fatal(self) -> None: with tempfile.TemporaryDirectory() as directory: day = datetime.now().astimezone().date().isoformat() (Path(directory) / f"{day}.json").write_text("{not json", encoding="utf-8") with patch.object(judgement, "JUDGEMENT_DIR", Path(directory)), \ patch.object(brief, "REGISTER_PATH", Path(directory) / "nope"): text = brief.render(self._collection(), [self._act_finding()], 0) self.assertIn("VETASSESS Application", text) self.assertNotIn("WORTH A THOUGHT", text) def test_stale_judgement_is_ignored(self) -> None: payload = { "written_at_utc": ( datetime.now(timezone.utc) - timedelta(hours=48) ).isoformat(), "lines": ["a thought from two days ago"], } with tempfile.TemporaryDirectory() as directory: text = self._render_with_judgement(payload, directory) self.assertNotIn("WORTH A THOUGHT", text) self.assertIn("VETASSESS Application", text) def test_judgement_cannot_emit_more_than_three_lines(self) -> None: payload = { "written_at_utc": datetime.now(timezone.utc).isoformat(), "lines": [f"line {n}" for n in range(20)], } with tempfile.TemporaryDirectory() as directory: text = self._render_with_judgement(payload, directory) self.assertEqual(3, sum(1 for line in text.splitlines() if line.startswith("- line "))) def test_judgement_module_writes_nowhere_but_its_own_directory(self) -> None: """Reading the inputs must not mutate them.""" before = {} for path in (judgement.FINDINGS_DIR, judgement.REGISTER_PATH): if path.exists(): before[path] = ( sorted((p.name, p.stat().st_mtime, p.stat().st_size) for p in path.glob("*.jsonl")) if path.is_dir() else (path.stat().st_mtime, path.stat().st_size) ) judgement.build_prompt() for path, snapshot in before.items(): now = ( sorted((p.name, p.stat().st_mtime, p.stat().st_size) for p in path.glob("*.jsonl")) if path.is_dir() else (path.stat().st_mtime, path.stat().st_size) ) self.assertEqual(snapshot, now, f"{path} was modified by the judgement pass") def test_no_write_verb_targets_findings_or_the_register(self) -> None: """Enforced by absence, the same way the send ban is.""" source = Path(judgement.__file__).read_text(encoding="utf-8") for forbidden in ("FINDINGS_DIR /", "REGISTER_PATH.open", "REGISTER_PATH.write", "FINDINGS_DIR.mkdir", "unlink", "rmtree"): self.assertNotIn(forbidden, source, f"judgement.py contains {forbidden!r}") if __name__ == "__main__": unittest.main()