import copy import json import tempfile import unittest from collections import Counter from pathlib import Path from dataset import ( ReviewItem, Verdict, accept_mapping, compute_summaries, event, pilot_cases, save_run, ) class DatasetTests(unittest.TestCase): def test_mapping_preserves_evidence_and_quarantines_changes(self): record = next( r for r in pilot_cases(per_scenario=1) if r["audit"]["group_id"] == "takeover" ) mapped = copy.deepcopy(record["input"]) mapped["texts"]["listing_note"] = "Listing submitted for review." accepted = accept_mapping(record, json.dumps(mapped)) self.assertEqual(accepted["target"], record["target"]) self.assertEqual(accepted["input"]["texts"]["listing_note"], "") self.assertFalse(accepted["metadata"]["source"]["text_written_by_llm"]) self.assertTrue(accepted["audit"]["mapped_by_llm"]) mapped["recent_events"][1]["details"]["attack_ip"] = False with self.assertRaises(ValueError): accept_mapping(record, json.dumps(mapped)) with tempfile.TemporaryDirectory() as tmp: output = Path(tmp) / "run" report = save_run([record], [{"text": json.dumps(mapped)}], output, {}) self.assertEqual((report["accepted"], report["rejected"]), (0, 1)) self.assertEqual((output / "candidates.jsonl").read_text(), "") self.assertIn( "LLM changed evidence", (output / "rejected.jsonl").read_text() ) def test_summaries_at_review_time_and_24h_boundary(self): item = next(pilot_cases(per_scenario=1))["input"] item["review_target"]["details"].update(shares=20000, price_per_share=105.0) item["company"]["last_trade_price"] = 114.0 item["recent_events"] = [ event("bank_account_changed", 1440), event("bid_placed", 1439), event("bid_cancelled", 1438), event("login", 100, device="new", success=False), event("login", 80, device="known", success=True), ] actual = compute_summaries(item) self.assertEqual( actual, { "listing_vs_typical_holding": 2.5, "price_vs_last_trade_pct": -7.9, "orders_last_24h": 2, "orders_cancelled_last_24h": 1, "failed_logins_last_24h": 1, "bank_changed_within_24h": False, "new_device_within_24h": True, "high_risk_country": False, }, ) item["recent_events"][0]["minutes_ago"] = 1439 self.assertTrue(compute_summaries(item)["bank_changed_within_24h"]) def test_rejects_leakage_bad_arithmetic_and_future_events(self): original = next(pilot_cases(per_scenario=1))["input"] for change in ("label", "summary", "time"): item = copy.deepcopy(original) if change == "label": item["fraud_type"] = "ownership_fraud" elif change == "summary": item["summaries"]["orders_last_24h"] += 1 else: item["recent_events"][0]["minutes_ago"] = -1 with self.subTest(change=change), self.assertRaises(ValueError): ReviewItem.model_validate(item) def test_verdict_uses_only_fixed_codes(self): for reasons in ( ["INVENTED_REASON"], ["NO_CONCERN", "BANK_CHANGED_RECENTLY"], ["SHARED_BANK", "SHARED_BANK"], ): with self.subTest(reasons=reasons), self.assertRaises(ValueError): Verdict(risk_level=3, reasons=reasons) for risk in (0, 6, True): with self.subTest(risk=risk), self.assertRaises(ValueError): Verdict(risk_level=risk, reasons=["NO_CONCERN"]) def test_reproducible_pilot_contains_benign_controls_without_claiming_outcomes( self, ): records = list(pilot_cases(seed=19, per_scenario=2)) self.assertEqual(records, list(pilot_cases(seed=19, per_scenario=2))) self.assertEqual(len({r["input"]["case_id"] for r in records}), len(records)) groups = {r["audit"]["group_id"]: r for r in records} self.assertFalse( groups["benign_bank_change"]["metadata"]["label"]["is_suspicious"] ) self.assertTrue(groups["takeover"]["metadata"]["label"]["is_suspicious"]) self.assertTrue( all( r["audit"]["synthetic"] and not r["audit"]["human_reviewed"] for r in records ) ) self.assertEqual({r["metadata"]["split"] for r in records}, {"train"}) def test_source_sampling_preserves_original_rows_and_both_labels(self): import zipfile from prepare_sources import sample_archive with tempfile.TemporaryDirectory() as tmp: archive = Path(tmp) / "source.zip" csv_text = "fraud_bool,amount\n" + "".join( f"{i % 2},{1000 + i}\n" for i in range(40) ) with zipfile.ZipFile(archive, "w") as z: z.writestr("Base.csv", csv_text) sampled, counts = sample_archive(archive, "Base.csv", "fraud_bool", 4, 42) self.assertEqual(counts, {"0": 20, "1": 20}) self.assertEqual(len(sampled), 8) self.assertEqual( sampled, sample_archive(archive, "Base.csv", "fraud_bool", 4, 42)[0] ) self.assertTrue(any(row["csv_data_row"] > 8 for row in sampled)) for row in sampled: self.assertEqual( row["fields"]["amount"], str(999 + row["csv_data_row"]) ) self.assertEqual( row["original_label"], str((row["csv_data_row"] - 1) % 2) ) def test_quality_batch_mixture_source_semantics_and_documentary_support(self): import re from quality_batch import audit_records, build_cases sources = [] for source in ("saml_d", "baf"): for i in range(200): fields = ( { "Amount": str(17.25 + i), "Payment_currency": "UK pounds", "Received_currency": "Euro", } if source == "saml_d" else { "month": str(i % 8), "device_distinct_emails_8w": "2", "session_length_in_minutes": "8.3", "phone_home_valid": "1", "phone_mobile_valid": "0", "days_since_request": "-1", } ) sources.append( { "source": source, "original_id": f"{source}:{i}", "original_label": str(i % 2), "fields": fields, } ) with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "source.jsonl" path.write_text("".join(json.dumps(r) + "\n" for r in sources)) records = build_cases(path) self.assertEqual(records, build_cases(path)) self.assertEqual( Counter(r["audit"]["mixture_class"] for r in records), {"ordinary": 125, "benign_unusual": 125, "suspicious": 250}, ) self.assertEqual(audit_records(records)["failures"], []) level_five = [r for r in records if r["target"]["risk_level"] == 5] self.assertEqual(len(level_five), 25) for row in level_five: total, committed, available = map( int, re.search( r"holds (\d+) shares; (\d+) shares.*shares: (\d+)", row["input"]["texts"]["document_text"], ).groups(), ) self.assertEqual(available, total - committed) self.assertGreater( row["input"]["review_target"]["details"]["shares"], available ) for row in records: root = row["audit"]["source_record"] if root["source"] == "saml_d": details = next( e["details"] for e in row["input"]["recent_events"] if e["minutes_ago"] == 180 ) self.assertEqual(details["amount"], float(root["fields"]["Amount"])) self.assertEqual(details["currency"], "unspecified_in_source") self.assertEqual(details["source_payment_currency"], "UK pounds") self.assertEqual(details["source_received_currency"], "Euro") self.assertNotIn("sender_matches_account", details) elif root["source"] == "baf": month = int(root["fields"]["month"]) self.assertEqual( row["metadata"]["split"], "train" if month < 6 else "validation" if month == 6 else "test", ) self.assertGreater(row["input"]["account"]["age_days"], 0) for source in sources: source["original_label"] = str(1 - int(source["original_label"])) path.write_text("".join(json.dumps(r) + "\n" for r in sources)) changed = build_cases(path) self.assertEqual( [(r["input"], r["target"]) for r in records], [(r["input"], r["target"]) for r in changed], ) expanded = build_cases(path, seed=29, expansion_batch=1) self.assertEqual(audit_records(expanded)["failures"], []) self.assertEqual( Counter(r["audit"]["mixture_class"] for r in expanded), {"ordinary": 125, "benign_unusual": 125, "suspicious": 250}, ) ordinary = [ r for r in expanded if r["audit"]["mixture_class"] == "ordinary" ] self.assertEqual( {r["input"]["account"]["role"] for r in ordinary}, {"buyer", "seller"} ) ownership = [ r for r in expanded if r["audit"]["case_family"] == "ownership" ] self.assertEqual({r["target"]["risk_level"] for r in ownership}, {3, 4}) for row in ownership: oversized = ( row["input"]["summaries"]["listing_vs_typical_holding"] > 1.5 ) self.assertEqual(row["target"]["risk_level"], 4 if oversized else 3) self.assertEqual( "LISTING_ABOVE_TYPICAL_HOLDING" in row["target"]["reasons"], oversized, ) other = build_cases(path, seed=30, expansion_batch=2) generated = lambda rows: { r["audit"]["group_id"] for r in rows if r["metadata"]["source"]["origin"] == "generated" } self.assertFalse(generated(expanded) & generated(other)) def test_teacher_notes_preserve_evidence_targets_and_reject_verdict_injection(self): from quality_batch import accept_proposals record = next(pilot_cases(per_scenario=1)) original = copy.deepcopy(record) proposal = { "case_id": record["input"]["case_id"], "listing_note": "I would like to sell my common shares through this platform.", } accepted, rejected = accept_proposals( [record], json.dumps({"cases": [proposal]}) ) self.assertEqual(rejected, []) self.assertEqual(record, original) self.assertEqual(accepted[0]["target"], original["target"]) accepted[0]["input"]["texts"]["listing_note"] = "" self.assertEqual(accepted[0]["input"], original["input"]) proposal["verdict"] = {"risk_level": 3, "reasons": ["NO_OWNERSHIP_PROOF"]} with self.assertRaises(ValueError): accept_proposals([record], json.dumps({"cases": [proposal]})) proposal.pop("verdict") proposal["listing_note"] = ( "My verified shares are approved and guaranteed to be legitimate." ) accepted, rejected = accept_proposals( [record], json.dumps({"cases": [proposal]}) ) self.assertEqual(len(accepted), 0) self.assertEqual(len(rejected), 1) proposal["listing_note"] = ( "I intend to list this property for rent to find a suitable tenant." ) accepted, rejected = accept_proposals( [record], json.dumps({"cases": [proposal]}) ) self.assertEqual((len(accepted), len(rejected)), (0, 1)) proposal["case_id"] = "unknown-case" with self.assertRaises(ValueError): accept_proposals([record], json.dumps({"cases": [proposal]})) def test_verdict_review_repairs_violations_and_preserves_the_case(self): from unittest.mock import patch from review_batch import Run record = next(pilot_cases(per_scenario=1)) original = copy.deepcopy(record) valid = { "verdict": record["target"], "evidence": [ { "reason": "NO_CONCERN", "path": "/account/age_days", "quote": str(record["input"]["account"]["age_days"]), } ], } invalid = copy.deepcopy(valid) invalid["evidence"][0]["quote"] = "invented evidence" audit = {"violations": [], "suggested_verdict": record["target"]} for fault in ("quote", "partial_numeric", "auditor", "reference"): first = invalid if fault == "quote" else valid first_audit = copy.deepcopy(audit) if fault == "partial_numeric": first = copy.deepcopy(valid) first["evidence"][0]["quote"] = first["evidence"][0]["quote"][1:] if fault == "auditor": first_audit["violations"] = [ "Unsupported reasoning despite correct label" ] if fault == "reference": first = copy.deepcopy(valid) first["verdict"]["risk_level"] = 3 first_audit["suggested_verdict"]["risk_level"] = 3 replies = [first, first_audit, valid, audit] with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp: directory = Path(tmp) (directory / "case-audits").mkdir() with patch.object( Run, "call", side_effect=[ (json.dumps(x), str(i)) for i, x in enumerate(replies) ], ): result = Run(directory).review(record) self.assertEqual(result["status"], "accepted") self.assertEqual(result["accepted"]["input"], original["input"]) self.assertEqual(result["accepted"]["metadata"], original["metadata"]) self.assertEqual(result["accepted"]["target"], original["target"]) self.assertEqual(len(result["history"]), 2) self.assertEqual(record, original) with tempfile.TemporaryDirectory() as tmp: directory = Path(tmp) (directory / "case-audits").mkdir() replies = [invalid, audit] * 3 with patch.object( Run, "call", side_effect=[(json.dumps(x), str(i)) for i, x in enumerate(replies)], ): result = Run(directory).review(record) self.assertEqual(result["status"], "quarantined") self.assertIsNone(result["accepted"]) self.assertEqual(len(result["history"]), 3) def test_droid_stream_accepts_reasoning_and_rejects_tool_use_or_wrong_model(self): from types import SimpleNamespace from unittest.mock import patch from review_batch import DroidRun, Proposal for fault in (None, "tool", "model"): events = [ { "type": "system", "subtype": "init", "model": "gpt-6-luna", "reasoning_effort": "medium", }, {"type": "reasoning", "text": "Assessing supplied facts."}, { "type": "completion", "numTurns": 1, "finalText": '{"ok":true}', "session_id": "test-session", }, ] if fault == "tool": events.insert(2, {"type": "tool_call", "name": "Read"}) if fault == "model": events[0]["model"] = "some-other-model" response = SimpleNamespace( returncode=0, stdout="\n".join(json.dumps(e) for e in events), stderr="" ) with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp: directory = Path(tmp) (directory / "responses").mkdir() (directory / "removed-tools.txt").write_text("read-cli") runner = DroidRun(directory) messages = [ {"role": "system", "content": "Review supplied facts."}, {"role": "user", "content": "{}"}, ] with patch("review_batch.subprocess.run", return_value=response): if fault is None: self.assertEqual( runner.call("case", messages, Proposal), ('{"ok":true}', "test-session"), ) else: with self.assertRaises(RuntimeError): runner.call("case", messages, Proposal) self.assertTrue((directory / "responses" / "case.json").exists()) self.assertFalse((directory / "responses" / "case.pending").exists()) def test_decision_labels_and_triage_metrics(self): from decision_modal import ( auroc, binary_metrics, calibration, decision, full_recall_threshold, ) self.assertEqual([decision(level) for level in range(1, 6)], [0, 0, 1, 1, 1]) labels = [0, 0, 1, 1] probs = [0.1, 0.6, 0.4, 0.9] self.assertEqual(auroc(labels, probs), 0.75) self.assertEqual(auroc([1, 1], [0.2, 0.3]), None) at_half = binary_metrics(labels, probs, 0.5) self.assertEqual(at_half["confusion"], {"tp": 1, "fp": 1, "fn": 1, "tn": 1}) threshold = full_recall_threshold(labels, probs) self.assertEqual(threshold, 0.4) self.assertEqual(full_recall_threshold([0, 1], [0.0, 1.0]), 0.5) self.assertEqual(full_recall_threshold([0], [0.2]), 0.5) self.assertEqual(binary_metrics(labels, probs, threshold)["recall"], 1.0) perfect = calibration([0, 1], [0.0, 1.0]) self.assertEqual(perfect["brier"], 0.0) self.assertEqual(perfect["expected_calibration_error"], 0.0) self.assertAlmostEqual( calibration([1], [0.8])["expected_calibration_error"], 0.2 ) if __name__ == "__main__": unittest.main()