Spaces:
Running
Running
| from __future__ import annotations | |
| from pathlib import Path | |
| from uuid import uuid4 | |
| import pytest | |
| from pydantic import ValidationError | |
| from app.copilot.actions import CopilotActionRegistry | |
| from app.copilot.errors import CopilotInvalidRequestError | |
| from app.copilot.planner import CopilotPlanner | |
| from app.copilot.schemas import ( | |
| CopilotContext, | |
| CopilotEditorSummary, | |
| CopilotPlan, | |
| ) | |
| def context(*, capabilities: list[str], asset: bool = False, clip: bool = False): | |
| project_id = uuid4() | |
| return CopilotContext( | |
| workspace_id=str(uuid4()), | |
| project_id=project_id, | |
| selected_asset_ids=[uuid4()] if asset else [], | |
| selected_clip_ids=["clip-1"] if clip else [], | |
| editor_summary=CopilotEditorSummary( | |
| revision=4, duration_ms=30_000, track_count=1, clip_count=1 | |
| ), | |
| available_capabilities=capabilities, | |
| ) | |
| def test_planner_fails_closed_for_unavailable_transcription() -> None: | |
| plan = CopilotPlanner().plan( | |
| "Turn this podcast into a TikTok", | |
| context(capabilities=["editor.render"], asset=True), | |
| ) | |
| assert not plan.executable | |
| assert plan.unsupported_capabilities == ["ai.transcribe"] | |
| assert plan.actions == [] | |
| def test_planner_requires_confirmation_for_render_and_generation() -> None: | |
| render = CopilotPlanner().plan("Render this project", context(capabilities=["editor.render"])) | |
| assert render.executable and render.requires_confirmation | |
| assert render.actions[0].type == "editor.render" | |
| image = CopilotPlanner().plan( | |
| "Generate an image of a lighthouse", | |
| context(capabilities=["ai.generate_image"]), | |
| ) | |
| assert image.executable and image.requires_confirmation | |
| assert image.actions[0].type == "ai.generate_image" | |
| def test_action_plan_rejects_unknown_model_generated_structures() -> None: | |
| with pytest.raises(ValidationError): | |
| CopilotPlan.model_validate( | |
| { | |
| "intent": "unsafe", | |
| "explanation": "unsafe", | |
| "actions": [ | |
| { | |
| "id": "a", | |
| "type": "shell.execute", | |
| "arguments": {"command": "rm -rf /"}, | |
| "reason": "unsafe", | |
| "requires_confirmation": False, | |
| "destructive": False, | |
| "external_side_effect": False, | |
| "required_permission": "admin", | |
| "required_capability": "shell", | |
| } | |
| ], | |
| "missing_information": [], | |
| "unsupported_capabilities": [], | |
| "executable": True, | |
| "requires_confirmation": False, | |
| } | |
| ) | |
| def test_action_registry_rejects_policy_metadata_tampering() -> None: | |
| registry = CopilotActionRegistry( | |
| projects=None, # type: ignore[arg-type] | |
| assets=None, # type: ignore[arg-type] | |
| editor=None, # type: ignore[arg-type] | |
| renders=None, # type: ignore[arg-type] | |
| ai=None, # type: ignore[arg-type] | |
| templates=None, # type: ignore[arg-type] | |
| ) | |
| plan = CopilotPlanner().plan( | |
| "Generate an image of a lighthouse", | |
| context(capabilities=["ai.generate_image"]), | |
| ) | |
| tampered = plan.actions[0].model_copy(update={"requires_confirmation": False}) | |
| with pytest.raises(CopilotInvalidRequestError): | |
| registry.validate(tampered) | |
| def test_copilot_migration_is_additive_and_tenant_isolated() -> None: | |
| migration = ( | |
| (Path(__file__).resolve().parents[1] / "app/projects/migrations/0005_ai_copilot.sql") | |
| .read_text(encoding="utf-8") | |
| .lower() | |
| ) | |
| for expected in ( | |
| "create table if not exists copilot_runs", | |
| "unique (workspace_id, idempotency_key)", | |
| "enable row level security", | |
| "force row level security", | |
| "create policy copilot_runs_select", | |
| "create policy copilot_runs_insert", | |
| "create policy copilot_runs_update", | |
| "copilot run identity fields are immutable", | |
| ): | |
| assert expected in migration | |
| assert "drop table" not in migration | |