#!/usr/bin/env python3 """test_jsonl_validate.py -- pins the exit-code and null-semantics contract of `jsonl_validate.py` (Q104). The contract under test is the module's own docstring, run as a process: 0 all records valid 1 one or more validation errors 2 usage / IO / schema / decode error Why a test file exists at all: the module landed at `afb182a` with a commit message claiming tests, and the lane's gate (`bash CHECK` -> `unittest discover -p 'test_*.py'`) reached none of them because there were none. Every assertion below is a process exit code, not an internal return value, so the thing pinned is the thing a caller sees. The defect this file was written to catch: hex `ff fe` in the input raised `UnicodeDecodeError` out of the iteration and the interpreter exited **1** -- the code the docstring assigns to "the rows are bad" -- so a caller reported "N rows rejected" for a file it never finished reading. It now exits 2, prints no summary line, and dumps no traceback. Run directly (`python3 test_jsonl_validate.py`) or as part of the lane gate. """ import json import os import subprocess import sys import tempfile import unittest HERE = os.path.dirname(os.path.abspath(__file__)) TOOL = os.path.join(HERE, "jsonl_validate.py") SCHEMA = { "required": ["id", "name"], "fields": { "id": "int", "name": "str", "score": "float?", "tags": "list?", "active": "bool?", "kind": "str", }, } class JsonlValidateContract(unittest.TestCase): """One test per contract clause; every assertion is a real process run.""" def setUp(self): self._td = tempfile.TemporaryDirectory() self.dir = self._td.name self.schema_path = self.write_text("schema.json", json.dumps(SCHEMA)) def tearDown(self): self._td.cleanup() # -- helpers --------------------------------------------------------- def write_text(self, name, text, encoding="utf-8"): path = os.path.join(self.dir, name) with open(path, "w", encoding=encoding) as f: f.write(text) return path def write_bytes(self, name, data): path = os.path.join(self.dir, name) with open(path, "wb") as f: f.write(data) return path def write_records(self, name, records): return self.write_text(name, "".join(json.dumps(r) + "\n" for r in records)) def run_tool(self, *args): proc = subprocess.run( [sys.executable, TOOL, *args], capture_output=True, text=True, ) return proc.returncode, proc.stdout, proc.stderr def assert_no_traceback(self, stderr): self.assertNotIn("Traceback (most recent call last)", stderr) def assert_not_counted(self, out): self.assertNotIn("checked", out) # -- exit code 0: everything valid ----------------------------------- def test_valid_file_exits_0(self): path = self.write_records("ok.jsonl", [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}]) rc, out, err = self.run_tool(self.schema_path, path) self.assertEqual(rc, 0, out + err) self.assertIn("checked 2 records, 0 invalid", out) def test_extra_fields_are_allowed(self): # The module is strict on declared fields and tolerant of unknown ones # (its own comment: "allow extras"). Pinned so that stays a decision. path = self.write_records("extra.jsonl", [{"id": 1, "name": "a", "unknown": [1, 2]}]) rc, out, _ = self.run_tool(self.schema_path, path) self.assertEqual(rc, 0, out) def test_blank_lines_are_skipped_and_not_counted(self): path = self.write_text("blank.jsonl", '{"id": 1, "name": "a"}\n\n \n\n{"id": 2, "name": "b"}\n') rc, out, _ = self.run_tool(self.schema_path, path) self.assertEqual(rc, 0, out) self.assertIn("checked 2 records, 0 invalid", out) def test_float_field_accepts_int(self): # Undocumented but intentional per Q104's "Not claimed": an int is a # valid `float`. Pinned to make it visible, not to endorse it. path = self.write_records("float.jsonl", [{"id": 1, "name": "a", "score": 3}]) rc, out, _ = self.run_tool(self.schema_path, path) self.assertEqual(rc, 0, out) # -- exit code 1: the rows are bad ----------------------------------- def test_type_mismatch_exits_1(self): path = self.write_records("bad.jsonl", [{"id": "x", "name": "a"}]) rc, out, err = self.run_tool(self.schema_path, path) self.assertEqual(rc, 1) self.assertIn("field 'id': expected int, got str", out) self.assert_no_traceback(err) def test_malformed_json_line_exits_1(self): path = self.write_text("broken.jsonl", '{"id": 1, "name": "a"}\n{not json}\n') rc, out, err = self.run_tool(self.schema_path, path) self.assertEqual(rc, 1) self.assertIn("line 2: invalid JSON", out) self.assertIn("checked 2 records, 1 invalid", out) self.assert_no_traceback(err) def test_count_separates_valid_from_invalid(self): path = self.write_records( "mixed.jsonl", [{"id": 1, "name": "a"}, {"id": "x", "name": "b"}, {"id": 2, "name": "c"}] ) rc, out, _ = self.run_tool(self.schema_path, path) self.assertEqual(rc, 1) self.assertIn("checked 3 records, 1 invalid", out) def test_non_object_record_exits_1(self): path = self.write_text("list.jsonl", "[1, 2, 3]\n") rc, out, _ = self.run_tool(self.schema_path, path) self.assertEqual(rc, 1) self.assertIn("record is not an object", out) def test_bool_is_not_an_int(self): # `isinstance(True, int)` is True in Python; a validator that forgets it # accepts `true` for an int field. Documented behaviour, asserted here. path = self.write_records("bool.jsonl", [{"id": True, "name": "a"}]) rc, out, _ = self.run_tool(self.schema_path, path) self.assertEqual(rc, 1) self.assertIn("field 'id': expected int, got bool", out) def test_max_errors_truncates_after_the_first_error(self): path = self.write_records("many.jsonl", [{"id": "a", "name": "x"}] * 4) rc, out, _ = self.run_tool(self.schema_path, path, "--max-errors", "1") self.assertEqual(rc, 1) self.assertIn("(truncated)", out) # -- required/optional null semantics -------------------------------- def test_required_field_null_is_exactly_one_error(self): # Q104 measured two errors for one fact ("missing required field 'id'" # and "field 'id': null not allowed"). One fact, one error. path = self.write_records("nullreq.jsonl", [{"id": None, "name": "a"}]) rc, out, _ = self.run_tool(self.schema_path, path) self.assertEqual(rc, 1) error_lines = [ln for ln in out.splitlines() if ln.startswith("line 1:")] self.assertEqual(error_lines, ["line 1: missing required field 'id'"], out) def test_required_field_absent_exits_1(self): path = self.write_records("absent.jsonl", [{"name": "a"}]) rc, out, _ = self.run_tool(self.schema_path, path) self.assertEqual(rc, 1) self.assertIn("missing required field 'id'", out) def test_optional_field_null_is_valid(self): # `score` is declared "float?" -- present-as-null is "missing", not bad. path = self.write_records("nullopt.jsonl", [{"id": 1, "name": "a", "score": None}]) rc, out, _ = self.run_tool(self.schema_path, path) self.assertEqual(rc, 0, out) def test_declared_non_optional_field_null_exits_1(self): # `kind` is declared "str" (not required, not optional): null is an # error, while the same field absent from the record is fine. absent = self.write_records("kind_absent.jsonl", [{"id": 1, "name": "a"}]) as_null = self.write_records("kind_null.jsonl", [{"id": 1, "name": "a", "kind": None}]) self.assertEqual(self.run_tool(self.schema_path, absent)[0], 0) rc, out, _ = self.run_tool(self.schema_path, as_null) self.assertEqual(rc, 1) self.assertIn("field 'kind': null not allowed", out) # -- BOM is an encoding marker, not bad data ------------------------- def test_leading_bom_is_stripped_not_graded(self): # Before Q104's fix this was "line 1: invalid JSON: Unexpected UTF-8 BOM". path = os.path.join(self.dir, "bom.jsonl") with open(path, "wb") as f: f.write(b"\xef\xbb\xbf" + json.dumps({"id": 1, "name": "a"}).encode("utf-8") + b"\n") rc, out, err = self.run_tool(self.schema_path, path) self.assertEqual(rc, 0, out + err) self.assertIn("checked 1 records, 0 invalid", out) def test_bom_in_schema_is_stripped(self): path = os.path.join(self.dir, "bom_schema.json") with open(path, "wb") as f: f.write(b"\xef\xbb\xbf" + json.dumps(SCHEMA).encode("utf-8")) records = self.write_records("ok2.jsonl", [{"id": 1, "name": "a"}]) rc, out, err = self.run_tool(path, records) self.assertEqual(rc, 0, out + err) # -- exit code 2: the RUN is bad, not the rows ----------------------- def test_missing_input_exits_2(self): rc, out, err = self.run_tool(self.schema_path, os.path.join(self.dir, "nope.jsonl")) self.assertEqual(rc, 2) self.assertIn("input error:", err) self.assert_no_traceback(err) self.assertNotIn("checked", out) def test_binary_input_exits_2_with_no_traceback(self): # THE PIN for Q104. At afb182a: rc 1 + UnicodeDecodeError traceback. path = self.write_bytes("bin.jsonl", b"\xff\xfe\n") rc, out, err = self.run_tool(self.schema_path, path) self.assertEqual(rc, 2, out + err) self.assert_no_traceback(err) self.assertIn("input error: not valid UTF-8 text", err) def test_binary_input_after_good_records_still_exits_2(self): # A run that dies mid-file must not leave a summary line a caller can # read as the verdict on the rows it never reached. path = self.write_bytes( "bin2.jsonl", json.dumps({"id": 1, "name": "a"}).encode("utf-8") + b"\n" + b"\xff\xfe\n", ) rc, out, err = self.run_tool(self.schema_path, path) self.assertEqual(rc, 2, out + err) self.assert_no_traceback(err) self.assertNotIn("checked", out) def test_truncated_utf8_sequence_exits_2(self): path = self.write_bytes("bin3.jsonl", b'{"id":1,"name":"\xc3"}') rc, out, err = self.run_tool(self.schema_path, path) self.assertEqual(rc, 2, out + err) self.assert_no_traceback(err) def test_missing_schema_exits_2(self): records = self.write_records("ok3.jsonl", [{"id": 1, "name": "a"}]) rc, out, err = self.run_tool(os.path.join(self.dir, "nope.json"), records) self.assertEqual(rc, 2) self.assertIn("schema error:", err) self.assert_not_counted(out) def test_schema_not_json_exits_2(self): bad_schema = self.write_text("bad_schema.json", "{schema: nope}") records = self.write_records("ok4.jsonl", [{"id": 1, "name": "a"}]) rc, _, err = self.run_tool(bad_schema, records) self.assertEqual(rc, 2) self.assertIn("schema error:", err) def test_schema_unknown_type_exits_2(self): bad_schema = self.write_text("bad_type.json", json.dumps({"fields": {"a": "bogus"}})) records = self.write_records("ok5.jsonl", [{"a": 1}]) rc, _, err = self.run_tool(bad_schema, records) self.assertEqual(rc, 2) self.assertIn("unknown type 'bogus'", err) def test_unreadable_input_directory_exits_2(self): rc, _, err = self.run_tool(self.schema_path, self.dir) self.assertEqual(rc, 2) self.assertIn("input error:", err) def test_usage_error_exits_2(self): rc, _, err = self.run_tool() self.assertEqual(rc, 2) self.assertNotIn("checked", err) if __name__ == "__main__": unittest.main(verbosity=2)