troupe-standup/wizard-work/spark/test_json_sha256_validator.py

#!/usr/bin/env python3
"""Cells for wizard-work/spark/json_sha256_validator.py (Q80).

Black box: every CLI reading runs the shipped module as a subprocess, so the
cells cannot agree with the code by sharing an implementation. The one
in-process cell imports the module by path to count how many times the subject
is read -- the property Q80 asks for, which a subprocess cannot observe.

Run: python3 wizard-work/spark/test_json_sha256_validator.py
Exit 0 iff every cell passes.
"""
import hashlib
import importlib.util
import io
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

TOOL = Path(__file__).resolve().with_name("json_sha256_validator.py")


def run(*args, cwd=None):
    proc = subprocess.run([sys.executable, str(TOOL), *args],
                          capture_output=True, text=True, cwd=cwd)
    return proc.returncode, proc.stdout, proc.stderr


class Base(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.dir = Path(self._tmp.name)
        self.ok = self.write("ok.json", b'{"a": [1, 2, 3], "b": "x"}\n')
        self.nan = self.write("nan.json", b'{"a": NaN}\n')
        self.inf = self.write("inf.json", b'{"a": Infinity}\n')
        self.neginf = self.write("neginf.json", b'{"a": -Infinity}\n')
        self.dup = self.write("dup.json", b'{"a": 1, "a": 2}\n')
        self.broken = self.write("broken.json", b'{"a": 1,\n')
        self.bom = self.write("bom.json", b'\xef\xbb\xbf{"a": 1}\n')

    def tearDown(self):
        self._tmp.cleanup()

    def write(self, name, data):
        path = self.dir / name
        path.write_bytes(data)
        return path

    def digest(self, path):
        return hashlib.sha256(Path(path).read_bytes()).hexdigest()


class Contract(Base):
    def test_usage_error_is_rc_2(self):
        rc, out, err = run()
        self.assertEqual(rc, 2, out + err)

    def test_valid_document_rc_0_and_digest_covers_the_file(self):
        rc, out, err = run(str(self.ok))
        self.assertEqual(rc, 0, out + err)
        self.assertIn("VALID", out)
        self.assertIn(f"SHA256={self.digest(self.ok)}", out)

    def test_nan_is_invalid_rc_1(self):
        rc, out, err = run(str(self.nan))
        self.assertEqual(rc, 1, out + err)
        self.assertIn("INVALID", out)
        self.assertIn("NaN", out)
        self.assertNotIn("SHA256", out)

    def test_infinity_and_negative_infinity_are_invalid(self):
        for path, name in ((self.inf, "Infinity"), (self.neginf, "Infinity")):
            rc, out, err = run(str(path))
            self.assertEqual(rc, 1, f"{path.name}: {out}{err}")
            self.assertIn("INVALID", out)
            self.assertIn(name, out)

    def test_duplicate_member_names_are_refused_by_default(self):
        rc, out, err = run(str(self.dup))
        self.assertEqual(rc, 1, out + err)
        self.assertIn("duplicate object member name", out)
        self.assertNotIn("SHA256", out)

    def test_allow_duplicates_accepts_them(self):
        rc, out, err = run("--allow-duplicates", str(self.dup))
        self.assertEqual(rc, 0, out + err)
        self.assertIn(f"SHA256={self.digest(self.dup)}", out)

    def test_syntax_error_is_rc_1(self):
        rc, out, err = run(str(self.broken))
        self.assertEqual(rc, 1, out + err)
        self.assertIn("INVALID", out)

    def test_bom_is_tolerated_and_the_choice_is_documented(self):
        rc, out, err = run(str(self.bom))
        self.assertEqual(rc, 0, out + err)
        self.assertIn(f"SHA256={self.digest(self.bom)}", out)
        self.assertIn("byte order mark", TOOL.read_text(encoding="utf-8").lower())

    def test_undecodable_bytes_are_invalid_rc_1(self):
        bad = self.write("bad.bin", b'\xff\xfe\x00\x01')
        rc, out, err = run(str(bad))
        self.assertEqual(rc, 1, out + err)
        self.assertIn("not decodable JSON text", out)


class Ungradable(Base):
    def test_missing_input_is_rc_2(self):
        rc, out, err = run(str(self.dir / "nope.json"))
        self.assertEqual(rc, 2, out + err)
        self.assertIn("UNGRADABLE", out)
        self.assertNotIn("VALID", out)
        self.assertNotIn("SHA256", out)

    def test_directory_input_is_rc_2(self):
        rc, out, err = run(str(self.dir))
        self.assertEqual(rc, 2, out + err)
        self.assertIn("UNGRADABLE", out)

    def test_non_directory_in_path_is_rc_2(self):
        rc, out, err = run(str(self.ok / "child.json"))
        self.assertEqual(rc, 2, out + err)
        self.assertIn("UNGRADABLE", out)

    def test_ungradable_suppresses_every_verdict_and_digest(self):
        rc, out, err = run(str(self.ok), str(self.dir / "nope.json"))
        self.assertEqual(rc, 2, out + err)
        self.assertNotIn("VALID", out)
        self.assertNotIn("INVALID", out)
        self.assertNotIn("SHA256", out)
        self.assertIn("no verdict and no digest printed", out)


class SingleRead(Base):
    def test_one_read_serves_both_the_parse_and_the_digest(self):
        """Serve payload A first and B after; the digest must be A's, read once."""
        target = self.dir / "race.json"
        target.write_bytes(b'{"a": 1}\n')
        a = b'{"a": 1}\n'
        b = b'{"a": 999}\n'
        counts = {"read_bytes": 0, "open": 0}

        spec = importlib.util.spec_from_file_location("json_sha256_validator", TOOL)
        if spec is None or spec.loader is None:
            self.fail(f"cannot load the module under test at {TOOL}")
        loader = spec.loader
        mod = importlib.util.module_from_spec(spec)
        loader.exec_module(mod)

        real_read_bytes = Path.read_bytes
        import builtins
        real_builtin_open = builtins.open

        def fake_read_bytes(self):
            counts["read_bytes"] += 1
            return a if counts["read_bytes"] == 1 else b

        def fake_open(path, mode="r", *args, **kwargs):
            if str(path) == str(target):
                counts["open"] += 1
                if "b" in mode:
                    return io.BytesIO(b)
                return io.StringIO(b.decode())
            return real_builtin_open(path, mode, *args, **kwargs)

        Path.read_bytes = fake_read_bytes
        builtins.open = fake_open
        try:
            record = mod.grade(str(target))
        finally:
            Path.read_bytes = real_read_bytes
            builtins.open = real_builtin_open

        reads = counts["read_bytes"] + counts["open"]
        self.assertEqual(record["verdict"], "valid", record)
        self.assertEqual(reads, 1, f"subject read {reads} times: {counts}")
        self.assertEqual(record["sha256"], hashlib.sha256(a).hexdigest())


class Results(Base):
    def test_results_file_carries_the_invocation_and_each_blob_sha(self):
        out_path = self.dir / "run.json"
        rc, out, err = run("--results", str(out_path), str(self.ok), str(self.dup))
        self.assertIn(rc, (0, 1), out + err)
        payload = json.loads(out_path.read_text(encoding="utf-8"))
        self.assertEqual(payload["tool"], TOOL.name)
        self.assertIn(str(TOOL.resolve()), payload["invocation"])
        self.assertIn(str(self.ok), payload["argv"])
        by_path = {item["path"]: item for item in payload["inputs"]}
        self.assertEqual(by_path[str(self.ok)]["sha256"], self.digest(self.ok))
        self.assertEqual(by_path[str(self.ok)]["verdict"], "valid")
        self.assertEqual(by_path[str(self.dup)]["verdict"], "invalid")


if __name__ == "__main__":
    unittest.main(verbosity=2)