import asyncio import json import pytest from fastapi import HTTPException from investibook.api import deps as deps_api from investibook.api import sources as sources_api from investibook.api import workspaces as ws_api from investibook.core.agent_runtime import _load_all_cluster_metadata from investibook.core.system_prompt import build_dynamic_context from investibook.db import queries as dbq from investibook.models.requests import ( CreateAlertWorkspaceRequest, SourcesPayload, WorkspaceCreate, ) @pytest.fixture def isolated_db(tmp_path, monkeypatch): db_path = tmp_path / "investibook_test.db" monkeypatch.setattr(dbq, "DB_PATH", db_path) dbq._db_init() return dbq @pytest.fixture def patched_workspace_io(tmp_path, monkeypatch): monkeypatch.setattr(ws_api, "WORKSPACES_DIR", tmp_path / "workspaces") monkeypatch.setattr(ws_api, "TEMPLATES_DIR", tmp_path / "templates") monkeypatch.setattr(ws_api, "_ensure_workspace_dir_initialized", lambda _wid: None) def test_from_alert_creates_workspace_and_source( isolated_db, patched_workspace_io, monkeypatch ): owner = isolated_db._db_create_user("o11y-service@pingcap.com", "pass-123") sources_api._prepare_tasks.clear() async def _noop_prepare_task(*_args, **_kwargs): return None monkeypatch.setattr(sources_api, "_prepare_issue_source_task", _noop_prepare_task) body = CreateAlertWorkspaceRequest( alert_id="O11Y-12345", alert_type="o11y", cluster_id="cluster-1", ) resp = asyncio.run(ws_api.create_workspace_from_alert(body, current_user=owner)) asyncio.run(asyncio.sleep(0)) assert resp["status"] == "created" workspace_id = resp["workspace_id"] assert workspace_id.startswith("ws-") assert resp["owner_user_id"] == owner["id"] assert resp["source_prepare_task_id"] in sources_api._prepare_tasks ws_row = isolated_db._db_get_workspace(workspace_id) assert ws_row is not None assert ws_row["user_id"] == owner["id"] assert ws_row["name"] == "O11Y-12345 / cluster-1" sources = isolated_db._db_get_blob("workspace_sources", workspace_id) assert isinstance(sources, list) assert len(sources) == 2 assert any(s["type"] == "cluster_meta" and s["name"] == "cluster-1" for s in sources) ticket_source = next(s for s in sources if s["type"] == "jira_issue") assert ticket_source["issueId"] == "O11Y-12345" assert ticket_source["taskId"] == resp["source_prepare_task_id"] assert ticket_source["meta"]["skip_jira_comments"] is False assert resp["automation_resumed"] is False def test_from_entry_persists_request_level_comment_exclusion( isolated_db, patched_workspace_io, monkeypatch ): owner = isolated_db._db_create_user("ao-service@pingcap.com", "pass-123") sources_api._prepare_tasks.clear() async def _noop_prepare_task(*_args, **_kwargs): return None monkeypatch.setattr(sources_api, "_prepare_issue_source_task", _noop_prepare_task) resp = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="ST-2814", skip_jira_comments=True, ), current_user=owner, ) ) task = sources_api._prepare_tasks[resp["source_prepare_task_id"]] source = next( item for item in isolated_db._db_get_blob("workspace_sources", resp["workspace_id"]) if item["type"] == "jira_issue" ) assert task["skip_jira_comments"] is True assert source["meta"]["skip_jira_comments"] is True def test_from_entry_persists_alt_as_alertstore_source( isolated_db, patched_workspace_io, monkeypatch ): owner = isolated_db._db_create_user("alertstore-service@pingcap.com", "pass-123") sources_api._prepare_tasks.clear() async def _noop_prepare_task(*_args, **_kwargs): return None monkeypatch.setattr(sources_api, "_prepare_issue_source_task", _noop_prepare_task) resp = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest(alert_id="ALT-2608-1742"), current_user=owner, ) ) task = sources_api._prepare_tasks[resp["source_prepare_task_id"]] source = next( item for item in isolated_db._db_get_blob("workspace_sources", resp["workspace_id"]) if item["type"] == "alertstore_alert" ) assert source["issueId"] == "ALT-2608-1742" assert source["meta"]["skip_jira_comments"] is True assert task["skip_jira_comments"] is True def test_alert_ingest_service_token_has_route_scoped_identity( isolated_db, monkeypatch ): monkeypatch.setattr(deps_api, "ALERT_INGEST_SERVICE_TOKEN", "ingest-secret") monkeypatch.setattr( deps_api, "ALERT_INGEST_SERVICE_EMAIL", "alertstore-service@pingcap.com", ) principal = asyncio.run( deps_api.get_alert_ingest_principal( authorization=None, x_api_key="ingest-secret", ) ) assert principal["email"] == "alertstore-service@pingcap.com" assert principal["principal_type"] == "service" assert principal["service_scope"] == "alert_ingest" assert isolated_db._db_get_user_by_id(principal["id"]) is not None def test_alert_ingest_service_token_is_not_general_user_auth( isolated_db, monkeypatch ): monkeypatch.setattr(deps_api, "ALERT_INGEST_SERVICE_TOKEN", "ingest-secret") monkeypatch.setattr( deps_api, "ALERT_INGEST_SERVICE_EMAIL", "alertstore-service@pingcap.com", ) with pytest.raises(HTTPException) as exc_info: asyncio.run(deps_api.get_current_user("Bearer ingest-secret")) assert exc_info.value.status_code == 401 def test_from_entry_creates_workspace_from_cluster_only( isolated_db, patched_workspace_io, monkeypatch ): owner = isolated_db._db_create_user("cluster-entry@pingcap.com", "pass-123") sources_api._prepare_tasks.clear() async def _noop_prepare_task(*_args, **_kwargs): return None monkeypatch.setattr(sources_api, "_prepare_issue_source_task", _noop_prepare_task) resp = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest(cluster_id="cluster-only"), current_user=owner, ) ) assert resp["status"] == "created" assert resp["source_prepare_task_id"] is None workspace_id = resp["workspace_id"] ws_row = isolated_db._db_get_workspace(workspace_id) assert ws_row["name"] == "Cluster cluster-only" sources = isolated_db._db_get_blob("workspace_sources", workspace_id) assert sources == [ { "id": "src-cluster-cluster-only", "type": "cluster_meta", "name": "cluster-only", "description": "Cluster context from o11y entry", "readonly": True, "createdAt": sources[0]["createdAt"], "meta": { "alert_type": "o11y", "cluster_id": "cluster-only", }, } ] workspace_dir = ws_api.WORKSPACES_DIR / workspace_id cluster_meta = json.loads( (workspace_dir / "cluster_meta.json").read_text(encoding="utf-8") ) assert cluster_meta == {"cluster_id": "cluster-only", "org_id": ""} loaded_metadata = _load_all_cluster_metadata(workspace_dir) assert loaded_metadata[0]["cluster_id"] == "cluster-only" assert "- Cluster ID: cluster-only" in build_dynamic_context(loaded_metadata) def test_from_alert_idempotency_is_scoped_to_owner( isolated_db, patched_workspace_io, monkeypatch ): owner = isolated_db._db_create_user("svc-idem@pingcap.com", "pass-123") responder = isolated_db._db_create_user("responder-idem@pingcap.com", "pass-123") sources_api._prepare_tasks.clear() async def _noop_prepare_task(*_args, **_kwargs): return None monkeypatch.setattr(sources_api, "_prepare_issue_source_task", _noop_prepare_task) key = "idem-001" first = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="TCOC-99", idempotency_key=key, ), current_user=owner, ) ) repeated = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="TCOC-99", idempotency_key=key, ), current_user=owner, ) ) second_owner = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="TCOC-99", idempotency_key=key, ), current_user=responder, ) ) assert first["status"] == "created" assert repeated["status"] == "existing" assert repeated["workspace_id"] == first["workspace_id"] assert second_owner["status"] == "created" assert second_owner["workspace_id"] != first["workspace_id"] access_row, is_owner, can_edit = asyncio.run( deps_api.require_workspace_access(second_owner["workspace_id"], responder) ) assert access_row["id"] == second_owner["workspace_id"] assert is_owner is True assert can_edit is True def test_from_alert_idempotency_reuses_same_owner_legacy_record( isolated_db, patched_workspace_io, monkeypatch ): owner = isolated_db._db_create_user("legacy-idem@pingcap.com", "pass-123") sources_api._prepare_tasks.clear() async def _noop_prepare_task(*_args, **_kwargs): return None monkeypatch.setattr(sources_api, "_prepare_issue_source_task", _noop_prepare_task) key = "legacy-idem-001" workspace_id = "ws-legacy-idem" now = "2026-09-01T00:00:00+00:00" isolated_db._db_upsert_workspace( workspace_id, owner["id"], "Legacy entry", None, False, now, now, ) isolated_db._db_create_alert_ingestion(key, workspace_id, "TCOC-100") workspace_dir = ws_api.WORKSPACES_DIR / workspace_id workspace_dir.mkdir(parents=True) (workspace_dir / "cluster_meta.json").write_text( json.dumps( { "cluster_id": "stale-cluster", "org_id": "org-legacy", "tidb_version": "v8.5.3", } ), encoding="utf-8", ) result = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="TCOC-100", cluster_id="cluster-legacy", idempotency_key=key, ), current_user=owner, ) ) assert result["status"] == "existing" assert result["workspace_id"] == workspace_id assert json.loads( (workspace_dir / "cluster_meta.json").read_text(encoding="utf-8") ) == { "cluster_id": "cluster-legacy", "org_id": "org-legacy", "tidb_version": "v8.5.3", } def test_put_sources_preserves_readonly_cluster_context( isolated_db, patched_workspace_io ): owner = isolated_db._db_create_user("source-owner@pingcap.com", "pass-123") workspace_id = "ws-source-preserve" now = "2026-09-01T00:00:00+00:00" isolated_db._db_upsert_workspace( workspace_id, owner["id"], "Cluster c1", None, False, now, now, ) cluster_source = { "id": "src-cluster-c1", "type": "cluster_meta", "name": "c1", "readonly": True, "meta": {"cluster_id": "c1"}, } isolated_db._db_set_blob( "workspace_sources", workspace_id, [cluster_source, {"id": "old-source", "type": "git_repo"}], ) asyncio.run( ws_api.put_sources( workspace_id, SourcesPayload( sources=[{"id": "skill-clinic", "type": "skill", "readonly": True}] ), current_user=owner, ) ) assert isolated_db._db_get_blob("workspace_sources", workspace_id) == [ {"id": "skill-clinic", "type": "skill", "readonly": True}, cluster_source, ] def test_from_entry_force_new_bypasses_default_idempotency( isolated_db, patched_workspace_io, monkeypatch ): owner = isolated_db._db_create_user("force-new@pingcap.com", "pass-123") sources_api._prepare_tasks.clear() async def _noop_prepare_task(*_args, **_kwargs): return None monkeypatch.setattr(sources_api, "_prepare_issue_source_task", _noop_prepare_task) first = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest(alert_id="O11Y-FORCE", cluster_id="c-force"), current_user=owner, ) ) second = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="O11Y-FORCE", cluster_id="c-force", force_new=True, ), current_user=owner, ) ) assert first["status"] == "created" assert second["status"] == "created" assert second["workspace_id"] != first["workspace_id"] def test_existing_workspace_resume_recreates_lost_source_prepare_task( isolated_db, patched_workspace_io, monkeypatch ): owner = isolated_db._db_create_user("resume-source@pingcap.com", "pass-123") sources_api._prepare_tasks.clear() async def _noop_prepare_task(*_args, **_kwargs): return None monkeypatch.setattr(sources_api, "_prepare_issue_source_task", _noop_prepare_task) key = "auto-oncall:v1:ST-2814" first = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="ST-2814", auto_questions=["diagnose"], idempotency_key=key, skip_jira_comments=True, ), current_user=owner, ) ) old_task_id = first["source_prepare_task_id"] # Simulate a backend restart: durable source state remains, task registry is lost. sources_api._prepare_tasks.clear() resumed = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="ST-2814", auto_questions=["diagnose"], idempotency_key=key, skip_jira_comments=True, resume_automation=True, ), current_user=owner, ) ) assert resumed["status"] == "existing" assert resumed["workspace_id"] == first["workspace_id"] assert resumed["automation_resumed"] is True assert resumed["source_prepare_task_id"] != old_task_id assert resumed["source_prepare_task_id"] in sources_api._prepare_tasks source = next( item for item in isolated_db._db_get_blob("workspace_sources", first["workspace_id"]) if item["type"] == "jira_issue" ) assert source["taskId"] == resumed["source_prepare_task_id"] assert source["meta"]["skip_jira_comments"] is True rows, total = isolated_db._db_get_workspaces(owner["id"]) assert total == 1 assert [row["id"] for row in rows] == [first["workspace_id"]] def test_existing_workspace_resume_restarts_lost_running_autochat( isolated_db, patched_workspace_io, monkeypatch ): owner = isolated_db._db_create_user("resume-chat@pingcap.com", "pass-123") sources_api._prepare_tasks.clear() async def _noop_prepare_task(*_args, **_kwargs): return None monkeypatch.setattr(sources_api, "_prepare_issue_source_task", _noop_prepare_task) key = "auto-oncall:v1:TCID-1" first = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="TCID-1", auto_questions=["diagnose"], idempotency_key=key, ), current_user=owner, ) ) workspace_id = first["workspace_id"] sources = isolated_db._db_get_blob("workspace_sources", workspace_id) jira_source = next(item for item in sources if item["type"] == "jira_issue") jira_source.update(status="ready", issueDir="issues/TCID-1", taskId=None) isolated_db._db_set_blob("workspace_sources", workspace_id, sources) isolated_db._db_update_workspace_diagnostics(workspace_id, status="running") started: list[str] = [] def _fake_start(workspace_id: str) -> None: started.append(workspace_id) async def _fake_callback(*_args, **_kwargs) -> None: return None import investibook.core.auto_chat as auto_chat_core monkeypatch.setattr(auto_chat_core, "start_auto_chat_task", _fake_start) monkeypatch.setattr(auto_chat_core, "_emit_status_callback", _fake_callback) resumed = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="TCID-1", auto_questions=["diagnose"], idempotency_key=key, resume_automation=True, ), current_user=owner, ) ) assert resumed["workspace_id"] == workspace_id assert resumed["automation_resumed"] is True assert resumed["source_prepare_task_id"] is None assert started == [workspace_id] def test_existing_ready_source_is_reprepared_when_comments_are_newly_excluded( isolated_db, patched_workspace_io, monkeypatch ): owner = isolated_db._db_create_user("resume-exclude@pingcap.com", "pass-123") sources_api._prepare_tasks.clear() async def _noop_prepare_task(*_args, **_kwargs): return None monkeypatch.setattr(sources_api, "_prepare_issue_source_task", _noop_prepare_task) key = "auto-oncall:v1:ST-2815" first = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="ST-2815", auto_questions=["diagnose"], idempotency_key=key, ), current_user=owner, ) ) workspace_id = first["workspace_id"] sources = isolated_db._db_get_blob("workspace_sources", workspace_id) jira_source = next(item for item in sources if item["type"] == "jira_issue") jira_source.update(status="ready", issueDir="issues/ST-2815", taskId=None) assert jira_source["meta"]["skip_jira_comments"] is False isolated_db._db_set_blob("workspace_sources", workspace_id, sources) isolated_db._db_update_workspace_diagnostics(workspace_id, status="running") sources_api._prepare_tasks.clear() resumed = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest( alert_id="ST-2815", auto_questions=["diagnose"], idempotency_key=key, skip_jira_comments=True, resume_automation=True, ), current_user=owner, ) ) assert resumed["workspace_id"] == workspace_id assert resumed["automation_resumed"] is True assert resumed["source_prepare_task_id"] in sources_api._prepare_tasks source = next( item for item in isolated_db._db_get_blob("workspace_sources", workspace_id) if item["type"] == "jira_issue" ) assert source["status"] == "loading" assert source["meta"]["skip_jira_comments"] is True assert isolated_db._db_get_workspace_diagnostics(workspace_id)["status"] == "pending" def test_workspace_access_is_readable_for_all_but_editable_only_by_owner( isolated_db, patched_workspace_io, monkeypatch ): owner = isolated_db._db_create_user("owner@pingcap.com", "pass-owner") outsider = isolated_db._db_create_user("outsider@pingcap.com", "pass-outsider") sources_api._prepare_tasks.clear() async def _noop_prepare_task(*_args, **_kwargs): return None monkeypatch.setattr(sources_api, "_prepare_issue_source_task", _noop_prepare_task) created = asyncio.run( ws_api.create_workspace_from_alert( CreateAlertWorkspaceRequest(alert_id="O11Y-777"), current_user=owner, ) ) workspace_id = created["workspace_id"] row = asyncio.run(deps_api.require_workspace_editor(workspace_id, owner)) assert row["id"] == workspace_id access_row, is_owner, can_edit = asyncio.run( deps_api.require_workspace_access(workspace_id, outsider) ) assert access_row["id"] == workspace_id assert is_owner is False assert can_edit is False with pytest.raises(HTTPException): asyncio.run(deps_api.require_workspace_editor(workspace_id, outsider)) private_created = asyncio.run( ws_api.create_workspace_api( WorkspaceCreate(name="Manual Workspace"), current_user=owner, ) ) private_workspace_id = private_created["id"] access_row, is_owner, can_edit = asyncio.run( deps_api.require_workspace_access(private_workspace_id, outsider) ) assert access_row["id"] == private_workspace_id assert is_owner is False assert can_edit is False outsider_rows, _total = isolated_db._db_get_workspaces(outsider["id"]) assert {row["id"] for row in outsider_rows} == set() with pytest.raises(HTTPException): asyncio.run(deps_api.require_workspace_editor(private_workspace_id, outsider)) def test_transfer_makes_new_owner_only_editor( isolated_db, patched_workspace_io ): owner = isolated_db._db_create_user("transfer-owner@pingcap.com", "pass-owner") new_owner = isolated_db._db_create_user("transfer-new@pingcap.com", "pass-new") created = asyncio.run( ws_api.create_workspace_api( WorkspaceCreate(name="Transfer Workspace"), current_user=owner, ) ) workspace_id = created["id"] asyncio.run( ws_api.transfer_workspace_api( workspace_id, body=ws_api.TransferOwnerPayload(new_owner_email=new_owner["email"]), current_user=owner, ) ) row = isolated_db._db_get_workspace(workspace_id) assert row["user_id"] == new_owner["id"] assert asyncio.run(deps_api.require_workspace_editor(workspace_id, new_owner))["id"] == workspace_id old_access_row, old_is_owner, old_can_edit = asyncio.run( deps_api.require_workspace_access(workspace_id, owner) ) assert old_access_row["id"] == workspace_id assert old_is_owner is False assert old_can_edit is False with pytest.raises(HTTPException): asyncio.run(deps_api.require_workspace_editor(workspace_id, owner)) @pytest.mark.parametrize( ("source_type", "issue_id"), [("jira_issue", "O11Y-4242"), ("alertstore_alert", "ALT-2608-1742")], ) def test_prepare_task_persists_source_status_without_frontend_polling( isolated_db, patched_workspace_io, source_type, issue_id ): owner = isolated_db._db_create_user("status-sync@pingcap.com", "pass-123") created = asyncio.run( ws_api.create_workspace_api( WorkspaceCreate(name="Status Sync"), current_user=owner, ) ) workspace_id = created["id"] task_id = "task-sync-001" isolated_db._db_set_blob( "workspace_sources", workspace_id, [ { "id": "src-1", "type": source_type, "name": issue_id, "issueId": issue_id, "status": "loading", "statusMessage": "Preparing data...", "taskId": task_id, "createdAt": "2026-01-01T00:00:00+00:00", } ], ) sources_api._prepare_tasks[task_id] = { "status": "completed", "issue_id": issue_id, "workspace_id": workspace_id, "files": None, "cluster_meta": None, "issue_dir": f"issues/{issue_id}", "error": None, } sources_api._sync_source_status_from_task(task_id) sources = isolated_db._db_get_blob("workspace_sources", workspace_id) or [] assert len(sources) == 1 source = sources[0] assert source["status"] == "ready" assert source.get("taskId") is None assert source["issueDir"] == f"issues/{issue_id}"