FossilRepo

fossilrepo / tests / test_api_coverage.py
Source Blame History 1511 lines
254b467… ragelink 1 """Tests covering uncovered code paths in fossil/api_views.py.
254b467… ragelink 2
254b467… ragelink 3 Targets: batch API, workspace CRUD (list/create/detail/commit/merge/abandon),
254b467… ragelink 4 workspace ownership checks, SSE event stream internals, and _resolve_batch_route.
254b467… ragelink 5 Existing test_agent_coordination.py covers ticket claim/release/submit and review
254b467… ragelink 6 CRUD -- this file does NOT duplicate those.
254b467… ragelink 7 """
254b467… ragelink 8
254b467… ragelink 9 import json
254b467… ragelink 10 from unittest.mock import MagicMock, patch
254b467… ragelink 11
254b467… ragelink 12 import pytest
254b467… ragelink 13 from django.contrib.auth.models import User
254b467… ragelink 14 from django.test import Client, RequestFactory
254b467… ragelink 15
254b467… ragelink 16 from fossil.agent_claims import TicketClaim
254b467… ragelink 17 from fossil.branch_protection import BranchProtection
254b467… ragelink 18 from fossil.ci import StatusCheck
254b467… ragelink 19 from fossil.code_reviews import CodeReview
254b467… ragelink 20 from fossil.models import FossilRepository
254b467… ragelink 21 from fossil.workspaces import AgentWorkspace
254b467… ragelink 22 from organization.models import Team
254b467… ragelink 23 from projects.models import ProjectTeam
254b467… ragelink 24
254b467… ragelink 25 # ---- Fixtures ----
254b467… ragelink 26
254b467… ragelink 27
254b467… ragelink 28 @pytest.fixture
254b467… ragelink 29 def fossil_repo_obj(sample_project):
254b467… ragelink 30 """Return the auto-created FossilRepository for sample_project."""
254b467… ragelink 31 return FossilRepository.objects.get(project=sample_project, deleted_at__isnull=True)
254b467… ragelink 32
254b467… ragelink 33
254b467… ragelink 34 @pytest.fixture
254b467… ragelink 35 def writer_user(db, admin_user, sample_project):
254b467… ragelink 36 """Non-admin user with write access to the project."""
254b467… ragelink 37 writer = User.objects.create_user(username="writer_cov", password="testpass123")
254b467… ragelink 38 team = Team.objects.create(name="Cov Writers", organization=sample_project.organization, created_by=admin_user)
254b467… ragelink 39 team.members.add(writer)
254b467… ragelink 40 ProjectTeam.objects.create(project=sample_project, team=team, role="write", created_by=admin_user)
254b467… ragelink 41 return writer
254b467… ragelink 42
254b467… ragelink 43
254b467… ragelink 44 @pytest.fixture
254b467… ragelink 45 def writer_client(writer_user):
254b467… ragelink 46 c = Client()
254b467… ragelink 47 c.login(username="writer_cov", password="testpass123")
254b467… ragelink 48 return c
254b467… ragelink 49
254b467… ragelink 50
254b467… ragelink 51 @pytest.fixture
254b467… ragelink 52 def reader_user(db, admin_user, sample_project):
254b467… ragelink 53 """User with read-only access to the project."""
254b467… ragelink 54 reader = User.objects.create_user(username="reader_cov", password="testpass123")
254b467… ragelink 55 team = Team.objects.create(name="Cov Readers", organization=sample_project.organization, created_by=admin_user)
254b467… ragelink 56 team.members.add(reader)
254b467… ragelink 57 ProjectTeam.objects.create(project=sample_project, team=team, role="read", created_by=admin_user)
254b467… ragelink 58 return reader
254b467… ragelink 59
254b467… ragelink 60
254b467… ragelink 61 @pytest.fixture
254b467… ragelink 62 def reader_client(reader_user):
254b467… ragelink 63 c = Client()
254b467… ragelink 64 c.login(username="reader_cov", password="testpass123")
254b467… ragelink 65 return c
254b467… ragelink 66
254b467… ragelink 67
254b467… ragelink 68 @pytest.fixture
254b467… ragelink 69 def workspace(fossil_repo_obj, admin_user):
254b467… ragelink 70 """An active agent workspace with a checkout path."""
254b467… ragelink 71 return AgentWorkspace.objects.create(
254b467… ragelink 72 repository=fossil_repo_obj,
254b467… ragelink 73 name="ws-test-1",
254b467… ragelink 74 branch="workspace/ws-test-1",
254b467… ragelink 75 agent_id="claude-test",
254b467… ragelink 76 status="active",
254b467… ragelink 77 checkout_path="/tmp/fake-checkout",
254b467… ragelink 78 created_by=admin_user,
254b467… ragelink 79 )
254b467… ragelink 80
254b467… ragelink 81
254b467… ragelink 82 def _api_url(slug, path):
254b467… ragelink 83 return f"/projects/{slug}/fossil/{path}"
254b467… ragelink 84
254b467… ragelink 85
254b467… ragelink 86 # ---- Helper to build a mock subprocess.run result ----
254b467… ragelink 87
254b467… ragelink 88
254b467… ragelink 89 def _make_proc(returncode=0, stdout="", stderr=""):
254b467… ragelink 90 result = MagicMock()
254b467… ragelink 91 result.returncode = returncode
254b467… ragelink 92 result.stdout = stdout
254b467… ragelink 93 result.stderr = stderr
254b467… ragelink 94 return result
254b467… ragelink 95
254b467… ragelink 96
254b467… ragelink 97 class _SSEBreakError(Exception):
254b467… ragelink 98 """Raised from mocked time.sleep to break the SSE infinite loop."""
254b467… ragelink 99
254b467… ragelink 100
254b467… ragelink 101 def _drain_sse_one_iteration(response):
254b467… ragelink 102 """Read one iteration of the SSE generator, collecting yielded chunks.
254b467… ragelink 103
254b467… ragelink 104 The SSE event_stream is an infinite while-True generator with time.sleep(5)
254b467… ragelink 105 at the end of each iteration. We mock time.sleep to raise _SSEBreakError after
254b467… ragelink 106 yielding events from the first poll cycle.
254b467… ragelink 107 """
254b467… ragelink 108 events = []
254b467… ragelink 109 with patch("fossil.api_views.time.sleep", side_effect=_SSEBreakError):
254b467… ragelink 110 try:
254b467… ragelink 111 for chunk in response.streaming_content:
254b467… ragelink 112 # StreamingHttpResponse wraps generator output in map() for
254b467… ragelink 113 # encoding; chunks are bytes.
254b467… ragelink 114 if isinstance(chunk, bytes):
254b467… ragelink 115 chunk = chunk.decode("utf-8", errors="replace")
254b467… ragelink 116 events.append(chunk)
254b467… ragelink 117 except (_SSEBreakError, RuntimeError):
254b467… ragelink 118 pass
254b467… ragelink 119 return events
254b467… ragelink 120
254b467… ragelink 121
254b467… ragelink 122 def _drain_sse_n_iterations(response, n=3):
254b467… ragelink 123 """Read n iterations of the SSE generator."""
254b467… ragelink 124 call_count = 0
254b467… ragelink 125
254b467… ragelink 126 def _count_and_break(_seconds):
254b467… ragelink 127 nonlocal call_count
254b467… ragelink 128 call_count += 1
254b467… ragelink 129 if call_count >= n:
254b467… ragelink 130 raise _SSEBreakError
254b467… ragelink 131
254b467… ragelink 132 events = []
254b467… ragelink 133 with patch("fossil.api_views.time.sleep", side_effect=_count_and_break):
254b467… ragelink 134 try:
254b467… ragelink 135 for chunk in response.streaming_content:
254b467… ragelink 136 if isinstance(chunk, bytes):
254b467… ragelink 137 chunk = chunk.decode("utf-8", errors="replace")
254b467… ragelink 138 events.append(chunk)
254b467… ragelink 139 except (_SSEBreakError, RuntimeError):
254b467… ragelink 140 pass
254b467… ragelink 141 return events
254b467… ragelink 142
254b467… ragelink 143
254b467… ragelink 144 # ================================================================
254b467… ragelink 145 # Batch API
254b467… ragelink 146 # ================================================================
254b467… ragelink 147
254b467… ragelink 148
254b467… ragelink 149 @pytest.mark.django_db
254b467… ragelink 150 class TestBatchAPI:
254b467… ragelink 151 """Tests for POST /projects/<slug>/fossil/api/batch (lines 636-706)."""
254b467… ragelink 152
254b467… ragelink 153 def test_batch_success_with_multiple_sub_requests(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 154 """Batch call dispatches multiple GET sub-requests and returns combined results."""
254b467… ragelink 155 with patch("fossil.api_views.FossilReader") as mock_reader_cls:
254b467… ragelink 156 reader = mock_reader_cls.return_value
254b467… ragelink 157 reader.__enter__ = MagicMock(return_value=reader)
254b467… ragelink 158 reader.__exit__ = MagicMock(return_value=False)
254b467… ragelink 159 reader.get_timeline.return_value = []
254b467… ragelink 160 reader.get_checkin_count.return_value = 0
254b467… ragelink 161 reader.get_tickets.return_value = []
254b467… ragelink 162
254b467… ragelink 163 response = admin_client.post(
254b467… ragelink 164 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 165 data=json.dumps(
254b467… ragelink 166 {
254b467… ragelink 167 "requests": [
254b467… ragelink 168 {"method": "GET", "path": "/api/timeline"},
254b467… ragelink 169 {"method": "GET", "path": "/api/tickets"},
254b467… ragelink 170 ]
254b467… ragelink 171 }
254b467… ragelink 172 ),
254b467… ragelink 173 content_type="application/json",
254b467… ragelink 174 )
254b467… ragelink 175
254b467… ragelink 176 assert response.status_code == 200
254b467… ragelink 177 data = response.json()
254b467… ragelink 178 assert len(data["responses"]) == 2
254b467… ragelink 179 assert data["responses"][0]["status"] == 200
254b467… ragelink 180 assert "checkins" in data["responses"][0]["body"]
254b467… ragelink 181 assert data["responses"][1]["status"] == 200
254b467… ragelink 182 assert "tickets" in data["responses"][1]["body"]
254b467… ragelink 183
254b467… ragelink 184 def test_batch_wrong_method(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 185 """GET to batch endpoint returns 405."""
254b467… ragelink 186 response = admin_client.get(_api_url(sample_project.slug, "api/batch"))
254b467… ragelink 187 assert response.status_code == 405
254b467… ragelink 188
254b467… ragelink 189 def test_batch_invalid_json(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 190 """Non-JSON body returns 400."""
254b467… ragelink 191 response = admin_client.post(
254b467… ragelink 192 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 193 data="not json",
254b467… ragelink 194 content_type="application/json",
254b467… ragelink 195 )
254b467… ragelink 196 assert response.status_code == 400
254b467… ragelink 197 assert "Invalid JSON" in response.json()["error"]
254b467… ragelink 198
254b467… ragelink 199 def test_batch_requests_not_list(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 200 """'requests' must be a list."""
254b467… ragelink 201 response = admin_client.post(
254b467… ragelink 202 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 203 data=json.dumps({"requests": "not-a-list"}),
254b467… ragelink 204 content_type="application/json",
254b467… ragelink 205 )
254b467… ragelink 206 assert response.status_code == 400
254b467… ragelink 207 assert "'requests' must be a list" in response.json()["error"]
254b467… ragelink 208
254b467… ragelink 209 def test_batch_exceeds_max_requests(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 210 """More than 25 sub-requests returns 400."""
254b467… ragelink 211 response = admin_client.post(
254b467… ragelink 212 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 213 data=json.dumps({"requests": [{"method": "GET", "path": "/api/project"}] * 26}),
254b467… ragelink 214 content_type="application/json",
254b467… ragelink 215 )
254b467… ragelink 216 assert response.status_code == 400
254b467… ragelink 217 assert "Maximum 25" in response.json()["error"]
254b467… ragelink 218
254b467… ragelink 219 def test_batch_empty_requests(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 220 """Empty requests list returns empty responses."""
254b467… ragelink 221 response = admin_client.post(
254b467… ragelink 222 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 223 data=json.dumps({"requests": []}),
254b467… ragelink 224 content_type="application/json",
254b467… ragelink 225 )
254b467… ragelink 226 assert response.status_code == 200
254b467… ragelink 227 assert response.json()["responses"] == []
254b467… ragelink 228
254b467… ragelink 229 def test_batch_non_get_rejected(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 230 """Non-GET sub-requests are rejected with 405."""
254b467… ragelink 231 response = admin_client.post(
254b467… ragelink 232 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 233 data=json.dumps({"requests": [{"method": "POST", "path": "/api/project"}]}),
254b467… ragelink 234 content_type="application/json",
254b467… ragelink 235 )
254b467… ragelink 236 assert response.status_code == 200
254b467… ragelink 237 sub = response.json()["responses"][0]
254b467… ragelink 238 assert sub["status"] == 405
254b467… ragelink 239 assert "Only GET" in sub["body"]["error"]
254b467… ragelink 240
254b467… ragelink 241 def test_batch_unknown_path(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 242 """Unknown API path in batch returns 404 sub-response."""
254b467… ragelink 243 response = admin_client.post(
254b467… ragelink 244 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 245 data=json.dumps({"requests": [{"method": "GET", "path": "/api/nonexistent"}]}),
254b467… ragelink 246 content_type="application/json",
254b467… ragelink 247 )
254b467… ragelink 248 assert response.status_code == 200
254b467… ragelink 249 sub = response.json()["responses"][0]
254b467… ragelink 250 assert sub["status"] == 404
254b467… ragelink 251 assert "Unknown API path" in sub["body"]["error"]
254b467… ragelink 252
254b467… ragelink 253 def test_batch_missing_path(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 254 """Sub-request without 'path' returns 400 sub-response."""
254b467… ragelink 255 response = admin_client.post(
254b467… ragelink 256 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 257 data=json.dumps({"requests": [{"method": "GET"}]}),
254b467… ragelink 258 content_type="application/json",
254b467… ragelink 259 )
254b467… ragelink 260 assert response.status_code == 200
254b467… ragelink 261 sub = response.json()["responses"][0]
254b467… ragelink 262 assert sub["status"] == 400
254b467… ragelink 263 assert "Missing 'path'" in sub["body"]["error"]
254b467… ragelink 264
254b467… ragelink 265 def test_batch_non_dict_sub_request(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 266 """Non-dict items in requests list return 400 sub-response."""
254b467… ragelink 267 response = admin_client.post(
254b467… ragelink 268 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 269 data=json.dumps({"requests": ["not-a-dict"]}),
254b467… ragelink 270 content_type="application/json",
254b467… ragelink 271 )
254b467… ragelink 272 assert response.status_code == 200
254b467… ragelink 273 sub = response.json()["responses"][0]
254b467… ragelink 274 assert sub["status"] == 400
254b467… ragelink 275 assert "must be an object" in sub["body"]["error"]
254b467… ragelink 276
254b467… ragelink 277 def test_batch_dynamic_route_ticket_detail(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 278 """Batch can route to dynamic ticket detail path."""
254b467… ragelink 279 with patch("fossil.api_views.FossilReader") as mock_reader_cls:
254b467… ragelink 280 reader = mock_reader_cls.return_value
254b467… ragelink 281 reader.__enter__ = MagicMock(return_value=reader)
254b467… ragelink 282 reader.__exit__ = MagicMock(return_value=False)
254b467… ragelink 283 ticket = MagicMock()
254b467… ragelink 284 ticket.uuid = "abc123"
254b467… ragelink 285 ticket.title = "Test"
254b467… ragelink 286 ticket.status = "Open"
254b467… ragelink 287 ticket.type = "Bug"
254b467… ragelink 288 ticket.subsystem = ""
254b467… ragelink 289 ticket.priority = ""
254b467… ragelink 290 ticket.severity = ""
254b467… ragelink 291 ticket.resolution = ""
254b467… ragelink 292 ticket.body = ""
254b467… ragelink 293 ticket.created = None
254b467… ragelink 294 reader.get_ticket_detail.return_value = ticket
254b467… ragelink 295 reader.get_ticket_comments.return_value = []
254b467… ragelink 296
254b467… ragelink 297 response = admin_client.post(
254b467… ragelink 298 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 299 data=json.dumps({"requests": [{"method": "GET", "path": "/api/tickets/abc123"}]}),
254b467… ragelink 300 content_type="application/json",
254b467… ragelink 301 )
254b467… ragelink 302
254b467… ragelink 303 assert response.status_code == 200
254b467… ragelink 304 sub = response.json()["responses"][0]
254b467… ragelink 305 assert sub["status"] == 200
254b467… ragelink 306 assert sub["body"]["uuid"] == "abc123"
254b467… ragelink 307
254b467… ragelink 308 def test_batch_dynamic_route_wiki_page(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 309 """Batch can route to dynamic wiki page path."""
254b467… ragelink 310 with patch("fossil.api_views.FossilReader") as mock_reader_cls:
254b467… ragelink 311 reader = mock_reader_cls.return_value
254b467… ragelink 312 reader.__enter__ = MagicMock(return_value=reader)
254b467… ragelink 313 reader.__exit__ = MagicMock(return_value=False)
254b467… ragelink 314 page = MagicMock()
254b467… ragelink 315 page.name = "Home"
254b467… ragelink 316 page.content = "# Home"
254b467… ragelink 317 page.last_modified = None
254b467… ragelink 318 page.user = "admin"
254b467… ragelink 319 reader.get_wiki_page.return_value = page
254b467… ragelink 320
254b467… ragelink 321 with patch("fossil.views._render_fossil_content", return_value="<h1>Home</h1>"):
254b467… ragelink 322 response = admin_client.post(
254b467… ragelink 323 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 324 data=json.dumps({"requests": [{"method": "GET", "path": "/api/wiki/Home"}]}),
254b467… ragelink 325 content_type="application/json",
254b467… ragelink 326 )
254b467… ragelink 327
254b467… ragelink 328 assert response.status_code == 200
254b467… ragelink 329 sub = response.json()["responses"][0]
254b467… ragelink 330 assert sub["status"] == 200
254b467… ragelink 331 assert sub["body"]["name"] == "Home"
254b467… ragelink 332
254b467… ragelink 333 def test_batch_denied_for_anon(self, client, sample_project, fossil_repo_obj):
254b467… ragelink 334 """Anonymous users cannot use the batch API."""
254b467… ragelink 335 response = client.post(
254b467… ragelink 336 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 337 data=json.dumps({"requests": []}),
254b467… ragelink 338 content_type="application/json",
254b467… ragelink 339 )
254b467… ragelink 340 assert response.status_code == 401
254b467… ragelink 341
254b467… ragelink 342 def test_batch_sub_request_exception_returns_500(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 343 """When a sub-request raises an exception, we get a 500 sub-response."""
254b467… ragelink 344 with patch("fossil.api_views.FossilReader") as mock_reader_cls:
254b467… ragelink 345 mock_reader_cls.side_effect = RuntimeError("boom")
254b467… ragelink 346
254b467… ragelink 347 response = admin_client.post(
254b467… ragelink 348 _api_url(sample_project.slug, "api/batch"),
254b467… ragelink 349 data=json.dumps({"requests": [{"method": "GET", "path": "/api/timeline"}]}),
254b467… ragelink 350 content_type="application/json",
254b467… ragelink 351 )
254b467… ragelink 352
254b467… ragelink 353 assert response.status_code == 200
254b467… ragelink 354 sub = response.json()["responses"][0]
254b467… ragelink 355 assert sub["status"] == 500
254b467… ragelink 356 assert "Internal error" in sub["body"]["error"]
254b467… ragelink 357
254b467… ragelink 358
254b467… ragelink 359 # ================================================================
254b467… ragelink 360 # Workspace List
254b467… ragelink 361 # ================================================================
254b467… ragelink 362
254b467… ragelink 363
254b467… ragelink 364 @pytest.mark.django_db
254b467… ragelink 365 class TestWorkspaceList:
254b467… ragelink 366 """Tests for GET /projects/<slug>/fossil/api/workspaces (lines 749-786)."""
254b467… ragelink 367
254b467… ragelink 368 def test_list_workspaces_empty(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 369 """Empty workspace list returns zero results."""
254b467… ragelink 370 response = admin_client.get(_api_url(sample_project.slug, "api/workspaces"))
254b467… ragelink 371 assert response.status_code == 200
254b467… ragelink 372 data = response.json()
254b467… ragelink 373 assert data["workspaces"] == []
254b467… ragelink 374
254b467… ragelink 375 def test_list_workspaces_returns_all(self, admin_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 376 """Lists all workspaces for the repo."""
254b467… ragelink 377 AgentWorkspace.objects.create(
254b467… ragelink 378 repository=fossil_repo_obj, name="ws-1", branch="workspace/ws-1", agent_id="a1", created_by=admin_user
254b467… ragelink 379 )
254b467… ragelink 380 AgentWorkspace.objects.create(
254b467… ragelink 381 repository=fossil_repo_obj, name="ws-2", branch="workspace/ws-2", agent_id="a2", created_by=admin_user
254b467… ragelink 382 )
254b467… ragelink 383
254b467… ragelink 384 response = admin_client.get(_api_url(sample_project.slug, "api/workspaces"))
254b467… ragelink 385 assert response.status_code == 200
254b467… ragelink 386 data = response.json()
254b467… ragelink 387 assert len(data["workspaces"]) == 2
254b467… ragelink 388 names = {ws["name"] for ws in data["workspaces"]}
254b467… ragelink 389 assert names == {"ws-1", "ws-2"}
254b467… ragelink 390
254b467… ragelink 391 def test_list_workspaces_filter_by_status(self, admin_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 392 """Status filter returns only matching workspaces."""
254b467… ragelink 393 AgentWorkspace.objects.create(repository=fossil_repo_obj, name="ws-active", branch="b/a", status="active", created_by=admin_user)
254b467… ragelink 394 AgentWorkspace.objects.create(repository=fossil_repo_obj, name="ws-merged", branch="b/m", status="merged", created_by=admin_user)
254b467… ragelink 395
254b467… ragelink 396 response = admin_client.get(_api_url(sample_project.slug, "api/workspaces") + "?status=active")
254b467… ragelink 397 assert response.status_code == 200
254b467… ragelink 398 data = response.json()
254b467… ragelink 399 assert len(data["workspaces"]) == 1
254b467… ragelink 400 assert data["workspaces"][0]["name"] == "ws-active"
254b467… ragelink 401
254b467… ragelink 402 def test_list_workspaces_wrong_method(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 403 """POST to workspace list returns 405."""
254b467… ragelink 404 response = admin_client.post(
254b467… ragelink 405 _api_url(sample_project.slug, "api/workspaces"),
254b467… ragelink 406 content_type="application/json",
254b467… ragelink 407 )
254b467… ragelink 408 assert response.status_code == 405
254b467… ragelink 409
254b467… ragelink 410 def test_list_workspaces_denied_for_anon(self, client, sample_project, fossil_repo_obj):
254b467… ragelink 411 """Anonymous users cannot list workspaces."""
254b467… ragelink 412 response = client.get(_api_url(sample_project.slug, "api/workspaces"))
254b467… ragelink 413 assert response.status_code == 401
254b467… ragelink 414
254b467… ragelink 415 def test_list_workspaces_response_shape(self, admin_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 416 """Verify the response includes all expected fields."""
254b467… ragelink 417 AgentWorkspace.objects.create(
254b467… ragelink 418 repository=fossil_repo_obj,
254b467… ragelink 419 name="ws-shape",
254b467… ragelink 420 branch="workspace/ws-shape",
254b467… ragelink 421 agent_id="claude-shape",
254b467… ragelink 422 description="test workspace",
254b467… ragelink 423 files_changed=3,
254b467… ragelink 424 commits_made=2,
254b467… ragelink 425 created_by=admin_user,
254b467… ragelink 426 )
254b467… ragelink 427 response = admin_client.get(_api_url(sample_project.slug, "api/workspaces"))
254b467… ragelink 428 ws = response.json()["workspaces"][0]
254b467… ragelink 429 assert ws["name"] == "ws-shape"
254b467… ragelink 430 assert ws["branch"] == "workspace/ws-shape"
254b467… ragelink 431 assert ws["status"] == "active"
254b467… ragelink 432 assert ws["agent_id"] == "claude-shape"
254b467… ragelink 433 assert ws["description"] == "test workspace"
254b467… ragelink 434 assert ws["files_changed"] == 3
254b467… ragelink 435 assert ws["commits_made"] == 2
254b467… ragelink 436 assert ws["created_at"] is not None
254b467… ragelink 437
254b467… ragelink 438
254b467… ragelink 439 # ================================================================
254b467… ragelink 440 # Workspace Detail
254b467… ragelink 441 # ================================================================
254b467… ragelink 442
254b467… ragelink 443
254b467… ragelink 444 @pytest.mark.django_db
254b467… ragelink 445 class TestWorkspaceDetail:
254b467… ragelink 446 """Tests for GET /projects/<slug>/fossil/api/workspaces/<name> (lines 904-934)."""
254b467… ragelink 447
254b467… ragelink 448 def test_get_workspace_detail(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 449 """Workspace detail returns all fields."""
254b467… ragelink 450 response = admin_client.get(_api_url(sample_project.slug, "api/workspaces/ws-test-1"))
254b467… ragelink 451 assert response.status_code == 200
254b467… ragelink 452 data = response.json()
254b467… ragelink 453 assert data["name"] == "ws-test-1"
254b467… ragelink 454 assert data["branch"] == "workspace/ws-test-1"
254b467… ragelink 455 assert data["agent_id"] == "claude-test"
254b467… ragelink 456 assert data["status"] == "active"
254b467… ragelink 457 assert data["updated_at"] is not None
254b467… ragelink 458
254b467… ragelink 459 def test_get_workspace_not_found(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 460 """Non-existent workspace returns 404."""
254b467… ragelink 461 response = admin_client.get(_api_url(sample_project.slug, "api/workspaces/nonexistent"))
254b467… ragelink 462 assert response.status_code == 404
254b467… ragelink 463 assert "not found" in response.json()["error"].lower()
254b467… ragelink 464
254b467… ragelink 465 def test_get_workspace_wrong_method(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 466 """POST to workspace detail returns 405."""
254b467… ragelink 467 response = admin_client.post(
254b467… ragelink 468 _api_url(sample_project.slug, "api/workspaces/ws-test-1"),
254b467… ragelink 469 content_type="application/json",
254b467… ragelink 470 )
254b467… ragelink 471 assert response.status_code == 405
254b467… ragelink 472
254b467… ragelink 473 def test_get_workspace_denied_for_anon(self, client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 474 """Anonymous users cannot view workspace details."""
254b467… ragelink 475 response = client.get(_api_url(sample_project.slug, "api/workspaces/ws-test-1"))
254b467… ragelink 476 assert response.status_code == 401
254b467… ragelink 477
254b467… ragelink 478
254b467… ragelink 479 # ================================================================
254b467… ragelink 480 # Workspace Create
254b467… ragelink 481 # ================================================================
254b467… ragelink 482
254b467… ragelink 483
254b467… ragelink 484 @pytest.mark.django_db
254b467… ragelink 485 class TestWorkspaceCreate:
254b467… ragelink 486 """Tests for POST /projects/<slug>/fossil/api/workspaces/create (lines 789-901)."""
254b467… ragelink 487
254b467… ragelink 488 def test_create_workspace_success(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 489 """Creating a workspace opens a Fossil checkout and creates DB record."""
254b467… ragelink 490 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls:
254b467… ragelink 491 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 492 mock_cli_cls.return_value._env = {}
254b467… ragelink 493 # All three subprocess calls succeed: open, branch new, update
254b467… ragelink 494 mock_run.return_value = _make_proc(stdout="checkout opened")
254b467… ragelink 495
254b467… ragelink 496 response = admin_client.post(
254b467… ragelink 497 _api_url(sample_project.slug, "api/workspaces/create"),
254b467… ragelink 498 data=json.dumps({"name": "agent-fix-99", "description": "Fix bug #99", "agent_id": "claude-99"}),
254b467… ragelink 499 content_type="application/json",
254b467… ragelink 500 )
254b467… ragelink 501
254b467… ragelink 502 assert response.status_code == 201
254b467… ragelink 503 data = response.json()
254b467… ragelink 504 assert data["name"] == "agent-fix-99"
254b467… ragelink 505 assert data["branch"] == "workspace/agent-fix-99"
254b467… ragelink 506 assert data["status"] == "active"
254b467… ragelink 507 assert data["agent_id"] == "claude-99"
254b467… ragelink 508
254b467… ragelink 509 # Verify DB state
254b467… ragelink 510 ws = AgentWorkspace.objects.get(repository=fossil_repo_obj, name="agent-fix-99")
254b467… ragelink 511 assert ws.branch == "workspace/agent-fix-99"
254b467… ragelink 512 assert ws.description == "Fix bug #99"
254b467… ragelink 513
254b467… ragelink 514 def test_create_workspace_missing_name(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 515 """Workspace name is required."""
254b467… ragelink 516 response = admin_client.post(
254b467… ragelink 517 _api_url(sample_project.slug, "api/workspaces/create"),
254b467… ragelink 518 data=json.dumps({"description": "no name"}),
254b467… ragelink 519 content_type="application/json",
254b467… ragelink 520 )
254b467… ragelink 521 assert response.status_code == 400
254b467… ragelink 522 assert "name" in response.json()["error"].lower()
254b467… ragelink 523
254b467… ragelink 524 def test_create_workspace_invalid_name(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 525 """Invalid workspace name (special chars) returns 400."""
254b467… ragelink 526 response = admin_client.post(
254b467… ragelink 527 _api_url(sample_project.slug, "api/workspaces/create"),
254b467… ragelink 528 data=json.dumps({"name": "../../etc/passwd"}),
254b467… ragelink 529 content_type="application/json",
254b467… ragelink 530 )
254b467… ragelink 531 assert response.status_code == 400
254b467… ragelink 532 assert "Invalid workspace name" in response.json()["error"]
254b467… ragelink 533
254b467… ragelink 534 def test_create_workspace_name_starts_with_dot(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 535 """Workspace name starting with a dot is rejected by the regex."""
254b467… ragelink 536 response = admin_client.post(
254b467… ragelink 537 _api_url(sample_project.slug, "api/workspaces/create"),
254b467… ragelink 538 data=json.dumps({"name": ".hidden"}),
254b467… ragelink 539 content_type="application/json",
254b467… ragelink 540 )
254b467… ragelink 541 assert response.status_code == 400
254b467… ragelink 542
254b467… ragelink 543 def test_create_workspace_duplicate_name(self, admin_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 544 """Duplicate workspace name returns 409."""
254b467… ragelink 545 AgentWorkspace.objects.create(repository=fossil_repo_obj, name="dup-ws", branch="workspace/dup-ws", created_by=admin_user)
254b467… ragelink 546
254b467… ragelink 547 response = admin_client.post(
254b467… ragelink 548 _api_url(sample_project.slug, "api/workspaces/create"),
254b467… ragelink 549 data=json.dumps({"name": "dup-ws"}),
254b467… ragelink 550 content_type="application/json",
254b467… ragelink 551 )
254b467… ragelink 552 assert response.status_code == 409
254b467… ragelink 553 assert "already exists" in response.json()["error"]
254b467… ragelink 554
254b467… ragelink 555 def test_create_workspace_invalid_json(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 556 """Invalid JSON body returns 400."""
254b467… ragelink 557 response = admin_client.post(
254b467… ragelink 558 _api_url(sample_project.slug, "api/workspaces/create"),
254b467… ragelink 559 data="not json",
254b467… ragelink 560 content_type="application/json",
254b467… ragelink 561 )
254b467… ragelink 562 assert response.status_code == 400
254b467… ragelink 563 assert "Invalid JSON" in response.json()["error"]
254b467… ragelink 564
254b467… ragelink 565 def test_create_workspace_fossil_open_fails(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 566 """When fossil open fails, return 500 and clean up."""
254b467… ragelink 567 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls, patch("shutil.rmtree"):
254b467… ragelink 568 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 569 mock_cli_cls.return_value._env = {}
254b467… ragelink 570 mock_run.return_value = _make_proc(returncode=1, stderr="open failed")
254b467… ragelink 571
254b467… ragelink 572 response = admin_client.post(
254b467… ragelink 573 _api_url(sample_project.slug, "api/workspaces/create"),
254b467… ragelink 574 data=json.dumps({"name": "fail-open"}),
254b467… ragelink 575 content_type="application/json",
254b467… ragelink 576 )
254b467… ragelink 577
254b467… ragelink 578 assert response.status_code == 500
254b467… ragelink 579 assert "Failed to open" in response.json()["error"]
254b467… ragelink 580
254b467… ragelink 581 def test_create_workspace_branch_creation_fails(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 582 """When branch creation fails, return 500 and clean up checkout."""
254b467… ragelink 583 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls, patch("shutil.rmtree"):
254b467… ragelink 584 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 585 mock_cli_cls.return_value._env = {}
254b467… ragelink 586 # First call (open) succeeds, second (branch new) fails
254b467… ragelink 587 mock_run.side_effect = [
254b467… ragelink 588 _make_proc(returncode=0), # open
254b467… ragelink 589 _make_proc(returncode=1, stderr="branch error"), # branch new
254b467… ragelink 590 _make_proc(returncode=0), # close --force (cleanup)
254b467… ragelink 591 ]
254b467… ragelink 592
254b467… ragelink 593 response = admin_client.post(
254b467… ragelink 594 _api_url(sample_project.slug, "api/workspaces/create"),
254b467… ragelink 595 data=json.dumps({"name": "fail-branch"}),
254b467… ragelink 596 content_type="application/json",
254b467… ragelink 597 )
254b467… ragelink 598
254b467… ragelink 599 assert response.status_code == 500
254b467… ragelink 600 assert "Failed to create branch" in response.json()["error"]
254b467… ragelink 601
254b467… ragelink 602 def test_create_workspace_update_fails(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 603 """When switching to the new branch fails, return 500 and clean up."""
254b467… ragelink 604 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls, patch("shutil.rmtree"):
254b467… ragelink 605 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 606 mock_cli_cls.return_value._env = {}
254b467… ragelink 607 mock_run.side_effect = [
254b467… ragelink 608 _make_proc(returncode=0), # open
254b467… ragelink 609 _make_proc(returncode=0), # branch new
254b467… ragelink 610 _make_proc(returncode=1, stderr="update failed"), # update branch
254b467… ragelink 611 _make_proc(returncode=0), # close --force (cleanup)
254b467… ragelink 612 ]
254b467… ragelink 613
254b467… ragelink 614 response = admin_client.post(
254b467… ragelink 615 _api_url(sample_project.slug, "api/workspaces/create"),
254b467… ragelink 616 data=json.dumps({"name": "fail-update"}),
254b467… ragelink 617 content_type="application/json",
254b467… ragelink 618 )
254b467… ragelink 619
254b467… ragelink 620 assert response.status_code == 500
254b467… ragelink 621 assert "Failed to switch to branch" in response.json()["error"]
254b467… ragelink 622
254b467… ragelink 623 def test_create_workspace_wrong_method(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 624 """GET to create endpoint returns 405."""
254b467… ragelink 625 response = admin_client.get(_api_url(sample_project.slug, "api/workspaces/create"))
254b467… ragelink 626 assert response.status_code == 405
254b467… ragelink 627
254b467… ragelink 628 def test_create_workspace_denied_for_reader(self, reader_client, sample_project, fossil_repo_obj):
254b467… ragelink 629 """Read-only users cannot create workspaces."""
254b467… ragelink 630 response = reader_client.post(
254b467… ragelink 631 _api_url(sample_project.slug, "api/workspaces/create"),
254b467… ragelink 632 data=json.dumps({"name": "denied-ws"}),
254b467… ragelink 633 content_type="application/json",
254b467… ragelink 634 )
254b467… ragelink 635 assert response.status_code == 403
254b467… ragelink 636
254b467… ragelink 637 def test_create_workspace_denied_for_anon(self, client, sample_project, fossil_repo_obj):
254b467… ragelink 638 """Anonymous users cannot create workspaces."""
254b467… ragelink 639 response = client.post(
254b467… ragelink 640 _api_url(sample_project.slug, "api/workspaces/create"),
254b467… ragelink 641 data=json.dumps({"name": "anon-ws"}),
254b467… ragelink 642 content_type="application/json",
254b467… ragelink 643 )
254b467… ragelink 644 assert response.status_code == 401
254b467… ragelink 645
254b467… ragelink 646
254b467… ragelink 647 # ================================================================
254b467… ragelink 648 # Workspace Commit
254b467… ragelink 649 # ================================================================
254b467… ragelink 650
254b467… ragelink 651
254b467… ragelink 652 @pytest.mark.django_db
254b467… ragelink 653 class TestWorkspaceCommit:
254b467… ragelink 654 """Tests for POST /projects/<slug>/fossil/api/workspaces/<name>/commit (lines 937-1034)."""
254b467… ragelink 655
254b467… ragelink 656 def test_commit_success(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 657 """Successful commit increments commits_made and returns output."""
254b467… ragelink 658 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls:
254b467… ragelink 659 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 660 mock_cli_cls.return_value._env = {}
254b467… ragelink 661 # addremove then commit
254b467… ragelink 662 mock_run.side_effect = [
254b467… ragelink 663 _make_proc(returncode=0), # addremove
254b467… ragelink 664 _make_proc(returncode=0, stdout="New_Version: abc123"), # commit
254b467… ragelink 665 ]
254b467… ragelink 666
254b467… ragelink 667 response = admin_client.post(
254b467… ragelink 668 _api_url(sample_project.slug, "api/workspaces/ws-test-1/commit"),
254b467… ragelink 669 data=json.dumps({"message": "Fix bug", "agent_id": "claude-test"}),
254b467… ragelink 670 content_type="application/json",
254b467… ragelink 671 )
254b467… ragelink 672
254b467… ragelink 673 assert response.status_code == 200
254b467… ragelink 674 data = response.json()
254b467… ragelink 675 assert data["message"] == "Fix bug"
254b467… ragelink 676 assert data["commits_made"] == 1
254b467… ragelink 677
254b467… ragelink 678 workspace.refresh_from_db()
254b467… ragelink 679 assert workspace.commits_made == 1
254b467… ragelink 680
254b467… ragelink 681 def test_commit_with_specific_files(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 682 """Committing specific files adds them individually."""
254b467… ragelink 683 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls:
254b467… ragelink 684 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 685 mock_cli_cls.return_value._env = {}
254b467… ragelink 686 mock_run.side_effect = [
254b467… ragelink 687 _make_proc(returncode=0), # add file1
254b467… ragelink 688 _make_proc(returncode=0), # add file2
254b467… ragelink 689 _make_proc(returncode=0, stdout="New_Version: def456"), # commit
254b467… ragelink 690 ]
254b467… ragelink 691
254b467… ragelink 692 response = admin_client.post(
254b467… ragelink 693 _api_url(sample_project.slug, "api/workspaces/ws-test-1/commit"),
254b467… ragelink 694 data=json.dumps({"message": "Add files", "files": ["a.py", "b.py"], "agent_id": "claude-test"}),
254b467… ragelink 695 content_type="application/json",
254b467… ragelink 696 )
254b467… ragelink 697
254b467… ragelink 698 assert response.status_code == 200
254b467… ragelink 699
254b467… ragelink 700 def test_commit_nothing_to_commit(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 701 """When fossil says nothing changed, return 409."""
254b467… ragelink 702 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls:
254b467… ragelink 703 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 704 mock_cli_cls.return_value._env = {}
254b467… ragelink 705 mock_run.side_effect = [
254b467… ragelink 706 _make_proc(returncode=0), # addremove
254b467… ragelink 707 _make_proc(returncode=1, stderr="nothing has changed"), # commit
254b467… ragelink 708 ]
254b467… ragelink 709
254b467… ragelink 710 response = admin_client.post(
254b467… ragelink 711 _api_url(sample_project.slug, "api/workspaces/ws-test-1/commit"),
254b467… ragelink 712 data=json.dumps({"message": "no change", "agent_id": "claude-test"}),
254b467… ragelink 713 content_type="application/json",
254b467… ragelink 714 )
254b467… ragelink 715
254b467… ragelink 716 assert response.status_code == 409
254b467… ragelink 717 assert "Nothing to commit" in response.json()["error"]
254b467… ragelink 718
254b467… ragelink 719 def test_commit_fossil_error(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 720 """When fossil commit fails (not nothing-changed), return 500."""
254b467… ragelink 721 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls:
254b467… ragelink 722 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 723 mock_cli_cls.return_value._env = {}
254b467… ragelink 724 mock_run.side_effect = [
254b467… ragelink 725 _make_proc(returncode=0), # addremove
254b467… ragelink 726 _make_proc(returncode=1, stderr="lock failed"), # commit
254b467… ragelink 727 ]
254b467… ragelink 728
254b467… ragelink 729 response = admin_client.post(
254b467… ragelink 730 _api_url(sample_project.slug, "api/workspaces/ws-test-1/commit"),
254b467… ragelink 731 data=json.dumps({"message": "fail commit", "agent_id": "claude-test"}),
254b467… ragelink 732 content_type="application/json",
254b467… ragelink 733 )
254b467… ragelink 734
254b467… ragelink 735 assert response.status_code == 500
254b467… ragelink 736 assert "Commit failed" in response.json()["error"]
254b467… ragelink 737
254b467… ragelink 738 def test_commit_missing_message(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 739 """Commit without message returns 400."""
254b467… ragelink 740 response = admin_client.post(
254b467… ragelink 741 _api_url(sample_project.slug, "api/workspaces/ws-test-1/commit"),
254b467… ragelink 742 data=json.dumps({"agent_id": "claude-test"}),
254b467… ragelink 743 content_type="application/json",
254b467… ragelink 744 )
254b467… ragelink 745 assert response.status_code == 400
254b467… ragelink 746 assert "message" in response.json()["error"].lower()
254b467… ragelink 747
254b467… ragelink 748 def test_commit_workspace_not_found(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 749 """Commit to non-existent workspace returns 404."""
254b467… ragelink 750 response = admin_client.post(
254b467… ragelink 751 _api_url(sample_project.slug, "api/workspaces/nonexistent/commit"),
254b467… ragelink 752 data=json.dumps({"message": "fix"}),
254b467… ragelink 753 content_type="application/json",
254b467… ragelink 754 )
254b467… ragelink 755 assert response.status_code == 404
254b467… ragelink 756
254b467… ragelink 757 def test_commit_workspace_not_active(self, admin_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 758 """Commit to a merged workspace returns 409."""
254b467… ragelink 759 AgentWorkspace.objects.create(
254b467… ragelink 760 repository=fossil_repo_obj,
254b467… ragelink 761 name="ws-merged",
254b467… ragelink 762 branch="workspace/ws-merged",
254b467… ragelink 763 status="merged",
254b467… ragelink 764 created_by=admin_user,
254b467… ragelink 765 )
254b467… ragelink 766
254b467… ragelink 767 response = admin_client.post(
254b467… ragelink 768 _api_url(sample_project.slug, "api/workspaces/ws-merged/commit"),
254b467… ragelink 769 data=json.dumps({"message": "too late"}),
254b467… ragelink 770 content_type="application/json",
254b467… ragelink 771 )
254b467… ragelink 772 assert response.status_code == 409
254b467… ragelink 773 assert "merged" in response.json()["error"]
254b467… ragelink 774
254b467… ragelink 775 def test_commit_invalid_json(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 776 """Invalid JSON body returns 400."""
254b467… ragelink 777 response = admin_client.post(
254b467… ragelink 778 _api_url(sample_project.slug, "api/workspaces/ws-test-1/commit"),
254b467… ragelink 779 data="not json",
254b467… ragelink 780 content_type="application/json",
254b467… ragelink 781 )
254b467… ragelink 782 assert response.status_code == 400
254b467… ragelink 783
254b467… ragelink 784 def test_commit_wrong_method(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 785 """GET to commit endpoint returns 405."""
254b467… ragelink 786 response = admin_client.get(_api_url(sample_project.slug, "api/workspaces/ws-test-1/commit"))
254b467… ragelink 787 assert response.status_code == 405
254b467… ragelink 788
254b467… ragelink 789 def test_commit_denied_for_reader(self, reader_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 790 """Read-only users cannot commit."""
254b467… ragelink 791 response = reader_client.post(
254b467… ragelink 792 _api_url(sample_project.slug, "api/workspaces/ws-test-1/commit"),
254b467… ragelink 793 data=json.dumps({"message": "denied"}),
254b467… ragelink 794 content_type="application/json",
254b467… ragelink 795 )
254b467… ragelink 796 assert response.status_code == 403
254b467… ragelink 797
254b467… ragelink 798
254b467… ragelink 799 # ================================================================
254b467… ragelink 800 # Workspace Merge
254b467… ragelink 801 # ================================================================
254b467… ragelink 802
254b467… ragelink 803
254b467… ragelink 804 @pytest.mark.django_db
254b467… ragelink 805 class TestWorkspaceMerge:
254b467… ragelink 806 """Tests for POST /projects/<slug>/fossil/api/workspaces/<name>/merge (lines 1037-1185).
254b467… ragelink 807
254b467… ragelink 808 This endpoint is complex: it enforces branch protection, review gates,
254b467… ragelink 809 and runs three subprocess calls (update, merge, commit).
254b467… ragelink 810 """
254b467… ragelink 811
254b467… ragelink 812 def test_merge_success_admin_bypass(self, admin_client, sample_project, fossil_repo_obj, workspace, admin_user):
254b467… ragelink 813 """Admin can merge without an approved review (admin bypass of review gate)."""
254b467… ragelink 814 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls, patch("shutil.rmtree"):
254b467… ragelink 815 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 816 mock_cli_cls.return_value._env = {}
254b467… ragelink 817 mock_run.side_effect = [
254b467… ragelink 818 _make_proc(returncode=0), # update trunk
254b467… ragelink 819 _make_proc(returncode=0, stdout="merged ok"), # merge
254b467… ragelink 820 _make_proc(returncode=0, stdout="committed"), # commit
254b467… ragelink 821 _make_proc(returncode=0), # close --force
254b467… ragelink 822 ]
254b467… ragelink 823
254b467… ragelink 824 response = admin_client.post(
254b467… ragelink 825 _api_url(sample_project.slug, "api/workspaces/ws-test-1/merge"),
254b467… ragelink 826 data=json.dumps({"target_branch": "trunk", "agent_id": "claude-test"}),
254b467… ragelink 827 content_type="application/json",
254b467… ragelink 828 )
254b467… ragelink 829
254b467… ragelink 830 assert response.status_code == 200
254b467… ragelink 831 data = response.json()
254b467… ragelink 832 assert data["status"] == "merged"
254b467… ragelink 833 assert data["target_branch"] == "trunk"
254b467… ragelink 834
254b467… ragelink 835 workspace.refresh_from_db()
254b467… ragelink 836 assert workspace.status == "merged"
254b467… ragelink 837 assert workspace.checkout_path == ""
254b467… ragelink 838
254b467… ragelink 839 def test_merge_with_approved_review(self, writer_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 840 """Non-admin writer can merge if an approved review exists for the workspace."""
254b467… ragelink 841 ws = AgentWorkspace.objects.create(
254b467… ragelink 842 repository=fossil_repo_obj,
254b467… ragelink 843 name="ws-reviewed",
254b467… ragelink 844 branch="workspace/ws-reviewed",
254b467… ragelink 845 status="active",
254b467… ragelink 846 checkout_path="/tmp/fake",
254b467… ragelink 847 created_by=admin_user,
254b467… ragelink 848 )
254b467… ragelink 849 CodeReview.objects.create(
254b467… ragelink 850 repository=fossil_repo_obj,
254b467… ragelink 851 workspace=ws,
254b467… ragelink 852 title="Fix",
254b467… ragelink 853 diff="d",
254b467… ragelink 854 status="approved",
254b467… ragelink 855 created_by=admin_user,
254b467… ragelink 856 )
254b467… ragelink 857
254b467… ragelink 858 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls, patch("shutil.rmtree"):
254b467… ragelink 859 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 860 mock_cli_cls.return_value._env = {}
254b467… ragelink 861 mock_run.side_effect = [
254b467… ragelink 862 _make_proc(returncode=0), # update
254b467… ragelink 863 _make_proc(returncode=0), # merge
254b467… ragelink 864 _make_proc(returncode=0), # commit
254b467… ragelink 865 _make_proc(returncode=0), # close
254b467… ragelink 866 ]
254b467… ragelink 867
254b467… ragelink 868 response = writer_client.post(
254b467… ragelink 869 _api_url(sample_project.slug, "api/workspaces/ws-reviewed/merge"),
254b467… ragelink 870 data=json.dumps({"target_branch": "trunk"}),
254b467… ragelink 871 content_type="application/json",
254b467… ragelink 872 )
254b467… ragelink 873
254b467… ragelink 874 assert response.status_code == 200
254b467… ragelink 875 assert response.json()["status"] == "merged"
254b467… ragelink 876
254b467… ragelink 877 def test_merge_marks_linked_review_as_merged(self, admin_client, sample_project, fossil_repo_obj, workspace, admin_user):
254b467… ragelink 878 """Merging a workspace with an approved review updates the review status to merged."""
254b467… ragelink 879 review = CodeReview.objects.create(
254b467… ragelink 880 repository=fossil_repo_obj,
254b467… ragelink 881 workspace=workspace,
254b467… ragelink 882 title="ws review",
254b467… ragelink 883 diff="d",
254b467… ragelink 884 status="approved",
254b467… ragelink 885 created_by=admin_user,
254b467… ragelink 886 )
254b467… ragelink 887
254b467… ragelink 888 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls, patch("shutil.rmtree"):
254b467… ragelink 889 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 890 mock_cli_cls.return_value._env = {}
254b467… ragelink 891 mock_run.return_value = _make_proc(returncode=0)
254b467… ragelink 892
254b467… ragelink 893 admin_client.post(
254b467… ragelink 894 _api_url(sample_project.slug, "api/workspaces/ws-test-1/merge"),
254b467… ragelink 895 data=json.dumps({"agent_id": "claude-test"}),
254b467… ragelink 896 content_type="application/json",
254b467… ragelink 897 )
254b467… ragelink 898
254b467… ragelink 899 review.refresh_from_db()
254b467… ragelink 900 assert review.status == "merged"
254b467… ragelink 901
254b467… ragelink 902 def test_merge_blocked_no_review_non_admin(self, writer_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 903 """Non-admin cannot merge if no approved review exists for the workspace."""
254b467… ragelink 904 AgentWorkspace.objects.create(
254b467… ragelink 905 repository=fossil_repo_obj,
254b467… ragelink 906 name="ws-no-review",
254b467… ragelink 907 branch="workspace/ws-no-review",
254b467… ragelink 908 status="active",
254b467… ragelink 909 checkout_path="/tmp/fake",
254b467… ragelink 910 created_by=admin_user,
254b467… ragelink 911 )
254b467… ragelink 912
254b467… ragelink 913 response = writer_client.post(
254b467… ragelink 914 _api_url(sample_project.slug, "api/workspaces/ws-no-review/merge"),
254b467… ragelink 915 data=json.dumps({}),
254b467… ragelink 916 content_type="application/json",
254b467… ragelink 917 )
254b467… ragelink 918 assert response.status_code == 403
254b467… ragelink 919 assert "No approved code review" in response.json()["error"]
254b467… ragelink 920
254b467… ragelink 921 def test_merge_blocked_review_not_approved(self, writer_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 922 """Non-admin cannot merge if the linked review is still pending."""
254b467… ragelink 923 ws = AgentWorkspace.objects.create(
254b467… ragelink 924 repository=fossil_repo_obj,
254b467… ragelink 925 name="ws-pending-review",
254b467… ragelink 926 branch="workspace/ws-pending-review",
254b467… ragelink 927 status="active",
254b467… ragelink 928 checkout_path="/tmp/fake",
254b467… ragelink 929 created_by=admin_user,
254b467… ragelink 930 )
254b467… ragelink 931 CodeReview.objects.create(
254b467… ragelink 932 repository=fossil_repo_obj,
254b467… ragelink 933 workspace=ws,
254b467… ragelink 934 title="Pending",
254b467… ragelink 935 diff="d",
254b467… ragelink 936 status="pending",
254b467… ragelink 937 created_by=admin_user,
254b467… ragelink 938 )
254b467… ragelink 939
254b467… ragelink 940 response = writer_client.post(
254b467… ragelink 941 _api_url(sample_project.slug, "api/workspaces/ws-pending-review/merge"),
254b467… ragelink 942 data=json.dumps({}),
254b467… ragelink 943 content_type="application/json",
254b467… ragelink 944 )
254b467… ragelink 945 assert response.status_code == 403
254b467… ragelink 946 assert "must be approved" in response.json()["error"]
254b467… ragelink 947
254b467… ragelink 948 def test_merge_blocked_branch_protection_restrict_push(self, writer_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 949 """Branch protection with restrict_push blocks non-admin merges."""
254b467… ragelink 950 AgentWorkspace.objects.create(
254b467… ragelink 951 repository=fossil_repo_obj,
254b467… ragelink 952 name="ws-protected",
254b467… ragelink 953 branch="workspace/ws-protected",
254b467… ragelink 954 status="active",
254b467… ragelink 955 checkout_path="/tmp/fake",
254b467… ragelink 956 created_by=admin_user,
254b467… ragelink 957 )
254b467… ragelink 958 BranchProtection.objects.create(
254b467… ragelink 959 repository=fossil_repo_obj,
254b467… ragelink 960 branch_pattern="trunk",
254b467… ragelink 961 restrict_push=True,
254b467… ragelink 962 created_by=admin_user,
254b467… ragelink 963 )
254b467… ragelink 964
254b467… ragelink 965 response = writer_client.post(
254b467… ragelink 966 _api_url(sample_project.slug, "api/workspaces/ws-protected/merge"),
254b467… ragelink 967 data=json.dumps({"target_branch": "trunk"}),
254b467… ragelink 968 content_type="application/json",
254b467… ragelink 969 )
254b467… ragelink 970 assert response.status_code == 403
254b467… ragelink 971 assert "protected" in response.json()["error"].lower()
254b467… ragelink 972
254b467… ragelink 973 def test_merge_blocked_required_status_check_not_passed(self, writer_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 974 """Branch protection with required status checks blocks merge when check hasn't passed."""
254b467… ragelink 975 AgentWorkspace.objects.create(
254b467… ragelink 976 repository=fossil_repo_obj,
254b467… ragelink 977 name="ws-ci-fail",
254b467… ragelink 978 branch="workspace/ws-ci-fail",
254b467… ragelink 979 status="active",
254b467… ragelink 980 checkout_path="/tmp/fake",
254b467… ragelink 981 created_by=admin_user,
254b467… ragelink 982 )
254b467… ragelink 983 BranchProtection.objects.create(
254b467… ragelink 984 repository=fossil_repo_obj,
254b467… ragelink 985 branch_pattern="trunk",
254b467… ragelink 986 restrict_push=False,
254b467… ragelink 987 require_status_checks=True,
254b467… ragelink 988 required_contexts="ci/tests",
254b467… ragelink 989 created_by=admin_user,
254b467… ragelink 990 )
254b467… ragelink 991 # Status check is pending (not success)
254b467… ragelink 992 StatusCheck.objects.create(
254b467… ragelink 993 repository=fossil_repo_obj,
254b467… ragelink 994 checkin_uuid="some-uuid",
254b467… ragelink 995 context="ci/tests",
254b467… ragelink 996 state="pending",
254b467… ragelink 997 created_by=admin_user,
254b467… ragelink 998 )
254b467… ragelink 999
254b467… ragelink 1000 response = writer_client.post(
254b467… ragelink 1001 _api_url(sample_project.slug, "api/workspaces/ws-ci-fail/merge"),
254b467… ragelink 1002 data=json.dumps({"target_branch": "trunk"}),
254b467… ragelink 1003 content_type="application/json",
254b467… ragelink 1004 )
254b467… ragelink 1005 assert response.status_code == 403
254b467… ragelink 1006 assert "status check" in response.json()["error"].lower()
254b467… ragelink 1007
254b467… ragelink 1008 def test_merge_allowed_with_passing_status_check(self, writer_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 1009 """Branch protection with passing required status check allows merge."""
254b467… ragelink 1010 ws = AgentWorkspace.objects.create(
254b467… ragelink 1011 repository=fossil_repo_obj,
254b467… ragelink 1012 name="ws-ci-pass",
254b467… ragelink 1013 branch="workspace/ws-ci-pass",
254b467… ragelink 1014 status="active",
254b467… ragelink 1015 checkout_path="/tmp/fake",
254b467… ragelink 1016 created_by=admin_user,
254b467… ragelink 1017 )
254b467… ragelink 1018 BranchProtection.objects.create(
254b467… ragelink 1019 repository=fossil_repo_obj,
254b467… ragelink 1020 branch_pattern="trunk",
254b467… ragelink 1021 restrict_push=False,
254b467… ragelink 1022 require_status_checks=True,
254b467… ragelink 1023 required_contexts="ci/tests",
254b467… ragelink 1024 created_by=admin_user,
254b467… ragelink 1025 )
254b467… ragelink 1026 StatusCheck.objects.create(
254b467… ragelink 1027 repository=fossil_repo_obj,
254b467… ragelink 1028 checkin_uuid="some-uuid",
254b467… ragelink 1029 context="ci/tests",
254b467… ragelink 1030 state="success",
254b467… ragelink 1031 created_by=admin_user,
254b467… ragelink 1032 )
254b467… ragelink 1033 CodeReview.objects.create(
254b467… ragelink 1034 repository=fossil_repo_obj,
254b467… ragelink 1035 workspace=ws,
254b467… ragelink 1036 title="Fix",
254b467… ragelink 1037 diff="d",
254b467… ragelink 1038 status="approved",
254b467… ragelink 1039 created_by=admin_user,
254b467… ragelink 1040 )
254b467… ragelink 1041
254b467… ragelink 1042 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls, patch("shutil.rmtree"):
254b467… ragelink 1043 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 1044 mock_cli_cls.return_value._env = {}
254b467… ragelink 1045 mock_run.return_value = _make_proc(returncode=0)
254b467… ragelink 1046
254b467… ragelink 1047 response = writer_client.post(
254b467… ragelink 1048 _api_url(sample_project.slug, "api/workspaces/ws-ci-pass/merge"),
254b467… ragelink 1049 data=json.dumps({"target_branch": "trunk"}),
254b467… ragelink 1050 content_type="application/json",
254b467… ragelink 1051 )
254b467… ragelink 1052
254b467… ragelink 1053 assert response.status_code == 200
254b467… ragelink 1054
254b467… ragelink 1055 def test_merge_fossil_update_fails(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 1056 """When fossil update to target branch fails, return 500."""
254b467… ragelink 1057 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls:
254b467… ragelink 1058 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 1059 mock_cli_cls.return_value._env = {}
254b467… ragelink 1060 mock_run.return_value = _make_proc(returncode=1, stderr="update failed")
254b467… ragelink 1061
254b467… ragelink 1062 response = admin_client.post(
254b467… ragelink 1063 _api_url(sample_project.slug, "api/workspaces/ws-test-1/merge"),
254b467… ragelink 1064 data=json.dumps({"agent_id": "claude-test"}),
254b467… ragelink 1065 content_type="application/json",
254b467… ragelink 1066 )
254b467… ragelink 1067
254b467… ragelink 1068 assert response.status_code == 500
254b467… ragelink 1069 assert "Failed to switch" in response.json()["error"]
254b467… ragelink 1070
254b467… ragelink 1071 def test_merge_fossil_merge_fails(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 1072 """When fossil merge command fails, return 500."""
254b467… ragelink 1073 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls:
254b467… ragelink 1074 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 1075 mock_cli_cls.return_value._env = {}
254b467… ragelink 1076 mock_run.side_effect = [
254b467… ragelink 1077 _make_proc(returncode=0), # update
254b467… ragelink 1078 _make_proc(returncode=1, stderr="merge conflict"), # merge
254b467… ragelink 1079 ]
254b467… ragelink 1080
254b467… ragelink 1081 response = admin_client.post(
254b467… ragelink 1082 _api_url(sample_project.slug, "api/workspaces/ws-test-1/merge"),
254b467… ragelink 1083 data=json.dumps({"agent_id": "claude-test"}),
254b467… ragelink 1084 content_type="application/json",
254b467… ragelink 1085 )
254b467… ragelink 1086
254b467… ragelink 1087 assert response.status_code == 500
254b467… ragelink 1088 assert "Merge failed" in response.json()["error"]
254b467… ragelink 1089
254b467… ragelink 1090 def test_merge_commit_fails(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 1091 """When the merge commit fails, return 500 and don't close workspace."""
254b467… ragelink 1092 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls:
254b467… ragelink 1093 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 1094 mock_cli_cls.return_value._env = {}
254b467… ragelink 1095 mock_run.side_effect = [
254b467… ragelink 1096 _make_proc(returncode=0), # update
254b467… ragelink 1097 _make_proc(returncode=0), # merge
254b467… ragelink 1098 _make_proc(returncode=1, stderr="commit lock"), # commit
254b467… ragelink 1099 ]
254b467… ragelink 1100
254b467… ragelink 1101 response = admin_client.post(
254b467… ragelink 1102 _api_url(sample_project.slug, "api/workspaces/ws-test-1/merge"),
254b467… ragelink 1103 data=json.dumps({"agent_id": "claude-test"}),
254b467… ragelink 1104 content_type="application/json",
254b467… ragelink 1105 )
254b467… ragelink 1106
254b467… ragelink 1107 assert response.status_code == 500
254b467… ragelink 1108 assert "Merge commit failed" in response.json()["error"]
254b467… ragelink 1109
254b467… ragelink 1110 # Workspace should still be active (not closed on commit failure)
254b467… ragelink 1111 workspace.refresh_from_db()
254b467… ragelink 1112 assert workspace.status == "active"
254b467… ragelink 1113
254b467… ragelink 1114 def test_merge_workspace_not_found(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 1115 """Merging a non-existent workspace returns 404."""
254b467… ragelink 1116 response = admin_client.post(
254b467… ragelink 1117 _api_url(sample_project.slug, "api/workspaces/nonexistent/merge"),
254b467… ragelink 1118 data=json.dumps({}),
254b467… ragelink 1119 content_type="application/json",
254b467… ragelink 1120 )
254b467… ragelink 1121 assert response.status_code == 404
254b467… ragelink 1122
254b467… ragelink 1123 def test_merge_workspace_not_active(self, admin_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 1124 """Merging an already-merged workspace returns 409."""
254b467… ragelink 1125 AgentWorkspace.objects.create(
254b467… ragelink 1126 repository=fossil_repo_obj,
254b467… ragelink 1127 name="ws-already-merged",
254b467… ragelink 1128 branch="workspace/ws-already-merged",
254b467… ragelink 1129 status="merged",
254b467… ragelink 1130 created_by=admin_user,
254b467… ragelink 1131 )
254b467… ragelink 1132
254b467… ragelink 1133 response = admin_client.post(
254b467… ragelink 1134 _api_url(sample_project.slug, "api/workspaces/ws-already-merged/merge"),
254b467… ragelink 1135 data=json.dumps({}),
254b467… ragelink 1136 content_type="application/json",
254b467… ragelink 1137 )
254b467… ragelink 1138 assert response.status_code == 409
254b467… ragelink 1139 assert "merged" in response.json()["error"]
254b467… ragelink 1140
254b467… ragelink 1141 def test_merge_wrong_method(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 1142 """GET to merge endpoint returns 405."""
254b467… ragelink 1143 response = admin_client.get(_api_url(sample_project.slug, "api/workspaces/ws-test-1/merge"))
254b467… ragelink 1144 assert response.status_code == 405
254b467… ragelink 1145
254b467… ragelink 1146 def test_merge_denied_for_reader(self, reader_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 1147 """Read-only users cannot merge workspaces."""
254b467… ragelink 1148 response = reader_client.post(
254b467… ragelink 1149 _api_url(sample_project.slug, "api/workspaces/ws-test-1/merge"),
254b467… ragelink 1150 data=json.dumps({}),
254b467… ragelink 1151 content_type="application/json",
254b467… ragelink 1152 )
254b467… ragelink 1153 assert response.status_code == 403
254b467… ragelink 1154
254b467… ragelink 1155
254b467… ragelink 1156 # ================================================================
254b467… ragelink 1157 # Workspace Abandon
254b467… ragelink 1158 # ================================================================
254b467… ragelink 1159
254b467… ragelink 1160
254b467… ragelink 1161 @pytest.mark.django_db
254b467… ragelink 1162 class TestWorkspaceAbandon:
254b467… ragelink 1163 """Tests for DELETE /projects/<slug>/fossil/api/workspaces/<name>/abandon (lines 1188-1238)."""
254b467… ragelink 1164
254b467… ragelink 1165 def test_abandon_success(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 1166 """Abandoning a workspace closes checkout, cleans up directory, and updates status."""
254b467… ragelink 1167 with (
254b467… ragelink 1168 patch("subprocess.run") as mock_run,
254b467… ragelink 1169 patch("fossil.cli.FossilCLI") as mock_cli_cls,
254b467… ragelink 1170 patch("shutil.rmtree") as mock_rmtree,
254b467… ragelink 1171 ):
254b467… ragelink 1172 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 1173 mock_cli_cls.return_value._env = {}
254b467… ragelink 1174 mock_run.return_value = _make_proc(returncode=0)
254b467… ragelink 1175
254b467… ragelink 1176 response = admin_client.delete(
254b467… ragelink 1177 _api_url(sample_project.slug, "api/workspaces/ws-test-1/abandon"),
254b467… ragelink 1178 )
254b467… ragelink 1179
254b467… ragelink 1180 assert response.status_code == 200
254b467… ragelink 1181 data = response.json()
254b467… ragelink 1182 assert data["status"] == "abandoned"
254b467… ragelink 1183 assert data["name"] == "ws-test-1"
254b467… ragelink 1184
254b467… ragelink 1185 workspace.refresh_from_db()
254b467… ragelink 1186 assert workspace.status == "abandoned"
254b467… ragelink 1187 assert workspace.checkout_path == ""
254b467… ragelink 1188
254b467… ragelink 1189 # Verify cleanup was called
254b467… ragelink 1190 mock_rmtree.assert_called_once()
254b467… ragelink 1191
254b467… ragelink 1192 def test_abandon_no_checkout_path(self, admin_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 1193 """Abandoning a workspace with empty checkout path still works (no cleanup needed)."""
254b467… ragelink 1194 ws = AgentWorkspace.objects.create(
254b467… ragelink 1195 repository=fossil_repo_obj,
254b467… ragelink 1196 name="ws-no-path",
254b467… ragelink 1197 branch="workspace/ws-no-path",
254b467… ragelink 1198 status="active",
254b467… ragelink 1199 checkout_path="",
254b467… ragelink 1200 created_by=admin_user,
254b467… ragelink 1201 )
254b467… ragelink 1202
254b467… ragelink 1203 with patch("fossil.cli.FossilCLI"):
254b467… ragelink 1204 response = admin_client.delete(_api_url(sample_project.slug, "api/workspaces/ws-no-path/abandon"))
254b467… ragelink 1205
254b467… ragelink 1206 assert response.status_code == 200
254b467… ragelink 1207 ws.refresh_from_db()
254b467… ragelink 1208 assert ws.status == "abandoned"
254b467… ragelink 1209
254b467… ragelink 1210 def test_abandon_workspace_not_found(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 1211 """Abandoning a non-existent workspace returns 404."""
254b467… ragelink 1212 response = admin_client.delete(_api_url(sample_project.slug, "api/workspaces/nonexistent/abandon"))
254b467… ragelink 1213 assert response.status_code == 404
254b467… ragelink 1214
254b467… ragelink 1215 def test_abandon_workspace_already_abandoned(self, admin_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 1216 """Abandoning an already-abandoned workspace returns 409."""
254b467… ragelink 1217 AgentWorkspace.objects.create(
254b467… ragelink 1218 repository=fossil_repo_obj,
254b467… ragelink 1219 name="ws-gone",
254b467… ragelink 1220 branch="workspace/ws-gone",
254b467… ragelink 1221 status="abandoned",
254b467… ragelink 1222 created_by=admin_user,
254b467… ragelink 1223 )
254b467… ragelink 1224
254b467… ragelink 1225 response = admin_client.delete(_api_url(sample_project.slug, "api/workspaces/ws-gone/abandon"))
254b467… ragelink 1226 assert response.status_code == 409
254b467… ragelink 1227 assert "already abandoned" in response.json()["error"]
254b467… ragelink 1228
254b467… ragelink 1229 def test_abandon_wrong_method(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 1230 """POST to abandon endpoint returns 405 (DELETE required)."""
254b467… ragelink 1231 response = admin_client.post(
254b467… ragelink 1232 _api_url(sample_project.slug, "api/workspaces/ws-test-1/abandon"),
254b467… ragelink 1233 content_type="application/json",
254b467… ragelink 1234 )
254b467… ragelink 1235 assert response.status_code == 405
254b467… ragelink 1236
254b467… ragelink 1237 def test_abandon_denied_for_reader(self, reader_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 1238 """Read-only users cannot abandon workspaces."""
254b467… ragelink 1239 response = reader_client.delete(_api_url(sample_project.slug, "api/workspaces/ws-test-1/abandon"))
254b467… ragelink 1240 assert response.status_code == 403
254b467… ragelink 1241
254b467… ragelink 1242 def test_abandon_denied_for_anon(self, client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 1243 """Anonymous users cannot abandon workspaces."""
254b467… ragelink 1244 response = client.delete(_api_url(sample_project.slug, "api/workspaces/ws-test-1/abandon"))
254b467… ragelink 1245 assert response.status_code == 401
254b467… ragelink 1246
254b467… ragelink 1247
254b467… ragelink 1248 # ================================================================
254b467… ragelink 1249 # Workspace Ownership Checks
254b467… ragelink 1250 # ================================================================
254b467… ragelink 1251
254b467… ragelink 1252
254b467… ragelink 1253 @pytest.mark.django_db
254b467… ragelink 1254 class TestWorkspaceOwnership:
254b467… ragelink 1255 """Tests for _check_workspace_ownership (lines 722-747).
254b467… ragelink 1256
254b467… ragelink 1257 Token-based callers must supply matching agent_id.
254b467… ragelink 1258 Session-auth users (human oversight) are always allowed.
254b467… ragelink 1259 """
254b467… ragelink 1260
254b467… ragelink 1261 def test_session_user_always_allowed(self, admin_client, sample_project, fossil_repo_obj, workspace):
254b467… ragelink 1262 """Session-auth users bypass ownership check (human oversight).
254b467… ragelink 1263 Tested through the commit endpoint which calls _check_workspace_ownership.
254b467… ragelink 1264 """
254b467… ragelink 1265 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls:
254b467… ragelink 1266 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 1267 mock_cli_cls.return_value._env = {}
254b467… ragelink 1268 mock_run.side_effect = [
254b467… ragelink 1269 _make_proc(returncode=0), # addremove
254b467… ragelink 1270 _make_proc(returncode=0, stdout="committed"), # commit
254b467… ragelink 1271 ]
254b467… ragelink 1272
254b467… ragelink 1273 # Session user does not provide agent_id -- should still be allowed
254b467… ragelink 1274 response = admin_client.post(
254b467… ragelink 1275 _api_url(sample_project.slug, "api/workspaces/ws-test-1/commit"),
254b467… ragelink 1276 data=json.dumps({"message": "Human override"}),
254b467… ragelink 1277 content_type="application/json",
254b467… ragelink 1278 )
254b467… ragelink 1279
254b467… ragelink 1280 assert response.status_code == 200
254b467… ragelink 1281
254b467… ragelink 1282 def test_workspace_without_agent_id_allows_any_writer(self, admin_client, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 1283 """Workspace with empty agent_id allows any writer to operate."""
254b467… ragelink 1284 AgentWorkspace.objects.create(
254b467… ragelink 1285 repository=fossil_repo_obj,
254b467… ragelink 1286 name="ws-no-agent",
254b467… ragelink 1287 branch="workspace/ws-no-agent",
254b467… ragelink 1288 agent_id="",
254b467… ragelink 1289 status="active",
254b467… ragelink 1290 checkout_path="/tmp/fake",
254b467… ragelink 1291 created_by=admin_user,
254b467… ragelink 1292 )
254b467… ragelink 1293
254b467… ragelink 1294 with patch("subprocess.run") as mock_run, patch("fossil.cli.FossilCLI") as mock_cli_cls:
254b467… ragelink 1295 mock_cli_cls.return_value.binary = "/usr/local/bin/fossil"
254b467… ragelink 1296 mock_cli_cls.return_value._env = {}
254b467… ragelink 1297 mock_run.side_effect = [
254b467… ragelink 1298 _make_proc(returncode=0),
254b467… ragelink 1299 _make_proc(returncode=0, stdout="committed"),
254b467… ragelink 1300 ]
254b467… ragelink 1301
254b467… ragelink 1302 response = admin_client.post(
254b467… ragelink 1303 _api_url(sample_project.slug, "api/workspaces/ws-no-agent/commit"),
254b467… ragelink 1304 data=json.dumps({"message": "Anyone can commit"}),
254b467… ragelink 1305 content_type="application/json",
254b467… ragelink 1306 )
254b467… ragelink 1307
254b467… ragelink 1308 assert response.status_code == 200
254b467… ragelink 1309
254b467… ragelink 1310
254b467… ragelink 1311 # ================================================================
254b467… ragelink 1312 # SSE Events - Stream Content
254b467… ragelink 1313 # ================================================================
254b467… ragelink 1314
254b467… ragelink 1315
254b467… ragelink 1316 @pytest.mark.django_db
254b467… ragelink 1317 class TestSSEEventStream:
254b467… ragelink 1318 """Tests for GET /projects/<slug>/fossil/api/events (lines 1521-1653).
254b467… ragelink 1319
254b467… ragelink 1320 The SSE endpoint returns a StreamingHttpResponse. We verify the response
254b467… ragelink 1321 metadata and test the event generator for various event types.
254b467… ragelink 1322 """
254b467… ragelink 1323
254b467… ragelink 1324 def test_sse_response_headers(self, admin_client, sample_project, fossil_repo_obj):
254b467… ragelink 1325 """SSE endpoint sets correct headers for event streaming."""
254b467… ragelink 1326 with patch("fossil.api_views.FossilReader") as mock_reader_cls:
254b467… ragelink 1327 reader = mock_reader_cls.return_value
254b467… ragelink 1328 reader.__enter__ = MagicMock(return_value=reader)
254b467… ragelink 1329 reader.__exit__ = MagicMock(return_value=False)
254b467… ragelink 1330 reader.get_checkin_count.return_value = 0
254b467… ragelink 1331
254b467… ragelink 1332 response = admin_client.get(_api_url(sample_project.slug, "api/events"))
254b467… ragelink 1333
254b467… ragelink 1334 assert response.status_code == 200
254b467… ragelink 1335 assert response["Content-Type"] == "text/event-stream"
254b467… ragelink 1336 assert response["Cache-Control"] == "no-cache"
254b467… ragelink 1337 assert response["X-Accel-Buffering"] == "no"
254b467… ragelink 1338 assert response.streaming is True
6c1d3dd… ragelink 1339 # Close the streaming response to release the DB connection
6c1d3dd… ragelink 1340 response.close()
254b467… ragelink 1341
254b467… ragelink 1342 def test_sse_generator_yields_claim_events(self, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 1343 """The SSE generator detects new TicketClaims and yields claim events.
254b467… ragelink 1344
254b467… ragelink 1345 We test the generator directly rather than going through StreamingHttpResponse,
254b467… ragelink 1346 because the response wraps it in map(make_bytes, ...) which complicates
254b467… ragelink 1347 exception-based termination.
254b467… ragelink 1348 """
254b467… ragelink 1349 # Simulate what event_stream() does: snapshot state, create new objects, check
254b467… ragelink 1350 last_claim_id = TicketClaim.all_objects.filter(repository=fossil_repo_obj).order_by("-pk").values_list("pk", flat=True).first() or 0
254b467… ragelink 1351
254b467… ragelink 1352 # Create a claim after the snapshot
254b467… ragelink 1353 TicketClaim.objects.create(
254b467… ragelink 1354 repository=fossil_repo_obj,
254b467… ragelink 1355 ticket_uuid="sse-test-ticket",
254b467… ragelink 1356 agent_id="sse-agent",
254b467… ragelink 1357 created_by=admin_user,
254b467… ragelink 1358 )
254b467… ragelink 1359
254b467… ragelink 1360 # Query exactly as the generator does
254b467… ragelink 1361 new_claims = TicketClaim.all_objects.filter(repository=fossil_repo_obj, pk__gt=last_claim_id).order_by("pk")
254b467… ragelink 1362 events = []
254b467… ragelink 1363 for c in new_claims:
254b467… ragelink 1364 events.append(f"event: claim\ndata: {json.dumps({'ticket_uuid': c.ticket_uuid, 'agent_id': c.agent_id})}\n\n")
254b467… ragelink 1365
254b467… ragelink 1366 assert len(events) >= 1
254b467… ragelink 1367 assert "sse-test-ticket" in events[0]
254b467… ragelink 1368 assert "sse-agent" in events[0]
254b467… ragelink 1369
254b467… ragelink 1370 def test_sse_generator_yields_workspace_events(self, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 1371 """The SSE generator detects new AgentWorkspaces and yields workspace events."""
254b467… ragelink 1372 last_ws_id = AgentWorkspace.all_objects.filter(repository=fossil_repo_obj).order_by("-pk").values_list("pk", flat=True).first() or 0
254b467… ragelink 1373
254b467… ragelink 1374 AgentWorkspace.objects.create(
254b467… ragelink 1375 repository=fossil_repo_obj,
254b467… ragelink 1376 name="sse-ws",
254b467… ragelink 1377 branch="workspace/sse-ws",
254b467… ragelink 1378 agent_id="sse-agent",
254b467… ragelink 1379 created_by=admin_user,
254b467… ragelink 1380 )
254b467… ragelink 1381
254b467… ragelink 1382 new_ws = AgentWorkspace.all_objects.filter(repository=fossil_repo_obj, pk__gt=last_ws_id).order_by("pk")
254b467… ragelink 1383 events = []
254b467… ragelink 1384 for ws in new_ws:
254b467… ragelink 1385 events.append(f"event: workspace\ndata: {json.dumps({'name': ws.name, 'agent_id': ws.agent_id})}\n\n")
254b467… ragelink 1386
254b467… ragelink 1387 assert len(events) >= 1
254b467… ragelink 1388 assert "sse-ws" in events[0]
254b467… ragelink 1389
254b467… ragelink 1390 def test_sse_generator_yields_review_events(self, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 1391 """The SSE generator detects new CodeReviews and yields review events."""
254b467… ragelink 1392 last_review_id = CodeReview.all_objects.filter(repository=fossil_repo_obj).order_by("-pk").values_list("pk", flat=True).first() or 0
254b467… ragelink 1393
254b467… ragelink 1394 CodeReview.objects.create(
254b467… ragelink 1395 repository=fossil_repo_obj,
254b467… ragelink 1396 title="SSE review",
254b467… ragelink 1397 diff="d",
254b467… ragelink 1398 agent_id="sse-agent",
254b467… ragelink 1399 created_by=admin_user,
254b467… ragelink 1400 )
254b467… ragelink 1401
254b467… ragelink 1402 new_reviews = CodeReview.all_objects.filter(repository=fossil_repo_obj, pk__gt=last_review_id).order_by("pk")
254b467… ragelink 1403 events = []
254b467… ragelink 1404 for r in new_reviews:
254b467… ragelink 1405 events.append(f"event: review\ndata: {json.dumps({'title': r.title, 'agent_id': r.agent_id})}\n\n")
254b467… ragelink 1406
254b467… ragelink 1407 assert len(events) >= 1
254b467… ragelink 1408 assert "SSE review" in events[0]
254b467… ragelink 1409
254b467… ragelink 1410 def test_sse_generator_yields_checkin_events(self, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 1411 """The SSE generator detects new checkins and yields checkin events."""
254b467… ragelink 1412 from fossil.api_views import api_events
254b467… ragelink 1413
254b467… ragelink 1414 factory = RequestFactory()
254b467… ragelink 1415 request = factory.get(_api_url(sample_project.slug, "api/events"))
254b467… ragelink 1416 request.user = admin_user
254b467… ragelink 1417 request.session = {}
254b467… ragelink 1418
254b467… ragelink 1419 timeline_entry = MagicMock()
254b467… ragelink 1420 timeline_entry.uuid = "checkin-001"
254b467… ragelink 1421 timeline_entry.user = "dev"
254b467… ragelink 1422 timeline_entry.comment = "initial commit"
254b467… ragelink 1423 timeline_entry.branch = "trunk"
254b467… ragelink 1424 timeline_entry.timestamp = None
254b467… ragelink 1425
254b467… ragelink 1426 with patch("fossil.api_views.FossilReader") as mock_reader_cls:
254b467… ragelink 1427 reader = mock_reader_cls.return_value
254b467… ragelink 1428 reader.__enter__ = MagicMock(return_value=reader)
254b467… ragelink 1429 reader.__exit__ = MagicMock(return_value=False)
254b467… ragelink 1430 # First call during snapshot: 0 checkins. Second call during poll: 1 checkin
254b467… ragelink 1431 reader.get_checkin_count.side_effect = [0, 1]
254b467… ragelink 1432 reader.get_timeline.return_value = [timeline_entry]
254b467… ragelink 1433
254b467… ragelink 1434 response = api_events(request, slug=sample_project.slug)
254b467… ragelink 1435 events = _drain_sse_one_iteration(response)
254b467… ragelink 1436
254b467… ragelink 1437 checkin_events = [e for e in events if "event: checkin" in e]
254b467… ragelink 1438 assert len(checkin_events) >= 1
254b467… ragelink 1439 assert "checkin-001" in checkin_events[0]
254b467… ragelink 1440
254b467… ragelink 1441 def test_sse_generator_heartbeat(self, sample_project, fossil_repo_obj, admin_user):
254b467… ragelink 1442 """After 3 empty iterations the generator emits a heartbeat comment."""
254b467… ragelink 1443 from fossil.api_views import api_events
254b467… ragelink 1444
254b467… ragelink 1445 factory = RequestFactory()
254b467… ragelink 1446 request = factory.get(_api_url(sample_project.slug, "api/events"))
254b467… ragelink 1447 request.user = admin_user
254b467… ragelink 1448 request.session = {}
254b467… ragelink 1449
254b467… ragelink 1450 with patch("fossil.api_views.FossilReader") as mock_reader_cls:
254b467… ragelink 1451 reader = mock_reader_cls.return_value
254b467… ragelink 1452 reader.__enter__ = MagicMock(return_value=reader)
254b467… ragelink 1453 reader.__exit__ = MagicMock(return_value=False)
254b467… ragelink 1454 reader.get_checkin_count.return_value = 0
254b467… ragelink 1455
254b467… ragelink 1456 response = api_events(request, slug=sample_project.slug)
254b467… ragelink 1457 # Run 3 empty iterations so heartbeat triggers
254b467… ragelink 1458 events = _drain_sse_n_iterations(response, n=3)
254b467… ragelink 1459
254b467… ragelink 1460 heartbeats = [e for e in events if ": heartbeat" in e]
254b467… ragelink 1461 assert len(heartbeats) >= 1
254b467… ragelink 1462
254b467… ragelink 1463
254b467… ragelink 1464 # ================================================================
254b467… ragelink 1465 # _resolve_batch_route
254b467… ragelink 1466 # ================================================================
254b467… ragelink 1467
254b467… ragelink 1468
254b467… ragelink 1469 @pytest.mark.django_db
254b467… ragelink 1470 class TestResolveBatchRoute:
254b467… ragelink 1471 """Tests for _resolve_batch_route helper (lines 596-607)."""
254b467… ragelink 1472
254b467… ragelink 1473 def test_static_route_timeline(self):
254b467… ragelink 1474 """Static route /api/timeline resolves to the timeline view."""
254b467… ragelink 1475 from fossil.api_views import _resolve_batch_route, api_timeline
254b467… ragelink 1476
254b467… ragelink 1477 view_func, kwargs = _resolve_batch_route("/api/timeline")
254b467… ragelink 1478 assert view_func is api_timeline
254b467… ragelink 1479 assert kwargs == {}
254b467… ragelink 1480
254b467… ragelink 1481 def test_static_route_project(self):
254b467… ragelink 1482 """Static route /api/project resolves to the project view."""
254b467… ragelink 1483 from fossil.api_views import _resolve_batch_route, api_project
254b467… ragelink 1484
254b467… ragelink 1485 view_func, kwargs = _resolve_batch_route("/api/project")
254b467… ragelink 1486 assert view_func is api_project
254b467… ragelink 1487 assert kwargs == {}
254b467… ragelink 1488
254b467… ragelink 1489 def test_dynamic_route_ticket(self):
254b467… ragelink 1490 """Dynamic route /api/tickets/<uuid> resolves with ticket_uuid kwarg."""
254b467… ragelink 1491 from fossil.api_views import _resolve_batch_route, api_ticket_detail
254b467… ragelink 1492
254b467… ragelink 1493 view_func, kwargs = _resolve_batch_route("/api/tickets/abc-123-def")
254b467… ragelink 1494 assert view_func is api_ticket_detail
254b467… ragelink 1495 assert kwargs == {"ticket_uuid": "abc-123-def"}
254b467… ragelink 1496
254b467… ragelink 1497 def test_dynamic_route_wiki(self):
254b467… ragelink 1498 """Dynamic route /api/wiki/<name> resolves with page_name kwarg."""
254b467… ragelink 1499 from fossil.api_views import _resolve_batch_route, api_wiki_page
254b467… ragelink 1500
254b467… ragelink 1501 view_func, kwargs = _resolve_batch_route("/api/wiki/Getting-Started")
254b467… ragelink 1502 assert view_func is api_wiki_page
254b467… ragelink 1503 assert kwargs == {"page_name": "Getting-Started"}
254b467… ragelink 1504
254b467… ragelink 1505 def test_unknown_route(self):
254b467… ragelink 1506 """Unknown path returns (None, None)."""
254b467… ragelink 1507 from fossil.api_views import _resolve_batch_route
254b467… ragelink 1508
254b467… ragelink 1509 view_func, kwargs = _resolve_batch_route("/api/nonexistent")
254b467… ragelink 1510 assert view_func is None
254b467… ragelink 1511 assert kwargs is None

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button