mirror of
https://github.com/DeusData/codebase-memory-mcp.git
synced 2026-10-21 05:59:35 +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>
217 lines
11 KiB
Python
217 lines
11 KiB
Python
"""Contract tests for complete LLVM frontend profile text records."""
|
|
import dataclasses
|
|
import hashlib
|
|
from pathlib import Path
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / 'scripts/test-impact'))
|
|
import profile_text_parser as p
|
|
|
|
CAPTURES = ROOT / 'tests/fixtures/profile_text'
|
|
BIG = p.ParseLimits(2_000_000, 20_000, 100_000, 1_000_000)
|
|
MAX = 18446744073709551615
|
|
|
|
|
|
def wire(name=b'f', function_hash=24, counters=(0,)):
|
|
return (name + b'\n# Func Hash:\n' + str(function_hash).encode() +
|
|
b'\n# Num Counters:\n' + str(len(counters)).encode() +
|
|
b'\n# Counter Values:\n' +
|
|
b''.join(str(x).encode() + b'\n' for x in counters) + b'\n')
|
|
|
|
|
|
def values(rows):
|
|
return tuple((x.raw_name, x.function_hash, x.counters) for x in rows)
|
|
|
|
|
|
class ProfileTextContract(unittest.TestCase):
|
|
def test_native_dense_and_sparse_captures(self):
|
|
expected = {
|
|
'cold-dense.proftext': ('b8f1bd2ca4c7dd9e2ca567b410897ca8f5c0c648a1d73648f58724a708a93daf',
|
|
((b'cbm_profile_probe_hit',24,(0,)), (b'cbm_profile_probe_never',24,(0,)),
|
|
(b'main',490457921399630188,(1,0,1,1,1,0,0)))),
|
|
'hot-dense.proftext': ('3f77bbbda8d76b4d440109d24f0431f8b9bca6628bc0241470d1dbbb2c2695ec',
|
|
((b'cbm_profile_probe_hit',24,(1,)), (b'cbm_profile_probe_never',24,(0,)),
|
|
(b'main',490457921399630188,(1,0,1,1,0,1,1)))),
|
|
'cold-sparse.proftext': ('50979a6198fcfcdabcc3f09057009be6cc54053eea4af581aadf4ec91b3199a7',
|
|
((b'main',490457921399630188,(1,0,1,1,1,0,0)),)),
|
|
'hot-sparse.proftext': ('72bf344ccd0897cfe3c5625657e6912d645deb1fb495d6dea551b40b89f9dc68',
|
|
((b'cbm_profile_probe_hit',24,(1,)),
|
|
(b'main',490457921399630188,(1,0,1,1,0,1,1))))}
|
|
for name, (digest, oracle) in expected.items():
|
|
with self.subTest(name=name):
|
|
data = (CAPTURES/name).read_bytes()
|
|
self.assertEqual(hashlib.sha256(data).hexdigest(), digest)
|
|
self.assertEqual(values(p.parse_profile_text(data, BIG)), oracle)
|
|
|
|
def test_full_identity_order_and_all_counters(self):
|
|
rows = ((b'\xff:name\t #',MAX,(0,MAX)), (b'z',10,(0,)),
|
|
(b'z',2,(0,0)), (b' a ',0,(1,)), (b'# Func Hash:',24,(0,)))
|
|
data = b''.join(wire(*r) for r in rows)
|
|
parsed = p.parse_profile_text(data, BIG)
|
|
self.assertEqual(values(parsed), (rows[3], rows[4], rows[2], rows[1], rows[0]))
|
|
self.assertEqual(values(p.parse_profile_text(b''.join(wire(*r) for r in reversed(rows)), BIG)),
|
|
values(parsed))
|
|
self.assertTrue(any(parsed[-1].counters))
|
|
self.assertEqual(parsed[-1].counters[0], 0)
|
|
|
|
def test_exact_grammar_and_unsigned_decimals(self):
|
|
good = wire()
|
|
malformed = [b'', b'\n', b'\n'+good, good+b'\n', good+b'# unknown\n',
|
|
good.replace(b'\n',b'\r\n'), good.replace(b'f\n',b'f\x00\n',1),
|
|
good.replace(b'# Func Hash:',b'# Function Hash:'),
|
|
good.replace(b'# Num Counters:',b'# Num Counters: '),
|
|
good.replace(b'# Counter Values:',b'# Values:'),
|
|
good.replace(b'\n1\n# Counter Values:',b'\n0\n# Counter Values:'),
|
|
good.replace(b'\n0\n\n',b'\n0\n0\n\n'),
|
|
good.replace(b'\n0\n\n',b'\n\n'), b':irprof\n'+good]
|
|
for decimal in (b'', b'00', b'024', b'+24', b'-1', b' 24', b'24 ', b'0x18',
|
|
b'2.4', b'2e1', b'18446744073709551616'):
|
|
malformed.append(good.replace(b'\n24\n',b'\n'+decimal+b'\n'))
|
|
malformed.append(good.replace(b'\n0\n\n',b'\n'+decimal+b'\n\n'))
|
|
self.assertEqual(values(p.parse_profile_text(good, BIG)), ((b'f',24,(0,)),))
|
|
for index,data in enumerate(malformed):
|
|
with self.subTest(index=index):
|
|
with self.assertRaises(p.MalformedProfileError):
|
|
p.parse_profile_text(data, BIG)
|
|
|
|
def test_all_proper_prefixes_and_suffix_rejection(self):
|
|
data = wire(b'prefix:name',MAX,(0,1,MAX))
|
|
self.assertEqual(len(p.parse_profile_text(data,BIG)),1)
|
|
for length in range(len(data)):
|
|
with self.subTest(length=length):
|
|
with self.assertRaises(p.MalformedProfileError):
|
|
p.parse_profile_text(data[:length],BIG)
|
|
for tail in (b'x',b'\x00',wire()[:-1],b'# Num Value Kinds:\n0\n'):
|
|
with self.assertRaises(p.MalformedProfileError):
|
|
p.parse_profile_text(data+tail,BIG)
|
|
|
|
def test_duplicate_identity_and_hash_distinction(self):
|
|
one=wire(b'name',1,(0,)); other=wire(b'other',1,(1,))
|
|
self.assertEqual(len(p.parse_profile_text(one+wire(b'name',2,(0,0)),BIG)),2)
|
|
for duplicate in (one,wire(b'name',1,(1,)),wire(b'name',1,(0,0))):
|
|
with self.assertRaises(p.MalformedProfileError):
|
|
p.parse_profile_text(one+other+duplicate,BIG)
|
|
|
|
def test_inclusive_resource_limits_and_claimed_counts(self):
|
|
data=wire(b'abc',1,(0,1))+wire(b'de',2,(0,))+wire(b'f',3,(0,0))
|
|
exact=(len(data),3,5,6)
|
|
self.assertEqual(len(p.parse_profile_text(data,p.ParseLimits(*exact))),3)
|
|
for index in range(4):
|
|
limited=list(exact);limited[index]-=1
|
|
with self.subTest(index=index):
|
|
with self.assertRaises(p.LimitExceededError):
|
|
p.parse_profile_text(data,p.ParseLimits(*limited))
|
|
claimed=b'f\n# Func Hash:\n0\n# Num Counters:\n18446744073709551615\n# Counter Values:\n'
|
|
with self.assertRaises(p.LimitExceededError):
|
|
p.parse_profile_text(claimed,BIG)
|
|
with self.assertRaises(p.MalformedProfileError):
|
|
p.parse_profile_text(claimed.replace(str(MAX).encode(),str(MAX+1).encode()),BIG)
|
|
|
|
def test_argument_types_and_limit_validation(self):
|
|
good=wire()
|
|
for invalid in (None,good.decode(),bytearray(good),memoryview(good)):
|
|
with self.assertRaises(p.InvalidArgumentError):
|
|
p.parse_profile_text(invalid,BIG)
|
|
for value in (0,-1,True,1.0,None,sys.maxsize+1):
|
|
for index in range(4):
|
|
limits=[1000,1000,1000,1000];limits[index]=value
|
|
with self.assertRaises(p.InvalidArgumentError):
|
|
p.parse_profile_text(good,p.ParseLimits(*limits))
|
|
with self.assertRaises(p.InvalidArgumentError):
|
|
p.parse_profile_text(good,None)
|
|
for callback in (False,1,'callback'):
|
|
with self.assertRaises(p.InvalidArgumentError):
|
|
p.parse_profile_text(good,BIG,cancel=callback)
|
|
for answer in (None,0,1,'true'):
|
|
with self.assertRaises(p.InvalidArgumentError):
|
|
p.parse_profile_text(good,BIG,cancel=lambda _:answer)
|
|
|
|
def test_cancellation_and_exception_propagation_are_owner_local(self):
|
|
marker=object();calls=[]
|
|
def cancelled(context):
|
|
self.assertIs(context,marker);calls.append(context);return True
|
|
with self.assertRaises(p.ParseCancelledError):
|
|
p.parse_profile_text(b'not a profile',BIG,cancel=cancelled,cancel_context=marker)
|
|
self.assertEqual(len(calls),1)
|
|
class CallbackFailure(Exception): pass
|
|
sentinel=CallbackFailure('callback marker')
|
|
def explode(_): raise sentinel
|
|
with self.assertRaises(CallbackFailure) as caught:
|
|
p.parse_profile_text(wire(),BIG,cancel=explode)
|
|
self.assertIs(caught.exception,sentinel)
|
|
big=wire(b'x'*100_000,MAX,(0,1))
|
|
for cutoff in (2,3):
|
|
state=[0]
|
|
def later(_): state[0]+=1;return state[0]>=cutoff
|
|
with self.assertRaises(p.ParseCancelledError):
|
|
p.parse_profile_text(big,BIG,cancel=later)
|
|
self.assertEqual(state[0],cutoff)
|
|
self.assertEqual(values(p.parse_profile_text(wire(),BIG)),((b'f',24,(0,)),))
|
|
|
|
def test_cancellation_sweep_including_sort_and_final_publication(self):
|
|
rows=tuple((b'q'*5000+bytes((97+i,)),i,(0,1)) for i in reversed(range(8)))
|
|
data=b''.join(wire(*row) for row in rows)
|
|
calls=[0]
|
|
def observe(_): calls[0]+=1;return False
|
|
expected=values(p.parse_profile_text(data,BIG,cancel=observe))
|
|
self.assertEqual(expected,tuple(reversed(rows)))
|
|
self.assertGreater(calls[0],3)
|
|
self.assertLessEqual(calls[0],512) # Bounds the deterministic sweep itself.
|
|
for cutoff in range(1,calls[0]+1):
|
|
state=[0]
|
|
def cancel_at(_): state[0]+=1;return state[0]>=cutoff
|
|
with self.subTest(cutoff=cutoff):
|
|
with self.assertRaises(p.ParseCancelledError):
|
|
p.parse_profile_text(data,BIG,cancel=cancel_at)
|
|
self.assertEqual(state[0],cutoff)
|
|
self.assertEqual(values(p.parse_profile_text(data,BIG)),expected)
|
|
|
|
def test_held_owner_survives_concurrent_cancellation(self):
|
|
self.assertEqual(values(p.parse_profile_text(wire(),BIG)),((b'f',24,(0,)),))
|
|
entered=threading.Event();release=threading.Event();out=[];errors=[]
|
|
def run():
|
|
try:
|
|
owner=p.parse_profile_text(wire(b'held',MAX,(0,1)),BIG)
|
|
entered.set()
|
|
if not release.wait(10): raise AssertionError('release gate failed')
|
|
out.append(owner)
|
|
except BaseException as error: errors.append(error)
|
|
thread=threading.Thread(target=run);thread.start()
|
|
try:
|
|
self.assertTrue(entered.wait(10),'entry gate failed')
|
|
other=p.parse_profile_text(wire(b'other',2,(1,)),BIG)
|
|
with self.assertRaises(p.ParseCancelledError):
|
|
p.parse_profile_text(wire(b'cancelled'),BIG,cancel=lambda _:True)
|
|
self.assertEqual(values(other),((b'other',2,(1,)),))
|
|
finally:
|
|
release.set();thread.join(10)
|
|
self.assertFalse(thread.is_alive());self.assertEqual(errors,[])
|
|
self.assertEqual(len(out),1);self.assertEqual(values(out[0]),((b'held',MAX,(0,1)),))
|
|
|
|
def test_result_immutability_and_independent_owners(self):
|
|
a=p.parse_profile_text(wire(b'a',1,(0,MAX)),BIG)
|
|
b=p.parse_profile_text(wire(b'b',2,(1,)),BIG)
|
|
self.assertIsInstance(a,tuple);self.assertIsInstance(a[0].counters,tuple)
|
|
with self.assertRaises((dataclasses.FrozenInstanceError,AttributeError)):
|
|
a[0].raw_name=b'changed'
|
|
self.assertEqual(values(a),((b'a',1,(0,MAX)),));self.assertEqual(values(b),((b'b',2,(1,)),))
|
|
start=threading.Event();out=[None,None];errors=[]
|
|
def run(slot):
|
|
try:
|
|
if not start.wait(10): raise AssertionError('start gate failed')
|
|
out[slot]=p.parse_profile_text(wire(bytes((97+slot,)),slot,(0,)),BIG)
|
|
except BaseException as error: errors.append(error)
|
|
threads=[threading.Thread(target=run,args=(i,)) for i in range(2)]
|
|
for thread in threads: thread.start()
|
|
start.set()
|
|
for thread in threads: thread.join(10);self.assertFalse(thread.is_alive())
|
|
self.assertEqual(errors,[])
|
|
self.assertEqual(values(out[0]),((b'a',0,(0,)),));self.assertEqual(values(out[1]),((b'b',1,(0,)),))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|