Initial import

tests/test_probe.py

@@ -0,0 +1,259 @@
+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()