|
| 1 | +"""Explicit JSON selectors distinguish an empty key from an absent option.""" |
| 2 | +import json |
| 3 | +import os |
| 4 | +from pathlib import Path |
| 5 | +import subprocess |
| 6 | +import sys |
| 7 | +import tempfile |
| 8 | +import unittest |
| 9 | + |
| 10 | + |
| 11 | +ROOT = Path(__file__).resolve().parents[1] |
| 12 | + |
| 13 | + |
| 14 | +class ExplicitRecordsKeyTests(unittest.TestCase): |
| 15 | + def setUp(self): |
| 16 | + temporary = tempfile.TemporaryDirectory() |
| 17 | + self.addCleanup(temporary.cleanup) |
| 18 | + self.root = Path(temporary.name) |
| 19 | + self.contract = self.root / "contract.json" |
| 20 | + self.contract.write_text('{"columns":{"id":{"type":"string"}}}', encoding="utf-8") |
| 21 | + |
| 22 | + def run_cli(self, name, content, options=()): |
| 23 | + source = self.root / name |
| 24 | + source.write_bytes(content.encode("utf-8")) |
| 25 | + environment = dict(os.environ, PYTHONPATH=str(ROOT / "src"), |
| 26 | + PYTHONDONTWRITEBYTECODE="1", PYTHONIOENCODING="utf-8") |
| 27 | + return subprocess.run( |
| 28 | + [sys.executable, "-m", "public_data_sentinel.cli", str(source), |
| 29 | + "--contract", str(self.contract), *options], |
| 30 | + cwd=self.root, env=environment, capture_output=True, text=True, |
| 31 | + encoding="utf-8", timeout=30, |
| 32 | + ) |
| 33 | + |
| 34 | + def test_explicit_empty_key_selects_valid_and_empty_record_arrays(self): |
| 35 | + for records, expected in (([{"id": "001"}], 0), ([], 1)): |
| 36 | + with self.subTest(records=records): |
| 37 | + result = self.run_cli( |
| 38 | + "envelope.json", json.dumps({"": records, "other": [{"id": 123}]}), |
| 39 | + ("--records-key", ""), |
| 40 | + ) |
| 41 | + self.assertEqual(result.returncode, expected, result.stderr) |
| 42 | + self.assertEqual(result.stderr, "") |
| 43 | + report = json.loads(result.stdout) |
| 44 | + self.assertEqual(report["records_checked"], len(records)) |
| 45 | + self.assertEqual(report["valid"], expected == 0) |
| 46 | + if not records: |
| 47 | + self.assertEqual(report["issues"][0]["code"], "empty_data") |
| 48 | + |
| 49 | + def test_explicit_empty_key_rejects_missing_or_nonarray_selection(self): |
| 50 | + for content in ('[{"id":"001"}]', '{"items":[{"id":"001"}]}', '{"":{}}'): |
| 51 | + with self.subTest(content=content): |
| 52 | + result = self.run_cli("invalid-selection.json", content, ("--records-key", "")) |
| 53 | + self.assertEqual(result.returncode, 2, result.stderr) |
| 54 | + self.assertEqual(result.stdout, "") |
| 55 | + self.assertTrue(result.stderr.startswith("data-sentinel: ")) |
| 56 | + self.assertNotIn("Traceback", result.stderr) |
| 57 | + |
| 58 | + def test_delimited_inputs_reject_empty_and_named_selectors(self): |
| 59 | + for suffix in ("csv", "tsv"): |
| 60 | + for key in ("", "items"): |
| 61 | + with self.subTest(suffix=suffix, key=key): |
| 62 | + result = self.run_cli("data." + suffix, "id\n001\n", ("--records-key", key)) |
| 63 | + self.assertEqual(result.returncode, 2, result.stderr) |
| 64 | + self.assertEqual(result.stdout, "") |
| 65 | + self.assertIn("--records-key applies only to JSON", result.stderr) |
| 66 | + |
| 67 | + def test_absent_and_named_selectors_keep_existing_behavior(self): |
| 68 | + cases = ( |
| 69 | + ("array.json", '[{"id":"001"}]', ()), |
| 70 | + ("envelope.json", '{"items":[{"id":"001"}]}', ("--records-key", "items")), |
| 71 | + ("data.csv", "id\n001\n", ()), |
| 72 | + ("data.tsv", "id\n001\n", ()), |
| 73 | + ) |
| 74 | + for name, content, options in cases: |
| 75 | + with self.subTest(name=name, options=options): |
| 76 | + result = self.run_cli(name, content, options) |
| 77 | + self.assertEqual(result.returncode, 0, result.stderr) |
| 78 | + self.assertEqual(result.stderr, "") |
| 79 | + report = json.loads(result.stdout) |
| 80 | + self.assertTrue(report["valid"]) |
| 81 | + self.assertEqual(report["records_checked"], 1) |
| 82 | + self.assertEqual(report["error_count"], 0) |
| 83 | + |
| 84 | + |
| 85 | +if __name__ == "__main__": |
| 86 | + unittest.main() |
0 commit comments