372 lines
15 KiB
Python
Executable File
372 lines
15 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Invariant and committed-golden tests for the Macrostep 00 Python oracle."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import math
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
TOOLS = ROOT / "tools"
|
|
FIXTURES = ROOT / "tests" / "fixtures" / "contract"
|
|
sys.path.insert(0, str(TOOLS))
|
|
|
|
_ORACLE_SPEC = importlib.util.spec_from_file_location(
|
|
"contract_oracle", TOOLS / "contract_oracle.py"
|
|
)
|
|
if _ORACLE_SPEC is None or _ORACLE_SPEC.loader is None:
|
|
raise RuntimeError("could not load tools/contract_oracle.py")
|
|
oracle = importlib.util.module_from_spec(_ORACLE_SPEC)
|
|
sys.modules[_ORACLE_SPEC.name] = oracle
|
|
_ORACLE_SPEC.loader.exec_module(oracle)
|
|
|
|
|
|
def load_fixture(name: str) -> dict[str, Any]:
|
|
try:
|
|
with (FIXTURES / name).open("r", encoding="utf-8") as handle:
|
|
value = json.load(handle)
|
|
except (OSError, json.JSONDecodeError) as error:
|
|
raise AssertionError(
|
|
f"could not load committed fixture {name}: {error}"
|
|
) from error
|
|
if not isinstance(value, dict):
|
|
raise AssertionError(f"fixture {name} is not a JSON object")
|
|
return value
|
|
|
|
|
|
class IndexAndRuleTests(unittest.TestCase):
|
|
def test_x_fast_index_round_trip_and_signed_offsets(self) -> None:
|
|
for shape in ((8,), (5, 3), (4, 3, 2)):
|
|
for expected, coords in enumerate(oracle.iter_coords(shape)):
|
|
self.assertEqual(oracle.flat_index(coords, shape), expected)
|
|
self.assertEqual(oracle.unflatten_index(expected, shape), coords)
|
|
self.assertEqual(
|
|
[oracle.signed_offset(i, 8) for i in range(8)], [0, 1, 2, 3, -4, -3, -2, -1]
|
|
)
|
|
|
|
def test_curve_boundaries_overshoot_and_hard_window(self) -> None:
|
|
center, width = 0.5, 0.2
|
|
for curve_type in (1, 2, 3):
|
|
self.assertAlmostEqual(
|
|
oracle.rising_curve(curve_type, center - width / 2, center, width),
|
|
0.0,
|
|
places=14,
|
|
)
|
|
self.assertAlmostEqual(
|
|
oracle.rising_curve(curve_type, center + width / 2, center, width),
|
|
1.0,
|
|
places=14,
|
|
)
|
|
self.assertEqual(oracle.window_curve(0, 0.3, 0.3, 0.7, 0.1), 1.0)
|
|
self.assertEqual(
|
|
oracle.window_curve(0, math.nextafter(0.7, -math.inf), 0.3, 0.7, 0.1), 1.0
|
|
)
|
|
self.assertEqual(oracle.window_curve(0, 0.7, 0.3, 0.7, 0.1), 0.0)
|
|
self.assertLess(oracle.rising_curve(7, 0.364, center, width), 0.0)
|
|
self.assertGreater(oracle.rising_curve(7, 0.636, center, width), 1.0)
|
|
|
|
def test_rule_fixture_covers_every_valid_enum_triple(self) -> None:
|
|
fixture = load_fixture("rule_surface.json")
|
|
axes = fixture["axes"]
|
|
self.assertEqual(axes["sigmode"], [1, 2, 3, 4])
|
|
self.assertEqual(axes["sigtype"], list(range(10)))
|
|
self.assertEqual(axes["mixtype"], list(range(8)))
|
|
seen = 0
|
|
for modes in fixture["values"]:
|
|
self.assertEqual(len(modes), 10)
|
|
for types in modes:
|
|
self.assertEqual(len(types), 8)
|
|
for surface in types:
|
|
self.assertEqual((len(surface), len(surface[0])), (3, 3))
|
|
self.assertTrue(
|
|
all(math.isfinite(value) for row in surface for value in row)
|
|
)
|
|
seen += 1
|
|
self.assertEqual(seen, fixture["enum_combination_count"])
|
|
self.assertEqual(seen, 320)
|
|
|
|
|
|
class KernelAndFftTests(unittest.TestCase):
|
|
def test_complete_sampled_kernels_and_direct_convolution(self) -> None:
|
|
fixture = load_fixture("planar_fields.json")
|
|
for case in fixture["cases"]:
|
|
kernels = case["kernel"]
|
|
self.assertAlmostEqual(sum(kernels["normalized_disk"]), 1.0, places=14)
|
|
self.assertAlmostEqual(sum(kernels["normalized_ring"]), 1.0, places=14)
|
|
support_indices = {
|
|
oracle.flat_index(entry["coords"], case["shape"])
|
|
for entry in kernels["support"]
|
|
}
|
|
expected_support = {
|
|
index
|
|
for index, pair in enumerate(
|
|
zip(kernels["raw_disk"], kernels["raw_ring"], strict=True)
|
|
)
|
|
if pair != (0.0, 0.0)
|
|
}
|
|
self.assertEqual(support_indices, expected_support)
|
|
constant = [0.375] * oracle.product(case["shape"])
|
|
disk = oracle.circular_convolution(
|
|
constant, kernels["normalized_disk"], case["shape"]
|
|
)
|
|
ring = oracle.circular_convolution(
|
|
constant, kernels["normalized_ring"], case["shape"]
|
|
)
|
|
self.assertTrue(all(abs(value - 0.375) < 1e-14 for value in disk + ring))
|
|
known = fixture["known_2d_check"]
|
|
self.assertAlmostEqual(known["ring_sum"], 279.216312, places=6)
|
|
self.assertAlmostEqual(known["disk_sum"], 35.524035, places=6)
|
|
|
|
def test_packed_fft_bits_plans_and_convolution(self) -> None:
|
|
fixture = load_fixture("legacy_packed_fft.json")
|
|
self.assertEqual(
|
|
[case["shape"] for case in fixture["cases"]], [[8], [4, 4], [4, 2, 2]]
|
|
)
|
|
for case in fixture["cases"]:
|
|
for decimal, bits in zip(
|
|
case["input_f32"]["decimal"], case["input_f32"]["bits"], strict=True
|
|
):
|
|
decoded = struct.unpack("<f", struct.pack("<I", int(bits, 16)))[0]
|
|
self.assertEqual(decoded, decimal)
|
|
self.assertLess(case["max_abs_error_to_direct"], 2e-6)
|
|
self.assertEqual(
|
|
len(case["convolution_f32"]["decimal"]), math.prod(case["shape"])
|
|
)
|
|
first_plan = case["plans"]["x_forward_butterflies"][0]["entries"]
|
|
self.assertEqual(first_plan[0]["source"][0], 0)
|
|
self.assertEqual(first_plan[0]["twiddle"]["bits"][0], "0x3f800000")
|
|
# The 1-D Nx=8 stage-1 plan starts from 2-bit bit-reversed packed indices.
|
|
first = fixture["cases"][0]["plans"]["x_forward_butterflies"][0]["entries"]
|
|
self.assertEqual(
|
|
[entry["source"] for entry in first], [[0, 2], [0, 2], [1, 3], [1, 3]]
|
|
)
|
|
|
|
|
|
class RandomAndInitializerTests(unittest.TestCase):
|
|
def test_chacha12_known_block_and_word_composition(self) -> None:
|
|
rng = oracle.ChaCha12(0)
|
|
self.assertEqual(
|
|
[rng.u32() for _ in range(4)],
|
|
[0x6A9AF49B, 0x53F95507, 0x12CE1F81, 0xD583265F],
|
|
)
|
|
words = oracle.ChaCha12(1)
|
|
low, high = words.u32(), words.u32()
|
|
combined = oracle.ChaCha12(1).u64()
|
|
self.assertEqual(combined, low | (high << 32))
|
|
value = oracle.ChaCha12(1).float53()
|
|
self.assertEqual(value, (combined >> 11) * 2.0**-53)
|
|
|
|
def test_integer_rejection_skips_incomplete_high_residue(self) -> None:
|
|
class Scripted(oracle.ChaCha12):
|
|
def __init__(self) -> None:
|
|
self.values = iter([(1 << 64) - 1, 5])
|
|
|
|
def u64(self) -> int:
|
|
return next(self.values)
|
|
|
|
# For span 2^63+1, the first scripted value is above the acceptance limit.
|
|
self.assertEqual(Scripted().integer(0, (1 << 63) + 1), 5)
|
|
|
|
def test_initializer_arrays_are_topology_native_and_explicit(self) -> None:
|
|
fixture = load_fixture("initializers.json")
|
|
planar = fixture["planar_periodic_splats"]
|
|
self.assertEqual(
|
|
[entry["shape"] for entry in planar], [[12], [8, 6], [6, 5, 4]]
|
|
)
|
|
for entry in planar:
|
|
self.assertEqual(len(entry["field"]), math.prod(entry["shape"]))
|
|
self.assertTrue(set(entry["field"]) <= {0.0, 1.0})
|
|
self.assertEqual(len(entry["splats"]), entry["count"])
|
|
sphere = fixture["sphere_geodesic_overlays"]
|
|
self.assertEqual(sphere["draw_count"], 1000)
|
|
self.assertEqual(len(sphere["field"]), 6 * sphere["k"] * sphere["k"])
|
|
self.assertTrue(all(2 <= draw["radius"] <= 7 for draw in sphere["first_draws"]))
|
|
delayed = fixture["delayed_time_periodic_boxes"]
|
|
self.assertEqual(delayed["depth"], 16)
|
|
self.assertEqual(len(delayed["history"]), 16)
|
|
self.assertTrue(all(layer == delayed["frame"] for layer in delayed["history"]))
|
|
self.assertTrue(
|
|
all(
|
|
10 <= box["width"] <= 19 and 10 <= box["height"] <= 19
|
|
for box in delayed["first_boxes"]
|
|
)
|
|
)
|
|
|
|
|
|
class DynamicsAndTopologyTests(unittest.TestCase):
|
|
def test_integrator_startup_and_relaxation_references(self) -> None:
|
|
fixture = load_fixture("integration.json")
|
|
startup = fixture["ab3_startup"]
|
|
self.assertEqual(startup["methods"], ["Euler", "AB2", "AB3"])
|
|
dt = fixture["dt"]
|
|
for before, derivative, after in zip(
|
|
startup["states"][0],
|
|
startup["derivatives"][0],
|
|
startup["states"][1],
|
|
strict=True,
|
|
):
|
|
self.assertAlmostEqual(
|
|
after, oracle.clamp(before + dt * derivative), places=15
|
|
)
|
|
current, previous = startup["derivatives"][1], startup["derivatives"][0]
|
|
expected_ab2 = [
|
|
(3 * now - old) / 2 for now, old in zip(current, previous, strict=True)
|
|
]
|
|
self.assertEqual(startup["ab2_combination"], expected_ab2)
|
|
for stage in fixture["rk4"]["stage_states"]:
|
|
self.assertTrue(all(0.0 <= value <= 1.0 for value in stage))
|
|
stage_state = fixture["rk4_relaxation_stage_state"]
|
|
step_origin = fixture["rk4_relaxation_step_origin"]
|
|
self.assertEqual(stage_state["derivatives"][0], step_origin["derivatives"][0])
|
|
self.assertNotEqual(
|
|
stage_state["derivatives"][1], step_origin["derivatives"][1]
|
|
)
|
|
self.assertNotEqual(stage_state["next"], step_origin["next"])
|
|
|
|
def test_all_multiscale_combinations_and_composition_equations(self) -> None:
|
|
fixture = load_fixture("multiscale.json")
|
|
cases = fixture["cases"]
|
|
triples = {
|
|
(case["dynamics"], case["interpretation"], case["composition"])
|
|
for case in cases
|
|
}
|
|
self.assertEqual(len(cases), 12)
|
|
self.assertEqual(len(triples), 12)
|
|
for case in cases:
|
|
if case["composition"] == "ordered_clamped_sum":
|
|
state = fixture["field"][:]
|
|
for increment in case["increments"]:
|
|
state = [
|
|
oracle.clamp(value + change)
|
|
for value, change in zip(state, increment, strict=True)
|
|
]
|
|
self.assertEqual(state, case["next"])
|
|
elif case["composition"] == "mean_increment":
|
|
expected = [
|
|
oracle.clamp(value + sum(changes) / 3.0)
|
|
for value, changes in zip(
|
|
fixture["field"], zip(*case["increments"]), strict=True
|
|
)
|
|
]
|
|
self.assertEqual(expected, case["next"])
|
|
else:
|
|
self.assertEqual(case["next"], case["clamp_stages"][-1])
|
|
|
|
def test_corrected_and_legacy_sphere_invariants(self) -> None:
|
|
fixture = load_fixture("sphere.json")
|
|
geometry = fixture["geometry"]
|
|
self.assertTrue(
|
|
all(area > 0.0 and math.isfinite(area) for area in geometry["areas"])
|
|
)
|
|
self.assertAlmostEqual(
|
|
geometry["total_area"], geometry["analytic_area"], places=12
|
|
)
|
|
corrected = fixture["models"]["corrected"]
|
|
legacy = fixture["models"]["legacy"]
|
|
for channel in ("m", "n"):
|
|
self.assertTrue(
|
|
all(
|
|
abs(value - 0.375) < 5e-15
|
|
for value in corrected["constant_field"][channel]
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
any(abs(value - 0.375) > 1e-3 for value in legacy["constant_field"]["m"])
|
|
)
|
|
self.assertEqual(corrected["disk_denominators"], corrected["visited_disk_sums"])
|
|
self.assertNotEqual(legacy["disk_denominators"], legacy["visited_disk_sums"])
|
|
for mapping in fixture["legacy_side_and_mask_map"]:
|
|
if mapping["region"].startswith("masked"):
|
|
self.assertIsNone(mapping["mapped"])
|
|
else:
|
|
self.assertIsNotNone(mapping["mapped"])
|
|
for model in (corrected, legacy):
|
|
expected_length = 6 * fixture["k"] * fixture["k"]
|
|
self.assertEqual(len(model["m"]), expected_length)
|
|
self.assertEqual(len(model["constant_field"]["s"]), expected_length)
|
|
self.assertEqual(
|
|
len(model["constant_field"]["next_discrete"]), expected_length
|
|
)
|
|
self.assertEqual(
|
|
len(model["constant_field"]["next_smooth"]), expected_length
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
0.0 <= value <= 1.0
|
|
for value in model["next_discrete"] + model["next_smooth"]
|
|
)
|
|
)
|
|
|
|
def test_delayed_time_causal_shells_and_updates(self) -> None:
|
|
fixture = load_fixture("delayed_time.json")
|
|
evaluation = fixture["evaluation"]
|
|
self.assertEqual(evaluation["latest"], (fixture["head"] - 1) % fixture["depth"])
|
|
for shell in fixture["shell_offsets_by_rounded_age"]:
|
|
self.assertEqual(
|
|
shell["selected_layer"],
|
|
(evaluation["latest"] - shell["age"]) % fixture["depth"],
|
|
)
|
|
stencil = evaluation["stencil"]
|
|
expected_m = (
|
|
sum(
|
|
(((evaluation["latest"] - entry["delay"]) % fixture["depth"]) / 15.0)
|
|
* entry["disk"]
|
|
for entry in stencil["entries"]
|
|
)
|
|
/ stencil["disk_sum"]
|
|
)
|
|
expected_n = (
|
|
sum(
|
|
(((evaluation["latest"] - entry["delay"]) % fixture["depth"]) / 15.0)
|
|
* entry["ring"]
|
|
for entry in stencil["entries"]
|
|
)
|
|
/ stencil["ring_sum"]
|
|
)
|
|
self.assertTrue(
|
|
all(abs(value - expected_m) < 1e-14 for value in evaluation["m"])
|
|
)
|
|
self.assertTrue(
|
|
all(abs(value - expected_n) < 1e-14 for value in evaluation["n"])
|
|
)
|
|
latest_value = evaluation["latest"] / 15.0
|
|
for target, smooth in zip(
|
|
evaluation["s"], evaluation["next_smooth"], strict=True
|
|
):
|
|
self.assertAlmostEqual(
|
|
smooth, oracle.clamp(latest_value + 0.1 * (2 * target - 1)), places=15
|
|
)
|
|
|
|
|
|
class CommittedFixtureTests(unittest.TestCase):
|
|
def test_manifest_hashes(self) -> None:
|
|
manifest = load_fixture("manifest.json")
|
|
self.assertTrue(manifest["standard_library_only"])
|
|
for entry in manifest["files"]:
|
|
content = (FIXTURES / entry["path"]).read_bytes()
|
|
self.assertEqual(len(content), entry["bytes"])
|
|
self.assertEqual(hashlib.sha256(content).hexdigest(), entry["sha256"])
|
|
|
|
def test_generator_check_mode(self) -> None:
|
|
completed = subprocess.run(
|
|
[sys.executable, str(TOOLS / "generate_contract_fixtures.py"), "--check"],
|
|
cwd=ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(completed.returncode, 0, completed.stdout + completed.stderr)
|
|
self.assertIn("verified 11 contract fixture files", completed.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|