"""Focused deployment handoff fixes. No Runtime or external service calls.""" import copy import json from pathlib import Path import sys import tempfile import unittest from unittest.mock import patch from deployment_kit import bundle, release, diagnostics from deployment_kit.core import KitError, canonical, run, sha256, write_json from helpers import fixture_site, mini_repo, spec, IID from test_release import ReleaseTests class OutputTests(unittest.TestCase): def test_warning_does_not_contaminate_json(self): code,raw=run([sys.executable,'-c','import sys; print("warning",file=sys.stderr); print("{\\"ok\\":true}")']) self.assertEqual(code,0); self.assertEqual(json.loads(raw),{'ok':True}) def test_failure_keeps_classification_not_secret(self): with self.assertRaises(KitError) as c: run([sys.executable,'-c','import sys; print("permission denied password=TEST_SECRET",file=sys.stderr); sys.exit(4)']) self.assertEqual(c.exception.diagnostic['cause'],'PERMISSION') self.assertNotIn('TEST_SECRET',str(c.exception.diagnostic)+c.exception.explain()) def test_stderr_is_bounded(self): with self.assertRaises(KitError) as c: run([sys.executable,'-c','import sys; print("x"*2000,file=sys.stderr)'],limit=100) self.assertEqual(c.exception.code,'COMMAND_LIMIT') class ResumeTests(ReleaseTests): # Run only these named tests, not the inherited candidate suite. def test_resume_own_failed_first_install(self): a,h=self.archive(); env=self.env(fail_health=True) with self.assertRaises(KitError): release.deploy(self.sitepath,a,h,apply=True) env.close() p=release.state_dir(self.site)/'inflight.json'; record=json.loads(p.read_text()) row={'id':'f'*64,'service':'analytics','running':False,'status':'exited', 'image_id':IID,'transaction':record['transaction']} self.env() with patch.object(release,'containers',return_value=[row]): release.recover(self.site,expected=sha256(p),apply=True) a,h=self.archive('v2') result=release.deploy(self.sitepath,a,h,apply=True) self.assertEqual(result['release_id'],'v2') self.assertTrue((Path(self.site['install_root'])/'releases/v1').is_dir()) def test_foreign_container_cannot_be_adopted(self): row={'id':'f'*64,'service':'analytics','running':False,'image_id':IID,'transaction':'foreign'} with self.assertRaises(KitError) as c: release.verify_failed_ownership([row],{'stage':'HEALTH','transaction':'owned','intended_images':{'analytics':IID}}) self.assertEqual(c.exception.code,'FAILED_RESOURCE_OWNERSHIP') def test_failure_record_is_safe(self): a,h=self.archive(); self.env(fail_apply=True) with self.assertRaises(KitError): release.deploy(self.sitepath,a,h,apply=True) record=json.loads((release.state_dir(self.site)/'inflight.json').read_text()) self.assertEqual(record['failure']['code'],'FIXTURE_APPLY_FAILED') self.assertEqual(record['stage'],'APPLY') self.assertEqual(record['intended_images']['analytics'],IID) class ModelScopeTests(unittest.TestCase): def test_extra_visible_model_rejected(self): with tempfile.TemporaryDirectory() as tmp: site=fixture_site(tmp); models=Path(site['storage']['models']['physical'])/'active' (models/'RFDETRMedium').mkdir(parents=True) (models/'RFDETRMedium/config.pbtxt').write_text('name: "RFDETRMedium"') volumes=[{'type':'bind','source':site['storage'][k]['physical'], 'target':site['storage'][k]['container']} for k in ('runtime','artifacts')] volumes.append({'type':'bind','source':str(models),'target':'/opt/contextra/models/triton','read_only':True}) doc={'services':{s:{'image':IID} for s in site['services']}} doc['services']['analytics']['volumes']=volumes release.validate_rendered(doc,site,{s:IID for s in site['services']}) (models/'RFDETRFireSmoke').mkdir() with self.assertRaises(KitError) as c: release.validate_rendered(doc,site,{s:IID for s in site['services']}) self.assertEqual(c.exception.code,'MODEL_EXPOSURE_MISMATCH') def test_shared_archive_loaded_once(self): m={'kind':'FULL','images':[{'service':s,'id':IID,'reference':IID,'archive':'images/all.tar'} for s in ('analytics','redis')]} with patch.object(release,'docker') as dk, patch.object(release,'image_id',return_value=IID): release.prepare_images(m,Path('/fixture'),None,None) self.assertEqual(dk.call_count,1) class SnapshotTests(unittest.TestCase): def test_exact_uncommitted_snapshot_and_drift(self): with tempfile.TemporaryDirectory() as tmp: root=Path(tmp); repo,head=mini_repo(root/'repo') (repo/'app/example.py').write_text('VERSION=2\n') (repo/'deployment_kit/new.py').write_text('# new\n') s=spec(head) s['snapshot']={'candidate_sha':head,'files':{p.relative_to(repo).as_posix():sha256(p) for p in repo.rglob('*') if p.is_file() and bundle.allowed_source(p.relative_to(repo).as_posix())}} sp=root/'spec.json'; write_json(sp,s) out=root/'release.tar.gz'; bundle.create(repo,sp,out) m=bundle.verify(out) self.assertEqual(m['snapshot']['kind'],'EXACT_WORKTREE') self.assertIn('deployment_kit/new.py',m['files']) self.assertFalse(any('private' in p or 'AGENTS' in p for p in m['files'])) (repo/'app/example.py').write_text('VERSION=3\n') with self.assertRaises(KitError) as c: bundle.create(repo,sp,root/'drift.tar.gz') self.assertEqual(c.exception.code,'SOURCE_DIRTY') def test_snapshot_cannot_include_secrets(self): with tempfile.TemporaryDirectory() as tmp: root=Path(tmp); repo,head=mini_repo(root/'repo'); s=spec(head) s['snapshot']={'candidate_sha':head,'files':{p.relative_to(repo).as_posix():sha256(p) for p in repo.rglob('*') if p.is_file() and bundle.allowed_source(p.relative_to(repo).as_posix())}} s['snapshot']['files']['.env']='0'*64 sp=root/'spec.json'; write_json(sp,s) with self.assertRaises(KitError) as c: bundle.create(repo,sp,root/'bad.tar.gz') self.assertEqual(c.exception.code,'SNAPSHOT_SCOPE')