238 lines
10 KiB
Python
238 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""test_jsonl_validate.py — pins the exit-code and schema contract of jsonl_validate.py.
|
|
|
|
Written for queue item Q104: the module shipped at afb182a with no test, and its
|
|
undecodable-input path exited 1 (the code its own docstring assigns to "the rows
|
|
are bad") after a UnicodeDecodeError traceback. Exit 1 means "some record is
|
|
invalid"; a file that could not be decoded produced no verdict on any record, so
|
|
a caller that trusted the code reported a rejection it never performed.
|
|
|
|
Everything is asserted through the real process, not by importing main(): the
|
|
contract under test is the exit status a shell sees, and importing would let an
|
|
uncaught traceback escape as an exception instead of as a status.
|
|
|
|
unittest.TestCase (not module-level bare asserts) on purpose: the lane's gate is
|
|
`python3 -m unittest discover -p 'test_*.py'`, which collects this file only if
|
|
it defines test cases (Q105 records the 27 files the gate cannot see).
|
|
"""
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
SCRIPT = Path(__file__).resolve().with_name("jsonl_validate.py")
|
|
|
|
SCHEMA = {
|
|
"required": ["id"],
|
|
"fields": {
|
|
"id": "int",
|
|
"name": "str?",
|
|
"tags": "list?",
|
|
"score": "float?",
|
|
"active": "bool?",
|
|
"meta": "object?",
|
|
"note": "str", # present-and-null must be an error (not optional)
|
|
},
|
|
}
|
|
|
|
|
|
class JsonlValidatorContract(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.dir = Path(self._tmp.name)
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
# --- helpers -----------------------------------------------------------
|
|
def write(self, name, data):
|
|
path = self.dir / name
|
|
if isinstance(data, bytes):
|
|
path.write_bytes(data)
|
|
elif isinstance(data, (dict, list)):
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
else:
|
|
path.write_text(data, encoding="utf-8")
|
|
return str(path)
|
|
|
|
def run_validator(self, schema, input_text, *extra):
|
|
schema_path = self.write("schema.json", schema)
|
|
input_path = self.write("input.jsonl", input_text)
|
|
proc = subprocess.run(
|
|
[sys.executable, str(SCRIPT), schema_path, input_path, *extra],
|
|
capture_output=True, text=True,
|
|
)
|
|
return proc
|
|
|
|
def out_lines(self, proc):
|
|
return [ln for ln in proc.stdout.splitlines() if ln.strip()]
|
|
|
|
# --- exit-code contract: 0 valid, 1 bad rows, 2 the run itself failed ---
|
|
def test_valid_records_exit_0(self):
|
|
proc = self.run_validator(SCHEMA, '{"id": 1}\n{"id": 2, "name": "a"}\n')
|
|
self.assertEqual(proc.returncode, 0, proc.stderr)
|
|
self.assertIn("checked 2 records, 0 invalid", proc.stdout)
|
|
|
|
def test_invalid_record_exit_1(self):
|
|
proc = self.run_validator(SCHEMA, '{"id": 1}\n{"id": "x"}\n')
|
|
self.assertEqual(proc.returncode, 1, proc.stderr)
|
|
self.assertIn("checked 2 records, 1 invalid", proc.stdout)
|
|
self.assertIn("field 'id': expected int, got str", proc.stdout)
|
|
|
|
def test_missing_input_file_exit_2(self):
|
|
schema_path = self.write("schema.json", SCHEMA)
|
|
proc = subprocess.run(
|
|
[sys.executable, str(SCRIPT), schema_path, str(self.dir / "nope.jsonl")],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
|
|
self.assertIn("input error", proc.stderr)
|
|
|
|
def test_missing_schema_file_exit_2(self):
|
|
input_path = self.write("input.jsonl", '{"id": 1}\n')
|
|
proc = subprocess.run(
|
|
[sys.executable, str(SCRIPT), str(self.dir / "nope.json"), input_path],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
|
|
self.assertIn("schema error", proc.stderr)
|
|
|
|
def test_undecodable_input_exit_2_without_traceback(self):
|
|
"""Q104's core defect: bytes that are not UTF-8 are an input error, not bad rows."""
|
|
schema_path = self.write("schema.json", SCHEMA)
|
|
input_path = self.write("input.jsonl", b'\xff\xfe\n')
|
|
proc = subprocess.run(
|
|
[sys.executable, str(SCRIPT), schema_path, input_path],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
|
|
self.assertNotIn("Traceback", proc.stderr)
|
|
self.assertIn("input error", proc.stderr)
|
|
self.assertNotIn("invalid", proc.stdout.splitlines()[-1] if proc.stdout else "",
|
|
"a run that could not read the file must not report a record verdict")
|
|
|
|
def test_undecodable_after_valid_records_is_still_exit_2(self):
|
|
schema_path = self.write("schema.json", SCHEMA)
|
|
input_path = self.write("input.jsonl", b'{"id": 1}\n\xff\n')
|
|
proc = subprocess.run(
|
|
[sys.executable, str(SCRIPT), schema_path, input_path],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
|
|
self.assertNotIn("Traceback", proc.stderr)
|
|
|
|
def test_malformed_schema_json_exit_2(self):
|
|
proc = self.run_validator("{not json", '{"id": 1}\n')
|
|
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
|
|
self.assertIn("schema error", proc.stderr)
|
|
|
|
def test_unknown_type_exit_2(self):
|
|
proc = self.run_validator({"fields": {"a": "bogus"}}, '{"a": 1}\n')
|
|
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
|
|
self.assertIn("unknown type 'bogus'", proc.stderr)
|
|
|
|
def test_non_object_schema_exit_2(self):
|
|
proc = self.run_validator([1, 2], '{"id": 1}\n')
|
|
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
|
|
self.assertIn("schema must be an object", proc.stderr)
|
|
|
|
def test_required_not_list_of_strings_exit_2(self):
|
|
proc = self.run_validator({"required": "id"}, '{"id": 1}\n')
|
|
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
|
|
|
|
def test_max_errors_zero_is_usage_error(self):
|
|
"""--max-errors 0 used to grade nothing and still claim '(truncated)'."""
|
|
proc = self.run_validator(SCHEMA, '{"id": "x"}\n', "--max-errors", "0")
|
|
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
|
|
self.assertIn("must be >= 1", proc.stderr)
|
|
self.assertNotIn("truncated", proc.stdout)
|
|
|
|
def test_max_errors_negative_is_usage_error(self):
|
|
proc = self.run_validator(SCHEMA, '{"id": "x"}\n', "--max-errors", "-3")
|
|
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
|
|
|
|
# --- schema semantics --------------------------------------------------
|
|
def test_required_null_reports_one_error_not_two(self):
|
|
proc = self.run_validator(SCHEMA, '{"id": null}\n')
|
|
self.assertEqual(proc.returncode, 1)
|
|
lines = self.out_lines(proc)
|
|
self.assertEqual(len(lines), 2, lines) # one error + the summary
|
|
self.assertIn("missing required field 'id'", lines[0])
|
|
self.assertNotIn("null not allowed", proc.stdout,
|
|
"the same fact must not be reported twice")
|
|
|
|
def test_optional_field_null_is_allowed(self):
|
|
proc = self.run_validator(SCHEMA, '{"id": 1, "name": null, "tags": null}\n')
|
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
|
|
|
def test_optional_field_missing_is_allowed(self):
|
|
proc = self.run_validator(SCHEMA, '{"id": 1}\n')
|
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
|
|
|
def test_non_optional_field_null_is_one_error(self):
|
|
proc = self.run_validator(SCHEMA, '{"id": 1, "note": null}\n')
|
|
self.assertEqual(proc.returncode, 1)
|
|
self.assertEqual(self.out_lines(proc)[0], "line 1: field 'note': null not allowed")
|
|
|
|
def test_required_field_missing_is_error(self):
|
|
proc = self.run_validator(SCHEMA, '{"name": "a"}\n')
|
|
self.assertEqual(proc.returncode, 1)
|
|
self.assertIn("missing required field 'id'", proc.stdout)
|
|
|
|
def test_bool_is_not_an_int(self):
|
|
proc = self.run_validator(SCHEMA, '{"id": true}\n')
|
|
self.assertEqual(proc.returncode, 1)
|
|
self.assertIn("expected int, got bool", proc.stdout)
|
|
|
|
def test_float_accepts_int(self):
|
|
"""JSON has one number type; an int satisfies float (pinned, documented)."""
|
|
proc = self.run_validator(SCHEMA, '{"id": 1, "score": 3}\n')
|
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
|
|
|
def test_float_field_rejects_string(self):
|
|
proc = self.run_validator(SCHEMA, '{"id": 1, "score": "3"}\n')
|
|
self.assertEqual(proc.returncode, 1)
|
|
self.assertIn("field 'score': expected float, got str", proc.stdout)
|
|
self.assertNotIn("float?", proc.stdout,
|
|
"the '?' marks absence-allowed, not the type of a present value")
|
|
|
|
def test_non_object_record_is_rejected(self):
|
|
proc = self.run_validator(SCHEMA, '[1, 2]\n')
|
|
self.assertEqual(proc.returncode, 1)
|
|
self.assertIn("line 1: record is not an object", proc.stdout)
|
|
|
|
def test_blank_lines_are_skipped_and_uncounted(self):
|
|
proc = self.run_validator(SCHEMA, '\n{"id": 1}\n\n \n{"id": 2}\n\n')
|
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
|
self.assertIn("checked 2 records, 0 invalid", proc.stdout)
|
|
|
|
def test_utf8_bom_is_tolerated(self):
|
|
schema_path = self.write("schema.json", SCHEMA)
|
|
input_path = self.write("input.jsonl", b'\xef\xbb\xbf{"id": 1}\n')
|
|
proc = subprocess.run(
|
|
[sys.executable, str(SCRIPT), schema_path, input_path],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
|
self.assertIn("checked 1 records, 0 invalid", proc.stdout)
|
|
|
|
def test_extra_fields_are_allowed(self):
|
|
proc = self.run_validator(SCHEMA, '{"id": 1, "unlisted": {"deep": [1]}}\n')
|
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
|
|
|
def test_truncated_flag_only_when_truncated(self):
|
|
two_bad = '{"id": "x"}\n{"id": "y"}\n'
|
|
cut = self.run_validator(SCHEMA, two_bad, "--max-errors", "1")
|
|
self.assertEqual(cut.returncode, 1)
|
|
self.assertIn("(truncated)", cut.stdout)
|
|
whole = self.run_validator(SCHEMA, two_bad, "--max-errors", "50")
|
|
self.assertEqual(whole.returncode, 1)
|
|
self.assertNotIn("(truncated)", whole.stdout)
|
|
self.assertIn("checked 2 records, 2 invalid", whole.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|