from __future__ import annotations import io import tempfile import time import unittest import zipfile from pathlib import Path try: import flask # noqa: F401 except ModuleNotFoundError: flask = None @unittest.skipIf(flask is None, "Flask is not installed in the static-analysis environment") class AppFlowTests(unittest.TestCase): def setUp(self): from kimu.app import create_app from kimu.crypto import generate_master_key self.temp = tempfile.TemporaryDirectory() root = Path(self.temp.name) self.app = create_app( { "TESTING": True, "DATABASE": root / "kimu.sqlite3", "STORAGE_DIR": root / "storage", "MAX_UPLOAD_BYTES": 4 * 1024 * 1024, "SECRET_KEY": "test-secret-key", "FAKE_PROVIDER": True, "TOGETHER_API_KEY": "", "K27_CONFIGURED": True, "COOKIE_SECURE": False, "MASTER_KEY": generate_master_key(), "MASTER_KEY_VERSION": 1, } ) self.client = self.app.test_client() created = self.client.post( "/api/v1/setup", json={"username": "owner", "password": "correct horse battery staple"}, ) self.assertEqual(created.status_code, 201) self.csrf = created.get_json()["csrf_token"] self.headers = {"X-CSRF-Token": self.csrf} def tearDown(self): self.temp.cleanup() def wait_for_row(self, table: str, row_id: str, terminal: set[str]) -> str: deadline = time.monotonic() + 5 status = "queued" while time.monotonic() < deadline: with self.app.app_context(): from kimu.db import connect db = connect(self.app.config["DATABASE"]) row = db.execute( f"SELECT status FROM {table} WHERE id=?", (row_id,) ).fetchone() db.close() status = row[0] if row else "missing" if status in terminal: return status time.sleep(0.03) return status def upload(self, content: bytes, name: str, mime: str = "text/plain") -> dict: response = self.client.post( "/api/v1/files", data={"files": (io.BytesIO(content), name, mime)}, content_type="multipart/form-data", headers=self.headers, ) self.assertEqual(response.status_code, 201, response.get_data(as_text=True)) return response.get_json()["files"][0] def test_switchable_models_and_adaptive_settings(self): data = self.client.get("/api/v1/models").get_json()["models"] models = {item["key"]: item for item in data} self.assertTrue(models["k2.6"]["settings_schema"]["reasoning_enabled"]["editable"]) self.assertFalse(models["k2.7-code"]["settings_schema"]["reasoning_enabled"]["editable"]) saved = self.client.put( "/api/v1/models/k2.7-code/preferences", json={"reasoning_enabled": False, "temperature": 0.2}, headers=self.headers, ) self.assertEqual(saved.status_code, 200) preferences = saved.get_json()["preferences"] self.assertTrue(preferences["reasoning_enabled"]) self.assertEqual(preferences["temperature"], 1.0) def test_file_zip_attachment_and_chat_flow(self): notes = self.upload(b"alpha beta gamma\n", "notes.txt") archive_buffer = io.BytesIO() with zipfile.ZipFile(archive_buffer, "w", zipfile.ZIP_DEFLATED) as handle: handle.writestr("src/app.py", "print('working')\n") handle.writestr("README.md", "# Test\n") archive = self.upload( archive_buffer.getvalue(), "source.zip", "application/zip" ) detail = self.client.get(f"/api/v1/files/{archive['id']}") self.assertEqual(detail.status_code, 200) self.assertEqual(len(detail.get_json()["file"]["zip_inventory"]), 2) created = self.client.post( "/api/v1/conversations", json={"model_key": "k2.7-code"}, headers=self.headers, ) conversation_id = created.get_json()["conversation"]["id"] started = self.client.post( "/api/v1/chat/runs", json={ "conversation_id": conversation_id, "content": "Review both attachments.", "attachment_ids": [notes["id"], archive["id"]], }, headers=self.headers, ) self.assertEqual(started.status_code, 202, started.get_data(as_text=True)) run_id = started.get_json()["run"]["id"] self.assertEqual( self.wait_for_row("runs", run_id, {"complete", "failed", "cancelled", "partial"}), "complete", ) conversation = self.client.get( f"/api/v1/conversations/{conversation_id}" ).get_json() self.assertEqual([item["role"] for item in conversation["messages"]], ["user", "assistant"]) self.assertEqual(len(conversation["messages"][0]["attachments"]), 2) self.assertIn("KIMU chat is working", conversation["messages"][1]["content"]) self.assertIn("preserved-reasoning", conversation["messages"][1]["reasoning"]) export = self.client.get(f"/api/v1/conversations/{conversation_id}/export.txt") self.assertEqual(export.status_code, 200) self.assertIn("Attachments:", export.get_data(as_text=True)) def test_artifacts_are_encrypted_at_rest_and_decrypt_on_download(self): payload = b"private KIMU artifact body" uploaded = self.upload(payload, "private.txt") self.assertTrue(uploaded["encrypted"]) with self.app.app_context(): from kimu.db import connect from kimu.crypto import MAGIC db = connect(self.app.config["DATABASE"]) row = db.execute("SELECT * FROM files WHERE id=?", (uploaded["id"],)).fetchone() stored = Path(self.app.config["STORAGE_DIR"]) / row["stored_name"] raw = stored.read_bytes() db.close() self.assertTrue(raw.startswith(MAGIC)) self.assertNotIn(payload, raw) downloaded = self.client.get(uploaded["download_url"]) self.assertEqual(downloaded.status_code, 200) self.assertEqual(downloaded.data, payload) self.assertEqual(downloaded.headers["X-KIMU-Encrypted-At-Rest"], "1") def test_offline_chat_idempotency_incremental_sync_and_conflict(self): conversation_id = "cnv_local_offline_test" body = { "conversation_id": conversation_id, "content": "Queued exactly once", "attachment_ids": [], "create_if_missing": True, "model_key": "k2.6", "conversation_title": "Offline chat", "client_operation_id": "op_offline_exactly_once", } first = self.client.post("/api/v1/chat/runs", json=body, headers=self.headers) second = self.client.post("/api/v1/chat/runs", json=body, headers=self.headers) self.assertEqual(first.status_code, 202) self.assertEqual(second.status_code, 202) self.assertFalse(first.get_json()["idempotent"]) self.assertTrue(second.get_json()["idempotent"]) self.assertEqual(first.get_json()["run"]["id"], second.get_json()["run"]["id"]) with self.app.app_context(): from kimu.db import connect db = connect(self.app.config["DATABASE"]) count = db.execute( "SELECT COUNT(*) FROM messages WHERE conversation_id=? AND role='user'", (conversation_id,), ).fetchone()[0] db.close() self.assertEqual(count, 1) sync = self.client.get("/api/v1/sync/pull?after=0&limit=200") self.assertEqual(sync.status_code, 200) self.assertTrue(any(item["entity_id"] == conversation_id for item in sync.get_json()["changes"])) conversation = self.client.get(f"/api/v1/conversations/{conversation_id}").get_json()["conversation"] conflict = self.client.patch( f"/api/v1/conversations/{conversation_id}", json={"title": "Stale update", "base_version": 1}, headers=self.headers, ) self.assertEqual(conflict.status_code, 409) self.assertEqual(conflict.get_json()["error"]["code"], "KIMU_SYNC_CONFLICT") self.assertGreater(conversation["version"], 1) def test_unsafe_zip_is_rejected(self): archive_buffer = io.BytesIO() with zipfile.ZipFile(archive_buffer, "w") as handle: handle.writestr("../escape.txt", "blocked") response = self.client.post( "/api/v1/files", data={ "files": ( io.BytesIO(archive_buffer.getvalue()), "unsafe.zip", "application/zip", ) }, content_type="multipart/form-data", headers=self.headers, ) self.assertEqual(response.status_code, 400) self.assertEqual(response.get_json()["error"]["code"], "KIMU_ZIP_UNSAFE_PATH") def test_code_interpreter_durable_job(self): job_response = self.client.post( "/api/v1/code/runs", json={"code": "print('KIMU')", "file_ids": []}, headers=self.headers, ) self.assertEqual(job_response.status_code, 202) job_id = job_response.get_json()["job"]["id"] self.assertEqual( self.wait_for_row( "jobs", job_id, {"successful", "failed", "cancelled", "partial", "blocked"} ), "successful", ) job = self.client.get(f"/api/v1/jobs/{job_id}").get_json()["job"] self.assertEqual(job["progress"], 100) self.assertEqual(job["result"]["session_id"], "ses_fake_kimu") self.assertIn("fake-provider mode", job["result"]["outputs"][0]["data"]) self.assertEqual( self.client.delete("/api/v1/code/session", headers=self.headers).status_code, 200, ) def test_memory_skill_settings_and_admin(self): memory = self.client.post( "/api/v1/memories", json={"title": "Project rule", "content": "Use UK spelling.", "scope": "global"}, headers=self.headers, ) self.assertEqual(memory.status_code, 201) memory_id = memory.get_json()["memory"]["id"] updated = self.client.patch( f"/api/v1/memories/{memory_id}", json={"enabled": False}, headers=self.headers, ) self.assertEqual(updated.status_code, 200) self.assertEqual(updated.get_json()["memory"]["enabled"], 0) skill = self.client.post( "/api/v1/skills", json={ "name": "Audit project", "description": "Static project audit", "instructions": "Inspect files and report defects.", "input_schema": {"type": "object"}, "permitted_tools": ["workspace.file.read"], }, headers=self.headers, ) self.assertEqual(skill.status_code, 201) settings = self.client.put( "/api/v1/settings", json={"haptics": False, "offline_queue": True}, headers=self.headers, ) self.assertEqual(settings.status_code, 200) self.assertFalse(settings.get_json()["settings"]["haptics"]) admin = self.client.get("/api/v1/admin/summary") self.assertEqual(admin.status_code, 200) data = admin.get_json() self.assertEqual(data["counts"]["memories"], 1) self.assertEqual(data["counts"]["skills"], 1) self.assertGreaterEqual(len(data["audit"]), 4) self.assertTrue(data["readiness"]["database"]["integrity"] == "ok") def test_tools_runtime_approval_and_idempotency(self): tools = self.client.get("/api/v1/tools") self.assertEqual(tools.status_code, 200) tool_ids = {item["id"] for item in tools.get_json()["tools"]} self.assertIn("workspace.list", tool_ids) self.assertIn("workspace.write", tool_ids) workspace = self.client.post( "/api/v1/workspaces", json={"name": "HTTP tool test", "kind": "tools"}, headers=self.headers, ) self.assertEqual(workspace.status_code, 201) workspace_id = workspace.get_json()["workspace"]["id"] write = self.client.post( "/api/v1/tool-operations", json={ "workspace_id": workspace_id, "tool_id": "workspace.write", "arguments": {"path": "notes/result.txt", "content": "verified"}, "idempotency_key": "http_write_once", }, headers=self.headers, ) self.assertEqual(write.status_code, 202) operation = write.get_json()["operation"] self.assertEqual(operation["status"], "awaiting_approval") duplicate = self.client.post( "/api/v1/tool-operations", json={ "workspace_id": workspace_id, "tool_id": "workspace.write", "arguments": {"path": "notes/result.txt", "content": "verified"}, "idempotency_key": "http_write_once", }, headers=self.headers, ) self.assertTrue(duplicate.get_json()["idempotent"]) approved = self.client.post( f"/api/v1/tool-operations/{operation['id']}/approve", headers=self.headers, ) self.assertEqual(approved.status_code, 202) self.assertEqual( self.wait_for_row("tool_operations", operation["id"], {"successful", "failed", "cancelled"}), "successful", ) result = self.client.get(f"/api/v1/tool-operations/{operation['id']}").get_json() self.assertEqual(result["operation"]["result"]["path"], "notes/result.txt") def test_website_builder_create_generate_test_export_and_import(self): created = self.client.post( "/api/v1/websites", json={ "name": "Mobile Test Site", "project_type": "static", "model_key": "k2.7-code", "brief": "A mobile-first verified test website.", }, headers=self.headers, ) self.assertEqual(created.status_code, 201, created.get_data(as_text=True)) project_id = created.get_json()["project"]["id"] detail = self.client.get(f"/api/v1/websites/{project_id}").get_json()["project"] self.assertIn("index.html", {item["path"] for item in detail["files"]}) generated = self.client.post( f"/api/v1/websites/{project_id}/generate", json={"instruction": "Improve mobile spacing without changing the purpose."}, headers=self.headers, ) self.assertEqual(generated.status_code, 200, generated.get_data(as_text=True)) self.assertTrue(generated.get_json()["test"]["passed"]) tested = self.client.post(f"/api/v1/websites/{project_id}/test", headers=self.headers) self.assertEqual(tested.status_code, 200) exported = self.client.get(f"/api/v1/websites/{project_id}/export.zip") self.assertEqual(exported.status_code, 200) self.assertTrue(exported.data.startswith(b"PK")) imported = self.client.post( "/api/v1/websites/import", data={ "file": (io.BytesIO(exported.data), "exported.zip", "application/zip"), "name": "Imported copy", "model_key": "k2.7-code", "project_type": "static", }, content_type="multipart/form-data", headers=self.headers, ) self.assertEqual(imported.status_code, 201, imported.get_data(as_text=True)) def test_swarm_role_graph_checkpoints_and_release_gate(self): created = self.client.post( "/api/v1/swarms", json={ "objective": "Create and independently review a small mobile-first website release plan.", "max_tool_calls": 20, "max_seconds": 120, }, headers=self.headers, ) self.assertEqual(created.status_code, 201) swarm_id = created.get_json()["swarm"]["id"] started = self.client.post(f"/api/v1/swarms/{swarm_id}/start", headers=self.headers) self.assertEqual(started.status_code, 202) status = self.wait_for_row("swarms", swarm_id, {"successful", "failed", "cancelled", "partial"}) self.assertEqual(status, "successful") swarm = self.client.get(f"/api/v1/swarms/{swarm_id}").get_json()["swarm"] self.assertEqual( [task["role"] for task in swarm["tasks"]], ["coordinator", "architect", "builder", "tester", "reviewer", "release"], ) self.assertTrue(all(task["status"] == "successful" for task in swarm["tasks"])) self.assertEqual(swarm["result"]["checkpoints"], 6) def test_audio_config_is_whisper_sonic_websocket_and_barge_in_locked(self): response = self.client.get("/api/v1/audio/config") self.assertEqual(response.status_code, 200) audio = response.get_json() self.assertEqual(audio["stt"]["model"], "openai/whisper-large-v3") self.assertEqual(audio["tts"]["model"], "cartesia/sonic-3") self.assertTrue(audio["conversation_mode"]) self.assertTrue(audio["barge_in"]["cancels_audio"]) self.assertTrue(audio["barge_in"]["cancels_model"]) self.assertEqual(audio["websocket_path"], "/api/v1/audio/conversation/ws") def test_csrf_blocks_authenticated_mutation(self): response = self.client.post("/api/v1/conversations", json={"model_key": "k2.6"}) self.assertEqual(response.status_code, 403) self.assertEqual(response.get_json()["error"]["code"], "KIMU_CSRF_INVALID") if __name__ == "__main__": unittest.main()