File size: 6,715 Bytes
6aecb2e | 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 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | from datetime import datetime
from pathlib import Path
from types import SimpleNamespace
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
def _user(user_id: int, *, is_admin: bool = False):
return SimpleNamespace(id=user_id, is_admin=is_admin)
def _task(*, metadata=None, result=None, status="completed", task_type="pdf_generation"):
return SimpleNamespace(
task_id="task-1",
task_type=task_type,
status=SimpleNamespace(value=status),
progress=100.0,
created_at=datetime(2026, 4, 7, 12, 0, 0),
updated_at=datetime(2026, 4, 7, 12, 1, 0),
metadata=metadata or {},
result=result,
error=None,
)
def test_export_task_routes_require_authentication():
from landppt.web.route_modules import export_routes
app = FastAPI()
app.include_router(export_routes.router)
app.dependency_overrides[export_routes.get_current_user_required] = lambda: (_ for _ in ()).throw(
HTTPException(status_code=401, detail="Authentication required")
)
client = TestClient(app)
status_response = client.get("/api/landppt/tasks/task-1")
download_response = client.get("/api/landppt/tasks/task-1/download")
assert status_response.status_code == 401
assert download_response.status_code == 401
def test_export_task_status_hides_other_users_tasks(monkeypatch):
from landppt.services import background_tasks as background_tasks_module
from landppt.web.route_modules import export_routes
class FakeTaskManager:
async def get_task_async(self, task_id: str):
return _task(metadata={"project_id": "proj-1", "user_id": 22})
monkeypatch.setattr(background_tasks_module, "get_task_manager", lambda: FakeTaskManager())
app = FastAPI()
app.include_router(export_routes.router)
app.dependency_overrides[export_routes.get_current_user_required] = lambda: _user(11, is_admin=False)
client = TestClient(app)
response = client.get("/api/landppt/tasks/task-1")
assert response.status_code == 404
assert response.json()["detail"] == "Task not found"
def test_export_task_status_hides_ownerless_tasks_from_non_admin(monkeypatch):
from landppt.services import background_tasks as background_tasks_module
from landppt.web.route_modules import export_routes
class FakeTaskManager:
async def get_task_async(self, task_id: str):
return _task(metadata={"project_id": "proj-1"})
monkeypatch.setattr(background_tasks_module, "get_task_manager", lambda: FakeTaskManager())
app = FastAPI()
app.include_router(export_routes.router)
app.dependency_overrides[export_routes.get_current_user_required] = lambda: _user(11, is_admin=False)
client = TestClient(app)
response = client.get("/api/landppt/tasks/task-1")
assert response.status_code == 404
assert response.json()["detail"] == "Task not found"
def test_export_task_status_redacts_internal_paths(monkeypatch):
from landppt.services import background_tasks as background_tasks_module
from landppt.web.route_modules import export_routes
class FakeTaskManager:
async def get_task_async(self, task_id: str):
return _task(
metadata={
"project_id": "proj-1",
"project_topic": "Demo",
"user_id": 11,
"pdf_path": "/tmp/private.pdf",
"pptx_path": "/tmp/private.pptx",
"progress_message": "working",
},
result={
"success": True,
"pdf_path": "/tmp/private.pdf",
"pptx_path": "/tmp/private.pptx",
"project_topic": "Demo",
},
task_type="pdf_generation",
)
monkeypatch.setattr(background_tasks_module, "get_task_manager", lambda: FakeTaskManager())
app = FastAPI()
app.include_router(export_routes.router)
app.dependency_overrides[export_routes.get_current_user_required] = lambda: _user(11, is_admin=False)
client = TestClient(app)
response = client.get("/api/landppt/tasks/task-1")
assert response.status_code == 200
payload = response.json()
assert payload["message"] == "working"
assert payload["download_url"] == "/api/landppt/tasks/task-1/download"
assert payload["metadata"] == {
"project_id": "proj-1",
"project_topic": "Demo",
"user_id": 11,
"progress_message": "working",
}
assert payload["result"] == {
"success": True,
"project_topic": "Demo",
}
def test_export_task_download_allows_admin_bypass(monkeypatch, tmp_path):
from landppt.services import background_tasks as background_tasks_module
from landppt.web.route_modules import export_routes
pdf_path = tmp_path / "demo.pdf"
pdf_path.write_bytes(b"%PDF-1.4\n")
class FakeTaskManager:
async def get_task_async(self, task_id: str):
return _task(
metadata={"project_id": "proj-1", "project_topic": "Demo"},
result={"success": True, "pdf_path": str(pdf_path)},
task_type="pdf_generation",
)
monkeypatch.setattr(background_tasks_module, "get_task_manager", lambda: FakeTaskManager())
monkeypatch.setattr(background_tasks_module, "TaskStatus", SimpleNamespace(COMPLETED=_task().status))
app = FastAPI()
app.include_router(export_routes.router)
app.dependency_overrides[export_routes.get_current_user_required] = lambda: _user(1, is_admin=True)
client = TestClient(app)
response = client.get("/api/landppt/tasks/task-1/download")
assert response.status_code == 200
assert response.content.startswith(b"%PDF-1.4")
def test_export_task_route_source_includes_owner_scoping_changes():
source = Path("/root/clawd/src/landppt/web/route_modules/export_routes.py").read_text(encoding="utf-8")
assert 'metadata_filter={"project_id": project_id, "user_id": user.id}' in source
assert '"user_id": user.id' in source
assert '"metadata": _sanitize_task_mapping(task.metadata)' in source
assert 'response["result"] = _sanitize_task_mapping(task.result)' in source
assert "_ensure_task_access(task, user)" in source
def test_narration_task_route_source_includes_owner_scoping_changes():
source = Path("/root/clawd/src/landppt/web/route_modules/narration_routes.py").read_text(encoding="utf-8")
assert 'metadata_filter={"project_id": project_id, "language": language, "provider": provider, "user_id": user.id}' in source
assert '"user_id": user.id' in source
|