from __future__ import annotations

import json
import tempfile
import unittest
import zipfile
from pathlib import Path

from kimu.config import env_float
from kimu.db import connect, initialise
from kimu.tool_runtime import TOOL_DEFINITIONS, ToolRuntimeError, execute_tool, registry_payload


class Increment4144CoreTests(unittest.TestCase):
    def test_tool_registry_is_versioned_and_risk_classified(self):
        tools = registry_payload()
        self.assertGreaterEqual(len(tools), 9)
        self.assertEqual({item["risk_class"] for item in tools}, {"read", "write", "execute", "destructive"})
        self.assertTrue(all(item["version"] >= 1 for item in tools))
        self.assertTrue(TOOL_DEFINITIONS["workspace.write"].approval_required)
        self.assertFalse(TOOL_DEFINITIONS["workspace.read"].approval_required)

    def test_workspace_tools_enforce_scope_versions_and_zip(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            written = execute_tool(
                TOOL_DEFINITIONS["workspace.write"],
                root,
                {"path": "src/app.py", "content": "print('one')\n", "expected_sha256": None},
            )
            with self.assertRaises(ToolRuntimeError) as conflict:
                execute_tool(
                    TOOL_DEFINITIONS["workspace.patch"],
                    root,
                    {"path": "src/app.py", "old_text": "one", "new_text": "two", "expected_sha256": "0" * 64},
                )
            self.assertEqual(conflict.exception.code, "KIMU_TOOL_VERSION_CONFLICT")
            patched = execute_tool(
                TOOL_DEFINITIONS["workspace.patch"],
                root,
                {"path": "src/app.py", "old_text": "one", "new_text": "two", "expected_sha256": written["sha256"]},
            )
            self.assertNotEqual(patched["sha256"], written["sha256"])
            execute_tool(TOOL_DEFINITIONS["archive.create"], root, {"output_path": "release.zip", "paths": ["src"]})
            inspected = execute_tool(TOOL_DEFINITIONS["archive.inspect"], root, {"path": "release.zip", "limit": 100})
            self.assertEqual(inspected["entries"][0]["name"], "src/app.py")
            with self.assertRaises(ToolRuntimeError):
                execute_tool(TOOL_DEFINITIONS["workspace.read"], root, {"path": "../escape.txt"})

    def test_static_checks_find_invalid_source(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            (root / "valid.json").write_text('{"ok":true}', encoding="utf-8")
            (root / "broken.py").write_text("def broken(:\n", encoding="utf-8")
            result = execute_tool(TOOL_DEFINITIONS["checks.static"], root, {"path": ".", "limit": 50})
            self.assertFalse(result["passed"])
            self.assertEqual(result["failures"][0]["path"], "broken.py")

    def test_increment_44_database_contract(self):
        with tempfile.TemporaryDirectory() as directory:
            database = Path(directory) / "kimu.sqlite3"
            initialise(database)
            db = connect(database)
            tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
            required = {
                "workspaces", "tool_operations", "tool_events", "website_projects", "website_files",
                "website_versions", "website_builds", "website_test_results", "swarms", "swarm_tasks",
                "swarm_events", "swarm_checkpoints", "workspace_file_leases", "audio_sessions",
            }
            self.assertTrue(required.issubset(tables), sorted(required - tables))
            versions = {row[0] for row in db.execute("SELECT version FROM schema_migrations")}
            self.assertIn(10, versions)
            db.close()

    def test_audio_and_feature_lock_source_contract(self):
        root = Path(__file__).resolve().parents[1]
        gateway = (root / "src/kimu/conversation_gateway.py").read_text(encoding="utf-8")
        client = (root / "frontend/src/conversation-client.js").read_text(encoding="utf-8")
        advanced = (root / "src/kimu/advanced.py").read_text(encoding="utf-8")
        main = (root / "frontend/src/main.jsx").read_text(encoding="utf-8")
        locks = json.loads((root / "config/feature-locks.json").read_text(encoding="utf-8"))
        for phrase in ["openai/whisper-large-v3", "cartesia/sonic-3"]:
            self.assertIn(phrase, json.dumps(locks))
        for phrase in ["context.cancel", "input_text_buffer.clear", "KIMU_CSRF_FAILED", "context_id != self.turn_id", "response.close", "cancel_event=cancel"]:
            self.assertIn(phrase, gateway)
        self.assertEqual(gateway.count("await self.connect_upstreams()"), 1)
        for phrase in ["barge_in", "clearAudio", "getCsrf()", "pcm.buffer"]:
            self.assertIn(phrase, client)
        for phrase in ["/websites/import", "/websites/<project_id>/test", "workspace_file_leases", "swarm_checkpoints", "KIMU_WEBSITE_IMPORT_DUPLICATE", "KIMU_WEBSITE_IMPORT_ENCRYPTED"]:
            self.assertIn(phrase, advanced)
        self.assertIn('sandbox="allow-scripts"', main)
        self.assertNotIn('allow-same-origin', main)

    def test_invalid_vad_threshold_falls_back(self):
        import os
        old = os.environ.get("KIMU_TEST_FLOAT")
        try:
            os.environ["KIMU_TEST_FLOAT"] = "invalid"
            self.assertEqual(env_float("KIMU_TEST_FLOAT", 0.3, 0.05, 0.95), 0.3)
            os.environ["KIMU_TEST_FLOAT"] = "9"
            self.assertEqual(env_float("KIMU_TEST_FLOAT", 0.3, 0.05, 0.95), 0.95)
        finally:
            if old is None:
                os.environ.pop("KIMU_TEST_FLOAT", None)
            else:
                os.environ["KIMU_TEST_FLOAT"] = old


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