"""Offline protection checks for the host installer. No deployed credentials or network."""
import contextlib
import io
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch

from tools.secure_bridge_provision import check, environment, install_ca, main
from tools.secure_bridge_tls import TLSError, command, prepare


class ProvisionTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.work = tempfile.TemporaryDirectory(prefix='provision-fixture-')
        cls.root = Path(cls.work.name)
        cls.ca, cls.key = cls.root / 'ca.crt', cls.root / 'ca.key'
        command('req', '-x509', '-newkey', 'rsa:2048', '-nodes',
                '-keyout', str(cls.key), '-out', str(cls.ca), '-days', '1',
                '-subj', '/CN=offline-test-only', '-addext', 'basicConstraints=critical,CA:TRUE')

    @classmethod
    def tearDownClass(cls):
        cls.work.cleanup()

    def setUp(self):
        self.work = tempfile.TemporaryDirectory(dir=self.root)
        self.addCleanup(self.work.cleanup)
        self.path = Path(self.work.name)

    def test_missing_inputs_do_not_create_files_or_run_commands(self):
        with patch('tools.secure_bridge_provision.command') as run:
            result = check({})
        self.assertEqual(result['status'], 'INPUT_REQUIRED')
        self.assertFalse(result['network_tested'])
        run.assert_not_called()
        self.assertEqual(list(self.path.iterdir()), [])

    def test_literal_env_and_duplicate_protection(self):
        env = self.path / 'private.env'
        env.write_text('AI_CONTROL_SERVER_ID="AI-01"\n')
        env.chmod(0o600)
        self.assertEqual(environment(env)['AI_CONTROL_SERVER_ID'], 'AI-01')
        for text in ('A=1\nA=2\n', 'A=$(id)\n', 'A="unclosed\n'):
            env.write_text(text)
            with self.assertRaises(ValueError):
                environment(env)

    def test_install_is_atomic_no_overwrite_and_idempotent(self):
        self.assertEqual(install_ca(self.ca, self.path), 'CA_INSTALLED')
        target = self.path / 'manager-ca.crt'
        before = target.stat().st_ino, target.read_bytes()
        self.assertEqual(install_ca(self.ca, self.path), 'CA_UNCHANGED')
        self.assertEqual(before, (target.stat().st_ino, target.read_bytes()))
        target.write_text('existing trust must not be overwritten')
        with self.assertRaisesRegex(ValueError, 'EXISTING_CA_DIFFERS'):
            install_ca(self.ca, self.path)
        self.assertEqual(target.read_text(), 'existing trust must not be overwritten')
        self.assertEqual(list(self.path.glob('.manager-ca-*')), [])

    def test_private_key_and_symlink_rejected(self):
        with self.assertRaises(TLSError):
            install_ca(self.key, self.path)
        link = self.path / 'linked.crt'
        link.symlink_to(self.ca)
        with self.assertRaises(TLSError):
            install_ca(link, self.path)
        target = self.path / 'manager-ca.crt'
        target.symlink_to(self.ca)
        with self.assertRaises(TLSError):
            install_ca(self.ca, self.path)

    def test_concurrent_publish_never_replaces_winner(self):
        target = self.path / 'manager-ca.crt'
        def race(source, destination):
            target.write_text('another installer won')
            raise FileExistsError()
        with patch('tools.secure_bridge_provision.os.link', side_effect=race):
            with self.assertRaises(FileExistsError):
                install_ca(self.ca, self.path)
        self.assertEqual(target.read_text(), 'another installer won')
        self.assertEqual(list(self.path.glob('.manager-ca-*')), [])

    def test_cli_redacts_invalid_input_and_does_not_claim_login(self):
        env = self.path / 'private.env'
        env.write_text('PASSWORD=never-show-this\nPASSWORD=invalid\n')
        env.chmod(0o600)
        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            self.assertEqual(main(['check', '--env-file', str(env)]), 2)
        self.assertNotIn('never-show-this', output.getvalue())
        self.assertEqual(json.loads(output.getvalue())['status'], 'INPUT_INVALID')

    def test_complete_offline_contract_and_read_permissions(self):
        secret_root, tls_root = self.path / 'bridge', self.path / 'tls'
        secret_root.mkdir(mode=0o700)
        tls_root.mkdir(mode=0o700)
        install_ca(self.ca, secret_root)
        _, csr = prepare(tls_root, '127.0.0.1')
        ext = self.path / 'leaf.ext'
        ext.write_text('subjectAltName=IP:127.0.0.1\nextendedKeyUsage=serverAuth\nbasicConstraints=critical,CA:FALSE\n')
        command('x509', '-req', '-in', str(csr), '-CA', str(self.ca), '-CAkey', str(self.key),
                '-set_serial', '2', '-out', str(tls_root / 'control-tls.crt'), '-days', '1',
                '-extfile', str(ext))
        body = dict(loginId='offline-test', controlPort=9443,
                    advertisedIp='127.0.0.1', bboxStreamKey='fixture', connectionConfigVersion=1,
                    aiControlCa=self.ca.read_text(), redis=dict(host='fixture', port=6379, db=0,
                                               username=None, password=None, tls=False))
        for name, value in (('login-body.json', json.dumps(body)), ('password', 'offline-only'),
                            ('control-token', 'fixture-control-000000'), ('relay-token', 'fixture-relay-00000000')):
            p = secret_root / name
            p.write_text(value)
            p.chmod(0o600)
        values = dict(AI_CONTROL_MANAGER_BASE_URL='https://manager.example.invalid',
                      AI_CONTROL_SERVER_ID='AI-test', AI_CONTROL_HOST_PORT='9443',
                      AI_BRIDGE_SECRET_ROOT=str(secret_root), AI_BRIDGE_TLS_ROOT=str(tls_root),
                      CONTEXTRA_ARTIFACT_PUBLIC_BASE='https://127.0.0.1:9443/artifacts')
        # Fixture ownership is not a real Sidecar UID; exercise both outcomes explicitly.
        with patch('tools.secure_bridge_provision.readable_by_sidecar', return_value=False):
            with self.assertRaisesRegex(ValueError, 'UID10001'):
                check(values)
        with patch('tools.secure_bridge_provision.readable_by_sidecar', return_value=True):
            self.assertEqual(check(values)['status'], 'OFFLINE_INPUTS_VALID')
            values['AI_CONTROL_HOST_PORT'] = '9444'
            with self.assertRaisesRegex(ValueError, 'PUBLIC_ADDRESS'):
                check(values)


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