from __future__ import annotations import hashlib, json, os, shutil, subprocess, sys, tempfile, unittest, 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_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 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_manifest_no_standalone_target(self): with tempfile.TemporaryDirectory() as t: p=Path(t)/'target' rec=make_target(p,PROFILES[0],'aarch64','test-build',None,{'source_paths':[],'library_paths':[]}) 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,{'source_paths':[],'library_paths':[]}) 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,{'source_paths':[],'library_paths':[]}) 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,{'source_paths':[],'library_paths':[]},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()