"""No Docker/mount/GPU; filesystem and protocol boundaries use controlled fixtures.""" import copy import io import json import os from pathlib import Path import subprocess import sys import tarfile import tempfile import unittest from unittest.mock import patch from deployment_kit import core, storage, site, host, compose_snapshot as cs, diagnostics, cli, models from helpers import fixture_site, inventory, IID class CoreTests(unittest.TestCase): def test_duplicate_json(self): with self.assertRaises(core.KitError): core.parse_json('{"a":1,"a":2}') def test_nonfinite_json(self): for x in ('NaN','Infinity','-Infinity'): with self.subTest(x=x), self.assertRaises(core.KitError): core.parse_json(x) def test_unsafe_paths(self): for x in ('/', 'relative', '/tmp/../etc', '/tmp/a\nb'): with self.subTest(x=x), self.assertRaises(core.KitError): core.absolute(x) def test_identifier(self): for x in (None, '', '.', '..', 'a/b', 'a'*81): with self.subTest(x=x), self.assertRaises(core.KitError): core.identifier(x) def test_symlink_path(self): with tempfile.TemporaryDirectory() as d: p=Path(d); (p/'dir').mkdir(); (p/'link').symlink_to(p/'dir',target_is_directory=True) with self.assertRaises(core.KitError): core.absolute(p/'link/file',no_symlinks=True) def test_atomic_mode(self): with tempfile.TemporaryDirectory() as d: p=Path(d)/'state'; core.atomic_write(p,'text',new=True) self.assertEqual(p.stat().st_mode & 0o777,0o600) def test_no_shell(self): rc,text=core.run([sys.executable,'-c','import sys;print(sys.argv[1])','$(touch invalid)']) self.assertEqual(text.strip(),'$(touch invalid)'); self.assertEqual(rc,0) def test_timeout_cleanup(self): with self.assertRaises(core.KitError) as c: core.run([sys.executable,'-c','import time;time.sleep(20)'],timeout=.05) self.assertEqual(c.exception.code,'COMMAND_LIMIT') def test_output_bound(self): with self.assertRaises(core.KitError): core.run([sys.executable,'-c','print("x"*10000)'],limit=100) def test_error_redacts_child_output(self): with self.assertRaises(core.KitError) as c: core.run([sys.executable,'-c','print("SECRET_VALUE");raise SystemExit(3)']) self.assertNotIn('SECRET_VALUE',str(c.exception)) def test_apply_requires_root(self): with patch.object(core.os,'geteuid',return_value=1001), self.assertRaises(core.KitError): core.approved(True,True) def test_no_apply_no_confirmation(self): with patch('builtins.input',side_effect=AssertionError('prompt')): self.assertFalse(core.approved(False)) class SiteStorageTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup) self.s=fixture_site(Path(self.temp.name)) def fs(self,uuid='test',target='/',options='rw'): return lambda _: (0,json.dumps({'filesystems':[{'target':target,'source':'/dev/fake','fstype':'ext4','uuid':uuid,'options':options}]})) def test_valid_and_portable(self): self.assertEqual(site.validate(self.s),self.s) def test_secret_field(self): self.s['password']='secret' with self.assertRaises(core.KitError): site.validate(self.s) def test_roots_overlap(self): self.s['install_root']=self.s['data_root'] with self.assertRaises(core.KitError): site.validate(self.s) def test_project_case(self): self.s['project']='BAD' with self.assertRaises(core.KitError): site.validate(self.s) def test_container_path_fixed(self): self.s['storage']['models']['container']='/different' with self.assertRaises(core.KitError): site.validate(self.s) def test_env_relative_forbidden(self): self.s['env_files']=['local.env'] with self.assertRaises(core.KitError): site.validate(self.s) def test_base_first(self): self.s['compose_files']=['source/deploy/web-start.compose.yml'] with self.assertRaises(core.KitError): site.validate(self.s) def test_matching_filesystem(self): result=storage.check_entry(self.s['storage']['models'],runner=self.fs()) self.assertEqual(result['filesystem']['uuid'],'test') def test_wrong_disk(self): with self.assertRaises(core.KitError): storage.check_entry(self.s['storage']['models'],runner=self.fs('other')) def test_read_only(self): with self.assertRaises(core.KitError): storage.check_entry(self.s['storage']['models'],runner=self.fs(options='ro')) def test_root_requires_consent(self): e=self.s['storage']['models']; e['allow_root']=False with self.assertRaises(core.KitError): storage.check_entry(e,runner=self.fs()) def test_view_readmes_without_mount(self): storage.prepare(self.s,bind_views=False,runner=self.fs()) self.assertTrue((Path(self.s['data_root'])/'README.ko.md').is_file()) self.assertTrue((Path(self.s['data_root'])/'offline/inputs').is_dir()) storage.prepare(self.s,bind_views=False,runner=self.fs()) def test_storage_move_never_runs_rsync(self): with patch.object(storage,'checks',return_value={}),patch.object(storage,'run',side_effect=AssertionError('mutation')): p=storage.move_plan(self.s,'artifacts',str(Path(self.temp.name)/'new')) self.assertFalse(p['automatic_move']); self.assertNotIn('--delete',p['copy']) def test_storage_move_source_overlap(self): with patch.object(storage,'checks',return_value={}),self.assertRaises(core.KitError): storage.move_plan(self.s,'models',self.s['storage']['models']['physical']+'/child') def test_site_save_cas(self): path=Path(self.s['install_root'])/'state/site.json' with patch.object(storage,'checks',return_value={}): site.save(self.s,path) with self.assertRaises(core.KitError): site.save(self.s,path) site.save(self.s,path,expected=core.sha256(path)) class HostTests(unittest.TestCase): def test_requirements_pass(self): host.enforce_requirements(inventory(),{'driver_min':'595.58.03'}) def test_driver_does_not_auto_upgrade(self): inv=inventory(); inv['gpu'][0]['driver']='580.173.02' with self.assertRaises(core.KitError) as c: host.enforce_requirements(inv,{'driver_min':'595.58.03'}) self.assertEqual(c.exception.code,'DRIVER_REVIEW_REQUIRED') def test_requirements_absent_blocks(self): with self.assertRaises(core.KitError): host.enforce_requirements(inventory(),{}) def test_compose_old_blocks(self): inv=inventory(); inv['compose_version']='2.10.0' with self.assertRaises(core.KitError): host.enforce_requirements(inv,{'driver_min':'580.0'}) def test_docker_unavailable(self): inv=inventory(); inv['docker_access']='UNAVAILABLE_OR_SUDO_REQUIRED' with self.assertRaises(core.KitError): host.enforce_requirements(inv,{'driver_min':'580.0'}) def test_ip_change(self): with self.assertRaises(core.KitError): host.network_check({'ai_ip':'192.0.2.99'},inventory()) def test_interface_selection(self): data=[{'ifname':'eth0','operstate':'UP','addr_info':[{'family':'inet','scope':'global','local':'192.0.2.10','prefixlen':24}]}] self.assertEqual(host.addresses(lambda _:(0,json.dumps(data)))[0]['ip'],'192.0.2.10') def test_ports(self): text='LISTEN 0 4096 0.0.0.0:9443 0.0.0.0:*\nLISTEN 0 10 [::]:22 [::]:*\n' self.assertEqual(host.listening_ports(lambda _:(0,text)),[22,9443]) def test_bootstrap_plan_offers_no_group_or_driver_change(self): with patch.object(host,'run',side_effect=AssertionError('mutation')): result=host.bootstrap() self.assertIn('docker group',result['not_changed']) def test_bootstrap_requires_lock(self): with self.assertRaises(core.KitError): host.bootstrap(apply=True) class ComposeTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup) self.p=Path(self.temp.name) self.doc={'name':'fixture','services':{'analytics':{'image':IID,'environment':{'X':'hello$$world$$$$$${a}'}}}, 'volumes':{'data':{'name':'fixture_data'}},'networks':{'default':{'name':'fixture_default'}}} def test_literal_dollars_roundtrip(self): path=cs.seal(self.doc,self.p/'compose.json','fixture',runner=lambda *a,**k:(0,json.dumps(self.doc))) self.assertEqual(cs.document(path),self.doc) self.assertEqual(json.loads(path.read_text())['services']['analytics']['environment']['X'],'hello$$world$$$$$${a}') def test_bad_roundtrip_blocks(self): with self.assertRaises(core.KitError): cs.seal(self.doc,self.p/'compose.json','fixture',runner=lambda *a,**k:(0,'{}')) def test_resource_identity_required(self): self.doc['volumes']['data']={} with self.assertRaises(core.KitError): cs.normalized(self.doc,'fixture') def test_project_mismatch(self): with self.assertRaises(core.KitError): cs.normalized(self.doc,'other') def test_shell_env_not_inherited(self): with patch.dict(os.environ,{'CONTEXTRA_IMAGE':'bad','POSTGRES_PASSWORD':'secret'}): env=cs.command_environment() self.assertNotIn('CONTEXTRA_IMAGE',env); self.assertNotIn('POSTGRES_PASSWORD',env) def test_bound_file_change_blocked(self): f=self.p/'runtime.json'; f.write_text('{}') d={'services':{'analytics':{'volumes':[{'type':'bind','source':str(f),'target':'/etc/test','read_only':True}]}}} records=cs.file_bindings(d); f.write_text('changed') with self.assertRaises(core.KitError): cs.verify_bindings(records) def test_external_secret_is_not_snapshotted(self): d={'services':{'analytics':{'volumes':[{'type':'bind','source':'/nonexistent','target':'/run/bridge/token','read_only':True}]}}} self.assertEqual(cs.file_bindings(d),{}) def test_directory_addition_blocked(self): d={'services':{'analytics':{'volumes':[{'type':'bind','source':str(self.p),'target':'/etc/contextra','read_only':True}]}}} (self.p/'a.json').write_text('{}'); r=cs.file_bindings(d); (self.p/'b.json').write_text('{}') with self.assertRaises(core.KitError): cs.verify_bindings(r) def test_snapshot_hash(self): p=cs.seal(self.doc,self.p/'compose.json','fixture',runner=lambda *a,**k:(0,json.dumps(self.doc))) with self.assertRaises(core.KitError): cs.document(p,'0'*64) class DiagnosticsCliTests(unittest.TestCase): def test_no_site_is_not_runtime_pass(self): d=diagnostics.doctor(inv=inventory()) self.assertNotEqual(d['HOST_READY'],'PASS'); self.assertEqual(d['RUNTIME_READY'],'NOT_RUN') def test_redaction(self): d=diagnostics.redact({'password':'abc','text':'Bearer TOKEN rtsp://u:p@camera/x'}) self.assertNotIn('TOKEN',d['text']); self.assertNotIn('u:p',d['text']); self.assertEqual(d['password'],'[REDACTED]') def test_support_no_raw_private_sources(self): with tempfile.TemporaryDirectory() as d,patch.object(host,'inventory',return_value=inventory()): p=Path(d)/'support.tar.gz'; diagnostics.collect(None,p) with tarfile.open(p) as t: self.assertEqual(t.getnames(),['support.json']); raw=t.extractfile('support.json').read() self.assertIn(b'SAFE_PROJECTIONS_ONLY',raw); self.assertEqual(p.stat().st_mode & 0o777,0o600) def test_apply_default_false(self): a=cli.parser().parse_args(['model','build','medium','--onnx','/x','--output','/y','--image',IID]) self.assertFalse(getattr(a,'apply',False)) def test_nested_approval_kept(self): a=cli.parser().parse_args(['--site','/x/site.json','storage','prepare','--apply','--yes']) self.assertEqual(a.site,'/x/site.json'); self.assertTrue(a.apply and a.yes) def test_model_profile_plan(self): with patch.object(models,'docker',side_effect=AssertionError('docker')): result=cli.execute(cli.parser().parse_args(['model','profile','fire'])) self.assertEqual(result['name'],'RFDETRFireSmoke') def test_help_runs_without_runtime_imports(self): p=subprocess.run([sys.executable,'-B','deployctl','--help'],capture_output=True,text=True) self.assertEqual(p.returncode,0); self.assertIn('bootstrap',p.stdout)