"""Real bundle/files/journal transactions with Docker/host operations replaced by fakes.""" from contextlib import ExitStack import copy import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from deployment_kit import bundle, release, compose_snapshot as cs from deployment_kit.core import KitError, canonical, sha256, write_json from helpers import fixture_site, mini_repo, spec, IID, inventory class ReleaseTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup) self.root=Path(self.tmp.name);self.site=fixture_site(self.root/'site') self.sitepath=Path(self.site['install_root'])/'state/site.json';write_json(self.sitepath,self.site,new=True) self.repo,self.sha=mini_repo(self.root/'repo');self.calls=[] def archive(self,rid='v1',parent=None,kind='FULL'): s=spec(self.sha,kind);s['release_id']=rid if parent:s['parent_release']=parent p=self.root/(rid+'.spec.json');p.write_bytes(canonical(s));a=self.root/(rid+'.tar.gz') bundle.create(self.repo,p,a);return a,sha256(a) def fake_render(self,site,root,images,tx): doc={'name':site['project'],'services':{s:{'image':images[s]} for s in site['services']}} p=tx/'compose.json' cs.seal(doc,p,site['project'],runner=lambda *a,**k:(0,json.dumps(doc))) return p,{str(self.site['env_files'][0]):sha256(self.site['env_files'][0])} def env(self,busy=False,fail_health=False,fail_apply=False): st=ExitStack();self.addCleanup(st.close) st.enter_context(patch.object(release.os,'geteuid',return_value=0)) st.enter_context(patch.object(release.host,'inventory',return_value=inventory())) st.enter_context(patch.object(release.storage,'checks',return_value={})) st.enter_context(patch.object(release.storage,'prepare',return_value=None)) st.enter_context(patch.object(release,'containers',return_value=[{'service':'analytics','running':True}] if busy else [])) st.enter_context(patch.object(release,'prepare_images',side_effect=lambda m,*a:{x['service']:x['id'] for x in m['images']})) st.enter_context(patch.object(release,'render',side_effect=self.fake_render)) st.enter_context(patch.object(release,'model_receipts',return_value={'RFDETRMedium':{'hash':'fake'}})) st.enter_context(patch.object(release,'image_id',side_effect=lambda x:x)) def dk(*a,**k): self.calls.append(a) if fail_apply:raise KitError('FIXTURE_APPLY_FAILED','fake') return 0,'' st.enter_context(patch.object(release,'docker',side_effect=dk)) health=KitError('FIXTURE_HEALTH_FAILED','fake') if fail_health else None st.enter_context(patch.object(release,'health',side_effect=health,return_value={ 'HOST_READY':'PASS','DEPLOYED':'PASS','RUNTIME_READY':'PASS','EXTERNAL_READY':'NOT_RUN'})) return st def test_plan_no_host_actions(self): a,h=self.archive() with patch.object(release,'docker',side_effect=AssertionError('docker')): p=release.deploy(self.sitepath,a,h) self.assertEqual(p['scope'],'PLAN_ONLY');self.assertIsNone(release.active(self.site)) def test_first_full_transaction(self): a,h=self.archive();self.env();r=release.deploy(self.sitepath,a,h,apply=True) self.assertEqual(r['release_id'],'v1');self.assertEqual(release.active(self.site)['release_id'],'v1') self.assertTrue((Path(self.site['install_root'])/'deployctl').is_file()) journal=json.loads((release.state_dir(self.site)/'inflight.json').read_text());self.assertEqual(journal['status'],'DONE') self.assertIn('--no-build',self.calls[0]);self.assertIn('never',self.calls[0]) def test_live_target_is_not_stopped_automatically(self): a,h=self.archive();self.env(busy=True) with self.assertRaises(KitError):release.deploy(self.sitepath,a,h,apply=True) self.assertEqual(self.calls,[]) def test_apply_failure_not_activated(self): a,h=self.archive();self.env(fail_apply=True) with self.assertRaises(KitError):release.deploy(self.sitepath,a,h,apply=True) self.assertIsNone(release.active(self.site)) j=json.loads((release.state_dir(self.site)/'inflight.json').read_text());self.assertEqual(j['status'],'FAILED') def test_failed_health_no_auto_rollback_or_stop(self): a,h=self.archive();self.env(fail_health=True) with self.assertRaises(KitError):release.deploy(self.sitepath,a,h,apply=True) self.assertEqual(len(self.calls),1);self.assertIsNone(release.active(self.site)) def test_completed_update_and_explicit_rollback_preserve_data(self): a,h=self.archive();self.env();release.deploy(self.sitepath,a,h,apply=True) old=release.active(self.site);data=Path(self.site['storage']['artifacts']['physical'])/'keep.bin';data.write_bytes(b'persist') a,h=self.archive('v2','v1','CONFIG');release.deploy(self.sitepath,a,h,apply=True) self.assertEqual(release.active(self.site)['release_id'],'v2') release.rollback(self.sitepath,old['transaction'],apply=True) self.assertEqual(release.active(self.site)['release_id'],'v1');self.assertEqual(data.read_bytes(),b'persist') def test_failed_journal_blocks_retry(self): a,h=self.archive();self.env() write_json(release.state_dir(self.site)/'inflight.json',{'status':'FAILED'}) with self.assertRaises(KitError) as c:release.deploy(self.sitepath,a,h,apply=True) self.assertEqual(c.exception.code,'INTERRUPTED_DEPLOYMENT');self.assertFalse(self.calls) def test_wrong_parent_blocks(self): a,h=self.archive();self.env();release.deploy(self.sitepath,a,h,apply=True) a,h=self.archive('v2','not-v1','CONFIG') with self.assertRaises(KitError):release.deploy(self.sitepath,a,h) def test_rotated_private_env_blocks_old_snapshot(self): a,h=self.archive();self.env();release.deploy(self.sitepath,a,h,apply=True);old=release.active(self.site) a,h=self.archive('v2','v1','CONFIG');release.deploy(self.sitepath,a,h,apply=True) Path(self.site['env_files'][0]).write_text('EXAMPLE=new\n') with self.assertRaises(KitError) as c:release.rollback(self.sitepath,old['transaction'],apply=True) self.assertEqual(c.exception.code,'CONFIG_CHANGED') def test_activation_proposal_survives_before_current_switch(self): a,h=self.archive();self.env() with patch.object(release,'switch_current',side_effect=OSError('fixture')): with self.assertRaises(OSError):release.deploy(self.sitepath,a,h,apply=True) j=json.loads((release.state_dir(self.site)/'inflight.json').read_text()) self.assertTrue((release.state_dir(self.site)/'transactions'/j['transaction']/'activation-proposal.json').is_file()) def test_current_mismatch_is_not_silently_fixed(self): root=Path(self.site['install_root']);(root/'current').symlink_to('releases/missing') with self.assertRaises(KitError):release.active(self.site) def test_persistent_mapping_covers_runtime_and_artifact(self): doc={'services':{'analytics':{'volumes':[{'type':'bind','source':'/old','target':'/data/contextra/artifacts'}]}}} before=release.persistent_bindings(doc);doc['services']['analytics']['volumes'][0]['source']='/new' self.assertNotEqual(before,release.persistent_bindings(doc)) def test_legacy_offline_builder_flags(self): cmd=release.app_patch_command('base','tag',Path('/tmp/Dockerfile'),Path('/tmp/context')) self.assertIn('DOCKER_BUILDKIT=0',cmd);self.assertIn('--network=none',cmd);self.assertIn('--pull=false',cmd) def test_config_cannot_hide_app_change(self): m={'kind':'CONFIG','parent_release':'v1','compatibility_id':'c','build_contract':'x','images':[{'service':'analytics','id':IID}], 'files':{'source/app/x.py':{'sha256':'new'}}} old={'release_id':'v1','compatibility_id':'c','build_contract':'x','images':{'analytics':IID}} with self.assertRaises(KitError):release.check_patch(m,old,{'files':{'source/app/x.py':{'sha256':'old'}}})