"""Offline candidate contracts; no GPU execution or claims about model quality.""" import copy import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from deployment_kit import models from deployment_kit.core import KitError, canonical, sha256 from helpers import IID def candidate(root,name='medium'): root.mkdir(); p=models.profile(name) (root/'model.plan').write_bytes(b'fixture-not-an-engine') (root/'config.pbtxt').write_text(models.pbtxt(p)); (root/'profile.json').write_bytes(canonical(p)) m={'status':'CONTRACT_PASS','profile':p,'engine_sha256':sha256(root/'model.plan'), 'config_sha256':sha256(root/'config.pbtxt'),'smoke':{'shape_dtype_finite':'PASS'}} (root/'manifest.json').write_bytes(canonical(m)); return m class ModelTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup);self.root=Path(self.tmp.name) def test_all_profiles(self): for name in models.PROFILES: with self.subTest(name=name): p=models.profile(name); self.assertEqual((p['instances'],p['batch'],p['resolution']),(8,4,512 if name=='crowd' else 576)) c=models.pbtxt(p) self.assertIn('TYPE_FP16',c);self.assertIn('TYPE_FP32',c);self.assertIn('max_batch_size: 4',c) self.assertIn('count: 8',c);self.assertIn('5000',c) def test_output_contracts(self): self.assertEqual(models.profile('pose')['outputs']['keypoints'],[100,34,8]) self.assertEqual(models.profile('fire')['outputs']['labels'],[300,3]) def test_options_reject_bool_and_invalid(self): for kw in ({'instances':True},{'batch':0},{'workspace':0}): with self.subTest(kw=kw),self.assertRaises(KitError):models.profile('medium',**kw) def test_profile_returns_independent_data(self): p=models.profile('medium'); p['outputs']['dets'][0]=17 self.assertEqual(models.profile('medium')['outputs']['dets'],[300,4]) def test_trtexec_io_options(self): a=models.trtexec_args(models.profile('pose')) self.assertIn('--inputIOFormats=fp16:chw',a) self.assertIn('--outputIOFormats=fp32:chw,fp32:chw,fp32:chw',a) self.assertIn('--maxShapes=input:4x3x576x576',a) def test_build_is_plan_only(self): onnx=self.root/'source.onnx';onnx.write_bytes(b'fake') with patch.object(models,'docker',side_effect=AssertionError('docker')): p=models.build('medium',onnx,self.root/'out',IID,allow_new_onnx=True) self.assertFalse((self.root/'out').exists());self.assertEqual(p['network'],'none') def test_unknown_lineage_blocked(self): p=self.root/'s.onnx';p.write_bytes(b'fake') with self.assertRaises(KitError):models.build('medium',p,self.root/'out',IID) def test_candidate_install_plan(self): d=self.root/'candidate';candidate(d) p=models.install_candidate(d,self.root/'repository') self.assertFalse(p['live_load']);self.assertFalse((self.root/'repository').exists()) def test_explicit_install_version_preserves_candidate(self): d=self.root/'candidate';before=candidate(d) with patch.object(models.os,'geteuid',return_value=0): models.install_candidate(d,self.root/'repository',apply=True,model_version=8) target=self.root/'repository'/'RFDETRMedium' self.assertEqual(sha256(target/'8/model.plan'),before['engine_sha256']) self.assertFalse((target/'1').exists()) self.assertIn('versions: [8]',(target/'config.pbtxt').read_text()) receipt=json.loads((target/'deployment-manifest.json').read_text()) self.assertEqual(receipt['model_version'],8) self.assertEqual(receipt['config_sha256'],sha256(target/'config.pbtxt')) self.assertEqual(models.candidate_manifest(d),before) with self.assertRaises(KitError): models.install_candidate(d,self.root/'other',model_version='../8') def test_candidate_profile_path_tamper_blocked(self): d=self.root/'candidate';m=candidate(d);m['profile']['name']='../../escape' (d/'manifest.json').write_bytes(canonical(m)) with self.assertRaises(KitError):models.install_candidate(d,self.root/'repository') def test_profile_file_mismatch_blocked(self): d=self.root/'candidate';candidate(d) (d/'profile.json').write_text('{}') with self.assertRaises(KitError):models.verify_candidate(d,IID) def test_config_semantics_tamper_blocked(self): d=self.root/'candidate';m=candidate(d) (d/'config.pbtxt').write_text('name: "other"') m['config_sha256']=sha256(d/'config.pbtxt');(d/'manifest.json').write_bytes(canonical(m)) with self.assertRaises(KitError):models.install_candidate(d,self.root/'repository') def test_smoke_is_required(self): d=self.root/'candidate';m=candidate(d);m['smoke']='NOT_RUN';(d/'manifest.json').write_bytes(canonical(m)) with self.assertRaises(KitError):models.install_candidate(d,self.root/'repository') def test_mode_reads_actual_process(self): with patch.object(models,'docker',return_value=(0,json.dumps([['tritonserver']]))): self.assertEqual(models.control_mode('fixture')['mode'],'NONE') with patch.object(models,'docker',return_value=(0,json.dumps([['tritonserver','--model-control-mode=explicit']]))): self.assertEqual(models.control_mode('fixture')['mode'],'EXPLICIT') def test_ambiguous_process_mode_blocked(self): with patch.object(models,'docker',return_value=(0,'[]')),self.assertRaises(KitError):models.control_mode('fixture') def test_none_does_not_post_load(self): with patch.object(models,'control_mode',return_value={'mode':'NONE'}),self.assertRaises(KitError):models.load_model('fixture','medium') def test_cleanup_owner_check(self): response={'Name':'/other','Id':'a'*64,'Config':{'Labels':{}}} with patch.object(models,'docker',return_value=(0,json.dumps(response))) as d,self.assertRaises(KitError):models.cleanup_helper('my-helper','a'*64) self.assertEqual(d.call_count,1) def test_cleanup_does_not_hide_daemon_failure(self): with patch.object(models,'docker',return_value=(1,'Cannot connect to the Docker daemon')),self.assertRaises(KitError):models.cleanup_helper('my-helper') def test_cleanup_confirmed_missing(self): with patch.object(models,'docker',return_value=(1,'Error: No such object: my-helper')): models.cleanup_helper('my-helper') def test_helper_network_bounded(self): a=models._helper_args(IID,'fixture',[],0) self.assertEqual(a[a.index('--network')+1],'none');self.assertEqual(a[a.index('--pull')+1],'never') self.assertIn('--memory',a);self.assertNotIn('--privileged',a) def test_helper_blas_does_not_exhaust_triton_pid_budget(self): for gpu in (None,0): with self.subTest(gpu=gpu): a=models._helper_args(IID,'fixture',[],gpu) env={a[i+1] for i,value in enumerate(a) if value=='--env'} self.assertEqual(env,{'OPENBLAS_NUM_THREADS=1','OMP_NUM_THREADS=1','MKL_NUM_THREADS=1'}) self.assertEqual(a[a.index('--pids-limit')+1],'128') self.assertEqual(a[a.index('--cpus')+1],'4') self.assertEqual(a[a.index('--memory')+1],'12g') self.assertEqual(a[a.index('--cap-drop')+1],'ALL') self.assertEqual(a[-3:],['--entrypoint','python3',IID])