← Source tree
demo.py
"""Generate original synthetic data; run a demo and save reproducible QA evidence."""
import hashlib
import json
import platform
import subprocess
import sys
import tempfile
import unittest
from importlib.metadata import version
from pathlib import Path
from reviewqueue import review
def fixture():
reference = [{"record_id": f"R{i:03}", "currency": "USD", "quantity": "1",
"amount": f"{80 + i * 2}.00"} for i in range(40)]
records = [
{"record_id": "00001", "currency": "USD", "quantity": "1", "amount": "119.00", "note": "Synthetic normal record"},
{"record_id": "00002", "currency": "USD", "quantity": "20", "amount": "99000.00", "note": "Synthetic unusual record, not known error"},
{"record_id": "00003", "currency": "USD", "quantity": "1", "amount": "130.00"},
{"record_id": "00003", "currency": "USD", "quantity": "1", "amount": "130.00"},
{"record_id": "00004", "currency": "USD", "quantity": "1", "amount": "1,234.50"},
{"record_id": "00005", "currency": "UZS", "quantity": "1", "amount": "1200000.00"},
]
return {"reference": reference, "records": records}
class TestReview(unittest.TestCase):
def setUp(self):
self.data = fixture()
def result(self):
return review(self.data)
def test_all_original_values_preserved(self):
self.assertEqual([r["original"] for r in self.result()["rows"]], self.data["records"])
def test_leading_zero_and_order(self):
self.assertEqual(self.result()["rows"][0]["original"]["record_id"], "00001")
def test_no_mutation(self):
before = json.dumps(self.data)
self.result()
self.assertEqual(before, json.dumps(self.data))
def test_duplicates_kept_and_flagged(self):
r = self.result()
self.assertEqual(r["output_rows"], 6)
self.assertTrue(all("repeated_record_id" in r["rows"][i]["flags"] for i in [2, 3]))
def test_locale_not_guessed(self):
self.assertIn("ambiguous_or_invalid_number", self.result()["rows"][4]["flags"])
def test_currency_not_mixed(self):
self.assertEqual(self.result()["rows"][5]["ml"]["status"], "not_scored")
def test_model_detects_synthetic_outlier(self):
self.assertIn("model_unusual_for_reference", self.result()["rows"][1]["flags"])
def test_missing_number(self):
self.data["records"][0]["amount"] = None
self.assertIn("missing_or_invalid_number", self.result()["rows"][0]["flags"])
def test_nontext_id_not_coerced(self):
self.data["records"][0]["record_id"] = 12
self.assertIn("missing_or_nontext_record_id", self.result()["rows"][0]["flags"])
def test_zero_quantity(self):
self.data["records"][0]["quantity"] = "0"
self.assertIn("quantity_must_be_positive", self.result()["rows"][0]["flags"])
def test_small_reference_abstains(self):
self.data["reference"] = self.data["reference"][:5]
self.assertEqual(self.result()["rows"][0]["ml"]["status"], "not_scored")
def test_invalid_reference_fails_explicitly(self):
self.data["reference"][0]["amount"] = "not a number"
with self.assertRaises(ValueError):
self.result()
def test_reproducible(self):
self.assertEqual(self.result(), self.result())
def test_empty_input(self):
self.data["records"] = []
self.assertEqual(self.result()["output_rows"], 0)
def test_boolean_amount_is_not_a_number(self):
self.data["records"][0]["amount"] = True
self.assertIn("missing_or_invalid_number", self.result()["rows"][0]["flags"])
def test_derived_overflow_is_flagged(self):
self.data["records"][0]["quantity"] = "0." + "0" * 999 + "1"
self.assertIn("unsupported_derived_numeric_range", self.result()["rows"][0]["flags"])
def test_cli_preserves_bytes_and_refuses_overwrite(self):
with tempfile.TemporaryDirectory() as d:
source, target = Path(d) / "input.json", Path(d) / "output.json"
raw = json.dumps(self.data).encode()
source.write_bytes(raw)
cmd = [sys.executable, str(Path(__file__).with_name("reviewqueue.py")), str(source), "--output", str(target)]
first = subprocess.run(cmd, capture_output=True, text=True)
self.assertEqual(first.returncode, 0, first.stderr)
produced = target.read_bytes()
second = subprocess.run(cmd, capture_output=True, text=True)
self.assertNotEqual(second.returncode, 0)
self.assertEqual(source.read_bytes(), raw)
self.assertEqual(target.read_bytes(), produced)
self.assertEqual(json.loads(produced)["source_sha256"], hashlib.sha256(raw).hexdigest())
def test_no_network_during_review(self):
from unittest.mock import patch
with patch("socket.socket", side_effect=AssertionError("network attempt")):
self.result()
def parse(self, raw):
import reviewqueue
return reviewqueue.load_payload(raw)
def test_duplicate_keys_rejected(self):
with self.assertRaisesRegex(ValueError, "duplicate JSON key"):
self.parse('{"record_id":"001","record_id":"999"}')
def test_escaped_duplicate_keys_rejected(self):
with self.assertRaisesRegex(ValueError, "duplicate JSON key"):
self.parse(r'{"record_id":"001","\u0072ecord_id":"999"}')
def test_nested_duplicate_keys_rejected(self):
with self.assertRaisesRegex(ValueError, "duplicate JSON key"):
self.parse('{"note":{"a":1,"a":2}}')
def test_float_precision_loss_rejected(self):
with self.assertRaisesRegex(ValueError, "precision"):
self.parse('{"amount":1.0000000000000001}')
def test_float_overflow_rejected(self):
with self.assertRaisesRegex(ValueError, "precision"):
self.parse('{"amount":1e309}')
def test_float_underflow_rejected(self):
with self.assertRaisesRegex(ValueError, "precision"):
self.parse('{"amount":1e-1000}')
def test_regular_float_preserved(self):
self.assertEqual(self.parse('{"amount":0.1,"id":"001"}'),
{"amount":0.1,"id":"001"})
def cli_rejection(self, raw):
with tempfile.TemporaryDirectory() as d:
source, target = Path(d) / "input.json", Path(d) / "output.json"
source.write_bytes(raw)
cmd = [sys.executable, str(Path(__file__).with_name("reviewqueue.py")),
str(source), "--output", str(target)]
result = subprocess.run(cmd, capture_output=True, text=True)
self.assertNotEqual(result.returncode, 0)
self.assertNotIn("Traceback", result.stderr)
self.assertEqual(source.read_bytes(), raw)
self.assertFalse(target.exists())
def test_cli_rejects_lossy_input_without_writing(self):
for raw in [b'{"records":[],"records":[{}],"reference":[]}',
b'{"records":[],"reference":[],"note":1.0000000000000001}']:
with self.subTest(raw=raw):
self.cli_rejection(raw)
def test_cli_rejects_unpaired_surrogate_without_writing(self):
self.cli_rejection(b'{"records":[{"note":"\\ud800"}],"reference":[]}')
def test_cli_rejects_invalid_utf8_without_writing(self):
self.cli_rejection(b'{"records":[],"reference":[],"note":"\xff"}')
if __name__ == "__main__":
here = Path(__file__).parent
data = fixture()
(here / "sample_input.json").write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
raw = (here / "sample_input.json").read_bytes()
result = review(data)
result["source_sha256"] = hashlib.sha256(raw).hexdigest()
(here / "sample_output.json").write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestReview)
with (here / "qa_evidence.txt").open("w", encoding="utf-8") as f:
f.write(f"Python {platform.python_version()}; scikit-learn {version('scikit-learn')}\n")
tests = unittest.TextTestRunner(stream=f, verbosity=2).run(suite)
f.write(f"Source unchanged: {raw == (here / 'sample_input.json').read_bytes()}\n")
f.write(f"Rows: {result['input_rows']} -> {result['output_rows']}; review flags: {result['review_rows']}\n")
f.write("Synthetic data only. Tests do not establish real-world accuracy or contest eligibility.\n")
print(f"{tests.testsRun} tests; success={tests.wasSuccessful()}; {result['review_rows']} review rows")
raise SystemExit(0 if tests.wasSuccessful() else 1)