EscapistsReintroduced/tests/test_native_package_external.py

"""Exact synthetic external closure and packaging orchestration; never launch native executables."""
import argparse
import contextlib
import io
import json
from pathlib import Path
import shutil
import sys
import unittest
import uuid
from unittest.mock import Mock,patch
import test_native_package as base
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools'))
import native_package as n
import external_addons as ext

class ExternalPackageTests(unittest.TestCase):
 def setUp(self):
  self.fixture=base.NativePackageTests();self.fixture.setUp();self.addCleanup(self.fixture.doCleanups)
  self.source=self.fixture.external_fixture()
  self.root=self.fixture.source;self.session=self.fixture.session
  self.steam=self.fixture.steam;self.compat=self.fixture.compat
  self.context=n.external_request('core',self.source,self.root)
 def prepare(self):
  self.authored=n.copy_payload(self.root,self.session)
  n.capture_external(self.context,self.session)
 def profile(self,project):
  profile=self.session/'profiles'/project;(profile/'addons').mkdir(parents=True)
  (profile/'addons/ArmaReforger').symlink_to(self.steam/'steamapps/common/Arma Reforger/addons/data',target_is_directory=True)
  return profile
 def native_double(self,project='core',mutation=None,error='',native_exit=0,cleanup=True,write_status=True):
  child=Mock();child.poll.return_value=0
  def spawn(command,**kwargs):
   self.assertNotIn('WINEPREFIX',kwargs['env'])
   logs=self.session/'profiles'/project/'logs/logs_synthetic';logs.mkdir(parents=True)
   (logs/'console.log').write_text('SYNTHETIC PACKAGE DOUBLE; NOT NATIVE EVIDENCE\n'+error)
   (self.session/'packages'/project/'data.pak').write_bytes(b'SYNTHETIC SUCCESSOR OUTPUT')
   native=n.command_for(self.steam,self.session,project);native=native[native.index('--')+1:]
   if write_status:n.save(self.session/'evidence'/project/'native-exit.json',{'command':native,'exit_code':native_exit})
   if mutation:mutation()
   return child
  with patch.object(n,'assert_no_game'),patch.object(n,'assert_prefix_idle'), \
   patch.object(n.subprocess,'Popen',side_effect=spawn) as launch, \
   patch.object(n.subprocess,'check_output',side_effect=lambda args,**kw:'[]' if 'sink-inputs' in args else '7'), \
   patch.object(n.subprocess,'run'),patch.object(n,'cleanup_owned_run',return_value={'cleanup_confirmed':cleanup,'errors':[]}) as cleaned:
   report=n.run_project(self.steam,self.compat,self.session,project,30,self.authored,
    external=self.context if project in {'core'} else None)
  launch.assert_called_once();cleaned.assert_called_once()
  return report
 def args(self,project='core',external=True,continuation=True):
  session=Path('/tmp')/('erri-external-package-'+uuid.uuid4().hex)
  self.addCleanup(lambda:shutil.rmtree(session,ignore_errors=True))
  return argparse.Namespace(project=project,external_addons=self.source if external else None,
   session=session,steam=self.steam,compat=self.compat,timeout=30,continue_on_failure=continuation)
 def run_double(self,args,callback):
  with patch.object(n,'assert_no_game'),patch.object(n,'assert_prefix_idle'), \
   patch.object(n.shutil,'which',return_value='/mock'),patch.object(n,'run_project',side_effect=callback) as runs, \
   patch.object(n.subprocess,'Popen') as child,contextlib.redirect_stdout(io.StringIO()):
   code=n.run(args,self.root)
  child.assert_not_called()
  return code,json.loads((args.session/'package-report.json').read_text()),runs.call_args_list
 def test_option_is_required_only_for_external_projects_before_native_work(self):
  for project in ('core',):
   with self.subTest(project=project),patch.object(n,'assert_no_game') as game,patch.object(n,'copy_payload') as copy:
    with self.assertRaisesRegex(ValueError,'require --external-addons'):n.run(self.args(project,False),self.root)
    game.assert_not_called();copy.assert_not_called()
  for project in ('ai','intel','map','chat'):
   with self.assertRaises(ValueError):n.project_order(project)
 def test_cli_passes_explicit_installed_path_without_launch(self):
  with patch.object(n,'run',return_value=1) as run:
   self.assertEqual(n.main(['--project','core','--session','/tmp/not-created','--compat',str(self.compat),
    '--external-addons',str(self.source)]),1)
   self.assertEqual(run.call_args.args[0].external_addons,self.source)
 def test_exact_profiles_expose_only_reviewed_six_not_workshop_parent(self):
  unrelated=self.source/'original';unrelated.mkdir();(unrelated/'addon.gproj').write_text('GameProject { GUID "5F16D7E4A1CBE075" }')
  self.prepare()
  for project in ('core',):
   profile=self.profile(project);context=self.context if project in ('core',) else None
   records=n.expose_external(project,profile,self.steam,context)
   expected={'ArmaReforger'}|({'ERRI_'+guid for guid in ext.ALLOWED} if context else set())
   self.assertEqual({p.name for p in (profile/'addons').iterdir()},expected)
   self.assertEqual(len(records),len(ext.ALLOWED) if context else 0)
   for link in (profile/'addons').iterdir():
    self.assertTrue(link.is_symlink());self.assertNotEqual(link.resolve(),self.source)
    self.assertNotEqual(link.resolve(),unrelated)
   command=n.command_for(self.steam,self.session,project)
   self.assertEqual(command[command.index('-addonsDir')+1],'Z:'+str(self.session/'source/addons'))
   self.assertNotIn(str(self.source),command)
  self.assertFalse((self.session/'source/addons'/ext.DIRECT[0]).exists())
  self.assertEqual(json.loads((self.session/'external-addons-before.json').read_text()),self.context['records'])
 def test_native_helper_rejects_missing_or_wrong_context_before_creating_profile(self):
  self.prepare()
  with patch.object(n.subprocess,'Popen') as child:
   with self.assertRaises(ValueError):n.run_project(self.steam,self.compat,self.session,'core',30,self.authored)
   with self.assertRaises(ValueError):n.run_project(self.steam,self.compat,self.session,'ai',30,self.authored,external=self.context)
   child.assert_not_called()
  self.assertFalse((self.session/'profiles').exists())
 def test_missing_manifest_changed_policy_or_foreign_dependency_refuse(self):
  manifest=self.source/ext.DIRECT[0]/'addon.gproj';manifest.write_text(manifest.read_text()+'// altered')
  self.session.mkdir()
  with self.assertRaisesRegex(ValueError,'manifest changed'):n.capture_external(self.context,self.session)
  self.assertFalse((self.session/'profiles').exists())
 def test_missing_closure_and_resource_identity_refuse_before_exposure(self):
  database=self.source/ext.DIRECT[1]/'resourceDatabase.rdb';database.write_bytes(b'wrong resource identity')
  self.session.mkdir()
  with self.assertRaisesRegex(ValueError,'resource identity'):n.capture_external(self.context,self.session)
  shutil.rmtree(self.source/ext.DIRECT[0])
  with self.assertRaisesRegex(ValueError,'Missing installed'):n.capture_external(self.context,self.session)
 def test_unrelated_addon_wrong_link_and_whole_tree_link_are_refused(self):
  self.prepare();profile=self.profile('core');records=n.expose_external('core',profile,self.steam,self.context)
  addon=profile/'addons';extra=addon/'Original';extra.symlink_to(self.source,target_is_directory=True)
  with self.assertRaisesRegex(ValueError,'unrelated'):n.verify_package_profile(profile,self.steam,records)
  extra.unlink();link=addon/('ERRI_'+ext.DIRECT[0]);link.unlink();link.symlink_to(self.source,target_is_directory=True)
  with self.assertRaisesRegex(ValueError,'link changed'):n.verify_package_profile(profile,self.steam,records)
  link.unlink();link.mkdir()
  with self.assertRaisesRegex(ValueError,'link changed'):n.verify_package_profile(profile,self.steam,records)
 def test_native_clean_exit_binds_before_after_external_fingerprints(self):
  self.prepare();report=self.native_double()
  self.assertTrue(report['candidate_created']);self.assertTrue(report['external_addons_unchanged_after_run'])
  self.assertTrue(report['profile_addons_unchanged']);self.assertEqual(report['external_addon_count'],len(ext.ALLOWED))
  evidence=self.session/'evidence/core'
  before=json.loads((evidence/'external-addons-before.json').read_text())
  after=json.loads((evidence/'external-addons-after.json').read_text())
  self.assertTrue(after['unchanged']);self.assertEqual(after['records'],before)
  self.assertEqual(json.loads((evidence/'launch.json').read_text())['external_policy_sha256'],report['external_policy_sha256'])
  for key in ('release_qualified','package_contents_verified','native_consumption_verified','tool_completion_semantics_verified'):
   self.assertFalse(report[key])
 def test_external_payload_change_cannot_qualify_after_clean_native_exit(self):
  self.prepare();pak=self.source/ext.DIRECT[0]/'data.pak'
  report=self.native_double(mutation=lambda:pak.write_bytes(b'changed installed payload'))
  self.assertFalse(report['candidate_created']);self.assertFalse(report['external_addons_unchanged_after_run'])
  self.assertTrue(report['copied_authored_unchanged']);self.assertTrue(report['cleanup_confirmed'])
  self.assertTrue(any('External input validation failed' in e for e in report['errors']))
  after=json.loads((self.session/'evidence/core/external-addons-after.json').read_text())
  self.assertFalse(after['unchanged']);self.assertNotEqual(after['records'],self.context['records'])
 def test_cache_tampering_cannot_relabel_changed_payload_as_unchanged(self):
  self.prepare()
  self.context['cache_path'].write_text('{}')
  with self.assertRaisesRegex(ValueError,'fingerprint cache changed'):n.verify_external(self.context)
 def test_policy_mutation_or_deleted_input_cannot_qualify(self):
  self.prepare();path=self.context['policy_path']
  report=self.native_double(mutation=lambda:path.write_text(path.read_text()+'\n'))
  self.assertFalse(report['candidate_created']);self.assertFalse(report['external_addons_unchanged_after_run'])
  after=json.loads((self.session/'evidence/core/external-addons-after.json').read_text());self.assertIsNone(after['records'])
 def test_relinked_private_dependency_fails_without_altering_installed_inputs(self):
  self.prepare()
  def mutate():
   link=self.session/'profiles/core/addons'/('ERRI_'+ext.DIRECT[0]);link.unlink();link.symlink_to(self.source,target_is_directory=True)
  report=self.native_double(mutation=mutate)
  self.assertFalse(report['candidate_created']);self.assertFalse(report['profile_addons_unchanged'])
  self.assertTrue(report['external_addons_unchanged_after_run'])
 def test_unchanged_external_inputs_do_not_suppress_native_errors(self):
  self.prepare();report=self.native_double(error='RESOURCES (E): retained upstream localization error')
  self.assertFalse(report['candidate_created']);self.assertTrue(report['external_addons_unchanged_after_run'])
  self.assertTrue(any('retained upstream localization error' in e for e in report['errors']))
 def test_missing_native_exit_or_failed_cleanup_remain_disqualifying(self):
  self.prepare();report=self.native_double(write_status=False,cleanup=False)
  self.assertFalse(report['candidate_created']);self.assertIsNone(report['native_exit_code'])
  self.assertFalse(report['cleanup_confirmed']);self.assertTrue(report['external_addons_unchanged_after_run'])
 def test_context_only_passed_to_ai_and_chat_in_full_build(self):
  args=self.args()
  def result(*pos,**kw):
   self.assertEqual(kw['external'] is not None,pos[3] in ('core',))
   return base.NativePackageTests.completed_report(True)
  code,summary,calls=self.run_double(args,result)
  self.assertEqual(code,0);self.assertEqual([c.args[3] for c in calls],['core'])
  self.assertEqual(summary['external_addon_count'],len(ext.ALLOWED));self.assertTrue(summary['external_addons_unchanged_after_run'])
  self.assertFalse(summary['release_qualified']);self.assertFalse(summary['published'])
  self.assertTrue((args.session/'external-addons-after.json').is_file())
 def test_diagnostic_continuation_stops_on_mid_batch_external_mutation(self):
  args=self.args();seen=[]
  def result(*pos,**kw):
   seen.append(pos[3]);(self.source/ext.DIRECT[0]/'data.pak').write_bytes(b'changed between projects')
   return base.NativePackageTests.completed_report(True)
  code,summary,calls=self.run_double(args,result)
  self.assertEqual(code,1);self.assertEqual(seen,['core']);self.assertEqual(len(calls),1)
  self.assertFalse(summary['external_addons_unchanged_after_run']);self.assertFalse(summary['all_candidates_created'])
  self.assertIn('external dependency inputs changed',summary['stopped_reason'])
  self.assertEqual(summary['skipped_projects'],[])
 def test_diagnostic_continuation_cannot_ignore_unverified_external_report(self):
  args=self.args();reports=[base.NativePackageTests.completed_report(True,external_addons_unchanged_after_run=False)]
  code,summary,calls=self.run_double(args,lambda *a,**k:reports.pop(0))
  self.assertEqual(code,1);self.assertEqual(len(calls),1);self.assertIn('external dependency',summary['stopped_reason'])
  self.assertFalse(summary['all_candidates_created'])

if __name__=='__main__':unittest.main()