volt
dc461001b9
wizard-work/beam/test_jsonl_validate.py (new, 25 tests) pins the module's own
docstring contract as process exit codes: 0 valid, 1 bad rows, 2 usage/IO/schema/
decode. wizard-work/beam/jsonl_validate.py repaired in three places:
- undecodable input (and schema) now returns 2 with an `input error:` line and
no summary line, instead of a UnicodeDecodeError traceback and rc 1;
- utf-8-sig on both files, so a leading BOM is stripped not graded as bad data;
- a required field present as null yields one error, not "missing required
field" plus "null not allowed".
Measured: the new test file against the shipped module (md5 f53b6288...) fails
6 of 25 (3 decode, 2 BOM, 1 null); against the repair, 25/25 pass. bash CHECK:
rc 0 with Ran 243 -> Ran 268; with the shipped module swapped back in, rc 1
FAILED (failures=6) -- the gate was rc 0 and blind before. No edit to CHECK:
the file is a unittest.TestCase, so the existing -p 'test_*.py' leg collects it.
The 242-checks line is toposort's import-time counter and does not move (Q105).
verify.py --repo . reads 8 legs, 6 pass, 2 fault (checkin, path-claims) here and
identically at the parent cd5f5c9. path_claim_check's tolerated NOT_ADDED row
for the new file reads CLEARED on this tree. Q105's 27 uncollected files / 131
tests remain open and untouched.
278 lines
12 KiB
Python
278 lines
12 KiB
Python
#!/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)
|