"""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_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)