|
| 1 | +import uuid |
| 2 | + |
| 3 | +import pytest |
| 4 | +from fastapi import HTTPException |
| 5 | + |
| 6 | +from app.api import files as files_api |
| 7 | +from app.models.agent import Agent |
| 8 | +from app.models.user import User |
| 9 | + |
| 10 | + |
| 11 | +def make_user(**overrides): |
| 12 | + values = { |
| 13 | + "id": uuid.uuid4(), |
| 14 | + "display_name": "Alice", |
| 15 | + "role": "member", |
| 16 | + "tenant_id": uuid.uuid4(), |
| 17 | + "is_active": True, |
| 18 | + } |
| 19 | + values.update(overrides) |
| 20 | + return User(**values) |
| 21 | + |
| 22 | + |
| 23 | +def make_agent(creator_id: uuid.UUID, **overrides): |
| 24 | + values = { |
| 25 | + "id": uuid.uuid4(), |
| 26 | + "name": "Ops Bot", |
| 27 | + "role_description": "assistant", |
| 28 | + "creator_id": creator_id, |
| 29 | + "status": "idle", |
| 30 | + "agent_type": "native", |
| 31 | + } |
| 32 | + values.update(overrides) |
| 33 | + return Agent(**values) |
| 34 | + |
| 35 | + |
| 36 | +@pytest.mark.asyncio |
| 37 | +async def test_use_access_cannot_delete_agent_workspace_file(monkeypatch, tmp_path): |
| 38 | + user = make_user() |
| 39 | + agent = make_agent(uuid.uuid4(), tenant_id=user.tenant_id) |
| 40 | + workspace_file = tmp_path / str(agent.id) / "workspace" / "important.md" |
| 41 | + workspace_file.parent.mkdir(parents=True) |
| 42 | + workspace_file.write_text("do not delete", encoding="utf-8") |
| 43 | + |
| 44 | + async def fake_check_agent_access(_db, _current_user, _agent_id): |
| 45 | + return agent, "use" |
| 46 | + |
| 47 | + monkeypatch.setattr(files_api.settings, "AGENT_DATA_DIR", str(tmp_path)) |
| 48 | + monkeypatch.setattr(files_api, "check_agent_access", fake_check_agent_access) |
| 49 | + |
| 50 | + with pytest.raises(HTTPException) as exc: |
| 51 | + await files_api.delete_file( |
| 52 | + agent_id=agent.id, |
| 53 | + path="workspace/important.md", |
| 54 | + current_user=user, |
| 55 | + db=object(), |
| 56 | + ) |
| 57 | + |
| 58 | + assert exc.value.status_code == 403 |
| 59 | + assert workspace_file.exists() |
| 60 | + |
| 61 | + |
| 62 | +@pytest.mark.asyncio |
| 63 | +async def test_manage_access_can_delete_agent_workspace_file(monkeypatch, tmp_path): |
| 64 | + user = make_user() |
| 65 | + agent = make_agent(user.id, tenant_id=user.tenant_id) |
| 66 | + workspace_file = tmp_path / str(agent.id) / "workspace" / "obsolete.md" |
| 67 | + workspace_file.parent.mkdir(parents=True) |
| 68 | + workspace_file.write_text("delete me", encoding="utf-8") |
| 69 | + |
| 70 | + async def fake_check_agent_access(_db, _current_user, _agent_id): |
| 71 | + return agent, "manage" |
| 72 | + |
| 73 | + monkeypatch.setattr(files_api.settings, "AGENT_DATA_DIR", str(tmp_path)) |
| 74 | + monkeypatch.setattr(files_api, "check_agent_access", fake_check_agent_access) |
| 75 | + |
| 76 | + result = await files_api.delete_file( |
| 77 | + agent_id=agent.id, |
| 78 | + path="workspace/obsolete.md", |
| 79 | + current_user=user, |
| 80 | + db=object(), |
| 81 | + ) |
| 82 | + |
| 83 | + assert result == {"status": "ok", "path": "workspace/obsolete.md"} |
| 84 | + assert not workspace_file.exists() |
0 commit comments