diff --git a/conformance/binding/README.md b/conformance/binding/README.md index 2d9a811..f6939b4 100644 --- a/conformance/binding/README.md +++ b/conformance/binding/README.md @@ -7,6 +7,7 @@ Python 3.6+, standard library only, Apache-2.0. python3 conformance.py # the reference fake binding, level B2 python3 conformance.py --level B1 # the fake binding acting as B1 (plan always reloads) python3 conformance.py --level B2 -- ./my-binding --device 0 # your binding +python3 conformance.py --level B2 --fixtures /runs/fixtures.json --report /runs/result.json -- ./my-binding ``` Your binding is run as `COMMAND [args]` with JSON on stdin and stdout and a non-zero exit on error: @@ -22,6 +23,67 @@ Your binding is run as `COMMAND [args]` with JSON on stdin and stdou Schemas and examples: [`docs/schemas/`](../../docs/schemas/). +## Runtime artifact fixtures + +`--fixtures FILE` selects external artifact directories instead of the four synthetic switches. Options for the +kit go before `--`; the command and options for your binding go after it. A provider file has this shape: + +```json +{ + "schema": "onggi.binding-fixtures.v1", + "cases": [ + { + "name": "native-identical", + "base_dir": "/runs/artifacts/base", + "target_dir": "/runs/artifacts/base", + "switchplan": "identical.switchplan.json", + "expect_mode": "units" + }, + { + "name": "native-inplace", + "base_dir": "/runs/artifacts/base", + "target_dir": "/runs/artifacts/target", + "switchplan": "inplace.switchplan.json", + "expect_mode": "units" + } + ] +} +``` + +`cases` must be nonempty, with unique nonempty names. Every case requires all five fields shown. Relative +`base_dir`, `target_dir`, and `switchplan` paths resolve against the provider file's directory. `switchplan` may +instead contain an inline `onggi.switchplan.v1` object. Obtain it with `onggi diff --ranges --json BASE TARGET`; +the provider is responsible for supplying verified artifacts and their authentic manifest identities. The kit +checks manifest syntax, the complete target file inventory and sizes, paired/source base files, range partitions, +and byte counts before invoking the binding. Every `same` range must equal the paired base at the same offset; +every `elsewhere` range must equal its declared base source interval. Artifact directories must exist and contain regular files; symlinks +are rejected so no files are silently skipped or read outside those directories. + +`expect_mode` is exactly `units` or `reload`. At B2 each case must return its declared mode, and at least one case +must declare `units` with changed ranges (`switchplan.moved + switchplan.new > 0`) and an actual byte difference +between target files and the same-path base files (or paired base files when the path is new). A nonempty target +file without a base counterpart also counts. Reload-only or identical-only fixtures cannot establish B2, even +if unchanged bytes are labeled `new`. Individual `new` spans may include equal bytes, as chunk boundaries can +produce such spans. At B1 every case must declare `reload`, and the binding must always return it. +B2 retains coverage, apply readback, resident identity, +rollback restoration, and second-rollback refusal checks for these fixtures. A runtime supporting a bounded +lossless subset can provide native identical and in-place cases with `units`; a passing fixture run establishes +that declared subset, rather than support for every native switch. Default runs still require `units` for the +synthetic `identical` and `inplace` cases and still run all four synthetic cases. + +The checker uses file sizes from filesystem metadata and compares or hashes intervals with reads of at most 1 MiB. +It neither copies external artifacts nor reads a whole model into memory. Keep the directories unchanged while +the kit runs. Binding-side staging and memory use remain the binding's responsibility. + +`--report FILE` exclusively creates a new JSON file (`onggi.binding-report.v1`); an existing file is never +overwritten. Reports include the level, binding command, provider file SHA-256, and, for each executed case, +the resolved artifact directories, expected mode, artifact file sizes and SHA-256 hashes, and a SHA-256 of the +switchplan's canonical JSON (sorted keys, compact separators, UTF-8). Each operation records its full command, +JSON input, exit code and JSON output, or its execution error. The final `ok`, `failures`, and optional `error` +also record unsuccessful runs. Hashing the complete artifacts for a report is streamed. Without `--fixtures`, +reports describe the synthetic cases and omit the provider file field. Reports contain paths, commands, and +binding output; choose the destination accordingly. + ## What is in here - `golden/*.switchplan.json`: what `onggi diff --ranges --json` prints for four synthetic switches of a 2 MiB diff --git a/conformance/binding/conformance.py b/conformance/binding/conformance.py index 4335a3a..393fbe5 100644 --- a/conformance/binding/conformance.py +++ b/conformance/binding/conformance.py @@ -2,11 +2,12 @@ # Copyright 2026 The onggi authors. SPDX-License-Identifier: Apache-2.0 """Conformance tests for an onggi device binding (ADR-0010). Python 3.6+, standard library only. - python3 conformance.py [--level B1|B2] [-- BINDING COMMAND...] + python3 conformance.py [--level B1|B2] [--fixtures FILE] [--report FILE] [-- BINDING COMMAND...] Without a command it tests the reference fake binding. The binding command is run as `COMMAND [args]` with JSON on stdin and stdout; see README.md. """ +import argparse import hashlib import json import os @@ -44,6 +45,160 @@ def sha(b): return hashlib.sha256(b).hexdigest() +def file_size(data): + """A file is either synthetic bytes or an external path; external bytes stay on disk.""" + return os.path.getsize(data) if isinstance(data, str) else len(data) + + +def interval_sha(data, offset, length): + if not isinstance(data, str): + return sha(data[offset:offset + length]) + digest = hashlib.sha256() + with open(data, "rb") as f: + f.seek(offset) + while length: + chunk = f.read(min(length, 1024 * 1024)) + if not chunk: + raise ValueError("artifact shortened while hashing: %s" % data) + digest.update(chunk) + length -= len(chunk) + return digest.hexdigest() + + +def intervals_equal(left, left_offset, right, right_offset, length): + """Compare external file intervals without retaining model-sized buffers.""" + with open(left, "rb") as a, open(right, "rb") as b: + a.seek(left_offset) + b.seek(right_offset) + while length: + size = min(length, 1024 * 1024) + achunk, bchunk = a.read(size), b.read(size) + if len(achunk) != size or len(bchunk) != size: + raise ValueError("artifact shortened while comparing intervals") + if achunk != bchunk: + return False + length -= size + return True + + +def artifacts_differ(sp, base, target): + for f in sp["files"]: + source = base.get(f["path"]) or base.get(f.get("base")) + if source is None: + if f["size"] > 0: + return True + elif (file_size(source) != f["size"] + or not intervals_equal(target[f["path"]], 0, source, 0, f["size"])): + return True + return False + + +def artifact_files(root): + if not os.path.isdir(root): + raise ValueError("artifact directory does not exist: %s" % root) + files = {} + def walk_error(error): + raise error + for directory, dirs, names in os.walk(root, onerror=walk_error): + for name in dirs + names: + if os.path.islink(os.path.join(directory, name)): + raise ValueError("artifact symlinks are not supported: %s" % name) + for name in names: + path = os.path.join(directory, name) + if not os.path.isfile(path): + raise ValueError("artifact entry is not a regular file: %s" % path) + files[os.path.relpath(path, root).replace(os.sep, "/")] = path + return files + + +def validate_switch(sp, base, target): + """Check the provider's byte-space plan before trusting its coverage claims.""" + def require(condition, message): + if not condition: + raise ValueError("invalid switchplan: " + message) + + def integer(value, minimum=0): + return type(value) is int and value >= minimum + + require(isinstance(sp, dict) and sp.get("schema") == "onggi.switchplan.v1", "schema") + for field in ("base_manifest", "manifest"): + require(isinstance(sp.get(field), str) and re.fullmatch(r"sha256:[0-9a-f]{64}", sp[field]), field) + require(isinstance(sp.get("files"), list), "files must be a list") + paths, totals = set(), {"size": 0, "same": 0, "moved": 0, "new": 0} + for f in sp["files"]: + require(isinstance(f, dict), "file must be an object") + path = f.get("path") + require(isinstance(path, str) and path in target and path not in paths, "unknown or duplicate target path") + paths.add(path) + for field in totals: + require(integer(f.get(field)), "file " + field) + totals[field] += f[field] + require(f["size"] == file_size(target[path]), "target file size: " + path) + if "base" in f: + require(isinstance(f["base"], str) and f["base"] in base, "unknown paired base file") + require(isinstance(f.get("ranges"), list), "ranges are required (use diff --ranges)") + position, counts = 0, {"same": 0, "moved": 0, "new": 0} + for r in f["ranges"]: + require(isinstance(r, dict) and integer(r.get("offset")) and integer(r.get("length"), 1), "range interval") + require(r["offset"] == position and position + r["length"] <= f["size"], "ranges must partition each file") + kind = r.get("kind") + require(kind in ("same", "elsewhere", "new"), "range kind") + counts["moved" if kind == "elsewhere" else kind] += r["length"] + position += r["length"] + if kind == "same": + require(f.get("base") in base and position <= file_size(base[f["base"]]), "same range exceeds base file") + require(intervals_equal(target[path], r["offset"], base[f["base"]], r["offset"], r["length"]), + "same range bytes differ from paired base") + if kind == "elsewhere": + source = r.get("source_file") + require(isinstance(source, str) and source in base and integer(r.get("source_offset")), "elsewhere source") + require(r["source_offset"] + r["length"] <= file_size(base[source]), "elsewhere range exceeds base file") + require(intervals_equal(target[path], r["offset"], base[source], r["source_offset"], r["length"]), + "elsewhere range bytes differ from source") + else: + require("source_file" not in r and "source_offset" not in r, "source fields require elsewhere") + require(position == f["size"] and all(f[k] == counts[k] for k in counts), "range coverage or counts") + require(paths == set(target), "plan must cover every target file") + require(all(integer(sp.get(k)) and sp[k] == totals[k] for k in totals), "total sizes or counts") + + +def load_fixtures(path, level): + directory = os.path.dirname(os.path.abspath(path)) + def resolve(value): + if not isinstance(value, str) or not value.strip(): + raise ValueError("fixture paths must be nonempty strings") + return os.path.abspath(os.path.join(directory, value)) + + with open(path, encoding="utf-8") as f: + provider = json.load(f) + if not (isinstance(provider, dict) and provider.get("schema") == "onggi.binding-fixtures.v1" + and isinstance(provider.get("cases"), list) and provider["cases"]): + raise ValueError("fixtures require schema onggi.binding-fixtures.v1 and nonempty cases") + cases, names, changed_units = [], set(), False + for case in provider["cases"]: + if not isinstance(case, dict): + raise ValueError("fixture case must be an object") + name, mode = case.get("name"), case.get("expect_mode") + if not isinstance(name, str) or not name.strip() or name in names: + raise ValueError("fixture names must be nonempty and unique") + names.add(name) + if mode not in ("units", "reload") or (level == "B1" and mode != "reload"): + raise ValueError("fixture expect_mode must be units or reload (B1 requires reload)") + bdir, tdir = resolve(case.get("base_dir")), resolve(case.get("target_dir")) + base, target = artifact_files(bdir), artifact_files(tdir) + sp = case.get("switchplan") + if isinstance(sp, str): + with open(resolve(sp), encoding="utf-8") as f: + sp = json.load(f) + validate_switch(sp, base, target) + changed_units = changed_units or (mode == "units" and sp["moved"] + sp["new"] > 0 + and artifacts_differ(sp, base, target)) + cases.append((name, bdir, tdir, base, target, sp, mode)) + if level == "B2" and not changed_units: + raise ValueError("B2 requires at least one changed switch with expect_mode units") + return cases + + def covers(update, switch): """Every byte that is not `same` in the switch plan lies in a unit of the same file.""" units = {} @@ -79,14 +234,14 @@ def interval(unit): path, off, length = unit.get("file"), unit.get("offset"), unit.get("length") if (not isinstance(path, str) or path not in files or type(off) is not int or type(length) is not int - or off < 0 or length < 0 or off + length > len(files[path]) - or (length == 0 and not (mode == "reload" and off == 0 and len(files[path]) == 0))): + or off < 0 or length < 0 or off + length > file_size(files[path]) + or (length == 0 and not (mode == "reload" and off == 0 and file_size(files[path]) == 0))): return None return path, off, length # Reload rollback restores the base files, whose paths and sizes may differ from the target. required = update["units"] if mode == "units" else [ - {"file": path, "offset": 0, "length": len(data)} for path, data in files.items()] + {"file": path, "offset": 0, "length": file_size(data)} for path, data in files.items()] expected = [interval(u) for u in required] actual = [interval(u) for u in ver["units"]] ok = (None not in expected and None not in actual @@ -94,26 +249,14 @@ def interval(unit): and set(actual) == set(expected)) if ok: for unit, (path, off, length) in zip(ver["units"], actual): - ok = ok and unit.get("sha256") == sha(files[path][off:off + length]) + ok = ok and unit.get("sha256") == interval_sha(files[path], off, length) if "bytes_written" in ver: ok = (ok and type(ver["bytes_written"]) is int and ver["bytes_written"] >= 0) return check(ok, what) -def main(argv): - level, cmd = "B2", None - if "--" in argv: - cmd = argv[argv.index("--") + 1:] - argv = argv[:argv.index("--")] - if "--level" in argv: - level = argv[argv.index("--level") + 1] - work = tempfile.mkdtemp(prefix="onggi-binding-") - if not cmd: - os.environ["ONGGI_FAKE_DEVICE"] = os.path.join(work, "device") - if level == "B1": - os.environ["ONGGI_FAKE_LEVEL"] = "B1" - cmd = [sys.executable, os.path.join(HERE, "fake_binding.py")] +def synthetic_cases(work, level): for name in golden_data.CASES: base, target = golden_data.case(name) bdir, tdir = os.path.join(work, name, "base"), os.path.join(work, name, "target") @@ -121,49 +264,125 @@ def main(argv): golden_data.write(tdir, target) with open(os.path.join(HERE, "golden", name + ".switchplan.json")) as f: sp = json.load(f) + mode = "reload" if level == "B1" else "units" if name in ("identical", "inplace") else None + yield name, bdir, tdir, base, target, sp, mode + + +def file_hashes(files): + return {path: {"size": file_size(data), "sha256": interval_sha(data, 0, file_size(data))} + for path, data in sorted(files.items())} + + +def run_cases(level, cmd, cases, report): + for name, bdir, tdir, base, target, sp, mode in cases: + start = len(FAILS) + case_report = {"name": name, "base_dir": bdir, "target_dir": tdir, "expect_mode": mode, + "results": []} + if report is not None: + report["cases"].append(case_report) + case_report.update(base_files=file_hashes(base), target_files=file_hashes(target), + switchplan_sha256=sha(json.dumps(sp, sort_keys=True, separators=(",", ":")).encode("utf-8"))) + + def execute(op, *args, **kwargs): + result = {"command": cmd + [op] + list(args), "stdin": kwargs.get("stdin")} + if report is not None: + case_report["results"].append(result) + try: + rc, value = run(cmd, op, *args, **kwargs) + result.update(returncode=rc, stdout=value) + return rc, value + except Exception as error: + result["error"] = str(error) + raise + print("== %s (%s)" % (name, level)) + try: + _, prof = execute("describe", tdir) + check(prof.get("schema") == "onggi.profile.v1", "describe returns onggi.profile.v1") + + execute("conformance-load", bdir, sp["base_manifest"]) + _, res = execute("resident") + check(res.get("schema") == "onggi.resident.v1" and res.get("manifest") == sp["base_manifest"], + "resident names the loaded base") - _, prof = run(cmd, "describe", tdir) - check(prof.get("schema") == "onggi.profile.v1", "describe returns onggi.profile.v1") - - run(cmd, "conformance-load", bdir, sp["base_manifest"]) - _, res = run(cmd, "resident") - check(res.get("schema") == "onggi.resident.v1" and res.get("manifest") == sp["base_manifest"], - "resident names the loaded base") - - _, up1 = run(cmd, "plan", stdin=sp) - _, up2 = run(cmd, "plan", stdin=sp) - _, res2 = run(cmd, "resident") - check(up1 == up2 and res2 == res, "plan is pure: same answer twice, resident state unchanged") - check(up1.get("schema") == "onggi.updateplan.v1" and up1.get("mode") in ("units", "reload"), - "plan returns onggi.updateplan.v1 with mode units or reload") - check(up1["base_manifest"] == sp["base_manifest"] and up1["manifest"] == sp["manifest"], - "update plan names the switch plan's base and target") - if level == "B1": - check(up1["mode"] == "reload", "B1: plan always answers reload") - elif name in ("identical", "inplace"): - check(up1["mode"] == "units", "B2: a switch that keeps 97 % in place is applied as units") - if up1["mode"] == "units": - check(covers(up1, sp), "units cover every byte that is not at the same offset") - - wrong = dict(sp, base_manifest="sha256:" + "0" * 64) - rc, upw = run(cmd, "plan", stdin=wrong, expect_ok=False) - check(rc != 0 or (upw or {}).get("mode") == "reload", "plan against a base that is not resident refuses (reload)") - - if level == "B1": - continue - _, ver = run(cmd, "apply", "--target", tdir, stdin=up1) - verify_units(ver, target, "apply verifies: every planned unit matches the target bytes", up1, "apply") - _, res3 = run(cmd, "resident") - check(res3.get("manifest") == sp["manifest"], "after apply the target is resident") - - _, rb = run(cmd, "rollback") - verify_units(rb, base, "rollback restores: every planned unit matches the base bytes", up1, "rollback") - _, res4 = run(cmd, "resident") - check(res4.get("manifest") == sp["base_manifest"], "after rollback the base is resident") - rc, _ = run(cmd, "rollback", expect_ok=False) - check(rc != 0, "a second rollback has nothing to undo and fails") + _, up1 = execute("plan", stdin=sp) + _, up2 = execute("plan", stdin=sp) + _, res2 = execute("resident") + check(up1 == up2 and res2 == res, "plan is pure: same answer twice, resident state unchanged") + check(up1.get("schema") == "onggi.updateplan.v1" and up1.get("mode") in ("units", "reload"), + "plan returns onggi.updateplan.v1 with mode units or reload") + check(up1["base_manifest"] == sp["base_manifest"] and up1["manifest"] == sp["manifest"], + "update plan names the switch plan's base and target") + if mode is not None: + check(up1["mode"] == mode, "%s: expected mode %s" % (level, mode)) + if up1["mode"] == "units": + check(covers(up1, sp), "units cover every byte that is not at the same offset") + wrong_manifest = "sha256:" + ("1" if sp["base_manifest"] == "sha256:" + "0" * 64 else "0") * 64 + wrong = dict(sp, base_manifest=wrong_manifest) + rc, upw = execute("plan", stdin=wrong, expect_ok=False) + check(rc != 0 or (upw or {}).get("mode") == "reload", "plan against a base that is not resident refuses (reload)") + + if level == "B1": + continue + _, ver = execute("apply", "--target", tdir, stdin=up1) + verify_units(ver, target, "apply verifies: every planned unit matches the target bytes", up1, "apply") + _, res3 = execute("resident") + check(res3.get("manifest") == sp["manifest"], "after apply the target is resident") + + _, rb = execute("rollback") + verify_units(rb, base, "rollback restores: every planned unit matches the base bytes", up1, "rollback") + _, res4 = execute("resident") + check(res4.get("manifest") == sp["base_manifest"], "after rollback the base is resident") + rc, _ = execute("rollback", expect_ok=False) + check(rc != 0, "a second rollback has nothing to undo and fails") + finally: + case_report["failures"] = FAILS[start:] + + +def main(argv): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--level", choices=("B1", "B2"), default="B2") + parser.add_argument("--fixtures", help="external onggi.binding-fixtures.v1 provider JSON") + parser.add_argument("--report", help="write a JSON run report to a new file") + cmd = None + if "--" in argv: + cmd = argv[argv.index("--") + 1:] + argv = argv[:argv.index("--")] + options = parser.parse_args(argv) + FAILS[:] = [] + report = {"schema": "onggi.binding-report.v1", "level": options.level, + "argv": argv + (["--"] + cmd if cmd else []), "cases": []} + report_file = None + completed = False + try: + if options.report: + report_file = open(options.report, "x", encoding="utf-8") + if options.fixtures: + if report_file is not None: + report["fixtures"] = {"path": os.path.abspath(options.fixtures), + "sha256": interval_sha(options.fixtures, 0, os.path.getsize(options.fixtures))} + cases = load_fixtures(options.fixtures, options.level) + work = tempfile.mkdtemp(prefix="onggi-binding-") + if not options.fixtures: + cases = synthetic_cases(work, options.level) + if not cmd: + os.environ["ONGGI_FAKE_DEVICE"] = os.path.join(work, "device") + if options.level == "B1": + os.environ["ONGGI_FAKE_LEVEL"] = "B1" + cmd = [sys.executable, os.path.join(HERE, "fake_binding.py")] + report["binding_command"] = cmd + run_cases(options.level, cmd, cases, report if report_file is not None else None) + completed = True + except (OSError, ValueError, RuntimeError, KeyError, TypeError, AttributeError) as error: + report["error"] = str(error) + check(False, str(error)) + finally: + if report_file is not None: + report.update(ok=completed and not FAILS, failures=list(FAILS)) + with report_file: + json.dump(report, report_file, indent=2, sort_keys=True) + report_file.write("\n") print("\n%d failed" % len(FAILS)) return 1 if FAILS else 0 diff --git a/conformance/binding/test_runtime_fixtures.py b/conformance/binding/test_runtime_fixtures.py new file mode 100644 index 0000000..7b65698 --- /dev/null +++ b/conformance/binding/test_runtime_fixtures.py @@ -0,0 +1,250 @@ +# Copyright 2026 The onggi authors. SPDX-License-Identifier: Apache-2.0 +"""External artifact fixtures exercise the same checks without loading model files into RAM.""" +import contextlib +import copy +import hashlib +import io +import json +import os +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +import conformance + + +class RuntimeFixtureTests(unittest.TestCase): + def setUp(self): + self.work = tempfile.TemporaryDirectory() + self.addCleanup(self.work.cleanup) + self.root = self.work.name + self.old_fails = conformance.FAILS[:] + conformance.FAILS[:] = [] + self.addCleanup(lambda: conformance.FAILS.__setitem__(slice(None), self.old_fails)) + self.base = b'a' * 16384 + self.target = self.base[:4096] + b'b' * 100 + self.base[4196:] + for name, data in [('base', self.base), ('target', self.target)]: + os.mkdir(os.path.join(self.root, name)) + with open(os.path.join(self.root, name, 'model.bin'), 'wb') as f: + f.write(data) + self.switch = { + 'schema': 'onggi.switchplan.v1', 'base_manifest': 'sha256:' + '1' * 64, + 'manifest': 'sha256:' + '2' * 64, 'size': 16384, 'same': 16284, 'moved': 0, 'new': 100, + 'files': [{'path': 'model.bin', 'base': 'model.bin', 'size': 16384, + 'same': 16284, 'moved': 0, 'new': 100, + 'ranges': [{'offset': 0, 'length': 4096, 'kind': 'same'}, + {'offset': 4096, 'length': 100, 'kind': 'new'}, + {'offset': 4196, 'length': 12188, 'kind': 'same'}]}]} + self.case = {'name': 'native-inplace', 'base_dir': 'base', 'target_dir': 'target', + 'switchplan': self.switch, 'expect_mode': 'units'} + self.fixture = {'schema': 'onggi.binding-fixtures.v1', 'cases': [self.case]} + self.fixture_path = os.path.join(self.root, 'fixtures.json') + + def write_fixture(self, fixture=None): + with open(self.fixture_path, 'w') as f: + json.dump(self.fixture if fixture is None else fixture, f) + + def cli(self, *args, fixture=None, env=None): + self.write_fixture(fixture) + return subprocess.run([sys.executable, conformance.__file__, '--fixtures', self.fixture_path] + list(args), + stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) + + def test_external_units_and_report_record_inputs_and_commands(self): + report = os.path.join(self.root, 'report.json') + result = self.cli('--report', report) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertTrue(os.path.isfile(report), 'external fixture run must write the requested report') + with open(report) as f: + doc = json.load(f) + self.assertTrue(doc['ok']) + with open(self.fixture_path, 'rb') as f: + self.assertEqual(doc['fixtures']['sha256'], hashlib.sha256(f.read()).hexdigest()) + case = doc['cases'][0] + self.assertEqual(case['name'], 'native-inplace') + self.assertEqual(case['target_files']['model.bin']['sha256'], hashlib.sha256(self.target).hexdigest()) + self.assertEqual(case['target_files']['model.bin']['size'], 16384) + operations = [r['command'][-1] for r in case['results'] if r['command'][-1] in ('resident', 'rollback')] + self.assertIn('resident', operations) + self.assertEqual(operations.count('rollback'), 2) + self.assertIn(b'== native-inplace', result.stdout) + self.assertNotIn(b'== shifted', result.stdout) + + def test_path_switchplan_and_b1_reload(self): + switchpath = os.path.join(self.root, 'plan.json') + with open(switchpath, 'w') as f: + json.dump(self.switch, f) + case = dict(self.case, switchplan='plan.json', expect_mode='reload') + result = self.cli('--level', 'B1', fixture=dict(self.fixture, cases=[case])) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn(b'== native-inplace', result.stdout) + self.assertNotIn(b'== shifted', result.stdout) + + def test_b2_reload_verifies_apply_and_rollback(self): + switch = copy.deepcopy(self.switch) + switch.update(same=0, new=16384) + switch['files'][0].update(same=0, new=16384, + ranges=[{'offset': 0, 'length': 16384, 'kind': 'new'}]) + case = dict(self.case, name='native-reload', switchplan=switch, expect_mode='reload') + result = self.cli(fixture=dict(self.fixture, cases=[self.case, case])) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn(b'expected mode units', result.stdout) + self.assertIn(b'expected mode reload', result.stdout) + self.assertEqual(result.stdout.count(b'rollback restores'), 2) + + def test_b2_requires_a_changed_switch_expected_to_use_units(self): + identical = copy.deepcopy(self.switch) + identical.update(same=16384, new=0) + identical['files'][0].update(same=16384, new=0, + ranges=[{'offset': 0, 'length': 16384, 'kind': 'same'}]) + no_write = dict(self.case, name='native-identical', target_dir='base', switchplan=identical) + reload_case = dict(self.case, expect_mode='reload') + for cases in [[reload_case], [no_write], [no_write, reload_case]]: + with self.subTest(cases=cases): + self.write_fixture(dict(self.fixture, cases=cases)) + with self.assertRaisesRegex(ValueError, 'B2 requires.*changed.*units'): + conformance.load_fixtures(self.fixture_path, 'B2') + + def test_rejects_false_same_and_elsewhere_byte_claims(self): + for kind, counter in [('same', 'same'), ('elsewhere', 'moved')]: + switch = copy.deepcopy(self.switch) + switch['new'] = switch['files'][0]['new'] = 0 + switch[counter] += 100 + switch['files'][0][counter] += 100 + middle = switch['files'][0]['ranges'][1] + middle['kind'] = kind + if kind == 'elsewhere': + middle.update(source_file='model.bin', source_offset=0) + case = dict(self.case, switchplan=switch, expect_mode='reload') + with self.subTest(kind=kind): + self.write_fixture(dict(self.fixture, cases=[case])) + with self.assertRaisesRegex(ValueError, kind + ' range bytes differ'): + conformance.load_fixtures(self.fixture_path, 'B1') + + def test_b2_rejects_identical_bytes_reported_as_new(self): + case = dict(self.case, target_dir='base') + self.write_fixture(dict(self.fixture, cases=[case])) + with self.assertRaisesRegex(ValueError, 'B2 requires.*changed.*units'): + conformance.load_fixtures(self.fixture_path, 'B2') + + def test_accepts_true_elsewhere_bytes_with_changed_units(self): + for directory, data in [('base', self.base), ('target', self.target)]: + data = data[:8192] + b'b' * 100 + data[8292:] + with open(os.path.join(self.root, directory, 'model.bin'), 'wb') as f: + f.write(data) + self.switch['new'] = self.switch['files'][0]['new'] = 0 + self.switch['moved'] = self.switch['files'][0]['moved'] = 100 + self.switch['files'][0]['ranges'][1].update(kind='elsewhere', source_file='model.bin', source_offset=8192) + result = self.cli() + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_fixture_expectation_cannot_be_silently_downgraded(self): + env = dict(os.environ, ONGGI_FAKE_LEVEL='B1') + result = self.cli(env=env) + self.assertNotEqual(result.returncode, 0) + self.assertIn(b'expected mode units', result.stdout) + + def test_zero_base_manifest_still_probes_a_different_wrong_base(self): + self.switch['base_manifest'] = 'sha256:' + '0' * 64 + result = self.cli() + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_invalid_fixture_report_records_input_hash_and_failure(self): + report = os.path.join(self.root, 'report.json') + result = self.cli('--report', report, fixture=dict(self.fixture, cases=[])) + self.assertNotEqual(result.returncode, 0) + with open(report) as f: + doc = json.load(f) + self.assertFalse(doc['ok']) + self.assertIn('fixtures', doc) + self.assertEqual(doc['fixtures']['path'], self.fixture_path) + with open(self.fixture_path, 'rb') as f: + self.assertEqual(doc['fixtures']['sha256'], hashlib.sha256(f.read()).hexdigest()) + self.assertTrue(doc['failures']) + + def test_rejects_invalid_provider_before_running_binding(self): + invalid = [dict(self.fixture, schema='unknown'), dict(self.fixture, cases=[]), + dict(self.fixture, cases=[self.case, self.case])] + for changes in [{'name': ''}, {'base_dir': 'missing'}, {'target_dir': ''}, + {'expect_mode': 'anything'}, {'switchplan': 'missing.json'}]: + invalid.append(dict(self.fixture, cases=[dict(self.case, **changes)])) + for changes in [{'manifest': 'no digest'}, {'schema': 'unknown'}, {'files': []}]: + invalid.append(dict(self.fixture, cases=[dict(self.case, switchplan=dict(self.switch, **changes))])) + for key, value in [('path', '../model.bin'), ('size', 123), ('ranges', []), + ('ranges', [{'offset': 1, 'length': 16383, 'kind': 'same'}])]: + switch = copy.deepcopy(self.switch) + switch['files'][0][key] = value + invalid.append(dict(self.fixture, cases=[dict(self.case, switchplan=switch)])) + for fixture in invalid: + with self.subTest(fixture=fixture): + result = self.cli(fixture=fixture) + self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertNotIn(b'== ', result.stdout) + + def test_report_refuses_to_overwrite_existing_file(self): + report = os.path.join(self.root, 'report.json') + with open(report, 'w') as f: + f.write('keep') + result = self.cli('--report', report) + self.assertNotEqual(result.returncode, 0) + with open(report) as f: + self.assertEqual(f.read(), 'keep') + self.assertNotIn(b'== ', result.stdout) + + def test_failed_mode_check_is_in_report(self): + report = os.path.join(self.root, 'report.json') + result = self.cli('--report', report, env=dict(os.environ, ONGGI_FAKE_LEVEL='B1')) + self.assertNotEqual(result.returncode, 0) + self.assertTrue(os.path.isfile(report)) + with open(report) as f: + doc = json.load(f) + self.assertFalse(doc['ok']) + self.assertTrue(any('expected mode units' in failure for failure in doc['failures'])) + + def test_verifies_disk_intervals_with_bounded_reads(self): + path = os.path.join(self.root, 'large.bin') + data = b'0123456789' * 300000 + with open(path, 'wb') as f: + f.write(data) + plan = {'mode': 'reload', 'manifest': 'sha256:' + '2' * 64, 'units': []} + ver = {'schema': 'onggi.verification.v1', 'op': 'apply', 'mode': 'reload', + 'manifest': plan['manifest'], 'ok': True, + 'units': [{'file': 'model.bin', 'offset': 0, 'length': len(data), + 'sha256': hashlib.sha256(data).hexdigest()}]} + real_open = open + + class BoundedReader: + def __init__(self, f): + self.f = f + def __enter__(self): + return self + def __exit__(self, *args): + self.f.close() + def seek(self, *args): + return self.f.seek(*args) + def read(self, n=-1): + if not 0 <= n <= 1024 * 1024: + raise AssertionError('artifact reads must be bounded') + return self.f.read(n) + + def bounded_open(filename, *args, **kwargs): + f = real_open(filename, *args, **kwargs) + return BoundedReader(f) if filename == path else f + + with mock.patch('builtins.open', side_effect=bounded_open), contextlib.redirect_stdout(io.StringIO()): + self.assertTrue(conformance.intervals_equal(path, 0, path, 0, len(data))) + self.assertFalse(conformance.intervals_equal(path, 0, path, 1, len(data) - 1)) + with self.assertRaisesRegex(ValueError, 'shortened'): + conformance.intervals_equal(path, 0, path, 0, len(data) + 1) + self.assertTrue(conformance.verify_units(ver, {'model.bin': path}, 'disk reload', plan, 'apply')) + plan = dict(plan, mode='units', units=[{'file': 'model.bin', 'offset': 4096, 'length': 1200000}]) + ver = dict(ver, mode='units', units=[dict(plan['units'][0], + sha256=hashlib.sha256(data[4096:1204096]).hexdigest())]) + self.assertTrue(conformance.verify_units(ver, {'model.bin': path}, 'disk units', plan, 'apply')) + ver['units'][0]['sha256'] = '0' * 64 + self.assertFalse(conformance.verify_units(ver, {'model.bin': path}, 'bad disk digest', plan, 'apply')) + + +if __name__ == '__main__': + unittest.main()