Spaces:
Running
Running
File size: 4,189 Bytes
7cc81cb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | 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
|