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

import connection
import gpu_metrics
import work_record
import network_prepare
import network_apply
import workspace_access
import stat


class ToolTests(unittest.TestCase):
    def test_acl_does_not_unmask_other_users_or_add_execute(self):
        raw = '# file: /fixture\nuser::rw-\nuser:2002:rwx\ngroup::---\nmask::r--\nother::---\n'
        result = workspace_access.transform(raw, 1001, stat.S_IFREG | 0o640, 0)
        self.assertIn('user:2002:r--', result)
        self.assertIn('user:1001:rw-', result)
        self.assertIn('mask::rw-', result)

    def test_acl_unused_mask_execute_is_not_a_real_grant(self):
        raw = '# file: /fixture\nuser::rw-\nuser:1001:rw-\ngroup::---\nmask::rwx\nother::---\n'
        result = workspace_access.transform(raw, 1001, stat.S_IFREG | 0o670, 0)
        self.assertIn('user:1001:rw-', result)
        self.assertIn('mask::rw-', result)

    def test_acl_plain_file_and_owner_write(self):
        raw = '# file: /fixture\nuser::r--\ngroup::---\nother::---\n'
        result = workspace_access.transform(raw, 1001, stat.S_IFREG | 0o400, 1001)
        self.assertIn('user::rw-', result)
        self.assertIn('user:1001:rw-', result)
        self.assertIn('mask::rw-', result)

    def test_acl_directory_inherits_without_owner_headers(self):
        raw = '# file: /fixture\n# owner: 0\n# group: 0\n# flags: -s-\nuser::rwx\ngroup::---\nother::---\n'
        result = workspace_access.transform(raw, 1001, stat.S_IFDIR | 0o2700, 0)
        self.assertIn('default:user:1001:rwx', result)
        self.assertIn('# flags: -s-', result)
        self.assertNotIn('# owner:', result)

    def test_network_rewrite_is_scoped(self):
        original={"services": {k:dict(environment={},volumes=[]) for k in ("analytics","ai-control-sidecar","ai-gateway","redis","postgresql")}}
        original["services"]["ai-gateway"]["ports"]=[dict(published="38443",target=38443)]
        original["services"]["redis"]["ports"]=[dict(host_ip="192.0.2.1",published="16379",target=6379)]
        original["services"]["analytics"]["environment"]["DATABASE_URL"]="preserve-fixture"
        wanted=dict(ai_ip="192.0.2.4",web_ip="192.0.2.5",manager_base_url="https://192.0.2.5:8443",event_endpoint="http://192.0.2.5:8080/api/v1/events",
                    control_port=38444,redis_port=16380,artifact_lan_cidr="192.0.2.0/24")
        changed=network_prepare.rewrite(original,wanted,Path("/fixture/bridge"),Path("/fixture/tls"),Path("/fixture/session"))
        self.assertEqual(original["services"]["redis"]["ports"][0]["published"],"16379")
        self.assertEqual(changed["services"]["redis"]["ports"][0]["target"],6379)
        self.assertEqual(changed["services"]["analytics"]["environment"]["DATABASE_URL"],"preserve-fixture")
        self.assertEqual(changed["services"]["ai-control-sidecar"]["environment"]["AI_CONTROL_EVENT_ENDPOINT"],wanted["event_endpoint"])

    def test_firewall_input_and_scope(self):
        text=network_apply.firewall_script("192.0.2.4","192.0.2.5",16379)
        compile(text,"firewall-fixture","exec")
        self.assertIn("CTX-ICERTI-REDIS",text)
        self.assertNotIn("-F','INPUT",text)
        self.assertNotIn("-F','DOCKER-USER",text)
        with self.assertRaises(ValueError):network_apply.firewall_script("192.0.2.4;reboot","192.0.2.5",16379)

    def test_addresses(self):
        valid = connection.addresses("192.0.2.11", "192.0.2.12", "https://192.0.2.12:8443", 38443, 16379)
        self.assertEqual(valid["ai_ip"], "192.0.2.11")
        for manager in ("http://192.0.2.12", "https://x:y@192.0.2.12", "https://192.0.2.12/login"):
            with self.assertRaises(ValueError):
                connection.addresses("192.0.2.11", "192.0.2.12", manager, 38443, 16379)

    def test_invalid_ip_and_ports(self):
        for ai in ("0.0.0.0", "127.0.0.1", "ff00::1"):
            with self.assertRaises(ValueError):
                connection.addresses(ai, "192.0.2.12", "https://192.0.2.12", 38443, 16379)
        with self.assertRaises(ValueError):
            connection.addresses("192.0.2.11", "192.0.2.12", "https://192.0.2.12", 22, 16379)

    def test_inventory_excludes_secrets(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            (root / "code.py").write_text("print(1)\n")
            (root / "password.txt").write_text("do-not-collect")
            (root / ".env").write_text("SECRET=value")
            self.assertEqual(list(work_record.inventory(root)), ["code.py"])
            (root / "link.py").symlink_to(root / "code.py")
            with self.assertRaises(ValueError):
                work_record.inventory(root)

    def test_work_diff_add_change_delete(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            source = root / "source"
            source.mkdir()
            record = root / "20260928T120000-1"
            (record / "before").mkdir(parents=True)
            (record / "before/change.py").write_text("x=1\n")
            (record / "before/delete.py").write_text("y=1\n")
            baseline = work_record.inventory(record / "before")
            (record / "baseline.json").write_text(json.dumps(dict(source=str(source), files=baseline)))
            (record / "notes.jsonl").touch()
            (root / "active").write_text(record.name)
            (source / "change.py").write_text("x=2\n")
            (source / "add.py").write_text("z=3\n")
            old = work_record.STATE
            work_record.STATE = root
            try:
                work_record.end()
                result = json.loads((record / "result.json").read_text())
                self.assertEqual(len(result["changed"]), 3)
                self.assertFalse((root / "active").exists())
                with self.assertRaises(ValueError):
                    work_record.export(record.name)
                work_record.export(record.name, reviewed=True)
            finally:
                work_record.STATE = old


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