"""Real archive/Git/file tests; no external services."""
import copy
import hashlib
import io
import json
from pathlib import Path
import tarfile
import tempfile
import unittest
from unittest.mock import patch
from deployment_kit import bundle
from deployment_kit.core import KitError, canonical, sha256
from helpers import mini_repo, spec, git, IID

class BundleTests(unittest.TestCase):
    def setUp(self):
        self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
        self.root=Path(self.tmp.name);self.repo,self.sha=mini_repo(self.root/'repo')
        self.spec=self.root/'spec.json';self.spec.write_bytes(canonical(spec(self.sha)))
        self.out=self.root/'release.tar.gz'
    def create(self): return bundle.create(self.repo,self.spec,self.out)
    def test_allowlist_current_components(self):
        for p in ('docker/secure-gateway-entrypoint.sh','deploy/secure-bridge.compose.yml','deploy/web-start.compose.yml','configs/control_templates/medium.json'):
            self.assertTrue(bundle.allowed_source(p),p)
    def test_deny_developer_and_hidden_files(self):
        for p in ('.git/config','AGENTS.md','docs/profiles/private.md','docs/deployment-review/HANDOFF.ko.md','tools/scratch.py','app/.env','app/__pycache__/x.pyc'):
            self.assertFalse(bundle.allowed_source(p),p)
    def test_real_archive_roundtrip_and_modes(self):
        self.create();m=bundle.verify(self.out,expected_hash=sha256(self.out),extract_to=self.root/'unpacked')
        self.assertEqual(self.out.read_bytes()[8],4)  # gzip XFL: fastest compression for FULL
        self.assertEqual(bundle.verify_directory(self.root/'unpacked'),m)
        self.assertEqual((self.root/'unpacked/deployctl').stat().st_mode&0o777,0o755)
        self.assertNotIn('source/docs/profiles/ltb1021/private.md',m['files'])
    def test_deterministic_bytes(self):
        self.create();other=self.root/'other.tar.gz';bundle.create(self.repo,self.spec,other)
        self.assertEqual(sha256(self.out),sha256(other))
    def test_untracked_source_is_not_packed(self):
        (self.repo/'app/private.json').write_text('{"secret":"untracked"}')
        self.create();self.assertNotIn('source/app/private.json',bundle.verify(self.out)['files'])
    def test_dirty_committed_source_blocked(self):
        (self.repo/'app/example.py').write_text('changed')
        with self.assertRaises(KitError) as c:self.create()
        self.assertEqual(c.exception.code,'SOURCE_DIRTY')
    def test_symlink_source_blocked(self):
        (self.repo/'app/example.py').unlink();(self.repo/'app/example.py').symlink_to(self.repo/'deployctl')
        with self.assertRaises(KitError):self.create()
    def test_wrong_external_hash(self):
        self.create()
        with self.assertRaises(KitError) as c:bundle.verify(self.out,expected_hash='0'*64)
        self.assertEqual(c.exception.code,'BUNDLE_HASH')
    def test_refuses_overwriting_archive(self):
        self.create()
        with self.assertRaises(KitError):self.create()
    def test_private_metadata_field_blocked(self):
        d=spec(self.sha);d['token']='secret';self.spec.write_bytes(canonical(d))
        with self.assertRaises(KitError):self.create()
    def test_manifest_extra_file_blocked(self):
        self.create();bundle.verify(self.out,extract_to=self.root/'x')
        (self.root/'x/unlisted').write_text('bad')
        with self.assertRaises(KitError):bundle.verify_directory(self.root/'x')
    def test_extracted_content_tamper(self):
        self.create();bundle.verify(self.out,extract_to=self.root/'x')
        (self.root/'x/source/app/example.py').write_text('bad')
        with self.assertRaises(KitError):bundle.verify_directory(self.root/'x')
    def test_extracted_permissions_tamper(self):
        self.create();bundle.verify(self.out,extract_to=self.root/'x')
        (self.root/'x/deployctl').chmod(0o777)
        with self.assertRaises(KitError):bundle.verify_directory(self.root/'x')
    def tamper_archive(self,transform):
        self.create()
        with tarfile.open(self.out,'r:gz') as tar:
            items=[(m,tar.extractfile(m).read()) for m in tar.getmembers()]
        items=transform(items);bad=self.root/'bad.tar'
        with tarfile.open(bad,'w') as tar:
            for m,b in items:tar.addfile(m,io.BytesIO(b) if m.isfile() else None)
        return bad
    def test_missing_file(self):
        bad=self.tamper_archive(lambda x:x[:-1])
        with self.assertRaises(KitError):bundle.verify(bad)
    def test_duplicate_member(self):
        bad=self.tamper_archive(lambda x:x+[x[-1]])
        with self.assertRaises(KitError):bundle.verify(bad)
    def test_traversal_member(self):
        def f(items):
            m=tarfile.TarInfo('../escape');m.size=1;return items+[(m,b'x')]
        with self.assertRaises(KitError):bundle.verify(self.tamper_archive(f),extract_to=self.root/'x')
        self.assertFalse((self.root/'escape').exists())
    def test_symlink_member(self):
        def f(items):
            m=tarfile.TarInfo('link');m.type=tarfile.SYMTYPE;m.linkname='/etc';return items+[(m,b'')]
        with self.assertRaises(KitError):bundle.verify(self.tamper_archive(f))
    def test_file_checksum_mismatch(self):
        def f(items):
            m,b=items[-1];return items[:-1]+[(m,b'x'*len(b))]
        with self.assertRaises(KitError):bundle.verify(self.tamper_archive(f))
    def test_not_a_tar(self):
        self.out.write_bytes(b'not tar')
        with self.assertRaises(KitError):bundle.verify(self.out)
    def test_app_contract_change_requires_full(self):
        from deployment_kit.core import read_json
        d=spec(self.sha,'APP');d.update(parent_release='v0',app_patch={'service':'analytics','base_id':IID,'base_contract':'0'*64})
        self.spec.write_bytes(canonical(d))
        with self.assertRaises(KitError) as c:self.create()
        self.assertEqual(c.exception.code,'FULL_BUNDLE_REQUIRED')
    def test_filename_directory_conflict(self):
        self.create();m=bundle.verify(self.out);m['files']['source']=m['files']['deployctl']
        with self.assertRaises(KitError):bundle.validate_manifest(m)