← Source tree
test_conversation.py
"""Synthetic integration tests; run python -m unittest -v test_conversation."""
import copy
import hashlib
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from conversation import ReviewSession, canonical
from demo import fixture
from reviewqueue import review
class TestConversation(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.payload = fixture()
cls.expected = review(cls.payload)
def setUp(self):
self.session = ReviewSession(self.payload)
def confirmed(self):
viewed = self.session.view()
self.session.confirm(viewed["digest"])
return viewed
def test_real_model_report_is_not_stubbed(self):
document = self.session.view()["document"]
self.assertEqual(document["review"], self.expected)
self.assertEqual(document["review"]["reference_models"]["USD"]["status"], "trained")
self.assertIn("model_unusual_for_reference", document["review"]["rows"][1]["flags"])
def test_original_rows_order_ids_and_ambiguous_values_survive(self):
self.confirmed()
report = json.loads(self.session.export())["review"]
self.assertEqual([r["original"] for r in report["rows"]], self.payload["records"])
self.assertEqual(report["output_rows"], 6)
self.assertEqual(report["review_rows"], 4)
def test_report_digest_binds_flags_and_all_original_fields(self):
viewed = self.confirmed()
encoded = self.session.export()
self.assertEqual(hashlib.sha256(encoded).hexdigest(), viewed["digest"])
self.assertEqual(json.loads(encoded), viewed["document"])
def test_input_hash_is_canonical_not_raw_file_hash(self):
document = self.session.view()["document"]
self.assertEqual(document["canonical_input_sha256"],
hashlib.sha256(canonical(self.payload)).hexdigest())
self.assertNotIn("source_sha256", document)
def test_export_before_confirmation_refused(self):
with self.assertRaises(ValueError):
self.session.export()
def test_confirmation_before_review_refused(self):
with self.assertRaises(ValueError):
self.session.confirm("guessed")
def test_wrong_digest_refused(self):
self.session.view()
with self.assertRaises(ValueError):
self.session.confirm("0" * 64)
def test_new_review_invalidates_confirmation(self):
self.confirmed()
self.session.view()
with self.assertRaises(ValueError):
self.session.export()
def test_reset_invalidates_confirmation(self):
self.confirmed()
self.session.reset()
with self.assertRaises(ValueError):
self.session.export()
def test_view_copy_cannot_change_internal_export(self):
viewed = self.confirmed()
viewed["document"]["review"]["rows"][0]["original"]["record_id"] = "tampered"
self.assertEqual(json.loads(self.session.export())["review"], self.expected)
def test_input_mutation_cannot_change_internal_export(self):
data = copy.deepcopy(self.payload)
session = ReviewSession(data)
data["records"].clear()
viewed = session.view()
session.confirm(viewed["digest"])
self.assertEqual(json.loads(session.export())["review"], self.expected)
def test_session_does_not_mutate_input(self):
before = copy.deepcopy(self.payload)
self.confirmed()
self.session.export()
self.assertEqual(self.payload, before)
def test_network_is_not_used_by_model_or_session(self):
with patch("socket.socket", side_effect=AssertionError("network attempt")):
session = ReviewSession(self.payload)
viewed = session.view()
session.confirm(viewed["digest"])
session.export()
def test_unpaired_unicode_rejected(self):
data = copy.deepcopy(self.payload)
data["records"][0]["note"] = "\ud800"
with self.assertRaises(ValueError):
ReviewSession(data)
def test_empty_records_supported_without_fabricated_score(self):
session = ReviewSession({"reference": [], "records": []})
viewed = session.view()
self.assertEqual(viewed["document"]["review"]["output_rows"], 0)
session.confirm(viewed["digest"])
self.assertEqual(json.loads(session.export())["review"]["rows"], [])
def run_cli(self, source, target, commands):
return subprocess.run([sys.executable, str(Path(__file__).with_name("conversation.py")),
str(source), "--output", str(target)], input=commands,
text=True, capture_output=True, timeout=30)
def test_cli_unconfirmed_export_creates_nothing(self):
with tempfile.TemporaryDirectory() as tmp:
source, target = Path(tmp) / "input.json", Path(tmp) / "export.json"
raw = canonical(self.payload)
source.write_bytes(raw)
result = self.run_cli(source, target, "export\nconfirm wrong\nexport\nquit\n")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertFalse(target.exists())
self.assertEqual(source.read_bytes(), raw)
self.assertIn("Refused:", result.stdout)
def test_cli_confirmed_export_preserves_complete_review(self):
viewed = self.session.view()
with tempfile.TemporaryDirectory() as tmp:
source, target = Path(tmp) / "input.json", Path(tmp) / "export.json"
raw = canonical(self.payload)
source.write_bytes(raw)
result = self.run_cli(source, target,
f"review\nconfirm {viewed['digest']}\nexport\nquit\n")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(json.loads(target.read_bytes()), viewed["document"])
self.assertEqual(source.read_bytes(), raw)
self.assertIn("no submission", result.stdout)
def test_cli_does_not_overwrite_existing_file(self):
viewed = self.session.view()
with tempfile.TemporaryDirectory() as tmp:
source, target = Path(tmp) / "input.json", Path(tmp) / "export.json"
source.write_bytes(canonical(self.payload))
target.write_bytes(b"existing evidence")
result = self.run_cli(source, target,
f"review\nconfirm {viewed['digest']}\nexport\nquit\n")
self.assertEqual(target.read_bytes(), b"existing evidence")
self.assertIn("Refused:", result.stdout)
def test_cli_refuses_input_as_output(self):
with tempfile.TemporaryDirectory() as tmp:
source = Path(tmp) / "input.json"
raw = canonical(self.payload)
source.write_bytes(raw)
result = self.run_cli(source, source, "review\nexport\n")
self.assertNotEqual(result.returncode, 0)
self.assertEqual(source.read_bytes(), raw)
def test_cli_rejects_duplicate_keys_before_output(self):
with tempfile.TemporaryDirectory() as tmp:
source, target = Path(tmp) / "input.json", Path(tmp) / "export.json"
source.write_bytes(b'{"records":[],"records":[{}],"reference":[]}')
result = self.run_cli(source, target, "review\nexport\n")
self.assertNotEqual(result.returncode, 0)
self.assertNotIn("Traceback", result.stderr)
self.assertFalse(target.exists())
if __name__ == "__main__":
unittest.main(verbosity=2)