from __future__ import annotations import hashlib, json, os, shutil, subprocess, sys, tempfile, unittest, xml.etree.ElementTree as ET, zipfile from pathlib import Path from unittest.mock import patch sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools')) from common import ROOT, COMMIT, ProbeError, git_blob, safe_relative, sha256, write_json, run_command, classify from source_resolution import build_source_resolution, load_source_contract, validate_source_contract from source_fetch import selected, enumerate_blobs, verify from source_audit import inventory from generate import make_target, imports_source, asset_inventory, COMPILER_OPTIONS from artifacts import inspect_elf, inspect_apk, runtime_identity from runner import gate, selected_profiles, cge_command, PROFILES from device import own_report_paths, validate_runtime from helpers import elf, section_elf class PathsTests(unittest.TestCase): pass for i,bad in enumerate(('', '/etc/passwd','../x','a/../x','a//b','./a','a/./b','C:/x','a\\b','a\nname','x\0y','a/')): def test(self,bad=bad): with self.assertRaises(ProbeError):safe_relative(bad) setattr(PathsTests,f'test_reject_path_{i:02d}',test) for i,good in enumerate(('src/common/KM_Points.pas','Maps/Тест/Тест.map','A B/c.d','data/100%.txt')): def test(self,good=good):self.assertEqual(safe_relative(good),good) setattr(PathsTests,f'test_accept_path_{i:02d}',test) class SourceContractTests(unittest.TestCase): def contract(self, value): temp = tempfile.TemporaryDirectory(); self.addCleanup(temp.cleanup) path = Path(temp.name) / 'extra-paths.json'; path.write_text(json.dumps(value)) return path def test_loads_normalized_targets_and_deduplicated_paths(self): path = self.contract({'schema':'kam-source-resolution/1','adaptations':['src/common/KM_Example.pas'], 'source_paths':['tools','tools'],'library_paths':[str(ROOT/'config')]}) loaded = load_source_contract(path) self.assertEqual(loaded['adaptation_targets'], ('src/common/KM_Example.pas',)) self.assertEqual(loaded['source_paths'], ((ROOT/'tools').resolve(),)) self.assertEqual(loaded['library_paths'], ((ROOT/'config').resolve(),)) def test_rejects_legacy_and_unsafe_contracts(self): for value in ({'source_paths':[],'library_paths':[]}, {'schema':'kam-source-resolution/1','adaptations':['src\\bad.pas'],'source_paths':[],'library_paths':[]}): with self.subTest(value=value), self.assertRaises(ProbeError): load_source_contract(self.contract(value)) class SourceOwnershipTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name);self.vendor=self.root/'vendor';self.sparse=self.root/'port-source/kam_remake' self.vendor.joinpath('src').mkdir(parents=True);self.vendor.joinpath('src/KM_Test.pas').write_text('unit KM_Test; interface implementation end.') self.receipt={'files':[{'path':'src/KM_Test.pas'}]};self.inventory={'units':{'km_test':['src/KM_Test.pas']}} def tearDown(self):self.tmp.cleanup() def validate(self, contract): with patch('source_resolution.ROOT',self.root): return validate_source_contract(contract,self.vendor,self.receipt,self.inventory) def test_registered_adaptation_has_vendor_identity_and_unit(self): self.sparse.joinpath('src').mkdir(parents=True);self.sparse.joinpath('src/KM_Test.pas').write_text('unit km_test; interface implementation end.') result=self.validate({'adaptation_targets':('src/KM_Test.pas',),'source_paths':(),'library_paths':()}) self.assertEqual(result['adaptations'][0]['unit_name'],'km_test');self.assertEqual(result['vendor_unit_keys'],frozenset({'km_test'})) def test_rejects_unregistered_sparse_and_generic_shadowing(self): self.sparse.mkdir(parents=True);self.sparse.joinpath('other.inc').write_text('x') with self.assertRaises(ProbeError):self.validate({'adaptation_targets':(),'source_paths':(),'library_paths':()}) shutil.rmtree(self.sparse);generic=self.root/'generic';generic.mkdir();generic.joinpath('KM_Test.pas').write_text('not a unit') with self.assertRaises(ProbeError):self.validate({'adaptation_targets':(),'source_paths':(generic,),'library_paths':()}) class CompilerSourceResolutionTests(unittest.TestCase): def test_orders_deduplicates_and_guards_application_vendor_units(self): with tempfile.TemporaryDirectory() as t: root = Path(t); app = root/'app'; vendor = root/'vendor'; generic = root/'generic'; library = root/'library' for path in (app, vendor/'src', vendor/'ext', generic, library): path.mkdir(parents=True, exist_ok=True) (vendor/'src/KM_Vendor.pas').write_text('unit KM_Vendor; interface implementation end.') first, second = root/'adaptations/z', root/'adaptations/a' first.mkdir(parents=True); second.mkdir(parents=True) validated = {'vendor_unit_keys': frozenset({'km_vendor'}), 'adaptations': ( {'target':'z.pas', 'project_file':first/'z.pas'}, {'target':'A.pas', 'project_file':second/'A.pas'}, {'target':'a2.pas', 'project_file':second/'a2.pas'}), 'source_paths': (generic, second), 'library_paths': (library,)} result = build_source_resolution(validated, vendor, app) self.assertEqual(result['adaptation_dirs'], (second.resolve(), first.resolve())) self.assertEqual(result['compiler_source_paths'][:4], (app.resolve(), second.resolve(), first.resolve(), generic.resolve())) self.assertEqual(result['library_paths'], (library.resolve(),)) (app/'KM_VENDOR.pas').write_text('malformed declaration') with self.assertRaises(ProbeError): build_source_resolution(validated, vendor, app) class SourceSelectionTests(unittest.TestCase): def test_known_git_blob(self): self.assertEqual(git_blob(b'hello\n'),'ce013625030ba8dba906f756967f9e9ca394464a') def test_excludes_symlinks_and_gitlinks(self): tree={'tree':[{'path':'src/a.pas','type':'blob','mode':'120000','sha':'a','size':1}, {'path':'src/module','type':'commit','sha':'b'}]} self.assertEqual(enumerate_blobs(tree,'x/y',COMMIT),[]) def test_keeps_case_and_sha(self): tree={'tree':[{'path':'src/A.PAS','type':'blob','mode':'100644','sha':'abc','size':1}]} out=enumerate_blobs(tree,'x/y',COMMIT) self.assertEqual(out[0]['path'],'src/A.PAS');self.assertEqual(out[0]['git_blob_sha1'],'abc') for i,name in enumerate(('src/KM.pas','src/ext/a.pp','src/common/defs.inc','KaM_Remake.inc','src/ext/LICENSE','src/ext/License.txt','LICENSE.md','src/ext/native.c')): def test(self,name=name):self.assertTrue(selected(name)) setattr(SourceSelectionTests,f'test_source_allowed_{i:02d}',test) for i,name in enumerate(('data/tiles.rxx','data/a.pas','Maps/a.map','Campaigns/a.dat','src/a.res','src/a.dll','src/a.so','src/a.a','src/a.o','src/a.ppu','src/LICENSE.png','src/LICENSE.zip','src/Examples/a.pas','src/ext/tests/a.pas','Utils/editor.pas','src/a.lfm','src/a.png')): def test(self,name=name):self.assertFalse(selected(name)) setattr(SourceSelectionTests,f'test_binary_or_unneeded_excluded_{i:02d}',test) class ReceiptTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name) file=self.root/'src/test.pas';file.parent.mkdir();file.write_text('unit Test; interface implementation end.') self.receipt={'commit':COMMIT,'repository':'reyandme/kam_remake','source_only':True, 'files':[{'path':'src/test.pas','size':file.stat().st_size,'sha256':sha256(file),'git_blob_sha1':git_blob(file.read_bytes())}]} write_json(self.root/'SOURCE_RECEIPT.json',self.receipt) def tearDown(self):self.tmp.cleanup() def test_valid_receipt(self):self.assertEqual(len(verify(self.root)['files']),1) def test_modified_source_rejected(self): (self.root/'src/test.pas').write_text('changed') with self.assertRaises(ProbeError):verify(self.root) def test_unrecorded_source_rejected(self): (self.root/'src/other.pas').write_text('x') with self.assertRaises(ProbeError):verify(self.root) def test_cached_native_units_rejected(self): (self.root/'src/test.ppu').write_bytes(b'PPU') with self.assertRaises(ProbeError):verify(self.root) def test_wrong_commit_rejected(self): self.receipt['commit']='0'*40;write_json(self.root/'SOURCE_RECEIPT.json',self.receipt) with self.assertRaises(ProbeError):verify(self.root) def test_empty_receipt_rejected(self): self.receipt['files']=[];write_json(self.root/'SOURCE_RECEIPT.json',self.receipt) with self.assertRaises(ProbeError):verify(self.root) class ArtifactTests(unittest.TestCase): def test_arm64_elf(self):self.assertEqual(inspect_elf(elf(),'aarch64')['machine'],183) def test_arm_elf(self):self.assertEqual(inspect_elf(elf('arm'),'arm')['elf_class'],1) def test_x86_android_elf(self):self.assertEqual(inspect_elf(elf('x86_64'),'x86_64')['machine'],62) def test_wrong_abi_rejected(self): with self.assertRaises(ProbeError):inspect_elf(elf('arm'),'aarch64') def test_random_nonelf_rejected(self): with self.assertRaises(ProbeError):inspect_elf(b'PASS compiler'+'x'.encode()*90,'aarch64') def test_textrel_rejected(self): with self.assertRaises(ProbeError):inspect_elf(elf(textrel=True),'aarch64') def test_needed_parsed(self):self.assertEqual(inspect_elf(elf(needed=['libc.so','libm.so']),'aarch64')['needed'],['libc.so','libm.so']) def test_bad_headers_rejected(self): data=bytearray(elf());data[54]=0 with self.assertRaises(ProbeError):inspect_elf(bytes(data),'aarch64') def apk(self,root,data=None,dex=True): p=Path(root)/'mock.apk';data=data or elf(needed=['libc.so']) with zipfile.ZipFile(p,'w') as z: z.writestr('AndroidManifest.xml','TEST DOUBLE') if dex:z.writestr('classes.dex',b'dex\n035\0TEST DOUBLE') z.writestr('lib/arm64-v8a/libtest.so',data) return p,data def test_apk_container_accepts_matching_mock_library(self): with tempfile.TemporaryDirectory() as t: p,data=self.apk(t) r=inspect_apk(p,'aarch64','libcompiled_android_aarch64.so',hashlib.sha256(data).hexdigest()) self.assertEqual(r['abi'],'arm64-v8a') self.assertEqual(r['main_library_path'],'lib/arm64-v8a/libtest.so') def test_runtime_identity_ignores_debug_only_changes(self): for cpu in ('arm','aarch64','x86_64'): self.assertEqual(runtime_identity(section_elf(cpu,debug=b'long debug before strip'),cpu), runtime_identity(section_elf(cpu,debug=b''),cpu)) def test_runtime_identity_detects_changed_code(self): self.assertNotEqual(runtime_identity(section_elf(code=b'original'),'aarch64'), runtime_identity(section_elf(code=b'modified'),'aarch64')) def test_apk_debug_stripping_not_false_stale_failure(self): before=section_elf(debug=b'DEBUG SYMBOLS BEFORE STRIP') after=section_elf(debug=b'') with tempfile.TemporaryDirectory() as t: p,_=self.apk(t,after) result=inspect_apk(p,'aarch64','libtest.so',hashlib.sha256(before).hexdigest(),before) self.assertIn('stripping',result['main_library_identity']) def test_apk_runtime_change_rejected_with_stripping(self): before=section_elf(code=b'original');after=section_elf(debug=b'',code=b'modified') with tempfile.TemporaryDirectory() as t: p,_=self.apk(t,after) with self.assertRaises(ProbeError):inspect_apk(p,'aarch64','libtest.so',hashlib.sha256(before).hexdigest(),before) def test_identity_no_sections_cannot_bypass_sha(self): with self.assertRaises(ProbeError):runtime_identity(elf(),'aarch64') def test_stale_apk_library_rejected(self): with tempfile.TemporaryDirectory() as t: p,data=self.apk(t) with self.assertRaises(ProbeError):inspect_apk(p,'aarch64','libtest.so','0'*64) def test_missing_dex_rejected(self): with tempfile.TemporaryDirectory() as t: p,data=self.apk(t,dex=False) with self.assertRaises(ProbeError):inspect_apk(p,'aarch64','libtest.so',hashlib.sha256(data).hexdigest()) def test_unpackaged_dependency_rejected(self): with tempfile.TemporaryDirectory() as t: p,data=self.apk(t,elf(needed=['libmissing.so'])) with self.assertRaises(ProbeError):inspect_apk(p,'aarch64','libtest.so',hashlib.sha256(data).hexdigest()) class OrchestrationTests(unittest.TestCase): def rows(self): return [{'id':p['id'],'required':p['required'],'compile_link':{'status':'PASS'}, 'package':{'status':'PASS' if p['package'] else 'NOT_RUN'}} for p in PROFILES] def test_full_gate(self):self.assertEqual(gate(self.rows()),'ANDROID_BUILD_GATE_PASS') def test_no_vacuous_pass(self):self.assertEqual(gate([]),'PARTIAL_SELECTION') def test_platform_alone_is_not_core_success(self):self.assertEqual(gate(self.rows()[:1]),'PARTIAL_SELECTION') def test_missing_required_row(self): r=[x for x in self.rows() if x['id']!='integration'];self.assertEqual(gate(r),'PARTIAL_SELECTION') def test_core_failure_fails_gate(self): r=self.rows();r[2]['compile_link']['status']='FAIL';self.assertEqual(gate(r),'NOT_BUILDABLE_AS_IS') def test_blocked_is_not_pass(self): r=self.rows();r[2]['compile_link']['status']='BLOCKED';self.assertEqual(gate(r),'BLOCKED') def test_optional_legacy_failure_recorded_not_mandatory(self): r=self.rows();next(v for v in r if v['id']=='legacy_render')['compile_link']['status']='FAIL' self.assertEqual(gate(r),'ANDROID_BUILD_GATE_PASS') def test_missing_apk_does_not_pass(self): r=self.rows();next(v for v in r if v['id']=='integration')['package']['status']='NOT_RUN';self.assertEqual(gate(r),'BLOCKED') def test_optional_decision_failure_does_not_change_core_gate(self): r=self.rows();next(v for v in r if v['id']=='oq_host_callback')['compile_link']['status']='FAIL' self.assertEqual(gate(r),'ANDROID_BUILD_GATE_PASS') def test_platform_baseline_added(self):self.assertEqual(selected_profiles('core_types')[0]['id'],'platform') def test_invalid_profile_rejected(self): with self.assertRaises(ProbeError):selected_profiles('fake') def test_no_native_host_target(self): for cpu in ('aarch64','arm','x86_64'): for stage in ('compile','package'): cmd=cge_command(cpu,stage) self.assertIn('--os=android',cmd);self.assertIn('--cpu='+cpu,cmd) self.assertNotIn('-Cn',' '.join(cmd)) def test_linker_strictness(self):self.assertIn('-k--no-undefined',COMPILER_OPTIONS) def test_no_global_delphi_mode(self):self.assertNotIn('-Mdelphi',COMPILER_OPTIONS) def test_no_smart_link_elimination(self): self.assertIn('-CX-',COMPILER_OPTIONS);self.assertIn('-XX-',COMPILER_OPTIONS) def test_failed_subprocess_not_swallowed(self): with tempfile.TemporaryDirectory() as t: r=run_command([sys.executable,'-c','raise SystemExit(7)'],Path(t),Path(t)/'log',30) self.assertEqual(r['status'],'FAIL');self.assertEqual(r['exit_code'],7) def test_missing_tool_is_blocked(self): with tempfile.TemporaryDirectory() as t: r=run_command(['/does-not-exist/kam-probe'],Path(t),Path(t)/'log',30) self.assertEqual(r['status'],'BLOCKED') def test_lcl_classification(self):self.assertEqual(classify("Fatal: Can't find unit Graphics used by KM_ResSprites"),'PLATFORM_BOUNDARY') class GenerationTests(unittest.TestCase): def test_empty_platform_has_no_core_imports(self): code,units=imports_source(PROFILES[0]);self.assertFalse(units);self.assertIn('NOT_RUN',code) def test_actual_point_calls(self): code,units=imports_source(PROFILES[1]);self.assertIn('KMPointRight(P)',code);self.assertIn('KM_Points',units) def test_actual_stream_calls(self): code,units=imports_source(PROFILES[2]);self.assertIn('TKMemoryStreamBinary.Create',code);self.assertIn('S.ReadW(U)',code) def test_upstream_manifest_uses_resolved_compiler_order(self): with tempfile.TemporaryDirectory() as t: root=Path(t);vendor=root/'vendor';generic=root/'generic';library=root/'library';adaptation=root/'adaptation' for path in (vendor/'src',generic,library,adaptation):path.mkdir(parents=True,exist_ok=True) (vendor/'src/KM_Points.pas').write_text('unit KM_Points; interface implementation end.') validated={'vendor_unit_keys':frozenset({'km_points'}),'adaptations':( {'target':'src/KM_Points.pas','project_file':adaptation/'KM_Points.pas'},), 'source_paths':(generic,),'library_paths':(library,)} p=root/'target';rec=make_target(p,PROFILES[1],'aarch64','id',vendor,validated) self.assertEqual(rec['source_paths'][:3],[str((p/'code').resolve()),str(adaptation.resolve()),str(generic.resolve())]) xml=ET.parse(p/'CastleEngineManifest.xml') self.assertEqual([node.attrib['value'] for node in xml.findall('./compiler_options/search_paths/path')][:3], ['code',os.path.relpath(adaptation,p).replace(os.sep,'/'),os.path.relpath(generic,p).replace(os.sep,'/')]) self.assertEqual([node.attrib['value'] for node in xml.findall('./compiler_options/library_paths/path')], [os.path.relpath(library,p).replace(os.sep,'/')]) def test_manifest_no_standalone_target(self): with tempfile.TemporaryDirectory() as t: p=Path(t)/'target' rec=make_target(p,PROFILES[0],'aarch64','test-build',None,None) xml=(p/'CastleEngineManifest.xml').read_text() self.assertNotIn('standalone_source',xml);self.assertNotIn('test_shims',xml) self.assertIn('game_units="GameInitialize"',xml) self.assertIn('Android-only',(p/'code/GameInitialize.pas').read_text()) self.assertFalse(rec['shims']) self.assertEqual(json.loads((p/'data/assets-manifest.json').read_text())['files'],[]) def test_decision_profile_uses_existing_generator_with_alternate_game_unit(self): with tempfile.TemporaryDirectory() as t: profile=next(x for x in PROFILES if x['id']=='oq_host_callback') p=Path(t)/'target';rec=make_target(p,profile,'aarch64','id',None,None) xml=(p/'CastleEngineManifest.xml').read_text() self.assertIn('game_units="DecisionHostCallback"',xml) self.assertEqual(rec['game_unit'],'DecisionHostCallback') self.assertTrue((p/'code/DecisionHostCallback.pas').is_file()) def test_selected_existing_asset_only(self): with tempfile.TemporaryDirectory() as t: root=Path(t)/'assets';root.mkdir();(root/'Maps').mkdir();(root/'data').mkdir() (root/'Maps/one.txt').write_text('map-marker') (root/'data/unused.txt').write_text('unused') rows=asset_inventory(root,['Maps/one.txt']) self.assertEqual([x['path'] for x in rows],['Maps/one.txt']) def test_fresh_target_required(self): with tempfile.TemporaryDirectory() as t: with self.assertRaises(ProbeError):make_target(Path(t),PROFILES[0],'aarch64','id',None,None) def test_optional_assets_preserved(self): with tempfile.TemporaryDirectory() as t: r=Path(t)/'assets';r.mkdir();(r/'Data').mkdir();raw=bytes(range(256)) (r/'Data/sample.dat').write_bytes(raw) entries=asset_inventory(r) p=Path(t)/'target';make_target(p,PROFILES[0],'aarch64','id',None,None,r,entries) self.assertEqual((p/'data/game/Data/sample.dat').read_bytes(),raw) def test_asset_symlinks_rejected(self): with tempfile.TemporaryDirectory() as t: r=Path(t);(r/'a').write_text('x');(r/'b').symlink_to(r/'a') with self.assertRaises(ProbeError):asset_inventory(r) def test_no_asset_network_call_by_default(self):self.assertEqual(asset_inventory(None),[]) def test_audit_flags_are_hints(self): with tempfile.TemporaryDirectory() as t: r=Path(t);(r/'test.pas').write_text('unit Test; interface uses Graphics; implementation end.') d=inventory(r);self.assertIn('test',d['units']);self.assertTrue(d['risks']);self.assertIn('Lexical',d['scope']) def test_audit_accepts_utf8_bom_unit(self): with tempfile.TemporaryDirectory() as t: r=Path(t);(r/'KM_Game.pas').write_bytes(b'\xef\xbb\xbfunit KM_Game; interface implementation end.') d=inventory(r);self.assertIn('km_game',d['units']);self.assertEqual(d['units']['km_game'],['KM_Game.pas']) class DeviceValidationTests(unittest.TestCase): def package(self):return {'build_id':'new','profile':'platform','cpu':'aarch64'} def test_scoped_report_path(self): self.assertEqual(own_report_paths('./files/kam-architecture-probe/runtime.json\n../secret\nother/runtime.json'),['files/kam-architecture-probe/runtime.json']) def test_stale_report_rejected(self): data={'schema':'kam-android-runtime/1','build_id':'old','profile':'platform','cpu_expected':'aarch64','automatic_smoke_status':'PASS','tests':[]} self.assertTrue(any('stale' in e for e in validate_runtime(data,self.package()))) def test_empty_pass_is_not_runtime_success(self): data={'schema':'kam-android-runtime/1','build_id':'new','profile':'platform','cpu_expected':'aarch64','automatic_smoke_status':'PASS','tests':[]} self.assertIn('No runtime tests',validate_runtime(data,self.package())) def test_missing_frame_evidence(self): data={'tests':[{'id':'android.startup','status':'PASS'}],'session':{'frames':0}} self.assertTrue(any('frames' in e for e in validate_runtime(data,self.package()))) def test_decision_callback_runtime_does_not_require_frames_or_upstream(self): pkg={'build_id':'new','profile':'oq_host_callback','cpu':'aarch64'} passed=['android.startup','abi.pointer','abi.scalar_layout','abi.endianness','files.native_utf8','files.seek','files.case_sensitive','fpc.zstream','fpc.thread','decision.host.callback'] observed=['decision.persistence.rename_replace','decision.persistence.append_reopen','decision.fixture.castle_stream','decision.fixture.materialized_native_path'] data={'schema':'kam-android-runtime/1','build_id':'new','profile':'oq_host_callback','cpu_expected':'aarch64','automatic_smoke_status':'PASS', 'tests':[{'id':x,'status':'PASS'} for x in passed]+[{'id':x,'status':'OBSERVED'} for x in observed]} self.assertEqual(validate_runtime(data,pkg,require_frames=False),[]) def test_text_decision_runtime_accepts_observation_results(self): pkg={'build_id':'new','profile':'platform','cpu':'aarch64'} passed=['android.startup','abi.pointer','abi.scalar_layout','abi.endianness','files.native_utf8','files.seek','files.case_sensitive','fpc.zstream','fpc.thread','decision.text.cwstring_unit'] data={'schema':'kam-android-runtime/1','build_id':'new','profile':'platform','cpu_expected':'aarch64','automatic_smoke_status':'PASS', 'tests':[{'id':x,'status':'PASS'} for x in passed]+[ {'id':'decision.persistence.rename_replace','status':'OBSERVED'},{'id':'decision.persistence.append_reopen','status':'OBSERVED'}, {'id':'decision.text.cp1251_decode','status':'OBSERVED'},{'id':'decision.text.cp1251_encode','status':'OBSERVED'}]} self.assertEqual(validate_runtime(data,pkg,require_frames=False),[]) if __name__=='__main__':unittest.main()