"""Release boundary checks: configuration identity, signing grants and feed order."""
import base64
import hashlib
import json
from pathlib import Path
import plistlib
import subprocess
import sys
import tempfile
import unittest
import zipfile
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from mac_updates import configure, read_configuration, sign_bundle
from publish_mac_update import check_build, publish, verify_download


class MacUpdatesTests(unittest.TestCase):
    def setUp(self):
        self.temporary = tempfile.TemporaryDirectory()
        self.addCleanup(self.temporary.cleanup)
        self.root = Path(self.temporary.name)
        self.app = self.root / 'Meet.app'
        (self.app / 'Contents').mkdir(parents=True)
        self.info = self.app / 'Contents/Info.plist'
        self.info.write_bytes(plistlib.dumps(dict(CFBundleIdentifier='net.amadan.oficina.meet', NSCameraUsageDescription='Camera', NSMicrophoneUsageDescription='Microphone')))
        self.config = self.root / 'updates.json'
        self.values = dict(bundle_id='net.amadan.oficina.meet', feed_url='https://raw.amadan.net/oficina/native-releases/main/meet/appcast.xml', public_ed_key=base64.b64encode(bytes(32)).decode(), keychain_account='oficina-test')
        self.config.write_text(json.dumps(self.values))

    def test_configuration_rejects_cross_app_and_insecure_feeds(self):
        for field, value in [('bundle_id', 'net.amadan.oficina.docs'), ('feed_url', 'http://example.com/feed'), ('feed_url', 'https://user:secret@example.com/feed'), ('public_ed_key', base64.b64encode(bytes(31)).decode())]:
            with self.subTest(field=field, value=value):
                self.config.write_text(json.dumps(self.values | {field: value}))
                with self.assertRaises(ValueError):
                    configure(self.app, self.config)

    def test_dev_build_removes_feed_without_forcing_preferences(self):
        configure(self.app, self.config, '1.2.3', '2')
        info = plistlib.loads(self.info.read_bytes())
        self.assertTrue(info['SUVerifyUpdateBeforeExtraction'])
        self.assertFalse(info['SUAllowsAutomaticUpdates'])
        self.assertNotIn('SUEnableAutomaticChecks', info)
        configure(self.app)
        self.assertNotIn('SUFeedURL', plistlib.loads(self.info.read_bytes()))

    def test_automatic_install_mode_sets_defaults_a_person_can_override(self):
        self.config.write_text(json.dumps(self.values | {'install': 'automatic'}))
        configure(self.app, self.config, '1.2.3', '2')
        info = plistlib.loads(self.info.read_bytes())
        # Info.plist values are defaults: Sparkle reads a stored preference first.
        self.assertTrue(info['SUAllowsAutomaticUpdates'])
        self.assertTrue(info['SUEnableAutomaticChecks'])
        self.assertTrue(info['SUAutomaticallyUpdate'])
        self.assertEqual(info['SUScheduledCheckInterval'], 86400)
        # Rebuilding in the default mode leaves none of them behind.
        self.config.write_text(json.dumps(self.values))
        configure(self.app, self.config)
        info = plistlib.loads(self.info.read_bytes())
        self.assertFalse(info['SUAllowsAutomaticUpdates'])
        for key in ['SUEnableAutomaticChecks', 'SUAutomaticallyUpdate', 'SUScheduledCheckInterval']:
            self.assertNotIn(key, info)

    def test_unknown_install_mode_is_rejected(self):
        self.config.write_text(json.dumps(self.values | {'install': 'silent'}))
        with self.assertRaises(ValueError):
            configure(self.app, self.config)

    def test_signing_keeps_device_grants_on_host_only(self):
        current = self.app / 'Contents/Frameworks/Sparkle.framework/Versions/Current'
        for path in ['XPCServices/Downloader.xpc', 'XPCServices/Installer.xpc', 'Autoupdate', 'Updater.app']:
            (current / path).mkdir(parents=True)
        calls = []
        entitlements = []
        def capture(command, **kwargs):
            calls.append(command)
            if '--entitlements' in command:
                entitlements.append(plistlib.loads(Path(command[command.index('--entitlements') + 1]).read_bytes()))
        with patch('mac_updates.subprocess.run', side_effect=capture):
            sign_bundle(self.app, 'Developer ID Application: Test')
        self.assertEqual(entitlements, [{'com.apple.security.device.camera': True, 'com.apple.security.device.audio-input': True}])
        self.assertEqual(calls[-2][-1], str(self.app))
        self.assertEqual(calls[-3][-1], str(self.app / 'Contents/Frameworks/Sparkle.framework'))
        self.assertIn('--preserve-metadata=entitlements', calls[0])
        self.assertIn('runtime', calls[-2])
        self.assertEqual(calls[-1][1:5], ['--verify', '--deep', '--strict', str(self.app)])

    def test_rejects_non_increasing_builds_in_current_and_old_appcast_formats(self):
        feed = self.root / 'appcast.xml'
        for version in ['<sparkle:version>7</sparkle:version>', '<enclosure sparkle:version="7"/>']:
            feed.write_text('<rss xmlns:sparkle="http://www.andymatuschak.org/xml-namespaces/sparkle"><channel><item>' + version + '</item></channel></rss>')
            for build in ['6', '7', 'invalid']:
                with self.assertRaises(ValueError):
                    check_build(feed, build)
            check_build(feed, '8')

    def test_public_archive_must_match_exact_signed_bytes(self):
        import hashlib
        import io
        def response(*args, **kwargs):
            stream = io.BytesIO(b'tampered')
            stream.url = 'https://raw.amadan.net/oficina/native-releases/commit/archive.zip'
            return stream
        with patch('publish_mac_update.urlopen', side_effect=response), patch('publish_mac_update.time.sleep'):
            with self.assertRaises(ValueError):
                verify_download('https://raw.amadan.net/archive.zip', 8, hashlib.sha256(b'original').hexdigest())


class PublicationBoundaryTests(unittest.TestCase):
    """Exercise publication failures without Git, Keychain, signing or network access."""

    def setUp(self):
        temporary = tempfile.TemporaryDirectory(prefix='sparkle-publication-test-')
        self.addCleanup(temporary.cleanup)
        self.root = Path(temporary.name)
        self.release = self.root / 'release'
        self.release.mkdir()
        self.repo = self.root / 'repository'
        (self.repo / 'meet').mkdir(parents=True)
        self.feed = self.repo / 'meet/appcast.xml'
        self.original_feed = '<rss xmlns:sparkle="http://www.andymatuschak.org/xml-namespaces/sparkle"><channel><item><sparkle:version>2</sparkle:version></item></channel></rss>'
        self.feed.write_text(self.original_feed)
        self.config = self.root / 'config.json'
        self.values = dict(bundle_id='net.amadan.oficina.meet', feed_url='https://raw.amadan.net/oficina/native-releases/main/meet/appcast.xml', public_ed_key=base64.b64encode(bytes(32)).decode(), keychain_account='test-only')
        self.config.write_text(json.dumps(self.values))
        self.archive = self.release / 'meet-3.zip'
        self.write_archive(SUAllowsAutomaticUpdates=False)
        self.manifest = dict(bundle_id=self.values['bundle_id'], feed_url=self.values['feed_url'], build='3', version='1.0', archive=self.archive.name, length=self.archive.stat().st_size, sha256=hashlib.sha256(self.archive.read_bytes()).hexdigest(), signature='test-signature')
        self.write_manifest()
        self.events = []
        self.generated_build = '3'
        self.download_error = None
        self.upload_returncode = 0
        self.existing_tag = ''
        self.downloads = []
        self.addCleanup(patch.stopall)
        patch('publish_mac_update.git', side_effect=self.git).start()
        patch('publish_mac_update.subprocess.check_output', return_value=self.values['public_ed_key']).start()
        patch('publish_mac_update.subprocess.run', side_effect=self.tool).start()
        patch('publish_mac_update.verify_download', side_effect=self.download).start()

    def write_archive(self, **install):
        info = dict(CFBundleIdentifier=self.values['bundle_id'], SUFeedURL=self.values['feed_url'], SUPublicEDKey=self.values['public_ed_key'], CFBundleVersion='3', CFBundleShortVersionString='1.0', **install)
        with zipfile.ZipFile(self.archive, 'w') as zipped:
            zipped.writestr('Meet.app/Contents/Info.plist', plistlib.dumps(info))

    def write_manifest(self):
        (self.release / 'release.json').write_text(json.dumps(self.manifest))

    def git(self, repo, *args):
        self.events.append(('git', *args))
        if args[:3] == ('remote', 'get-url', '--push'):
            return 'https://amadan.net/oficina/native-releases'
        if args[0] == 'branch':
            return 'main'
        if args[0] == 'rev-parse':
            return 'a' * 40
        if args[:2] == ('tag', '--list'):
            return self.existing_tag
        return ''

    def tool(self, command, **kwargs):
        tool = Path(command[0]).name
        self.events.append((tool, *command[1:3]) if tool == 'amadan' else (tool,))
        if tool == 'amadan':
            return subprocess.CompletedProcess(command, self.upload_returncode)
        if tool != 'generate_appcast':
            return
        prefix = command[command.index('--download-url-prefix') + 1]
        output = Path(command[command.index('-o') + 1])
        output.write_text(f'<rss xmlns:sparkle="http://www.andymatuschak.org/xml-namespaces/sparkle"><channel><item><sparkle:version>{self.generated_build}</sparkle:version><enclosure url="{prefix}{self.archive.name}" sparkle:edSignature="test-signature" length="{self.manifest["length"]}"/></item></channel></rss>')

    def download(self, *args):
        self.downloads.append(args)
        self.events.append(('verify_download',))
        if self.download_error:
            raise self.download_error

    def publish(self, enabled=True):
        return publish(self.release, self.config, self.repo, self.root / 'tools', enabled)

    def pushes(self):
        return [event for event in self.events if event[:2] == ('git', 'push')]

    def feed_pushes(self):
        return [event for event in self.pushes() if event[-1] == 'HEAD:refs/heads/main']

    def test_manifest_cannot_claim_a_newer_build_than_the_signed_archive(self):
        self.manifest['build'] = '4'
        self.write_manifest()
        with self.assertRaisesRegex(ValueError, 'metadata does not match'):
            self.publish()
        self.assertEqual(self.events, [])
        self.assertFalse((self.repo / 'meet/releases').exists())

    def test_archive_built_for_another_install_mode_is_rejected(self):
        # A quiet app's feed must not ship a build that asks, nor the reverse.
        self.config.write_text(json.dumps(self.values | {'install': 'automatic'}))
        with self.assertRaisesRegex(ValueError, 'metadata does not match'):
            self.publish()
        self.assertEqual(self.events, [])

    def test_automatic_install_mode_archive_publishes(self):
        self.config.write_text(json.dumps(self.values | {'install': 'automatic'}))
        self.write_archive(SUAllowsAutomaticUpdates=True, SUEnableAutomaticChecks=True, SUAutomaticallyUpdate=True, SUScheduledCheckInterval=86400)
        self.manifest.update(length=self.archive.stat().st_size, sha256=hashlib.sha256(self.archive.read_bytes()).hexdigest())
        self.write_manifest()
        self.assertEqual(self.publish(enabled=False)['build'], '3')

    def test_changed_archive_is_rejected_before_publication(self):
        with self.archive.open('ab') as archive:
            archive.write(b'changed after signing')
        with self.assertRaisesRegex(ValueError, 'archive changed'):
            self.publish()
        self.assertEqual(self.events, [])

    def test_plan_does_not_commit_push_or_copy_artifacts(self):
        plan = self.publish(enabled=False)
        self.assertEqual(plan['build'], '3')
        self.assertFalse((self.repo / 'meet/releases').exists())
        self.assertEqual(self.feed.read_text(), self.original_feed)
        self.assertFalse(any(event[0] == 'git' and event[1] in ('fetch', 'tag', 'add', 'commit', 'push') for event in self.events))
        self.assertNotIn(('amadan', 'release', 'create'), self.events)
        self.assertEqual(plan['download'], 'https://amadan.net/oficina/native-releases/-/releases/meet-3/meet-3.zip')

    def test_failed_public_download_leaves_existing_feed_untouched(self):
        self.download_error = ValueError('download checksum mismatch')
        with self.assertRaisesRegex(ValueError, 'checksum mismatch'):
            self.publish()
        self.assertEqual(self.feed_pushes(), [])
        self.assertNotIn(('generate_appcast',), self.events)
        self.assertEqual(self.feed.read_text(), self.original_feed)

    def test_generated_feed_with_wrong_version_is_never_published(self):
        self.generated_build = '4'
        with self.assertRaisesRegex(ValueError, 'build does not match'):
            self.publish()
        self.assertEqual(self.feed_pushes(), [])
        self.assertEqual(self.feed.read_text(), self.original_feed)

    def test_archive_is_publicly_verified_before_feed_publication(self):
        self.publish()
        tag = self.events.index(('git', 'push', 'origin', 'refs/tags/meet-3'))
        uploaded = self.events.index(('amadan', 'release', 'create'))
        verified = self.events.index(('verify_download',))
        generated = self.events.index(('generate_appcast',))
        feed = self.events.index(('git', 'push', 'origin', 'HEAD:refs/heads/main'))
        self.assertLess(tag, uploaded)
        self.assertLess(uploaded, verified)
        self.assertLess(verified, generated)
        self.assertLess(generated, feed)

    def test_archive_is_never_committed_to_git(self):
        # native-releases#1: zips in Git history took amadan.net down.
        self.publish()
        added = [event for event in self.events if event[:2] == ('git', 'add')]
        self.assertEqual(added, [('git', 'add', '--', 'meet/appcast.xml')])
        self.assertFalse((self.repo / 'meet/releases').exists())

    def test_verification_checks_the_release_asset_against_the_signed_bytes(self):
        self.publish()
        self.assertEqual(self.downloads, [('https://amadan.net/oficina/native-releases/-/releases/meet-3/meet-3.zip', self.manifest['length'], self.manifest['sha256'])])

    def test_first_release_of_an_app_creates_its_feed(self):
        self.feed.unlink()
        self.feed.parent.rmdir()
        self.publish()
        self.assertIn('<sparkle:version>3</sparkle:version>', self.feed.read_text())

    def test_rerun_reuses_the_tag_an_earlier_run_pushed(self):
        self.existing_tag = 'meet-3'
        self.publish()
        self.assertFalse(any(event[:2] == ('git', 'tag') and event[2] != '--list' for event in self.events))
        self.assertIn(('git', 'push', 'origin', 'refs/tags/meet-3'), self.events)

    def test_rerun_accepts_refused_upload_only_when_asset_matches(self):
        # An asset cannot be replaced, so a rerun's upload fails; the public bytes decide.
        self.upload_returncode = 1
        self.assertEqual(self.publish()['build'], '3')
        self.assertEqual(len(self.feed_pushes()), 1)

    def test_failed_upload_without_a_matching_asset_publishes_nothing(self):
        self.upload_returncode = 1
        self.download_error = OSError('HTTP Error 404: Not Found')
        with self.assertRaises(OSError):
            self.publish()
        self.assertEqual(self.feed_pushes(), [])
        self.assertEqual(self.feed.read_text(), self.original_feed)


if __name__ == '__main__':
    unittest.main()