import importlib.util
import json
import os
from pathlib import Path
import stat
import tempfile
import unittest

spec = importlib.util.spec_from_file_location("settings_guard", Path(__file__).with_name("check-settings.py"))
guard = importlib.util.module_from_spec(spec)
spec.loader.exec_module(guard)

SAFE = {"localAuthBypassEnabled": False, "apiRestricted": True, "autoUpdateEnabled": False}
KEY = "ab" * 32


class SettingsTests(unittest.TestCase):
    def test_safe_escaped_names_are_semantically_validated(self):
        raw = b'{"localAuthBypass\\u0045nabled":false,"api\\u0052estricted":true,"autoUpdateEnabled":false}'
        self.assertEqual(guard.validate(raw), SAFE)

    def test_unsafe_flags_defaults_and_wrong_types_refused(self):
        for field, bad_values in {
            "localAuthBypassEnabled": [True, None, 0, 1, "false"],
            "apiRestricted": [False, None, 0, 1, "true"],
            "autoUpdateEnabled": [True, None, 0, 1, "false"],
        }.items():
            for value in bad_values:
                with self.subTest(field=field, value=value):
                    with self.assertRaises(guard.UnsafeSettings):
                        guard.validate(json.dumps({**SAFE, field: value}).encode())
            absent = dict(SAFE)
            del absent[field]
            with self.assertRaises(guard.UnsafeSettings):
                guard.validate(json.dumps(absent).encode())

    def test_original_escaped_bypasses_refused(self):
        for raw in [
            b'{"localAuthBypass\\u0045nabled":true,"apiRestricted":true,"autoUpdateEnabled":false}',
            b'{"localAuthBypassEnabled":false,"api\\u0052estricted":false,"autoUpdateEnabled":false}',
        ]:
            with self.assertRaises(guard.UnsafeSettings):
                guard.validate(raw)

    def test_redirect_and_legacy_key_properties_refused_even_null(self):
        for key in ("userPath", "apiKey", "apiKeyPath"):
            for value in (None, "", "/another-location"):
                with self.assertRaises(guard.UnsafeSettings):
                    guard.validate(json.dumps({**SAFE, key: value}).encode())

    def test_ambiguous_non_json_and_oversized_inputs_refused(self):
        for raw in [b'{}', b'[]', b'null', b'{"x":NaN}', b'{"x":Infinity}',
                    b'{"x":1,"\\u0078":2}', b'{"x":{},"x":{}}',
                    b'{"x":1,}', b'# comment\n{}', b'{}{}', b'\xff',
                    b' ' * (guard.MAX_SETTINGS_BYTES + 1)]:
            with self.subTest(raw=raw[:40]):
                with self.assertRaises(guard.UnsafeSettings):
                    guard.validate(raw)

    def test_snapshot_is_private_independent_and_preserves_original(self):
        with tempfile.TemporaryDirectory() as tmp:
            source, dest = Path(tmp) / "operator.json", Path(tmp) / "runtime.json"
            original = json.dumps({**SAFE, "maxStorageCapacity": 21474836480}).encode()
            source.write_bytes(original)
            guard.snapshot(source, dest)
            self.assertEqual(source.read_bytes(), original)
            self.assertEqual(stat.S_IMODE(dest.stat().st_mode), 0o600)
            self.assertEqual(json.loads(dest.read_bytes())["maxStorageCapacity"], 21474836480)
            source.write_text('{"localAuthBypassEnabled":true}')
            self.assertEqual(json.loads(dest.read_bytes())["localAuthBypassEnabled"], False)
            with self.assertRaises(FileExistsError):
                guard.snapshot(dest, dest)

    def test_source_symlinks_and_output_collisions_refused(self):
        with tempfile.TemporaryDirectory() as tmp:
            source, dest, link = [Path(tmp) / x for x in ("source", "dest", "link")]
            source.write_text(json.dumps(SAFE))
            link.symlink_to(source)
            with self.assertRaises(OSError):
                guard.snapshot(link, dest)
            dest.symlink_to(source)
            with self.assertRaises(OSError):
                guard.snapshot(source, dest)
            self.assertEqual(json.loads(source.read_bytes()), SAFE)

    def test_invalid_settings_do_not_create_snapshot(self):
        with tempfile.TemporaryDirectory() as tmp:
            source, dest = Path(tmp) / "source", Path(tmp) / "dest"
            source.write_text('{"apiRestricted":false}')
            with self.assertRaises(guard.UnsafeSettings):
                guard.snapshot(source, dest)
            self.assertFalse(dest.exists())

    def test_key_is_private_idempotent_and_never_rotated(self):
        with tempfile.TemporaryDirectory() as tmp:
            key = Path(tmp) / "apikey.txt"
            guard.ensure_api_key(key, KEY)
            self.assertEqual(key.read_text(), KEY)
            self.assertEqual(stat.S_IMODE(key.stat().st_mode), 0o600)
            before = key.stat().st_mtime_ns
            guard.ensure_api_key(key, KEY)
            self.assertEqual(key.stat().st_mtime_ns, before)
            for value in ("", "x" * 64, KEY + "\n", "cd" * 32):
                with self.assertRaises(guard.UnsafeSettings):
                    guard.ensure_api_key(key, value)
                self.assertEqual(key.read_text(), KEY)

    def test_existing_key_permissions_are_not_silently_repaired(self):
        with tempfile.TemporaryDirectory() as tmp:
            key = Path(tmp) / "key"
            key.write_text(KEY)
            key.chmod(0o644)
            with self.assertRaises(guard.UnsafeSettings):
                guard.ensure_api_key(key, KEY)
            self.assertEqual(key.read_text(), KEY)
            self.assertEqual(stat.S_IMODE(key.stat().st_mode), 0o644)

    def test_key_symlink_and_fifo_refused_without_blocking(self):
        with tempfile.TemporaryDirectory() as tmp:
            target, link = Path(tmp) / "target", Path(tmp) / "link"
            target.write_text(KEY)
            link.symlink_to(target)
            with self.assertRaises(OSError):
                guard.ensure_api_key(link, KEY)
            fifo = Path(tmp) / "fifo"
            os.mkfifo(fifo)
            with self.assertRaises(guard.UnsafeSettings):
                guard.ensure_api_key(fifo, KEY)


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