mirror of
https://github.com/DeusData/codebase-memory-mcp.git
synced 2026-10-26 06:43:15 +08:00
Martin Vogel
a245618a64
Preserve the current selector, frozen-input inventory, result codec, coverage collection, runner filtering, and isolated Windows venue work. Component-level regression proofs are recorded in the PR handover. The live classified provider, public execution integration, artifact admission, fresh oracle/shadow evidence, and full platform gates remain incomplete. This checkpoint does not enable or claim verified narrowing. Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>
648 lines
33 KiB
Python
648 lines
33 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent frozen-contract tests for the pure single-image v2 producer.
|
|
|
|
Python 3.9 compatible. --module selects the producer under test; --parser can
|
|
select its unchanged real dependency for a scratch proof. --fixtures selects
|
|
the shared native golden fixture directory. No LLVM, compiler,
|
|
runner, network, admission or source-origin proof is performed by this suite.
|
|
"""
|
|
|
|
import argparse
|
|
from dataclasses import FrozenInstanceError, replace
|
|
import hashlib
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
|
|
|
|
M = None
|
|
P = None
|
|
FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures/profile_map_v2"
|
|
IMAGE_HEX = b"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
|
|
IMAGE = bytes.fromhex(IMAGE_HEX.decode("ascii"))
|
|
HIGH = 1 << 63
|
|
U64 = (1 << 64) - 1
|
|
GOLD_PROFILES = (
|
|
b"CBM_PROFILE_MAP\t2\t" + IMAGE_HEX + b"\t3\n"
|
|
b"0\t61\t0000000000000000\t2\n"
|
|
b"1\t61\t8000000000000000\t1\n"
|
|
b"2\tff3a09207a\tffffffffffffffff\t2\n"
|
|
)
|
|
GOLD_TESTS = (
|
|
b"s\xfe:*\tcomplete\t\t\n"
|
|
b"s\xfe:hit:raw\xfd\tcomplete\t\t0 2\n"
|
|
b"s\xfe:missing\tincomplete\tmissing_parent\t\n"
|
|
b"s\xfe:quiet\tcomplete\t\t\n"
|
|
)
|
|
GOLD_PROFILES_SHA = bytes.fromhex("ecc78b966bef4eb27a144eba98c2794ebf5f31a318f269e5e21d3453afd02d76")
|
|
GOLD_TESTS_SHA = bytes.fromhex("fe662544cf1d1eca44cc08ef6706ab04c59db76481b59aa3a4117a4c6c82da43")
|
|
|
|
|
|
def record(name, function_hash, counters):
|
|
"""Hand-shaped supported proftext fixture, not a producer-output oracle."""
|
|
return (name + b"\n# Func Hash:\n" + str(function_hash).encode("ascii") +
|
|
b"\n# Num Counters:\n" + str(len(counters)).encode("ascii") +
|
|
b"\n# Counter Values:\n" +
|
|
b"".join(str(value).encode("ascii") + b"\n" for value in counters) + b"\n")
|
|
|
|
|
|
def text(records):
|
|
return b"".join(record(*row) for row in records)
|
|
|
|
|
|
def ample():
|
|
return M.ProfileMapLimits(8 * 1024 * 1024, 512, 1024, 5000, 50000,
|
|
100000, 8 * 1024 * 1024, 10000, 16 * 1024 * 1024)
|
|
|
|
|
|
def claims(suite, test, flags=(True, True, True, True)):
|
|
return M.RowEvidence(suite, test, *flags)
|
|
|
|
|
|
def observation(suite, test, key, role, value):
|
|
return M.ProfileObservation(suite, test, key, role, value)
|
|
|
|
|
|
def gold_fixture():
|
|
suite = b"s\xfe"
|
|
zero = text(((b"a", 0, (0, 0)), (b"a", HIGH, (0,)), (b"\xff:\t z", U64, (0, 0))))
|
|
enumeration = text(((b"\xff:\t z", U64, (0, 0)), (b"a", HIGH, (0,)), (b"a", 0, (9, 0))))
|
|
parent = text(((b"a", 0, (0, 7)), (b"a", HIGH, (0,)), (b"\xff:\t z", U64, (0, 0))))
|
|
child = text(((b"a", 0, (0, 0)), (b"a", HIGH, (0,)), (b"\xff:\t z", U64, (0, 3))))
|
|
return dict(image_sha256=IMAGE, enumeration_text=enumeration,
|
|
expected_suites=(M.ExpectedSuite(suite, (b"quiet", b"missing", b"hit:raw\xfd"),
|
|
M.SuiteOutcome.PASSED),),
|
|
observations=(observation(suite, b"hit:raw\xfd", b"\xff:child/ opaque", M.ObservationRole.CHILD, child),
|
|
observation(suite, b"quiet", b"quiet", M.ObservationRole.PARENT, zero),
|
|
observation(suite, b"*", b"setup", M.ObservationRole.SETUP, zero),
|
|
observation(suite, b"hit:raw\xfd", b"a-parent", M.ObservationRole.PARENT, parent)),
|
|
row_evidence=tuple(claims(suite, name) for name in (b"quiet", b"*", b"hit:raw\xfd", b"missing")),
|
|
limits=ample())
|
|
|
|
|
|
def simple_fixture():
|
|
zero = text(((b"a", 1, (0,)), (b"b", 2, (0, 0))))
|
|
return dict(image_sha256=IMAGE,
|
|
enumeration_text=text(((b"a", 1, (5,)), (b"b", 2, (0, 0)))),
|
|
expected_suites=(M.ExpectedSuite(b"s", (b"t",), M.SuiteOutcome.PASSED),),
|
|
observations=(observation(b"s", b"*", b"s", M.ObservationRole.SETUP, zero),
|
|
observation(b"s", b"t", b"z", M.ObservationRole.PARENT, zero)),
|
|
row_evidence=(claims(b"s", b"*"), claims(b"s", b"t")), limits=ample())
|
|
|
|
|
|
SIMPLE_PROFILES = (b"CBM_PROFILE_MAP\t2\t" + IMAGE_HEX + b"\t2\n"
|
|
b"0\t61\t0000000000000001\t1\n1\t62\t0000000000000002\t2\n")
|
|
SIMPLE_TESTS = b"s:*\tcomplete\t\t\ns:t\tcomplete\t\t\n"
|
|
|
|
|
|
def row_view(wire):
|
|
"""Small assertion-side splitter for the already-expected four-field wire."""
|
|
if wire and (not wire.endswith(b"\n") or b"\r" in wire or b"\0" in wire):
|
|
raise AssertionError("producer test wire has unsupported record delimiters")
|
|
rows = {}
|
|
for line in wire.split(b"\n")[:-1]:
|
|
key, state, reason, ids = line.split(b"\t")
|
|
if key in rows:
|
|
raise AssertionError("producer duplicated an expected row")
|
|
rows[key] = (state, reason, ids)
|
|
return rows
|
|
|
|
|
|
class CountCancel:
|
|
def __init__(self, at=0, failure=None):
|
|
self.at = at
|
|
self.failure = failure
|
|
self.calls = 0
|
|
|
|
def __call__(self, context):
|
|
self.calls += 1
|
|
if self.at and self.calls >= self.at:
|
|
if self.failure is not None:
|
|
raise self.failure
|
|
return True
|
|
return False
|
|
|
|
|
|
class DependencyControl(unittest.TestCase):
|
|
def test_real_parser_preserves_full_identity_and_zero_records(self):
|
|
raw = text(((b"\xff:\t z", U64, (0, 0)), (b"a", HIGH, (0,)), (b"a", 0, (0, 7))))
|
|
limits = P.ParseLimits(4096, 3, 5, 7)
|
|
parsed = P.parse_profile_text(raw, limits)
|
|
self.assertEqual(tuple((r.raw_name, r.function_hash, r.counters) for r in parsed),
|
|
((b"a", 0, (0, 7)), (b"a", HIGH, (0,)), (b"\xff:\t z", U64, (0, 0))))
|
|
with self.assertRaises(P.MalformedProfileError):
|
|
P.parse_profile_text(raw[:-1], limits)
|
|
|
|
|
|
class ProfileMapV2Tests(unittest.TestCase):
|
|
def assert_result(self, result, profiles, tests):
|
|
self.assertIs(type(result), M.ProducedMap)
|
|
self.assertIs(type(result.profiles_tsv), bytes)
|
|
self.assertIs(type(result.tests_tsv), bytes)
|
|
self.assertIs(type(result.profiles_sha256), bytes)
|
|
self.assertIs(type(result.tests_sha256), bytes)
|
|
self.assertEqual(result.profiles_tsv, profiles)
|
|
self.assertEqual(result.tests_tsv, tests)
|
|
self.assertEqual(result.profiles_sha256, hashlib.sha256(profiles).digest())
|
|
self.assertEqual(result.tests_sha256, hashlib.sha256(tests).digest())
|
|
self.assertEqual(len(result.profiles_sha256), 32)
|
|
self.assertEqual(len(result.tests_sha256), 32)
|
|
|
|
def gold(self):
|
|
result = M.build_profile_map_v2(**gold_fixture())
|
|
self.assert_result(result, GOLD_PROFILES, GOLD_TESTS)
|
|
self.assertEqual(result.profiles_sha256, GOLD_PROFILES_SHA)
|
|
self.assertEqual(result.tests_sha256, GOLD_TESTS_SHA)
|
|
self.assertEqual(result.profiles_tsv, (FIXTURE_DIR / "profiles.tsv").read_bytes())
|
|
self.assertEqual(result.tests_tsv, (FIXTURE_DIR / "tests.tsv").read_bytes())
|
|
return result
|
|
|
|
def assert_row(self, fixture, key, state, reason, ids=b""):
|
|
result = M.build_profile_map_v2(**fixture)
|
|
self.assert_result(result, SIMPLE_PROFILES, result.tests_tsv)
|
|
rows = row_view(result.tests_tsv)
|
|
self.assertEqual(set(rows), {b"s:*", b"s:t"})
|
|
self.assertEqual(rows[key], (state, reason, ids))
|
|
return result
|
|
|
|
def test_exact_wire_full_identity_zero_ids_and_input_order(self):
|
|
self.gold()
|
|
f = gold_fixture()
|
|
f["expected_suites"] = (replace(f["expected_suites"][0],
|
|
tests=tuple(reversed(f["expected_suites"][0].tests))),)
|
|
f["observations"] = tuple(reversed(f["observations"]))
|
|
f["row_evidence"] = tuple(reversed(f["row_evidence"]))
|
|
self.assert_result(M.build_profile_map_v2(**f), GOLD_PROFILES, GOLD_TESTS)
|
|
# Distinct raw names may share the entire hash; neither is collapsed.
|
|
old = b"\n18446744073709551615\n# Num Counters:"
|
|
new = b"\n0\n# Num Counters:"
|
|
f = gold_fixture()
|
|
f["enumeration_text"] = f["enumeration_text"].replace(old, new)
|
|
f["observations"] = tuple(replace(o, text=o.text.replace(old, new)) for o in f["observations"])
|
|
self.assert_result(M.build_profile_map_v2(**f),
|
|
GOLD_PROFILES.replace(b"ffffffffffffffff", b"0000000000000000"), GOLD_TESTS)
|
|
empty = simple_fixture()
|
|
self.assert_result(M.build_profile_map_v2(**empty), SIMPLE_PROFILES, SIMPLE_TESTS)
|
|
empty["expected_suites"] = ()
|
|
empty["observations"] = ()
|
|
empty["row_evidence"] = ()
|
|
self.assert_result(M.build_profile_map_v2(**empty), SIMPLE_PROFILES, b"")
|
|
empty["expected_suites"] = (M.ExpectedSuite(b"s", (), M.SuiteOutcome.PASSED),)
|
|
self.assert_result(M.build_profile_map_v2(**empty), SIMPLE_PROFILES,
|
|
b"s:*\tincomplete\trow_evidence_missing;missing_setup\t\n")
|
|
empty["observations"] = simple_fixture()["observations"][:1]
|
|
empty["row_evidence"] = (claims(b"s", b"*"),)
|
|
self.assert_result(M.build_profile_map_v2(**empty), SIMPLE_PROFILES, b"s:*\tcomplete\t\t\n")
|
|
|
|
def test_expected_rows_primary_setup_outcomes_and_all_reason_codes(self):
|
|
self.gold()
|
|
base = simple_fixture()
|
|
self.assert_result(M.build_profile_map_v2(**base), SIMPLE_PROFILES, SIMPLE_TESTS)
|
|
for outcome, reason in ((M.SuiteOutcome.FAILED, b"suite_failed"),
|
|
(M.SuiteOutcome.TIMED_OUT, b"suite_timeout"),
|
|
(M.SuiteOutcome.UNKNOWN, b"suite_unknown")):
|
|
with self.subTest(outcome=outcome):
|
|
f = simple_fixture()
|
|
f["expected_suites"] = (replace(f["expected_suites"][0], outcome=outcome),)
|
|
result = self.assert_row(f, b"s:t", b"incomplete", reason)
|
|
self.assertEqual(row_view(result.tests_tsv)[b"s:*"], (b"incomplete", reason, b""))
|
|
fields = (("interval_complete", b"interval_unproved"),
|
|
("runtime_compatible", b"runtime_unproved"),
|
|
("same_image", b"image_unproved"), ("children_complete", b"children_unproved"))
|
|
for field, reason in fields:
|
|
with self.subTest(field=field):
|
|
f = simple_fixture()
|
|
f["row_evidence"] = (f["row_evidence"][0], replace(f["row_evidence"][1], **{field: False}))
|
|
self.assert_row(f, b"s:t", b"incomplete", reason)
|
|
f = simple_fixture()
|
|
f["row_evidence"] = f["row_evidence"][:1]
|
|
self.assert_row(f, b"s:t", b"incomplete", b"row_evidence_missing")
|
|
for test_name, reason in ((b"t", b"missing_parent"), (b"*", b"missing_setup")):
|
|
f = simple_fixture()
|
|
f["observations"] = tuple(o for o in f["observations"] if o.test != test_name)
|
|
self.assert_row(f, b"s:" + test_name, b"incomplete", reason)
|
|
for value, reason in ((None, b"missing_parent;observation_unavailable"),
|
|
(b"", b"missing_parent;observation_malformed")):
|
|
f = simple_fixture()
|
|
f["observations"] = (f["observations"][0], replace(f["observations"][1], text=value))
|
|
self.assert_row(f, b"s:t", b"incomplete", reason)
|
|
f = simple_fixture()
|
|
f["observations"] = (f["observations"][0], replace(f["observations"][1],
|
|
text=record(b"a", 1, (7,))))
|
|
self.assert_row(f, b"s:t", b"incomplete", b"partial_stream", b"0")
|
|
|
|
def test_stream_unions_child_does_not_replace_primary_and_reason_order(self):
|
|
self.gold()
|
|
full = text(((b"a", 1, (0,)), (b"b", 2, (0, U64))))
|
|
f = simple_fixture()
|
|
child = observation(b"s", b"t", b"child", M.ObservationRole.CHILD, full)
|
|
f["observations"] += (child, replace(child, key=b"child-again"))
|
|
f["limits"] = replace(ample(), max_id_references=1)
|
|
self.assert_row(f, b"s:t", b"complete", b"", b"1")
|
|
f["observations"] = (f["observations"][0], child)
|
|
self.assert_row(f, b"s:t", b"incomplete", b"missing_parent", b"1")
|
|
setup_child = replace(child, test=b"*", key=b"setup-child")
|
|
f = simple_fixture()
|
|
f["observations"] = (setup_child, f["observations"][1])
|
|
self.assert_row(f, b"s:*", b"incomplete", b"missing_setup", b"1")
|
|
f = simple_fixture()
|
|
f["observations"] += (replace(setup_child, role=M.ObservationRole.SETUP, key=b"second-setup"),)
|
|
self.assert_row(f, b"s:*", b"complete", b"", b"1")
|
|
f = simple_fixture()
|
|
f["expected_suites"] = (replace(f["expected_suites"][0], outcome=M.SuiteOutcome.FAILED),)
|
|
f["row_evidence"] = (f["row_evidence"][0], claims(b"s", b"t", (False,) * 4))
|
|
f["observations"] = (f["observations"][0],
|
|
replace(f["observations"][1], text=None),
|
|
replace(child, key=b"bad", text=b"not a complete record"),
|
|
replace(child, key=b"partial", text=record(b"a", 1, (2,))))
|
|
reason = (b"suite_failed;interval_unproved;runtime_unproved;image_unproved;"
|
|
b"children_unproved;missing_parent;observation_unavailable;"
|
|
b"observation_malformed;partial_stream")
|
|
self.assert_row(f, b"s:t", b"incomplete", reason, b"0")
|
|
f["row_evidence"] = f["row_evidence"][:1]
|
|
reason = (b"suite_failed;row_evidence_missing;missing_parent;observation_unavailable;"
|
|
b"observation_malformed;partial_stream")
|
|
self.assert_row(f, b"s:t", b"incomplete", reason, b"0")
|
|
|
|
def test_each_aggregate_cap_at_boundary_and_one_below(self):
|
|
self.gold()
|
|
f, expected, exact = budget_fixture()
|
|
f["limits"] = exact
|
|
self.assert_result(M.build_profile_map_v2(**f), SIMPLE_PROFILES, expected)
|
|
for field in M.ProfileMapLimits.__slots__:
|
|
value = getattr(exact, field)
|
|
self.assertGreater(value, 1, "one-below must remain a valid positive configured cap")
|
|
with self.subTest(cap=field):
|
|
f["limits"] = replace(exact, **{field: value - 1})
|
|
with self.assertRaises(M.ProfileMapLimitExceededError):
|
|
M.build_profile_map_v2(**f)
|
|
f["limits"] = exact
|
|
self.assert_result(M.build_profile_map_v2(**f), SIMPLE_PROFILES, expected)
|
|
|
|
def test_malformed_attempt_discard_and_zero_remaining_allowance(self):
|
|
self.gold()
|
|
f, expected, exact = budget_fixture()
|
|
f["limits"] = exact
|
|
self.assert_result(M.build_profile_map_v2(**f), SIMPLE_PROFILES, expected)
|
|
# The malformed child's positive prefix must not leak a hit or consume
|
|
# successful record/name/counter allowances needed by the later parent.
|
|
zero = simple_fixture()["observations"][1].text
|
|
malformed = record(b"a", 1, (7,)) + b"b\n# Wrong marker\n"
|
|
f["observations"] = (f["observations"][0], replace(f["observations"][1], text=zero),
|
|
observation(b"s", b"t", b"a", M.ObservationRole.CHILD, malformed))
|
|
f["limits"] = replace(exact, max_observations=3,
|
|
max_input_bytes=exact.max_input_bytes + len(malformed) + 3,
|
|
max_name_bytes=exact.max_name_bytes + 3, max_output_bytes=4096)
|
|
result = self.assert_row(f, b"s:t", b"incomplete", b"observation_malformed")
|
|
self.assertEqual(row_view(result.tests_tsv)[b"s:*"], (b"complete", b"", b"0"))
|
|
for field, consumed in (("max_records", 4), ("max_counter_values", 6), ("max_name_bytes", 16)):
|
|
f, _, _ = budget_fixture()
|
|
f["limits"] = replace(ample(), **{field: consumed})
|
|
f["observations"] = (f["observations"][0], replace(f["observations"][1], text=None))
|
|
with self.subTest(exhausted=field):
|
|
self.assert_row(f, b"s:t", b"incomplete", b"missing_parent;observation_unavailable")
|
|
f["observations"] = (f["observations"][0], replace(f["observations"][1], text=b""))
|
|
with self.assertRaises(M.ProfileMapLimitExceededError):
|
|
M.build_profile_map_v2(**f)
|
|
|
|
def test_whole_call_preflight_and_parser_limit_is_not_malformed(self):
|
|
self.gold()
|
|
f = simple_fixture()
|
|
f["enumeration_text"] = b""
|
|
f["observations"] += (observation(b"s", b"t", b"zz", M.ObservationRole.CHILD, b"x" * 4096),)
|
|
f["limits"] = replace(ample(), max_input_bytes=1024)
|
|
with self.assertRaises(M.ProfileMapLimitExceededError):
|
|
M.build_profile_map_v2(**f)
|
|
f = simple_fixture()
|
|
f["enumeration_text"] = b""
|
|
f["observations"] += (observation(b"s", b"unexpected", b"zz", M.ObservationRole.CHILD, None),)
|
|
with self.assertRaises(M.ProfileMapInvalidArgumentError):
|
|
M.build_profile_map_v2(**f)
|
|
# An attempt sees a genuine record cap before its later malformed tail.
|
|
f = simple_fixture()
|
|
f["observations"] = (f["observations"][0], replace(f["observations"][1],
|
|
text=record(b"a", 1, (0,)) + record(b"b", 2, (0, 0)) + b"broken"))
|
|
f["limits"] = replace(ample(), max_records=5)
|
|
with self.assertRaises(M.ProfileMapLimitExceededError) as caught:
|
|
M.build_profile_map_v2(**f)
|
|
self.assertIsInstance(caught.exception.__cause__, P.LimitExceededError)
|
|
|
|
def test_callback_provenance_invalid_results_and_entry_order(self):
|
|
held = self.gold()
|
|
f = simple_fixture()
|
|
observed = CountCancel()
|
|
self.assert_result(M.build_profile_map_v2(**f, cancel=observed), SIMPLE_PROFILES, SIMPLE_TESTS)
|
|
self.assertGreater(observed.calls, 0)
|
|
token = object()
|
|
|
|
def context_predicate(context):
|
|
if context is not token:
|
|
raise AssertionError("callback context was not preserved")
|
|
return False
|
|
|
|
self.assert_result(M.build_profile_map_v2(**f, cancel=context_predicate,
|
|
cancel_context=token), SIMPLE_PROFILES, SIMPLE_TESTS)
|
|
positions = sampled_positions(observed.calls)
|
|
failures = (ValueError("caller sentinel"), P.MalformedProfileError("caller malformed class"),
|
|
P.LimitExceededError("caller limit class"), P.ParseCancelledError("caller cancel class"),
|
|
M.ProfileMapEnumerationError("caller map class"), MemoryError("caller memory class"))
|
|
for failure in failures:
|
|
for position in positions:
|
|
with self.subTest(error=type(failure).__name__, position=position):
|
|
callback = CountCancel(position, failure)
|
|
caught = None
|
|
result = None
|
|
try:
|
|
result = M.build_profile_map_v2(**f, cancel=callback)
|
|
except BaseException as error:
|
|
caught = error
|
|
if callback.calls >= position:
|
|
self.assertIs(caught, failure)
|
|
self.assertIsNone(result)
|
|
else:
|
|
self.assertIsNone(caught)
|
|
self.assert_result(result, SIMPLE_PROFILES, SIMPLE_TESTS)
|
|
if position == 1:
|
|
self.assertIs(caught, failure)
|
|
for value in (None, 0, 1, b"", object()):
|
|
with self.subTest(callback_result=repr(value)):
|
|
with self.assertRaises(M.ProfileMapInvalidArgumentError):
|
|
M.build_profile_map_v2(**f, cancel=lambda context, answer=value: answer)
|
|
invalid = dict(f, image_sha256=b"short")
|
|
never = CountCancel(1)
|
|
with self.assertRaises(M.ProfileMapInvalidArgumentError):
|
|
M.build_profile_map_v2(**invalid, cancel=never)
|
|
self.assertEqual(never.calls, 0)
|
|
self.assert_result(held, GOLD_PROFILES, GOLD_TESTS)
|
|
self.assert_result(M.build_profile_map_v2(**f), SIMPLE_PROFILES, SIMPLE_TESTS)
|
|
|
|
def test_bounded_cancellation_immutable_results_and_independent_calls(self):
|
|
held = self.gold()
|
|
f, profiles, tests = stress_fixture()
|
|
observed = CountCancel()
|
|
result = M.build_profile_map_v2(**f, cancel=observed, cancel_context=object())
|
|
self.assert_result(result, profiles, tests)
|
|
self.assertGreater(observed.calls, 0)
|
|
for position in sampled_positions(observed.calls):
|
|
callback = CountCancel(position)
|
|
returned = None
|
|
cancelled = False
|
|
try:
|
|
returned = M.build_profile_map_v2(**f, cancel=callback)
|
|
except M.ProfileMapCancelledError:
|
|
cancelled = True
|
|
with self.subTest(position=position):
|
|
if callback.calls >= position:
|
|
self.assertTrue(cancelled)
|
|
self.assertIsNone(returned)
|
|
else:
|
|
self.assertFalse(cancelled)
|
|
self.assert_result(returned, profiles, tests)
|
|
if position == 1:
|
|
self.assertTrue(cancelled)
|
|
self.assert_result(result, profiles, tests)
|
|
self.assert_result(held, GOLD_PROFILES, GOLD_TESTS)
|
|
self.assert_result(M.build_profile_map_v2(**f), profiles, tests)
|
|
del f
|
|
self.assert_result(result, profiles, tests)
|
|
for field in M.ProducedMap.__slots__:
|
|
with self.subTest(immutable=field):
|
|
with self.assertRaises((FrozenInstanceError, AttributeError)):
|
|
setattr(result, field, b"changed")
|
|
self.assert_result(result, profiles, tests)
|
|
|
|
def test_strict_schema_membership_roles_names_and_exact_types(self):
|
|
self.gold()
|
|
for label, fixture in bad_schema_cases():
|
|
with self.subTest(case=label):
|
|
with self.assertRaises(M.ProfileMapInvalidArgumentError):
|
|
M.build_profile_map_v2(**fixture)
|
|
for label, fixture in bad_type_cases():
|
|
with self.subTest(case=label):
|
|
with self.assertRaises(M.ProfileMapInvalidArgumentError):
|
|
M.build_profile_map_v2(**fixture)
|
|
|
|
def test_enumeration_failure_and_observation_incompatibility_are_fatal(self):
|
|
self.gold()
|
|
valid = simple_fixture()["enumeration_text"]
|
|
malformed = (b"", valid[:-1], valid + b"# unknown section\n",
|
|
record(b"a\r", 1, (0,)), record(b"a\0", 1, (0,)),
|
|
record(b"a", 1, (0,)) * 2,
|
|
record(b"a", 1, (0,)) + record(b"a", 1, (0, 0)),
|
|
valid.replace(b"# Func Hash:", b"# Unknown:", 1),
|
|
valid.replace(b"\n1\n# Num", b"\n01\n# Num", 1),
|
|
record(b"a", U64 + 1, (0,)))
|
|
for value in malformed:
|
|
f = simple_fixture()
|
|
f["enumeration_text"] = value
|
|
with self.subTest(enumeration=value[:24]):
|
|
with self.assertRaises(M.ProfileMapEnumerationError) as caught:
|
|
M.build_profile_map_v2(**f)
|
|
self.assertIsInstance(caught.exception.__cause__, P.MalformedProfileError)
|
|
incompatible = (record(b"a", 3, (0,)), record(b"new", 1, (0,)),
|
|
record(b"a", 1, (0, 0)))
|
|
for value in incompatible:
|
|
for incomplete in (False, True):
|
|
f = simple_fixture()
|
|
f["observations"] = (f["observations"][0], replace(f["observations"][1], text=value))
|
|
if incomplete:
|
|
f["expected_suites"] = (replace(f["expected_suites"][0], outcome=M.SuiteOutcome.FAILED),)
|
|
f["row_evidence"] = (f["row_evidence"][0], claims(b"s", b"t", (False,) * 4))
|
|
with self.subTest(identity=value[:24], incomplete=incomplete):
|
|
with self.assertRaises(M.ProfileMapCompatibilityError):
|
|
M.build_profile_map_v2(**f)
|
|
|
|
|
|
def budget_fixture():
|
|
f = simple_fixture()
|
|
setup = text(((b"a", 1, (1,)), (b"b", 2, (0, 0))))
|
|
parent = text(((b"a", 1, (1,)), (b"b", 2, (0, 1))))
|
|
f["observations"] = (replace(f["observations"][0], text=setup),
|
|
replace(f["observations"][1], text=parent))
|
|
expected = b"s:*\tcomplete\t\t0\ns:t\tcomplete\t\t0 1\n"
|
|
# B = expected (1+1) + two observation routes (1+1+1)*2 +
|
|
# two evidence routes (1+1)*2 = 12. Three two-record streams contribute
|
|
# six names of one byte and nine counter values. Emitted ID refs = 1+2.
|
|
exact = M.ProfileMapLimits(32 + len(f["enumeration_text"]) + len(setup) + len(parent) + 12,
|
|
2, 2, 2, 6, 9, 18, 3, len(SIMPLE_PROFILES) + len(expected))
|
|
return f, expected, exact
|
|
|
|
|
|
def sampled_positions(calls):
|
|
# At most five observed positions, with no phase/count or time guarantee.
|
|
return tuple(sorted({1, calls // 4 + 1, calls // 2 + 1, calls - calls // 4, calls}))
|
|
|
|
|
|
def stress_fixture():
|
|
first = b"a" * 4097
|
|
second = b"a" * 4096 + b"b"
|
|
zero = text(((first, 0, (0, 0)), (second, U64, (0,))))
|
|
hit = text(((first, 0, (0, 1)), (second, U64, (0,))))
|
|
enumeration = text(((second, U64, (0,)), (first, 0, (9, 0))))
|
|
names = tuple(b"x" * 4097 + ("%03d" % i).encode("ascii") for i in range(128))
|
|
expected = (M.ExpectedSuite(b"s", tuple(reversed(names)), M.SuiteOutcome.PASSED),)
|
|
observations = (observation(b"s", b"*", b"setup", M.ObservationRole.SETUP, zero),) + tuple(
|
|
observation(b"s", name, ("k%03d" % i).encode("ascii"), M.ObservationRole.PARENT, hit)
|
|
for i, name in enumerate(reversed(names)))
|
|
evidence = (claims(b"s", b"*"),) + tuple(claims(b"s", name) for name in names)
|
|
profiles = (b"CBM_PROFILE_MAP\t2\t" + IMAGE_HEX + b"\t2\n0\t" + first.hex().encode("ascii") +
|
|
b"\t0000000000000000\t2\n1\t" + second.hex().encode("ascii") +
|
|
b"\tffffffffffffffff\t1\n")
|
|
tests = b"s:*\tcomplete\t\t\n" + b"".join(b"s:" + name + b"\tcomplete\t\t0\n" for name in names)
|
|
f = dict(image_sha256=IMAGE, enumeration_text=enumeration, expected_suites=expected,
|
|
observations=observations, row_evidence=evidence, limits=ample())
|
|
return f, profiles, tests
|
|
|
|
|
|
def renamed_fixture(suite=None, test=None):
|
|
f = simple_fixture()
|
|
suite = b"s" if suite is None else suite
|
|
test = b"t" if test is None else test
|
|
f["expected_suites"] = (M.ExpectedSuite(suite, (test,), M.SuiteOutcome.PASSED),)
|
|
f["observations"] = tuple(replace(o, suite=suite, test=test if o.test == b"t" else b"*")
|
|
for o in f["observations"])
|
|
f["row_evidence"] = tuple(replace(e, suite=suite, test=test if e.test == b"t" else b"*")
|
|
for e in f["row_evidence"])
|
|
return f
|
|
|
|
|
|
def bad_schema_cases():
|
|
for value in (b"", b"s:", b"s\0", b"s\t", b"s\r", b"s\n"):
|
|
yield ("suite-name-" + repr(value), renamed_fixture(suite=value))
|
|
for value in (b"", b"*", b"t\0", b"t\t", b"t\r", b"t\n"):
|
|
yield ("test-name-" + repr(value), renamed_fixture(test=value))
|
|
for value in (b"", b"k\0", b"k\t", b"k\r", b"k\n"):
|
|
f = simple_fixture()
|
|
f["observations"] = (f["observations"][0], replace(f["observations"][1], key=value))
|
|
yield ("observation-key-" + repr(value), f)
|
|
f = simple_fixture()
|
|
f["expected_suites"] *= 2
|
|
yield ("duplicate-suite", f)
|
|
f = simple_fixture()
|
|
f["expected_suites"] = (replace(f["expected_suites"][0], tests=(b"t", b"t")),)
|
|
yield ("duplicate-test", f)
|
|
for field in ("observations", "row_evidence"):
|
|
f = simple_fixture()
|
|
f[field] += (f[field][0],)
|
|
yield ("duplicate-" + field, f)
|
|
f = simple_fixture()
|
|
f["observations"] += (replace(f["observations"][1], key=b"another-parent"),)
|
|
yield ("two-parent-keys", f)
|
|
f = simple_fixture()
|
|
f["observations"] += (replace(f["observations"][1], role=M.ObservationRole.CHILD),)
|
|
yield ("same-key-different-role", f)
|
|
f = simple_fixture()
|
|
f["observations"] += (replace(f["observations"][1], text=None),)
|
|
yield ("same-key-different-text", f)
|
|
for target in ("observations", "row_evidence"):
|
|
for field, value in (("suite", b"other"), ("test", b"other")):
|
|
f = simple_fixture()
|
|
f[target] = (f[target][0], replace(f[target][1], **{field: value}))
|
|
yield ("unexpected-" + target + "-" + field, f)
|
|
for index, role in ((0, M.ObservationRole.PARENT), (1, M.ObservationRole.SETUP)):
|
|
f = simple_fixture()
|
|
observations = list(f["observations"])
|
|
observations[index] = replace(observations[index], role=role)
|
|
f["observations"] = tuple(observations)
|
|
yield ("invalid-role-" + str(index), f)
|
|
|
|
|
|
def bad_type_cases():
|
|
for field, value in (("image_sha256", b"x" * 31), ("image_sha256", b"x" * 33),
|
|
("image_sha256", bytearray(32)), ("enumeration_text", "text"),
|
|
("expected_suites", []), ("observations", []), ("row_evidence", []),
|
|
("limits", None), ("cancel", 17)):
|
|
f = simple_fixture()
|
|
f[field] = value
|
|
yield ("top-level-" + field + "-" + repr(type(value)), f)
|
|
derived_bytes = type("DerivedBytes", (bytes,), {})
|
|
derived_tuple = type("DerivedTuple", (tuple,), {})
|
|
for field, value in (("image_sha256", derived_bytes(IMAGE)),
|
|
("observations", derived_tuple(simple_fixture()["observations"]))):
|
|
f = simple_fixture()
|
|
f[field] = value
|
|
yield ("subclass-" + field, f)
|
|
for field, value in (("tests", [b"t"]), ("outcome", "passed"), ("suite", "s")):
|
|
f = simple_fixture()
|
|
f["expected_suites"] = (replace(f["expected_suites"][0], **{field: value}),)
|
|
yield ("expected-field-" + field, f)
|
|
for field, value in (("role", "parent"), ("text", bytearray()), ("key", "key"),
|
|
("suite", b"s".decode("ascii")), ("test", 1)):
|
|
f = simple_fixture()
|
|
f["observations"] = (f["observations"][0], replace(f["observations"][1], **{field: value}))
|
|
yield ("observation-field-" + field, f)
|
|
for field in ("interval_complete", "runtime_compatible", "same_image", "children_complete"):
|
|
f = simple_fixture()
|
|
f["row_evidence"] = (f["row_evidence"][0], replace(f["row_evidence"][1], **{field: 1}))
|
|
yield ("evidence-bool-" + field, f)
|
|
for field in ("expected_suites", "observations", "row_evidence"):
|
|
f = simple_fixture()
|
|
f[field] = (object(),)
|
|
yield ("record-type-" + field, f)
|
|
f = simple_fixture()
|
|
derived_expected = type("DerivedExpectedSuite", (M.ExpectedSuite,), {})
|
|
f["expected_suites"] = (derived_expected(b"s", (b"t",), M.SuiteOutcome.PASSED),)
|
|
yield ("dataclass-subclass", f)
|
|
for field in ("observations", "row_evidence"):
|
|
f = simple_fixture()
|
|
old = f[field][0]
|
|
derived = type("DerivedRecord", (type(old),), {})
|
|
f[field] = (derived(*(getattr(old, name) for name in type(old).__slots__)),) + f[field][1:]
|
|
yield ("record-subclass-" + field, f)
|
|
f = simple_fixture()
|
|
derived_limits = type("DerivedLimits", (M.ProfileMapLimits,), {})
|
|
f["limits"] = derived_limits(*(getattr(f["limits"], name) for name in M.ProfileMapLimits.__slots__))
|
|
yield ("limits-subclass", f)
|
|
f = simple_fixture()
|
|
derived_int = type("DerivedInt", (int,), {})
|
|
f["limits"] = replace(f["limits"], max_rows=derived_int(512))
|
|
yield ("limit-value-subclass", f)
|
|
for field in M.ProfileMapLimits.__slots__:
|
|
for value in (0, -1, True, sys.maxsize + 1):
|
|
f = simple_fixture()
|
|
f["limits"] = replace(f["limits"], **{field: value})
|
|
yield ("invalid-limit-" + field + "-" + repr(value), f)
|
|
f = simple_fixture()
|
|
f["limits"] = replace(f["limits"], max_ids=2147483648)
|
|
yield ("ids-int-representation", f)
|
|
|
|
|
|
def load_module(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, str(path))
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError("cannot load requested module: " + str(path))
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def main():
|
|
global M, P, FIXTURE_DIR
|
|
arguments = argparse.ArgumentParser(description=__doc__)
|
|
default_module = Path(__file__).resolve().parents[1] / "scripts/test-impact/profile_map_v2.py"
|
|
arguments.add_argument("--module", type=Path, default=default_module)
|
|
arguments.add_argument("--parser", type=Path,
|
|
help="real unchanged parser; default profile_text_parser.py beside --module")
|
|
arguments.add_argument("--fixtures", type=Path, default=FIXTURE_DIR,
|
|
help="shared golden directory; default fixtures/profile_map_v2 beside this test")
|
|
selected = arguments.parse_args()
|
|
FIXTURE_DIR = selected.fixtures.resolve()
|
|
module_path = selected.module.resolve()
|
|
parser_path = (selected.parser or module_path.with_name("profile_text_parser.py")).resolve()
|
|
if not module_path.is_file() or not parser_path.is_file():
|
|
arguments.error("--module and --parser must name existing source files")
|
|
sys.dont_write_bytecode = True
|
|
# Freeze the actual dependency before importing the chosen producer; both
|
|
# baseline and implementation use this real parser, never a fake grammar.
|
|
P = load_module("profile_text_parser", parser_path)
|
|
M = load_module("profile_map_v2_under_test", module_path)
|
|
suite = unittest.TestSuite((unittest.defaultTestLoader.loadTestsFromTestCase(DependencyControl),
|
|
unittest.defaultTestLoader.loadTestsFromTestCase(ProfileMapV2Tests)))
|
|
return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|