From b50a933d85c82c3ed1ed7e9af90f7f804a7f20bb Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Thu, 27 Aug 2026 13:09:09 +0200 Subject: [PATCH] =?UTF-8?q?chore(cleanup):=20toten=20AI-Copilot-Legacy=20e?= =?UTF-8?q?ntfernt=20=E2=80=94=20Router=20nie=20gemountet,=20Tabellen=20vo?= =?UTF-8?q?n=20Migration=200137=20gedroppt=20(Chat=20l=C3=A4uft=20seitdem?= =?UTF-8?q?=20=C3=BCber=20kommunikation/comm=5Fconversations);=20schemas/O?= =?UTF-8?q?penAPI-Tag=20bereinigt;=20Geister-Test=20test=5Fai=5Fcopilot.py?= =?UTF-8?q?=20geloescht=20(pytest.skip=20seit=20Phase=202);=20test=5Fconta?= =?UTF-8?q?cts=5Flifecycle=20Route-Anzahl-Failure=20als=20Vorbestand=20bew?= =?UTF-8?q?iesen=20(83=20Routen=20auch=20auf=20clean=20HEAD)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/main.py | 1 - app/models/ai_conversation.py | 53 -- app/routes/ai_copilot.py | 135 --- app/schemas/__init__.py | 9 - app/schemas/ai_copilot.py | 69 -- app/services/ai_copilot_service.py | 477 ---------- tests/test_ai_copilot.py | 1304 ---------------------------- 7 files changed, 2048 deletions(-) delete mode 100644 app/models/ai_conversation.py delete mode 100644 app/routes/ai_copilot.py delete mode 100644 app/schemas/ai_copilot.py delete mode 100644 app/services/ai_copilot_service.py delete mode 100644 tests/test_ai_copilot.py diff --git a/app/main.py b/app/main.py index f7bb289..b255c7e 100644 --- a/app/main.py +++ b/app/main.py @@ -438,7 +438,6 @@ def create_app() -> FastAPI: {"name": "entity-history", "description": "Audit trail and entity change history."}, {"name": "import-export", "description": "Bulk import and export of contacts and data."}, {"name": "plugins", "description": "Plugin management: list, install, activate, deactivate."}, - {"name": "ai-copilot", "description": "AI copilot: chat, suggestions, conversation history."}, {"name": "workflows", "description": "Workflow definitions, instances, and execution."}, {"name": "user-preferences", "description": "Per-user preference settings."}, {"name": "currencies", "description": "Currency management for multi-currency support."}, diff --git a/app/models/ai_conversation.py b/app/models/ai_conversation.py deleted file mode 100644 index edab655..0000000 --- a/app/models/ai_conversation.py +++ /dev/null @@ -1,53 +0,0 @@ -"""AI Conversation and Message models — tenant-scoped with RLS.""" - -from __future__ import annotations - -import uuid -from typing import Any - -from sqlalchemy import ForeignKey, Index, Integer, String, Text -from sqlalchemy.dialects.postgresql import JSONB -from sqlalchemy.dialects.postgresql import UUID as PGUUID -from sqlalchemy.orm import Mapped, mapped_column - -from app.core.db import Base, TenantMixin -from app.models.owned_mixin import OwnedMixin - - -class AIConversation(Base, TenantMixin, OwnedMixin): - """AI Copilot conversation thread — tenant-scoped.""" - - __tablename__ = "ai_conversations" - __table_args__ = (Index("ix_ai_conversations_tenant_user", "tenant_id", "user_id"),) - - id: Mapped[uuid.UUID] = mapped_column( - PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4 - ) - user_id: Mapped[uuid.UUID] = mapped_column( - PGUUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True - ) - title: Mapped[str] = mapped_column(String(255), nullable=False, default="Untitled") - context: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict, nullable=False) - - -class AIMessage(Base, TenantMixin, OwnedMixin): - """Individual messages within an AI conversation — user input, AI response, actions.""" - - __tablename__ = "ai_messages" - __table_args__ = (Index("ix_ai_messages_tenant_conversation", "tenant_id", "conversation_id"),) - - id: Mapped[uuid.UUID] = mapped_column( - PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4 - ) - conversation_id: Mapped[uuid.UUID] = mapped_column( - PGUUID(as_uuid=True), - ForeignKey("ai_conversations.id", ondelete="CASCADE"), - nullable=False, - index=True, - ) - role: Mapped[str] = mapped_column(String(20), nullable=False) # user, assistant, system - content: Mapped[str] = mapped_column(Text, nullable=False) - proposed_actions: Mapped[list[dict[str, Any]] | None] = mapped_column(JSONB, nullable=True) - executed_action: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) - execution_result: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) - message_index: Mapped[int] = mapped_column(Integer, nullable=False, default=0) diff --git a/app/routes/ai_copilot.py b/app/routes/ai_copilot.py deleted file mode 100644 index e28ff8a..0000000 --- a/app/routes/ai_copilot.py +++ /dev/null @@ -1,135 +0,0 @@ -"""AI Copilot routes — query, execute, history. - -.. deprecated:: Phase 2 (ARCH-059) - Legacy module, not wired into main.py and without frontend consumers. - Scheduled for removal together with app/services/ai_copilot_service.py - and the ``ai_conversations`` tables (see service docstring for plan). -""" - -from __future__ import annotations - -import uuid - -from fastapi import APIRouter, Depends, HTTPException, Query, Request, status -from sqlalchemy.ext.asyncio import AsyncSession - -from app.core.db import get_db -from app.core.rate_limit import RateLimitPolicy, check_rate_limit_policy -from app.deps import require_permission -from app.schemas.ai_copilot import ( - CopilotExecuteRequest, - CopilotQueryRequest, -) -from app.services import ai_copilot_service - -router = APIRouter(prefix="/api/v1/ai/copilot", tags=["ai-copilot"]) - - -@router.post("/query") -async def copilot_query( - body: CopilotQueryRequest, - request: Request, - db: AsyncSession = Depends(get_db), - current_user: dict = Depends(require_permission("ai:write")), -): - """Process a natural language query and return proposed actions. - - Returns proposed_actions array for user confirmation. - Does NOT execute any actions — user must call /execute to confirm. - """ - tenant_id = uuid.UUID(current_user["tenant_id"]) - user_id = uuid.UUID(current_user["user_id"]) - - # Rate limit — AI policy (cost-sensitive LLM call) - await check_rate_limit_policy( - f"rate:ai:copilot:{tenant_id}:{user_id}", - RateLimitPolicy.AI, - ) - - result = await ai_copilot_service.process_query( - db, - tenant_id, - user_id, - query=body.query, - conversation_id=body.conversation_id, - context=body.context, - ) - - if "error" in result and result.get("status_code") == 404: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail={"detail": result["error"], "code": "conversation_not_found"}, - ) - - return result - - -@router.post("/execute") -async def copilot_execute( - body: CopilotExecuteRequest, - request: Request, - db: AsyncSession = Depends(get_db), - current_user: dict = Depends(require_permission("ai:write")), -): - """Execute a proposed action after user confirmation. - - RBAC is enforced — the user must have permission for the action. - Returns 200 with execution result or 403 if RBAC blocks. - """ - tenant_id = uuid.UUID(current_user["tenant_id"]) - user_id = uuid.UUID(current_user["user_id"]) - - # Rate limit — AI policy (cost-sensitive LLM call) - await check_rate_limit_policy( - f"rate:ai:copilot:{tenant_id}:{user_id}", - RateLimitPolicy.AI, - ) - resolved = { - "permissions": current_user.get("permissions", []), - "denied": current_user.get("denied_permissions", []), - "field_permissions": current_user.get("field_permissions", {}), - "is_system_admin": current_user.get("is_system_admin", False), - } - - result = await ai_copilot_service.execute_action( - db, - tenant_id, - user_id, - resolved, - conversation_id=body.conversation_id, - action=body.action.model_dump(), - ) - - if result.get("status_code") == 404: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail={"detail": result.get("error", "Not found"), "code": "not_found"}, - ) - - if result.get("status_code") == 403: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail={"detail": result.get("error", "Insufficient permissions"), "code": "forbidden"}, - ) - - return result - - -@router.get("/history") -async def copilot_history( - page: int = Query(1, ge=1), - page_size: int = Query(20, ge=1, le=100), - db: AsyncSession = Depends(get_db), - current_user: dict = Depends(require_permission("ai:read")), -): - """Get paginated conversation history for the current user.""" - tenant_id = uuid.UUID(current_user["tenant_id"]) - user_id = uuid.UUID(current_user["user_id"]) - - return await ai_copilot_service.get_history( - db, - tenant_id, - user_id, - page=page, - page_size=page_size, - ) diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py index cf8942f..b5511d2 100644 --- a/app/schemas/__init__.py +++ b/app/schemas/__init__.py @@ -1,14 +1,5 @@ """Pydantic schemas package.""" -from app.schemas.ai_copilot import ( - CopilotAction, # noqa: F401 - CopilotExecuteRequest, # noqa: F401 - CopilotExecuteResponse, # noqa: F401 - CopilotHistoryResponse, # noqa: F401 - CopilotMessageResponse, # noqa: F401 - CopilotQueryRequest, # noqa: F401 - CopilotQueryResponse, # noqa: F401 -) from app.schemas.plugin import ( PluginActionResponse, # noqa: F401 PluginInfo, # noqa: F401 diff --git a/app/schemas/ai_copilot.py b/app/schemas/ai_copilot.py deleted file mode 100644 index 6f9b1f5..0000000 --- a/app/schemas/ai_copilot.py +++ /dev/null @@ -1,69 +0,0 @@ -"""AI Copilot schemas — query, execute, history.""" - -from __future__ import annotations - -from pydantic import BaseModel, Field - - -class CopilotQueryRequest(BaseModel): - """Natural language query to the AI copilot.""" - - query: str = Field(..., min_length=1, max_length=2000) - conversation_id: str | None = None - context: dict = Field(default_factory=dict) - - -class CopilotAction(BaseModel): - """A proposed API action derived from NL input.""" - - method: str = Field(..., pattern="^(GET|POST|PATCH|DELETE)$") - path: str = Field(..., min_length=1) - body: dict | None = None - description: str = "" - confidence: float = Field(0.0, ge=0.0, le=1.0) - - -class CopilotQueryResponse(BaseModel): - """Response from copilot query — proposed actions for user confirmation.""" - - conversation_id: str - message: str - proposed_actions: list[CopilotAction] = Field(default_factory=list) - - -class CopilotExecuteRequest(BaseModel): - """Execute a proposed action after user confirmation.""" - - conversation_id: str - action: CopilotAction - - -class CopilotExecuteResponse(BaseModel): - """Result of executing a proposed action.""" - - conversation_id: str - success: bool - status_code: int - data: dict | list | None = None - error: str | None = None - - -class CopilotMessageResponse(BaseModel): - """A single message in conversation history.""" - - id: str - role: str - content: str - proposed_actions: list[dict] | None = None - executed_action: dict | None = None - execution_result: dict | None = None - created_at: str | None = None - - -class CopilotHistoryResponse(BaseModel): - """Paginated conversation history.""" - - items: list[CopilotMessageResponse] - total: int - page: int - page_size: int diff --git a/app/services/ai_copilot_service.py b/app/services/ai_copilot_service.py deleted file mode 100644 index 18bc517..0000000 --- a/app/services/ai_copilot_service.py +++ /dev/null @@ -1,477 +0,0 @@ -"""AI Copilot service — NL query processing, action execution, RBAC, audit logging. - -.. deprecated:: Phase 2 - This module is a legacy leftover of the pre-kommunikation AI chat system. - It is NOT registered in main.py, has NO frontend consumers, and its test - suite is skipped ("ai_copilot routes removed in Phase 2"). - - Shutdown plan (ARCH-059): - 1. ✅ Marked deprecated (this notice) - 2. Remove routes/service/model + ``ai_conversations`` tables in a dedicated - migration once a release confirms zero traffic on /api/v1/ai/copilot/* - 3. Do NOT migrate to kommunikation — the plugin already covers chat via - CommConversation/CommMessage; parity work would be wasted effort. -""" - -from __future__ import annotations - -import uuid -import warnings -from typing import Any - -from sqlalchemy import desc, func, select -from sqlalchemy.ext.asyncio import AsyncSession - -from app.ai.llm_client import get_llm_client -from app.core.audit import log_audit -from app.core.permissions import check_permission -from app.core.visibility import apply_visibility_filter, check_single_entity_access - -try: - from app.models.ai_conversation import AIConversation, AIMessage -except ImportError: - AIConversation = None # type: ignore - AIMessage = None # type: ignore -from app.models.contact import Contact -from app.models.workflow import Workflow - -warnings.warn( - "app.services.ai_copilot_service is deprecated (ARCH-059) — " - "scheduled for removal; do not build new features on it", - DeprecationWarning, - stacklevel=2, -) - - -def _safe_iso(dt) -> str | None: - if dt is None: - return None - try: - return dt.isoformat() if hasattr(dt, "isoformat") else None - except Exception: - return None - - -def _get_attr(obj, name, default=None): - try: - val = getattr(obj, name) - return val if val is not None else default - except Exception: - return default - - -def _conversation_to_dict(c: AIConversation) -> dict[str, Any]: - return { - "id": str(c.id), - "title": c.title, - "context": c.context, - "created_at": _safe_iso(_get_attr(c, "created_at")), - "updated_at": _safe_iso(_get_attr(c, "updated_at")), - } - - -def _message_to_dict(m: AIMessage) -> dict[str, Any]: - return { - "id": str(m.id), - "role": m.role, - "content": m.content, - "proposed_actions": m.proposed_actions, - "executed_action": m.executed_action, - "execution_result": m.execution_result, - "created_at": _safe_iso(_get_attr(m, "created_at")), - } - - -async def process_query( - db: AsyncSession, - tenant_id: uuid.UUID, - user_id: uuid.UUID, - query: str, - conversation_id: str | None = None, - context: dict[str, Any] | None = None, - is_system_admin: bool = False, -) -> dict[str, Any]: - """Process a natural language query and return proposed actions. - - 1. Get or create conversation - 2. Store user message - 3. Call LLM client (mock or real) to get proposed actions - 4. Store assistant message with proposed actions - 5. Return response with conversation_id and proposed_actions - """ - context = context or {} - - # Get or create conversation - if conversation_id: - conv_uuid = uuid.UUID(conversation_id) - result = await db.execute( - select(AIConversation).where( - AIConversation.id == conv_uuid, - AIConversation.tenant_id == tenant_id, - ) - ) - conversation = result.scalar_one_or_none() - if conversation is None: - return {"error": "Conversation not found", "status_code": 404} - else: - conversation = AIConversation( - tenant_id=tenant_id, - user_id=user_id, - title=query[:100] if query else "Untitled", - context=context, - ) - db.add(conversation) - await db.flush() - await db.refresh(conversation) - - # Get next message index - count_q = select(func.count()).select_from( - select(AIMessage).where(AIMessage.conversation_id == conversation.id).subquery() - ) - count_result = await db.execute(count_q) - msg_index = count_result.scalar_one() - - # Store user message - user_msg = AIMessage( - tenant_id=tenant_id, - conversation_id=conversation.id, - role="user", - content=query, - message_index=msg_index, - ) - db.add(user_msg) - await db.flush() - await db.refresh(user_msg) - - # Call LLM client - llm = get_llm_client() - llm_response = await llm.generate(query, context) - - # Store assistant message - assistant_msg = AIMessage( - tenant_id=tenant_id, - conversation_id=conversation.id, - role="assistant", - content=llm_response.message, - proposed_actions=llm_response.proposed_actions, - message_index=msg_index + 1, - ) - db.add(assistant_msg) - await db.flush() - await db.refresh(assistant_msg) - - # Log to audit - await log_audit( - db, - tenant_id, - user_id, - action="query", - entity_type="ai_copilot", - entity_id=conversation.id, - changes={"query": query, "proposed_action_count": len(llm_response.proposed_actions)}, - ) - - return { - "conversation_id": str(conversation.id), - "message": llm_response.message, - "proposed_actions": llm_response.proposed_actions, - } - - -async def execute_action( - db: AsyncSession, - tenant_id: uuid.UUID, - user_id: uuid.UUID, - resolved: dict[str, Any], - conversation_id: str, - action: dict[str, Any], - is_system_admin: bool = False, -) -> dict[str, Any]: - """Execute a proposed action with RBAC enforcement. - - 1. Validate conversation belongs to tenant - 2. Check RBAC permissions for the action - 3. Execute the action (direct DB or API call) - 4. Store execution result in message - 5. Log to audit - """ - conv_uuid = uuid.UUID(conversation_id) - - # Validate conversation ownership - result = await db.execute( - select(AIConversation).where( - AIConversation.id == conv_uuid, - AIConversation.tenant_id == tenant_id, - ) - ) - conversation = result.scalar_one_or_none() - if conversation is None: - return {"error": "Conversation not found", "status_code": 404} - - method = action.get("method", "GET").upper() - path = action.get("path", "") - body = action.get("body") or {} - - # Determine module and action_type from path for RBAC - module, action_type = _derive_rbac_from_path(method, path) - required_perm = f"{module}:{action_type}" - - if not check_permission(resolved, required_perm): - return { - "error": "Insufficient permissions for this action", - "status_code": 403, - "success": False, - } - - # Check single entity access for write operations - if method in ("POST", "PATCH", "DELETE"): - parts = path.replace("/api/v1/", "").strip("/").split("/") - entity_type = parts[0] if parts else "" - entity_id = parts[1] if len(parts) > 1 else None - if entity_id: - try: - entity_uuid = uuid.UUID(entity_id) - except (ValueError, TypeError): - entity_uuid = None - if entity_uuid: - has_access = await check_single_entity_access( - db, entity_type, entity_uuid, user_id, tenant_id, - required_level="write", is_system_admin=is_system_admin, - ) - if not has_access: - return { - "error": "Insufficient access to this entity", - "status_code": 403, - "success": False, - } - - # Execute the action - try: - exec_result = await _execute_api_action(db, tenant_id, user_id, method, path, body, is_system_admin=is_system_admin) - except Exception as exc: - exec_result = {"error": str(exc), "status_code": 500} - - # Get next message index - count_q = select(func.count()).select_from( - select(AIMessage).where(AIMessage.conversation_id == conversation.id).subquery() - ) - count_result = await db.execute(count_q) - msg_index = count_result.scalar_one() - - # Store execution message - exec_msg = AIMessage( - tenant_id=tenant_id, - conversation_id=conversation.id, - role="assistant", - content=f"Executed {method} {path}", - executed_action=action, - execution_result=exec_result, - message_index=msg_index, - ) - db.add(exec_msg) - await db.flush() - await db.refresh(exec_msg) - - # Log to audit - await log_audit( - db, - tenant_id, - user_id, - action="execute", - entity_type="ai_copilot", - entity_id=conversation.id, - changes={"action": action, "result": exec_result}, - ) - - return { - "conversation_id": str(conversation.id), - "success": exec_result.get("success", True), - "status_code": exec_result.get("status_code", 200), - "data": exec_result.get("data"), - "error": exec_result.get("error"), - } - - -async def get_history( - db: AsyncSession, - tenant_id: uuid.UUID, - user_id: uuid.UUID, - page: int = 1, - page_size: int = 20, -) -> dict[str, Any]: - """Get paginated conversation history for the current user.""" - page = max(1, page) - page_size = max(1, min(100, page_size)) - - base = select(AIConversation).where( - AIConversation.tenant_id == tenant_id, - AIConversation.user_id == user_id, - ) - - count_q = select(func.count()).select_from(base.subquery()) - total_result = await db.execute(count_q) - total = total_result.scalar_one() - - offset = (page - 1) * page_size - paginated = base.order_by(desc(AIConversation.created_at)).offset(offset).limit(page_size) - result = await db.execute(paginated) - conversations = result.scalars().all() - - items: list[dict[str, Any]] = [] - for conv in conversations: - # Get messages for each conversation - msg_q = ( - select(AIMessage) - .where( - AIMessage.conversation_id == conv.id, - AIMessage.tenant_id == tenant_id, - ) - .order_by(AIMessage.message_index) - ) - msg_result = await db.execute(msg_q) - messages = msg_result.scalars().all() - - items.append( - { - **_conversation_to_dict(conv), - "messages": [_message_to_dict(m) for m in messages], - } - ) - - return { - "items": items, - "total": total, - "page": page, - "page_size": page_size, - } - - -async def _execute_api_action( - db: AsyncSession, - tenant_id: uuid.UUID, - user_id: uuid.UUID, - method: str, - path: str, - body: dict[str, Any], - is_system_admin: bool = False, -) -> dict[str, Any]: - """Execute an API action directly against the database. - - Supports contacts CRUD (including company-type contacts), plus workflow listing. - """ - # Parse path to determine entity and operation - parts = path.replace("/api/v1/", "").strip("/").split("/") - entity = parts[0] if parts else "" - entity_id = parts[1] if len(parts) > 1 else None - - if entity in ("companies", "contacts"): - return await _exec_contacts(db, tenant_id, user_id, method, entity_id, body, is_system_admin=is_system_admin) - elif entity == "workflows": - return await _exec_workflows(db, tenant_id, user_id, method, entity_id, body, is_system_admin=is_system_admin) - else: - return {"error": f"Unsupported entity: {entity}", "status_code": 400, "success": False} - - -async def _exec_contacts( - db: AsyncSession, - tenant_id: uuid.UUID, - user_id: uuid.UUID, - method: str, - entity_id: str | None, - body: dict[str, Any], - is_system_admin: bool = False, -) -> dict[str, Any]: - """Execute contact operations (unified: company + person).""" - if method == "GET": - query = select(Contact).where( - Contact.tenant_id == tenant_id, - Contact.deleted_at.is_(None), - ) - query = await apply_visibility_filter( - db, query, "contact", Contact, user_id, tenant_id, is_system_admin=is_system_admin - ) - result = await db.execute(query) - contacts = result.scalars().all() - return { - "success": True, - "status_code": 200, - "data": [{"id": str(c.id), "name": c.name or c.displayname, "email": c.email_1, "type": c.type} for c in contacts], - } - - elif method == "POST": - contact = Contact( - tenant_id=tenant_id, - type=body.get("type", "company"), - name=body.get("name", "Untitled"), - email_1=body.get("email"), - phone_1=body.get("phone"), - created_by=user_id, - updated_by=user_id, - ) - db.add(contact) - await db.flush() - return { - "success": True, - "status_code": 201, - "data": {"id": str(contact.id), "name": contact.name, "type": contact.type}, - } - - return {"error": f"Unsupported method: {method}", "status_code": 400, "success": False} - - -async def _exec_workflows( - db: AsyncSession, - tenant_id: uuid.UUID, - user_id: uuid.UUID, - method: str, - entity_id: str | None, - body: dict[str, Any], - is_system_admin: bool = False, -) -> dict[str, Any]: - """Execute workflow operations.""" - if method == "GET": - result = await db.execute( - select(Workflow).where( - Workflow.tenant_id == tenant_id, - Workflow.is_active.is_(True), - ) - ) - workflows = result.scalars().all() - return { - "success": True, - "status_code": 200, - "data": [ - {"id": str(w.id), "name": w.name, "trigger_event": w.trigger_event} - for w in workflows - ], - } - - return {"error": f"Unsupported method: {method}", "status_code": 400, "success": False} - - -def _derive_rbac_from_path(method: str, path: str) -> tuple[str, str]: - """Derive module and action_type from HTTP method and path for RBAC. - - Returns (module, action_type) suitable for check_permission(). - """ - parts = path.replace("/api/v1/", "").strip("/").split("/") - entity = parts[0] if parts else "" - - method_to_action = { - "GET": "read", - "POST": "create", - "PATCH": "update", - "DELETE": "delete", - } - action_type = method_to_action.get(method, "read") - - # Map path entities to permission modules - entity_to_module = { - "companies": "companies", - "contacts": "contacts", - "workflows": "workflows", - "ai": "ai_copilot", - } - module = entity_to_module.get(entity, entity) - - return module, action_type diff --git a/tests/test_ai_copilot.py b/tests/test_ai_copilot.py deleted file mode 100644 index e200fce..0000000 --- a/tests/test_ai_copilot.py +++ /dev/null @@ -1,1304 +0,0 @@ -"""Tests for AI Copilot — covers ACs 1-7.""" - -from __future__ import annotations - -from datetime import UTC - -import pytest -pytestmark = pytest.mark.skip(reason="ai_copilot routes removed in Phase 2") -from httpx import ASGITransport, AsyncClient - -from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users - - -@pytest.mark.asyncio -async def test_ac1_copilot_query_returns_proposed_actions(ai_client: AsyncClient, db_session): - """AC1: POST /api/v1/ai/copilot/query with NL input returns 200 + proposed_actions array.""" - await seed_tenant_and_users(db_session) - await login_client(ai_client, "admin@tenanta.com") - - resp = await ai_client.post( - "/api/v1/ai/copilot/query", - json={"query": "Create a company named Acme Corp"}, - ) - assert resp.status_code == 200 - data = resp.json() - assert "conversation_id" in data - assert "proposed_actions" in data - assert len(data["proposed_actions"]) > 0 - action = data["proposed_actions"][0] - assert action["method"] == "POST" - assert "/api/v1/contacts" in action["path"] - assert action["body"]["name"] == "Acme Corp" - - -@pytest.mark.asyncio -async def test_ac2_copilot_execute_action_success(ai_client: AsyncClient, db_session): - """AC2: POST /api/v1/ai/copilot/execute with proposed action returns 200 + API result (RBAC enforced).""" - await seed_tenant_and_users(db_session) - await login_client(ai_client, "admin@tenanta.com") - - # First query to get a conversation and proposed action - query_resp = await ai_client.post( - "/api/v1/ai/copilot/query", - json={"query": "Create a company named TestCorp"}, - ) - assert query_resp.status_code == 200 - conv_id = query_resp.json()["conversation_id"] - action = query_resp.json()["proposed_actions"][0] - - # Execute the proposed action - exec_resp = await ai_client.post( - "/api/v1/ai/copilot/execute", - json={"conversation_id": conv_id, "action": action}, - ) - assert exec_resp.status_code == 200 - exec_data = exec_resp.json() - assert exec_data["success"] is True - assert exec_data["status_code"] == 201 - assert exec_data["data"]["name"] == "TestCorp" - - -@pytest.mark.asyncio -async def test_ac3_copilot_execute_blocked_by_rbac(ai_client: AsyncClient, db_session): - """AC3: POST /api/v1/ai/copilot/execute as viewer with delete action returns 403 (RBAC blocks).""" - await seed_tenant_and_users(db_session) - await login_client(ai_client, "viewer@tenanta.com") - - # Query for a delete action - query_resp = await ai_client.post( - "/api/v1/ai/copilot/query", - json={ - "query": "Delete company", - "context": {"entity_id": "00000000-0000-0000-0000-000000000000"}, - }, - ) - assert query_resp.status_code == 200 - conv_id = query_resp.json()["conversation_id"] - actions = query_resp.json()["proposed_actions"] - assert len(actions) > 0 - action = actions[0] - assert action["method"] == "DELETE" - - # Viewer should be blocked from delete - exec_resp = await ai_client.post( - "/api/v1/ai/copilot/execute", - json={"conversation_id": conv_id, "action": action}, - ) - assert exec_resp.status_code == 403 - assert "forbidden" in exec_resp.json()["detail"]["code"] - - -@pytest.mark.asyncio -async def test_ac4_copilot_history_paginated(ai_client: AsyncClient, db_session): - """AC4: GET /api/v1/ai/copilot/history returns 200 + paginated conversation history.""" - await seed_tenant_and_users(db_session) - await login_client(ai_client, "admin@tenanta.com") - - # Create a conversation by querying - await ai_client.post( - "/api/v1/ai/copilot/query", - json={"query": "List companies"}, - ) - - resp = await ai_client.get("/api/v1/ai/copilot/history") - assert resp.status_code == 200 - data = resp.json() - assert "items" in data - assert "total" in data - assert "page" in data - assert "page_size" in data - assert data["total"] >= 1 - assert len(data["items"]) >= 1 - # Each item should have messages - assert "messages" in data["items"][0] - assert len(data["items"][0]["messages"]) >= 2 # user + assistant - - -@pytest.mark.asyncio -async def test_ac5_copilot_action_logged_in_audit(ai_client: AsyncClient, db_session): - """AC5: Copilot action logged in audit_log with entity_type=ai_copilot.""" - from sqlalchemy import select - - from app.models.audit import AuditLog - - await seed_tenant_and_users(db_session) - await login_client(ai_client, "admin@tenanta.com") - - # Execute a query to generate audit log - resp = await ai_client.post( - "/api/v1/ai/copilot/query", - json={"query": "List companies"}, - ) - assert resp.status_code == 200 - - # Check audit log - result = await db_session.execute(select(AuditLog).where(AuditLog.entity_type == "ai_copilot")) - logs = result.scalars().all() - assert len(logs) >= 1 - assert logs[0].action == "query" - assert logs[0].entity_type == "ai_copilot" - - -@pytest.mark.asyncio -async def test_ac6_copilot_tenant_isolation(ai_client: AsyncClient, db_session): - """AC6: Copilot respects tenant isolation — cross-tenant access returns 404.""" - await seed_tenant_and_users(db_session) - # Login as tenant A admin - await login_client(ai_client, "admin@tenanta.com") - - # Create a conversation in tenant A - query_resp = await ai_client.post( - "/api/v1/ai/copilot/query", - json={"query": "List companies"}, - ) - conv_id_a = query_resp.json()["conversation_id"] - - # Login as tenant B admin (different cookie jar) - AsyncClient(transport=ASGITransport(app=ai_client._transport.app), base_url="http://test") - # Need to use the same app — just re-login with a fresh client - # Actually we need a new client without tenant A cookies - from httpx import AsyncClient as AC # noqa: N817 - - # Use the same app instance - import app.main - - app_instance = app.main.app - - async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b: - await login_client(client_b, "admin@tenantb.com") - - # Try to execute action in tenant A conversation from tenant B - exec_resp = await client_b.post( - "/api/v1/ai/copilot/execute", - json={ - "conversation_id": conv_id_a, - "action": { - "method": "GET", - "path": "/api/v1/companies", - "description": "List", - "confidence": 0.9, - }, - }, - ) - assert exec_resp.status_code == 404 - - -@pytest.mark.asyncio -async def test_ac7_copilot_field_level_permissions(ai_client: AsyncClient, db_session): - """AC7: Copilot respects field-level permissions — hidden fields not in response.""" - from app.core.permissions import filter_fields_by_permission - - await seed_tenant_and_users(db_session) - - # Test the filter_fields_by_permission function directly - data = {"name": "Company A", "annual_revenue": 1000000, "industry": "IT"} - field_perms = {"annual_revenue": "hidden"} - - # Admin sees all fields - admin_filtered = filter_fields_by_permission(data, field_perms, "admin") - assert "annual_revenue" in admin_filtered - - # Viewer does not see hidden fields - viewer_filtered = filter_fields_by_permission(data, field_perms, "viewer") - assert "annual_revenue" not in viewer_filtered or "annual_revenue" in viewer_filtered # Field-level permissions may not be applied in service-level calls - assert "name" in viewer_filtered - assert "industry" in viewer_filtered - - -@pytest.mark.asyncio -async def test_copilot_query_with_existing_conversation(ai_client: AsyncClient, db_session): - """Edge case: Query with existing conversation_id appends to conversation.""" - await seed_tenant_and_users(db_session) - await login_client(ai_client, "admin@tenanta.com") - - # First query creates conversation - resp1 = await ai_client.post( - "/api/v1/ai/copilot/query", - json={"query": "List companies"}, - ) - conv_id = resp1.json()["conversation_id"] - - # Second query with same conversation_id - resp2 = await ai_client.post( - "/api/v1/ai/copilot/query", - json={"query": "Create a company named FooBar", "conversation_id": conv_id}, - ) - assert resp2.status_code == 200 - assert resp2.json()["conversation_id"] == conv_id - - -@pytest.mark.asyncio -async def test_copilot_query_invalid_conversation(ai_client: AsyncClient, db_session): - """Edge case: Query with invalid conversation_id returns 404.""" - await seed_tenant_and_users(db_session) - await login_client(ai_client, "admin@tenanta.com") - - resp = await ai_client.post( - "/api/v1/ai/copilot/query", - json={"query": "List companies", "conversation_id": "00000000-0000-0000-0000-000000000000"}, - ) - assert resp.status_code == 404 - - -@pytest.mark.asyncio -async def test_copilot_unauthenticated(ai_client: AsyncClient, db_session): - """Edge case: Unauthenticated request returns 403.""" - resp = await ai_client.post( - "/api/v1/ai/copilot/query", - json={"query": "List companies"}, - ) - assert resp.status_code == 403 - - -# ─── ActionMapper Unit Tests ─── - - -def test_action_mapper_create_company(): - """ActionMapper: 'create company named X' → POST /api/v1/companies.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Create a company named Acme Corp") - assert len(actions) == 1 - assert actions[0]["method"] == "POST" - assert actions[0]["path"] == "/api/v1/contacts" - assert actions[0]["body"]["name"] == "Acme Corp" - assert actions[0]["body"]["type"] == "company" - assert actions[0]["confidence"] == 0.9 - - -def test_action_mapper_create_company_no_name(): - """ActionMapper: 'create company' without name → default 'New Company'.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Add a new company") - assert len(actions) == 1 - assert actions[0]["body"]["name"] == "New Contact" - - -def test_action_mapper_delete_company_with_context(): - """ActionMapper: 'delete company' with entity_id in context → DELETE with specific ID.""" - from app.ai.action_mapper import map_query_to_actions - - test_id = "12345678-1234-1234-1234-123456789abc" - actions = map_query_to_actions("Delete company", context={"entity_id": test_id}) - assert len(actions) == 1 - assert actions[0]["method"] == "DELETE" - assert test_id in actions[0]["path"] - assert actions[0]["confidence"] == 0.9 - - -def test_action_mapper_delete_company_no_context(): - """ActionMapper: 'delete company' without context → DELETE with {id} placeholder.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Remove company") - assert len(actions) == 1 - assert actions[0]["method"] == "DELETE" - assert "{id}" in actions[0]["path"] - assert actions[0]["confidence"] == 0.5 - - -def test_action_mapper_update_company(): - """ActionMapper: 'update company' → PATCH with extracted fields.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Update company industry to Tech, name to FooBar") - assert len(actions) == 1 - assert actions[0]["method"] == "PATCH" - assert actions[0]["body"]["industry"] == "Tech" - assert actions[0]["body"]["name"] == "FooBar" - - -def test_action_mapper_update_company_with_context(): - """ActionMapper: 'update company' with entity_id → PATCH with specific path.""" - from app.ai.action_mapper import map_query_to_actions - - test_id = "12345678-1234-1234-1234-123456789abc" - actions = map_query_to_actions("Edit company", context={"entity_id": test_id}) - assert len(actions) == 1 - assert test_id in actions[0]["path"] - - -def test_action_mapper_list_companies(): - """ActionMapper: 'list companies' → GET /api/v1/companies.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Show all compan") - assert len(actions) == 1 - assert actions[0]["method"] == "GET" - assert actions[0]["path"] == "/api/v1/contacts" - - -def test_action_mapper_list_companies_with_search(): - """ActionMapper: 'find companies named X' → GET with search description.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Find compan named Acme") - assert len(actions) == 1 - assert "Acme" in actions[0]["description"] - - -def test_action_mapper_create_contact(): - """ActionMapper: 'create contact named X' → POST /api/v1/contacts.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Create a contact named John Doe") - assert len(actions) == 1 - assert actions[0]["method"] == "POST" - assert actions[0]["path"] == "/api/v1/contacts" - assert actions[0]["body"]["name"] == "John Doe" - - -def test_action_mapper_list_contacts(): - """ActionMapper: 'list contacts' → GET /api/v1/contacts.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Show all contact") - assert len(actions) == 1 - assert actions[0]["method"] == "GET" - assert actions[0]["path"] == "/api/v1/contacts" - - -def test_action_mapper_list_workflows(): - """ActionMapper: 'list workflows' → GET /api/v1/workflows.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Show all workflow") - assert len(actions) == 1 - assert actions[0]["method"] == "GET" - assert actions[0]["path"] == "/api/v1/workflows" - - -def test_action_mapper_create_workflow(): - """ActionMapper: 'create workflow named X' → POST /api/v1/workflows.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Create a new workflow named Approval Process") - assert len(actions) == 1 - assert actions[0]["method"] == "POST" - assert actions[0]["path"] == "/api/v1/workflows" - assert actions[0]["body"]["name"] == "Approval Process" - - -def test_action_mapper_help_intent(): - """ActionMapper: 'help' query returns demo action with low confidence.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("What can you do? Help me please") - assert len(actions) == 1 - assert actions[0]["confidence"] == 0.3 - - -def test_action_mapper_unknown_query(): - """ActionMapper: unrecognized query returns empty actions list.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("xyzzy flonk") - assert actions == [] - - -def test_action_mapper_update_company_phone_email(): - """ActionMapper: update company with phone and email fields.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Update company phone to 123456, email to test@examplecom") - assert len(actions) == 1 - assert actions[0]["body"]["phone_1"] == "123456" - assert actions[0]["body"]["email_1"] == "test@examplecom" - - -def test_action_mapper_update_company_no_fields(): - """ActionMapper: update company without explicit fields → default name.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Modify company details") - assert len(actions) == 1 - assert "name" in actions[0]["body"] - - -def test_action_mapper_company_list_all_pattern(): - """ActionMapper: 'companies all' matches list_company2 pattern.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("Show me companies all") - assert len(actions) == 1 - assert actions[0]["method"] == "GET" - - -def test_action_mapper_empty_query(): - """ActionMapper: empty query returns no actions.""" - from app.ai.action_mapper import map_query_to_actions - - actions = map_query_to_actions("") - assert actions == [] - - -# ─── LLMClient Unit Tests ─── - - -@pytest.mark.asyncio -async def test_llm_client_mock_mode_generate(): - """LLMClient: mock mode generates actions from keyword matching.""" - from app.ai.llm_client import LLMClient - - client = LLMClient(model=None, api_key=None) - assert client.is_mock is True - - response = await client.generate("Create a company named TestCo") - assert len(response.proposed_actions) > 0 - assert response.proposed_actions[0]["method"] == "POST" - assert "TestCo" in response.proposed_actions[0]["body"]["name"] - - -@pytest.mark.asyncio -async def test_llm_client_mock_mode_no_actions(): - """LLMClient: mock mode with unrecognized query returns empty actions.""" - from app.ai.llm_client import LLMClient - - client = LLMClient(model=None, api_key=None) - - response = await client.generate("xyzzy flonk") - assert response.proposed_actions == [] - assert "couldn't determine" in response.message.lower() - - -@pytest.mark.asyncio -async def test_llm_client_mock_mode_with_context(): - """LLMClient: mock mode passes context to action mapper.""" - from app.ai.llm_client import LLMClient - - client = LLMClient(model=None, api_key=None) - - response = await client.generate("Delete company", context={"entity_id": "abc123"}) - assert len(response.proposed_actions) > 0 - assert response.proposed_actions[0]["method"] == "DELETE" - - -def test_llm_client_api_mode_init(): - """LLMClient: with model and api_key set, is_mock is False.""" - from app.ai.llm_client import LLMClient - - client = LLMClient(model="gpt-4", api_key="test-key") - assert client.is_mock is False - assert client.model == "gpt-4" - assert client.api_key == "test-key" - - -def test_llm_client_api_base_default(): - """LLMClient: default api_base is OpenAI URL.""" - from app.ai.llm_client import LLMClient - - client = LLMClient(model=None, api_key=None) - assert client.api_base == "" - - -def test_llm_client_to_dict(): - """LLMResponse: to_dict returns structured response.""" - from app.ai.llm_client import LLMResponse - - resp = LLMResponse(message="Hello", proposed_actions=[{"method": "GET"}], confidence=0.9) - d = resp.to_dict() - assert d["message"] == "Hello" - assert d["proposed_actions"] == [{"method": "GET"}] - assert d["confidence"] == 0.9 - - -def test_llm_client_parse_valid_json(): - """LLMClient: _parse_llm_response with valid JSON returns structured response.""" - import json - - from app.ai.llm_client import LLMClient - - client = LLMClient(model=None, api_key=None) - content = json.dumps( - { - "message": "Here are actions", - "proposed_actions": [{"method": "GET", "path": "/api/v1/companies"}], - "confidence": 0.95, - } - ) - response = client._parse_llm_response(content) - assert response.message == "Here are actions" - assert len(response.proposed_actions) == 1 - assert response.confidence == 0.95 - - -def test_llm_client_parse_invalid_json(): - """LLMClient: _parse_llm_response with invalid JSON returns fallback response.""" - from app.ai.llm_client import LLMClient - - client = LLMClient(model=None, api_key=None) - response = client._parse_llm_response("not valid json at all") - assert response.proposed_actions == [] - assert response.confidence == 0.3 - - -def test_llm_client_build_system_prompt(): - """LLMClient: _build_system_prompt contains API endpoints and context.""" - from app.ai.llm_client import LLMClient - - client = LLMClient(model=None, api_key=None) - prompt = client._build_system_prompt({"page": "companies"}) - assert "LeoCRM" in prompt - assert "companies" in prompt - assert "proposed_actions" in prompt - - -def test_llm_client_build_system_prompt_empty_context(): - """LLMClient: _build_system_prompt with no context uses empty dict.""" - from app.ai.llm_client import LLMClient - - client = LLMClient(model=None, api_key=None) - prompt = client._build_system_prompt({}) - assert "LeoCRM" in prompt - - -def test_llm_client_get_and_reset(): - """LLMClient: get_llm_client returns singleton, reset clears it.""" - from app.ai.llm_client import get_llm_client, reset_llm_client - - reset_llm_client() - client1 = get_llm_client() - client2 = get_llm_client() - assert client1 is client2 - reset_llm_client() - client3 = get_llm_client() - assert client3 is not client1 - - -# ─── AI Copilot Service Unit Tests ─── - - -@pytest.mark.asyncio -async def test_service_process_query_new_conversation(db_session): - """Service: process_query creates new conversation and returns proposed actions.""" - from app.services.ai_copilot_service import process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - result = await process_query(db_session, tenant_id, admin_id, "Create a company named TestCorp") - assert "conversation_id" in result - assert len(result["proposed_actions"]) > 0 - assert result["proposed_actions"][0]["body"]["name"] == "TestCorp" - - -@pytest.mark.asyncio -async def test_service_process_query_existing_conversation(db_session): - """Service: process_query with existing conversation_id appends to conversation.""" - from app.services.ai_copilot_service import process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - # First query creates conversation - result1 = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = result1["conversation_id"] - - # Second query with same conversation_id - result2 = await process_query( - db_session, - tenant_id, - admin_id, - "Create a company named FooBar", - conversation_id=conv_id, - ) - assert result2["conversation_id"] == conv_id - - -@pytest.mark.asyncio -async def test_service_process_query_invalid_conversation(db_session): - """Service: process_query with invalid conversation_id returns 404 error.""" - from app.services.ai_copilot_service import process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - result = await process_query( - db_session, - tenant_id, - admin_id, - "List companies", - conversation_id="00000000-0000-0000-0000-000000000000", - ) - assert result["error"] == "Conversation not found" - assert result["status_code"] in (404, 403) - - -@pytest.mark.asyncio -async def test_service_process_query_empty_query(db_session): - """Service: process_query with empty query creates conversation with 'Untitled' title.""" - from app.services.ai_copilot_service import process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - result = await process_query(db_session, tenant_id, admin_id, "") - assert "conversation_id" in result - - -@pytest.mark.asyncio -async def test_service_execute_action_companies_get(db_session): - """Service: execute_action with GET /api/v1/companies returns list.""" - from app.services.ai_copilot_service import execute_action - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - # First create a conversation - from app.services.ai_copilot_service import process_query - - query_result = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "GET", "path": "/api/v1/companies", "body": None}, - ) - assert result["success"] is True - assert result["status_code"] == 200 - assert isinstance(result["data"], list) - - -@pytest.mark.asyncio -async def test_service_execute_action_companies_post(db_session): - """Service: execute_action with POST /api/v1/companies creates company.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - { - "method": "POST", - "path": "/api/v1/companies", - "body": {"name": "NewCo", "industry": "Tech"}, - }, - ) - assert result["success"] is True - assert result["status_code"] == 201 - assert result["data"]["name"] == "NewCo" - - -@pytest.mark.asyncio -async def test_service_execute_action_companies_patch(db_session): - """Service: execute_action with PATCH /api/v1/companies/{id} updates company.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - # Create a company first - query_result = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = query_result["conversation_id"] - - create_result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "POST", "path": "/api/v1/contacts", "body": {"name": "PatchCo", "type": "company"}}, - ) - company_id = create_result["data"]["id"] - - # Now patch it — PATCH is not supported by the copilot execute_action service - patch_result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - { - "method": "PATCH", - "path": f"/api/v1/contacts/{company_id}", - "body": {"name": "PatchedCo"}, - }, - ) - assert patch_result["success"] is False - assert patch_result["status_code"] in (400, 403) # May be 403 if RBAC check runs first - assert patch_result["success"] is False # PATCH not supported or RBAC blocked - - -@pytest.mark.asyncio -async def test_service_execute_action_companies_patch_not_found(db_session): - """Service: execute_action with PATCH non-existent company returns 404.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - { - "method": "PATCH", - "path": "/api/v1/contacts/00000000-0000-0000-0000-000000000000", - "body": {"name": "X"}, - }, - ) - assert result["success"] is False - assert result["status_code"] in (400, 403) - - -@pytest.mark.asyncio -async def test_service_execute_action_companies_patch_no_id(db_session): - """Service: execute_action with PATCH /api/v1/companies/{id} returns 400.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "PATCH", "path": "/api/v1/companies/{id}", "body": {"name": "X"}}, - ) - assert result["success"] is False - assert result["status_code"] in (400, 403) - - -@pytest.mark.asyncio -async def test_service_execute_action_companies_delete(db_session): - """Service: execute_action with DELETE /api/v1/companies/{id} soft-deletes company.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = query_result["conversation_id"] - - create_result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "POST", "path": "/api/v1/contacts", "body": {"name": "DeleteMe", "type": "company"}}, - ) - company_id = create_result["data"]["id"] - - del_result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "DELETE", "path": f"/api/v1/contacts/{company_id}", "body": None}, - ) - assert del_result["success"] is False - assert del_result["status_code"] in (400, 403) - assert del_result["success"] is False # DELETE not supported or RBAC blocked - - -@pytest.mark.asyncio -async def test_service_execute_action_companies_delete_not_found(db_session): - """Service: execute_action with DELETE non-existent company returns 404.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - { - "method": "DELETE", - "path": "/api/v1/contacts/00000000-0000-0000-0000-000000000000", - "body": None, - }, - ) - assert result["success"] is False - assert result["status_code"] in (400, 403) - - -@pytest.mark.asyncio -async def test_service_execute_action_companies_delete_no_id(db_session): - """Service: execute_action with DELETE without ID returns 400.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "DELETE", "path": "/api/v1/companies/{id}", "body": None}, - ) - assert result["success"] is False - assert result["status_code"] in (400, 403) - - -@pytest.mark.asyncio -async def test_service_execute_action_contacts_get(db_session): - """Service: execute_action with GET /api/v1/contacts returns list.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List contact") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "GET", "path": "/api/v1/contacts", "body": None}, - ) - assert result["success"] is True - assert result["status_code"] == 200 - - -@pytest.mark.asyncio -async def test_service_execute_action_contacts_post(db_session): - """Service: execute_action with POST /api/v1/contacts — Contact model uses first_name/last_name, - service passes 'name' which causes error, verify error handling works.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List contact") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - { - "method": "POST", - "path": "/api/v1/contacts", - "body": {"name": "John Doe", "email": "john@example.com"}, - }, - ) - # Unified contact model accepts 'name' field and creates the contact successfully - assert result["success"] is True - assert result["status_code"] == 201 - assert result["data"]["name"] == "John Doe" - - -@pytest.mark.asyncio -async def test_service_execute_action_contacts_unsupported_method(db_session): - """Service: execute_action with DELETE /api/v1/contacts returns 400 (unsupported).""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List contact") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "DELETE", "path": "/api/v1/contacts/123", "body": None}, - ) - assert result["success"] is False - assert result["status_code"] in (400, 403) - - -@pytest.mark.asyncio -async def test_service_execute_action_workflows_get(db_session): - """Service: execute_action with GET /api/v1/workflows returns list.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List workflow") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "GET", "path": "/api/v1/workflows", "body": None}, - ) - assert result["success"] is True - assert result["status_code"] == 200 - - -@pytest.mark.asyncio -async def test_service_execute_action_workflows_unsupported_method(db_session): - """Service: execute_action with POST /api/v1/workflows returns 400 (unsupported).""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List workflow") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "POST", "path": "/api/v1/workflows", "body": {"name": "test"}}, - ) - assert result["success"] is False - assert result["status_code"] in (400, 403) - - -@pytest.mark.asyncio -async def test_service_execute_action_unsupported_entity(db_session): - """Service: execute_action with unknown entity returns 400.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "GET", "path": "/api/v1/unknown", "body": None}, - ) - assert result["success"] is False - assert result["status_code"] in (400, 403) - assert "Unsupported entity" in result["error"] - - -@pytest.mark.asyncio -async def test_service_execute_action_companies_unsupported_method(db_session): - """Service: execute_action with PUT /api/v1/companies returns 400 (unsupported).""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": True, "permissions": ["*:*"], "denied": []}, - conv_id, - {"method": "PUT", "path": "/api/v1/companies", "body": {}}, - ) - assert result["success"] is False - assert result["status_code"] in (400, 403) - - -@pytest.mark.asyncio -async def test_service_execute_action_invalid_conversation(db_session): - """Service: execute_action with invalid conversation_id returns 404.""" - from app.services.ai_copilot_service import execute_action - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - result = await execute_action( - db_session, - tenant_id, - admin_id, - "admin", - "00000000-0000-0000-0000-000000000000", - {"method": "GET", "path": "/api/v1/companies", "body": None}, - ) - assert result["error"] == "Conversation not found" - assert result["status_code"] in (404, 403) - - -@pytest.mark.asyncio -async def test_service_execute_action_rbac_blocked(db_session): - """Service: execute_action as viewer with DELETE returns 403.""" - from app.services.ai_copilot_service import execute_action, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - query_result = await process_query(db_session, tenant_id, admin_id, "List companies") - conv_id = query_result["conversation_id"] - - result = await execute_action( - db_session, - tenant_id, - admin_id, - {"is_system_admin": False, "permissions": ["contacts:read"], "denied": []}, - conv_id, - { - "method": "DELETE", - "path": "/api/v1/companies/00000000-0000-0000-0000-000000000000", - "body": None, - }, - ) - assert result["status_code"] == 403 - assert result["success"] is False - - -@pytest.mark.asyncio -async def test_service_get_history_pagination(db_session): - """Service: get_history returns paginated results.""" - from app.services.ai_copilot_service import get_history, process_query - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - # Create a few conversations - await process_query(db_session, tenant_id, admin_id, "List companies") - await process_query(db_session, tenant_id, admin_id, "List contacts") - - result = await get_history(db_session, tenant_id, admin_id, page=1, page_size=10) - assert result["total"] >= 2 - assert len(result["items"]) >= 2 - assert all("messages" in item for item in result["items"]) - - -@pytest.mark.asyncio -async def test_service_get_history_empty(db_session): - """Service: get_history returns empty when no conversations exist.""" - from app.services.ai_copilot_service import get_history - - seed = await seed_tenant_and_users(db_session) - tenant_id = seed["tenant_a"].id - admin_id = seed["admin_a"].id - - result = await get_history(db_session, tenant_id, admin_id) - assert result["total"] == 0 - assert result["items"] == [] - - -def test_service_derive_rbac_from_path_companies_get(): - """Service: _derive_rbac_from_path for GET /api/v1/companies → (companies, read).""" - from app.services.ai_copilot_service import _derive_rbac_from_path - - module, action = _derive_rbac_from_path("GET", "/api/v1/companies") - assert module == "companies" - assert action == "read" - - -def test_service_derive_rbac_from_path_contacts_post(): - """Service: _derive_rbac_from_path for POST /api/v1/contacts → (contacts, create).""" - from app.services.ai_copilot_service import _derive_rbac_from_path - - module, action = _derive_rbac_from_path("POST", "/api/v1/contacts") - assert module == "contacts" - assert action == "create" - - -def test_service_derive_rbac_from_path_workflows_patch(): - """Service: _derive_rbac_from_path for PATCH /api/v1/workflows → (workflows, update).""" - from app.services.ai_copilot_service import _derive_rbac_from_path - - module, action = _derive_rbac_from_path("PATCH", "/api/v1/workflows") - assert module == "workflows" - assert action == "update" - - -def test_service_derive_rbac_from_path_unknown_entity(): - """Service: _derive_rbac_from_path for unknown entity returns entity as module.""" - from app.services.ai_copilot_service import _derive_rbac_from_path - - module, action = _derive_rbac_from_path("DELETE", "/api/v1/foobar") - assert module == "foobar" - assert action == "delete" - - -def test_service_derive_rbac_from_path_ai_entity(): - """Service: _derive_rbac_from_path for /api/v1/ai → (ai_copilot, read).""" - from app.services.ai_copilot_service import _derive_rbac_from_path - - module, action = _derive_rbac_from_path("GET", "/api/v1/ai/copilot/query") - assert module == "ai_copilot" - assert action == "read" - - -def test_service_safe_iso_none(): - """Service: _safe_iso with None returns None.""" - from app.services.ai_copilot_service import _safe_iso - - assert _safe_iso(None) is None - - -def test_service_safe_iso_datetime(): - """Service: _safe_iso with datetime returns ISO string.""" - from datetime import datetime - - from app.services.ai_copilot_service import _safe_iso - - dt = datetime(2026, 1, 1, 12, 0, 0, tzinfo=UTC) - result = _safe_iso(dt) - assert result is not None - assert "2026-01-01" in result - - -def test_service_safe_iso_exception(): - """Service: _safe_iso with object that raises on isoformat returns None.""" - from app.services.ai_copilot_service import _safe_iso - - class Bad: - def isoformat(self): - raise ValueError("bad") - - assert _safe_iso(Bad()) is None - - -def test_service_get_attr_missing(): - """Service: _get_attr with missing attribute returns default.""" - from app.services.ai_copilot_service import _get_attr - - obj = type("Obj", (), {"x": 1})() - assert _get_attr(obj, "x") == 1 - assert _get_attr(obj, "y", "default") == "default" - - -def test_service_get_attr_none(): - """Service: _get_attr with None value returns default.""" - from app.services.ai_copilot_service import _get_attr - - obj = type("Obj", (), {"x": None})() - assert _get_attr(obj, "x", "fallback") == "fallback" - - -# ─── Route Error Path Tests for AI Copilot ─── - - -@pytest.mark.asyncio -async def test_route_copilot_execute_not_found(ai_client: AsyncClient, db_session): - """Route: POST /execute with invalid conversation_id returns 404.""" - await seed_tenant_and_users(db_session) - await login_client(ai_client, "admin@tenanta.com") - - resp = await ai_client.post( - "/api/v1/ai/copilot/execute", - json={ - "conversation_id": "00000000-0000-0000-0000-000000000000", - "action": { - "method": "GET", - "path": "/api/v1/companies", - "description": "List", - "confidence": 0.9, - }, - }, - ) - assert resp.status_code == 404 - - -@pytest.mark.asyncio -async def test_route_copilot_execute_rbac_blocked(ai_client: AsyncClient, db_session): - """Route: POST /execute as viewer with delete action returns 403.""" - await seed_tenant_and_users(db_session) - await login_client(ai_client, "viewer@tenanta.com") - - # First create a conversation as viewer - query_resp = await ai_client.post( - "/api/v1/ai/copilot/query", - json={ - "query": "Delete company", - "context": {"entity_id": "00000000-0000-0000-0000-000000000000"}, - }, - ) - if query_resp.status_code == 403: - return # RBAC blocked - expected - conv_id = query_resp.json()["conversation_id"] - action = query_resp.json()["proposed_actions"][0] - - exec_resp = await ai_client.post( - "/api/v1/ai/copilot/execute", - json={"conversation_id": conv_id, "action": action}, - ) - assert exec_resp.status_code == 403 - - -@pytest.mark.asyncio -async def test_route_copilot_history_unauthenticated(ai_client: AsyncClient, db_session): - """Route: GET /history without auth returns 401.""" - resp = await ai_client.get("/api/v1/ai/copilot/history") - assert resp.status_code == 401 - - -@pytest.mark.asyncio -async def test_route_copilot_execute_unauthenticated(ai_client: AsyncClient, db_session): - """Route: POST /execute without auth returns 403.""" - resp = await ai_client.post( - "/api/v1/ai/copilot/execute", - json={ - "conversation_id": "00000000-0000-0000-0000-000000000000", - "action": { - "method": "GET", - "path": "/api/v1/companies", - "description": "List", - "confidence": 0.9, - }, - }, - ) - assert resp.status_code == 403