"""Single-user policy tests; no real credentials, models or production services.""" import json from pathlib import Path import sys import tempfile import unittest from unittest import mock from datetime import datetime,timezone,timedelta sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import local_mode as local import controller as c import test_controller as fixture class LocalPolicyTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='jobjakt-single-user-test-') self.base = Path(self.temp.name) self.root = self.base / 'control' self.repo = self.base / 'worker' self.repo.mkdir() self.config = dict(mode='single-user-local', executionPolicy='operator-controlled', ownerSid='S-1-5-21-100', runnerSid='S-1-5-21-100', repository=str(self.repo), workerCodeXHome=str(self.root / 'auth'), normalCodeXHome=str(self.base / 'normal-codex'), workerTemp=str(self.base / 'worker-temp'), toolReadRoots=[], trustedCodexOverrides=[]) def tearDown(self): self.temp.cleanup() def test_same_windows_identity_supported(self): local.validate_identity(self.config, self.config['ownerSid']) def test_different_or_missing_sid_rejected(self): with self.assertRaises(ValueError): local.validate_identity(self.config, 'S-1-5-21-999') def test_existing_normal_auth_home_is_rejected(self): self.config['workerCodeXHome'] = self.config['normalCodeXHome'] with self.assertRaisesRegex(ValueError, 'auth'): local.validate_config(self.root, self.config) def test_auth_home_inside_worker_is_rejected(self): self.config['workerCodeXHome'] = str(self.repo / 'auth') with self.assertRaises(ValueError): local.validate_config(self.root,self.config) def test_profile_denies_root_control_auth_and_normal_configuration(self): rules = local.filesystem_rules(self.root,self.config) self.assertEqual('deny',rules[':root']) self.assertEqual('read',rules[':minimal']) for path in [self.root,Path(self.config['workerCodeXHome']),Path(self.config['normalCodeXHome'])]: self.assertEqual('deny',rules[path.resolve().as_posix()]) self.assertEqual('write',rules[self.repo.resolve().as_posix()]) self.assertEqual('read',rules[(self.repo / '.git').resolve().as_posix()]) def test_mxc_is_explicit_without_legacy_or_unrestricted_fallback(self): args = local.permission_args(self.root,self.config,'exec') self.assertIn('windows.sandbox="mxc"',args) self.assertIn('permissions.jobjakt-local.network.enabled=false',args) self.assertNotIn('--sandbox',args) self.assertNotIn('--dangerously-bypass-approvals-and-sandbox',args) for value in ('features.apps=false','features.plugins=false','features.remote_plugin=false','mcp_servers={}'): self.assertIn(value,args) def test_plugins_are_not_an_implicit_same_user_capability(self): self.config['trustedCodexOverrides'] = [{'key':'mcp_servers.any.command','toml':'"unsafe"'}] with self.assertRaisesRegex(ValueError,'plugin'): local.validate_config(self.root,self.config) def test_tool_or_temp_roots_cannot_reopen_protected_parent(self): self.config['toolReadRoots']=[str(self.base)] with self.assertRaises(ValueError): local.validate_config(self.root,self.config) self.config['toolReadRoots']=[] self.config['workerTemp']=str(self.base) with self.assertRaises(ValueError): local.validate_config(self.root,self.config) def test_operator_controlled_policy_cannot_launch_a_worker(self): with self.assertRaisesRegex(ValueError,'operator-controlled'): local.require_unattended(self.root,self.config) def test_missing_or_stale_qualification_never_enables_unattended(self): self.config['executionPolicy'] = 'qualified-unattended' with self.assertRaises(ValueError): local.require_unattended(self.root,self.config) def test_probe_summary_rejects_partial_and_failed_boundaries(self): result = dict(positiveControl=True,networkDenied=True,signingKeyDenied=True, files=[dict(readDenied=True,writeDenied=False)]) self.assertFalse(local.probes_passed(result)) result['files'][0]['writeDenied'] = True self.assertFalse(local.probes_passed(result)) # required categories cannot be omitted def test_same_chatgpt_auth_is_explicit_and_tokens_are_not_copied(self): normal = Path(self.config['normalCodeXHome']) normal.mkdir() (normal/'auth.json').write_text('synthetic-existing-token-do-not-copy') with mock.patch.object(local,'detect_identity',return_value=('synthetic-user',self.config['ownerSid'])), \ mock.patch.dict(local.os.environ,{'CODEX_HOME':str(normal)}): config = local.configure_current_user(self.root,dict(self.config)) self.assertEqual(config['ownerSid'],config['runnerSid']) self.assertEqual([],list(Path(config['workerCodeXHome']).iterdir())) args = local.auth_command({**config,'codexExecutable':'codex'}) self.assertIn('forced_login_method="chatgpt"',args) self.assertIn('cli_auth_credentials_store="file"',args) self.assertNotIn('--with-api-key',args) self.assertEqual('synthetic-existing-token-do-not-copy',(normal/'auth.json').read_text()) def test_failed_probe_and_unconfigured_owner_key_cannot_qualify(self): self.root.mkdir() self.config['executionPolicy']='qualified-unattended' valid = dict(positiveControl=True,networkDenied=True,signingKeyDenied=True, files=[dict(category=k,readDenied=True,writeDenied=True) for k in sorted(local.CATEGORIES)]) report = dict(status='PASS',failure=None,binding=local.binding(self.root,self.config),issuedUtc=datetime.now(timezone.utc).isoformat(),probes=valid,ownerKeyDenied=False) c.atomic(self.root/'qualification.json',report) with self.assertRaises(ValueError): local.require_unattended(self.root,self.config) report['ownerKeyDenied']=True report['probes']['networkDenied']=False c.atomic(self.root/'qualification.json',report) with self.assertRaises(ValueError): local.require_unattended(self.root,self.config) report['probes']['networkDenied']=True c.atomic(self.root/'qualification.json',report) local.require_unattended(self.root,self.config) report['issuedUtc']=(datetime.now(timezone.utc)-timedelta(days=2)).isoformat() c.atomic(self.root/'qualification.json',report) with self.assertRaises(ValueError): local.require_unattended(self.root,self.config) def test_blocked_status_rejects_otherwise_successful_probes(self): self.root.mkdir() self.config['executionPolicy']='qualified-unattended' c.atomic(self.root/'qualification.json',dict(status='BLOCKED',failure='cleanup failed', binding=local.binding(self.root,self.config),issuedUtc=datetime.now(timezone.utc).isoformat(),ownerKeyDenied=True, probes=dict(positiveControl=True,networkDenied=True,signingKeyDenied=True, files=[dict(category=k,readDenied=True,writeDenied=True) for k in sorted(local.CATEGORIES)]))) with self.assertRaises(ValueError): local.require_unattended(self.root,self.config) def test_noninteractive_approval_is_refused(self): c.atomic(self.root/'config.json',self.config) c.atomic(self.root/'backlog.json',{}) with mock.patch.object(c.sys.stdin,'isatty',return_value=False): with self.assertRaisesRegex(c.GateError,'interactive terminal'): c.main(['approve','--control-root',str(self.root)]) class LocalFlowTests(unittest.TestCase): tearDown=fixture.EngineTests.tearDown git_at=staticmethod(fixture.EngineTests.git_at) run_engine=fixture.EngineTests.run_engine def setUp(self): fixture.EngineTests.setUp(self) self.config.update(mode='single-user-local',executionPolicy='qualified-unattended', ownerSid='S-1-5-21-100',runnerSid='S-1-5-21-100', normalCodeXHome=str(self.root/'normal-codex'),workerCodeXHome=str(self.control/'auth'), workerTemp=str(self.root/'worker-temp'),toolReadRoots=[],trustedCodexOverrides=[]) Path(self.config['workerCodeXHome']).mkdir() Path(self.config['workerTemp']).mkdir() c.atomic(self.control/'qualification.json',dict(status='PASS',failure=None,binding=local.binding(self.control,self.config), issuedUtc=datetime.now(timezone.utc).isoformat(),ownerKeyDenied=True, probes=dict(positiveControl=True,networkDenied=True,signingKeyDenied=True, files=[dict(category=k,readDenied=True,writeDenied=True) for k in sorted(local.CATEGORIES)]))) def test_single_user_subprocess_protocol_and_ordered_publication(self): self.engine.runner=fixture.ProtocolRunner() self.assertEqual('COMPLETED',self.run_engine()['stages']['TWO']['status']) def test_single_user_missing_approval_stops_before_worker(self): self.engine.approval_verifier=lambda request:False self.assertEqual('AWAITING_APPROVAL',self.run_engine()['stages']['ONE']['status']) self.assertEqual(0,self.runner.calls) def test_single_user_pause_and_publication_recovery(self): self.engine.pause() self.run_engine() self.assertEqual(0,self.runner.calls) self.engine.resume() publish=self.engine.publish self.engine.publish=lambda *args: (_ for _ in ()).throw(c.PublicationError('synthetic outage')) self.assertEqual('BLOCKED',self.run_engine()['stages']['ONE']['status']) self.assertEqual(1,self.runner.calls) self.engine.publish=publish self.assertEqual('COMPLETED',self.run_engine()['stages']['TWO']['status']) self.assertEqual(2,self.runner.calls) def test_single_user_concurrent_controller_and_interrupted_guard(self): with c.RepoLock(self.repo): with self.assertRaises(c.GateError): self.run_engine() state=self.engine.store.read() state['stages']['ONE']['status']='RUNNING' self.engine.store.write(state) with self.assertRaises(c.GateError): self.run_engine() def test_single_user_main_checkout_is_still_prohibited(self): self.git_at(self.repo,'switch','-c','main') self.assertEqual('BLOCKED',self.run_engine()['stages']['ONE']['status']) self.assertEqual(0,self.runner.calls) def test_pending_recovery_can_be_signed_from_idle_operator_terminal(self): self.config['ownerCertificateThumbprint']='synthetic-owner-key' c.atomic(self.control/'config.json',self.config) c.atomic(self.control/'backlog.json',self.manifest) state=self.engine.store.read() state['stages']['ONE']['status']='RUNNING' self.engine.store.write(state) self.engine.approval_verifier=lambda r:False def answer(prompt): if prompt.startswith('Type '): return prompt.split()[1] if prompt.startswith('Disposition'): return 'Approve' return 'Operator inspected the stopped synthetic process tree' with mock.patch.object(c,'Engine',return_value=self.engine): self.assertEqual(3,c.main(['recover','--control-root',str(self.control)])) run=c.subprocess.run def operator_command(argv,**kwargs): return mock.Mock(returncode=0) if 'Sign' in argv else run(argv,**kwargs) with mock.patch.object(c.sys.stdin,'isatty',return_value=True), \ mock.patch.object(local,'detect_identity',return_value=('synthetic-user',self.config['ownerSid'])), \ mock.patch('builtins.input',side_effect=answer), \ mock.patch.object(c.subprocess,'run',side_effect=operator_command) as signed: self.assertEqual(0,c.main(['approve','--control-root',str(self.control)])) self.assertIn('Sign',signed.call_args.args[0]) self.assertTrue(self.engine.store.read()['recoveryRequired']) # signing alone never marks recovery complete if __name__ == '__main__': unittest.main(verbosity=2)