FossilRepo

fossilrepo / tests / test_ci_status.py
Source Blame History 349 lines
c588255… ragelink 1 import json
c588255… ragelink 2
c588255… ragelink 3 import pytest
c588255… ragelink 4 from django.contrib.auth.models import User
c588255… ragelink 5 from django.test import Client
c588255… ragelink 6
c588255… ragelink 7 from fossil.api_tokens import APIToken
c588255… ragelink 8 from fossil.ci import StatusCheck
c588255… ragelink 9 from fossil.models import FossilRepository
c588255… ragelink 10 from organization.models import Team
c588255… ragelink 11 from projects.models import ProjectTeam
c588255… ragelink 12
c588255… ragelink 13
c588255… ragelink 14 @pytest.fixture
c588255… ragelink 15 def fossil_repo_obj(sample_project):
c588255… ragelink 16 """Return the auto-created FossilRepository for sample_project."""
c588255… ragelink 17 return FossilRepository.objects.get(project=sample_project, deleted_at__isnull=True)
c588255… ragelink 18
c588255… ragelink 19
c588255… ragelink 20 @pytest.fixture
c588255… ragelink 21 def api_token(fossil_repo_obj, admin_user):
c588255… ragelink 22 """Create an API token and return (APIToken instance, raw_token)."""
c588255… ragelink 23 raw, token_hash, prefix = APIToken.generate()
c588255… ragelink 24 token = APIToken.objects.create(
c588255… ragelink 25 repository=fossil_repo_obj,
c588255… ragelink 26 name="CI Token",
c588255… ragelink 27 token_hash=token_hash,
c588255… ragelink 28 token_prefix=prefix,
c588255… ragelink 29 permissions="status:write",
c588255… ragelink 30 created_by=admin_user,
c588255… ragelink 31 )
c588255… ragelink 32 return token, raw
c588255… ragelink 33
c588255… ragelink 34
c588255… ragelink 35 @pytest.fixture
c588255… ragelink 36 def status_check(fossil_repo_obj):
c588255… ragelink 37 return StatusCheck.objects.create(
c588255… ragelink 38 repository=fossil_repo_obj,
c588255… ragelink 39 checkin_uuid="abc123def456",
c588255… ragelink 40 context="ci/tests",
c588255… ragelink 41 state="success",
c588255… ragelink 42 description="All 42 tests passed",
c588255… ragelink 43 target_url="https://ci.example.com/build/1",
c588255… ragelink 44 )
c588255… ragelink 45
c588255… ragelink 46
c588255… ragelink 47 @pytest.fixture
c588255… ragelink 48 def writer_user(db, admin_user, sample_project):
c588255… ragelink 49 """User with write access but not admin."""
c588255… ragelink 50 writer = User.objects.create_user(username="writer_ci", password="testpass123")
c588255… ragelink 51 team = Team.objects.create(name="CI Writers", organization=sample_project.organization, created_by=admin_user)
c588255… ragelink 52 team.members.add(writer)
c588255… ragelink 53 ProjectTeam.objects.create(project=sample_project, team=team, role="write", created_by=admin_user)
c588255… ragelink 54 return writer
c588255… ragelink 55
c588255… ragelink 56
c588255… ragelink 57 @pytest.fixture
c588255… ragelink 58 def writer_client(writer_user):
c588255… ragelink 59 client = Client()
c588255… ragelink 60 client.login(username="writer_ci", password="testpass123")
c588255… ragelink 61 return client
c588255… ragelink 62
c588255… ragelink 63
c588255… ragelink 64 # --- StatusCheck Model Tests ---
c588255… ragelink 65
c588255… ragelink 66
c588255… ragelink 67 @pytest.mark.django_db
c588255… ragelink 68 class TestStatusCheckModel:
c588255… ragelink 69 def test_create_status_check(self, status_check):
c588255… ragelink 70 assert status_check.pk is not None
c588255… ragelink 71 assert str(status_check) == "ci/tests: success @ abc123def4"
c588255… ragelink 72
c588255… ragelink 73 def test_soft_delete(self, status_check, admin_user):
c588255… ragelink 74 status_check.soft_delete(user=admin_user)
c588255… ragelink 75 assert status_check.is_deleted
c588255… ragelink 76 assert StatusCheck.objects.filter(pk=status_check.pk).count() == 0
c588255… ragelink 77 assert StatusCheck.all_objects.filter(pk=status_check.pk).count() == 1
c588255… ragelink 78
c588255… ragelink 79 def test_unique_together(self, fossil_repo_obj):
c588255… ragelink 80 StatusCheck.objects.create(
c588255… ragelink 81 repository=fossil_repo_obj,
c588255… ragelink 82 checkin_uuid="unique123",
c588255… ragelink 83 context="ci/lint",
c588255… ragelink 84 state="pending",
c588255… ragelink 85 )
c588255… ragelink 86 from django.db import IntegrityError
c588255… ragelink 87
c588255… ragelink 88 with pytest.raises(IntegrityError):
c588255… ragelink 89 StatusCheck.objects.create(
c588255… ragelink 90 repository=fossil_repo_obj,
c588255… ragelink 91 checkin_uuid="unique123",
c588255… ragelink 92 context="ci/lint",
c588255… ragelink 93 state="success",
c588255… ragelink 94 )
c588255… ragelink 95
c588255… ragelink 96 def test_ordering(self, fossil_repo_obj):
c588255… ragelink 97 c1 = StatusCheck.objects.create(repository=fossil_repo_obj, checkin_uuid="ord1", context="ci/first", state="pending")
c588255… ragelink 98 c2 = StatusCheck.objects.create(repository=fossil_repo_obj, checkin_uuid="ord2", context="ci/second", state="success")
c588255… ragelink 99 checks = list(StatusCheck.objects.filter(repository=fossil_repo_obj))
c588255… ragelink 100 assert checks[0] == c2 # newest first
c588255… ragelink 101 assert checks[1] == c1
c588255… ragelink 102
c588255… ragelink 103 def test_state_choices(self):
c588255… ragelink 104 assert "pending" in StatusCheck.State.values
c588255… ragelink 105 assert "success" in StatusCheck.State.values
c588255… ragelink 106 assert "failure" in StatusCheck.State.values
c588255… ragelink 107 assert "error" in StatusCheck.State.values
c588255… ragelink 108
c588255… ragelink 109
c588255… ragelink 110 # --- Status Check API POST Tests ---
c588255… ragelink 111
c588255… ragelink 112
c588255… ragelink 113 @pytest.mark.django_db
c588255… ragelink 114 class TestStatusCheckAPIPost:
c588255… ragelink 115 def test_post_creates_status_check(self, client, sample_project, fossil_repo_obj, api_token):
c588255… ragelink 116 token_obj, raw_token = api_token
c588255… ragelink 117 response = client.post(
c588255… ragelink 118 f"/projects/{sample_project.slug}/fossil/api/status",
c588255… ragelink 119 data=json.dumps(
c588255… ragelink 120 {
c588255… ragelink 121 "checkin": "deadbeef123",
c588255… ragelink 122 "context": "ci/tests",
c588255… ragelink 123 "state": "success",
c588255… ragelink 124 "description": "All tests passed",
c588255… ragelink 125 "target_url": "https://ci.example.com/build/42",
c588255… ragelink 126 }
c588255… ragelink 127 ),
c588255… ragelink 128 content_type="application/json",
c588255… ragelink 129 HTTP_AUTHORIZATION=f"Bearer {raw_token}",
c588255… ragelink 130 )
c588255… ragelink 131 assert response.status_code == 201
c588255… ragelink 132 data = response.json()
c588255… ragelink 133 assert data["context"] == "ci/tests"
c588255… ragelink 134 assert data["state"] == "success"
c588255… ragelink 135 assert data["created"] is True
c588255… ragelink 136
c588255… ragelink 137 check = StatusCheck.objects.get(repository=fossil_repo_obj, checkin_uuid="deadbeef123", context="ci/tests")
c588255… ragelink 138 assert check.state == "success"
c588255… ragelink 139 assert check.description == "All tests passed"
c588255… ragelink 140
c588255… ragelink 141 def test_post_updates_existing_check(self, client, sample_project, fossil_repo_obj, api_token):
c588255… ragelink 142 token_obj, raw_token = api_token
c588255… ragelink 143 # Create initial check
c588255… ragelink 144 StatusCheck.objects.create(repository=fossil_repo_obj, checkin_uuid="update123", context="ci/tests", state="pending")
c588255… ragelink 145 # Update it
c588255… ragelink 146 response = client.post(
c588255… ragelink 147 f"/projects/{sample_project.slug}/fossil/api/status",
c588255… ragelink 148 data=json.dumps(
c588255… ragelink 149 {
c588255… ragelink 150 "checkin": "update123",
c588255… ragelink 151 "context": "ci/tests",
c588255… ragelink 152 "state": "success",
c588255… ragelink 153 "description": "Now passing",
c588255… ragelink 154 }
c588255… ragelink 155 ),
c588255… ragelink 156 content_type="application/json",
c588255… ragelink 157 HTTP_AUTHORIZATION=f"Bearer {raw_token}",
c588255… ragelink 158 )
c588255… ragelink 159 assert response.status_code == 200
c588255… ragelink 160 data = response.json()
c588255… ragelink 161 assert data["created"] is False
c588255… ragelink 162 assert data["state"] == "success"
c588255… ragelink 163
c588255… ragelink 164 def test_post_without_token_returns_401(self, client, sample_project, fossil_repo_obj):
c588255… ragelink 165 response = client.post(
c588255… ragelink 166 f"/projects/{sample_project.slug}/fossil/api/status",
c588255… ragelink 167 data=json.dumps({"checkin": "abc", "context": "ci/tests", "state": "success"}),
c588255… ragelink 168 content_type="application/json",
c588255… ragelink 169 )
c588255… ragelink 170 assert response.status_code == 401
c588255… ragelink 171
c588255… ragelink 172 def test_post_with_invalid_token_returns_401(self, client, sample_project, fossil_repo_obj):
c588255… ragelink 173 response = client.post(
c588255… ragelink 174 f"/projects/{sample_project.slug}/fossil/api/status",
c588255… ragelink 175 data=json.dumps({"checkin": "abc", "context": "ci/tests", "state": "success"}),
c588255… ragelink 176 content_type="application/json",
c588255… ragelink 177 HTTP_AUTHORIZATION="Bearer invalid_token_xyz",
c588255… ragelink 178 )
c588255… ragelink 179 assert response.status_code == 401
c588255… ragelink 180
c588255… ragelink 181 def test_post_with_wrong_repo_token_returns_401(self, client, sample_project, fossil_repo_obj, admin_user, org):
c588255… ragelink 182 """Token scoped to a different repo should fail."""
c588255… ragelink 183 from projects.models import Project
c588255… ragelink 184
c588255… ragelink 185 other_project = Project.objects.create(name="Other Project", organization=org, visibility="private", created_by=admin_user)
c588255… ragelink 186 other_repo = FossilRepository.objects.get(project=other_project, deleted_at__isnull=True)
c588255… ragelink 187 raw, token_hash, prefix = APIToken.generate()
c588255… ragelink 188 APIToken.objects.create(
c588255… ragelink 189 repository=other_repo, name="Other Token", token_hash=token_hash, token_prefix=prefix, created_by=admin_user
c588255… ragelink 190 )
c588255… ragelink 191
c588255… ragelink 192 response = client.post(
c588255… ragelink 193 f"/projects/{sample_project.slug}/fossil/api/status",
c588255… ragelink 194 data=json.dumps({"checkin": "abc", "context": "ci/tests", "state": "success"}),
c588255… ragelink 195 content_type="application/json",
c588255… ragelink 196 HTTP_AUTHORIZATION=f"Bearer {raw}",
c588255… ragelink 197 )
c588255… ragelink 198 assert response.status_code == 401
c588255… ragelink 199
c588255… ragelink 200 def test_post_missing_checkin_returns_400(self, client, sample_project, fossil_repo_obj, api_token):
c588255… ragelink 201 _, raw_token = api_token
c588255… ragelink 202 response = client.post(
c588255… ragelink 203 f"/projects/{sample_project.slug}/fossil/api/status",
c588255… ragelink 204 data=json.dumps({"context": "ci/tests", "state": "success"}),
c588255… ragelink 205 content_type="application/json",
c588255… ragelink 206 HTTP_AUTHORIZATION=f"Bearer {raw_token}",
c588255… ragelink 207 )
c588255… ragelink 208 assert response.status_code == 400
c588255… ragelink 209 assert "checkin" in response.json()["error"]
c588255… ragelink 210
c588255… ragelink 211 def test_post_missing_context_returns_400(self, client, sample_project, fossil_repo_obj, api_token):
c588255… ragelink 212 _, raw_token = api_token
c588255… ragelink 213 response = client.post(
c588255… ragelink 214 f"/projects/{sample_project.slug}/fossil/api/status",
c588255… ragelink 215 data=json.dumps({"checkin": "abc123", "state": "success"}),
c588255… ragelink 216 content_type="application/json",
c588255… ragelink 217 HTTP_AUTHORIZATION=f"Bearer {raw_token}",
c588255… ragelink 218 )
c588255… ragelink 219 assert response.status_code == 400
c588255… ragelink 220 assert "context" in response.json()["error"]
c588255… ragelink 221
c588255… ragelink 222 def test_post_invalid_state_returns_400(self, client, sample_project, fossil_repo_obj, api_token):
c588255… ragelink 223 _, raw_token = api_token
c588255… ragelink 224 response = client.post(
c588255… ragelink 225 f"/projects/{sample_project.slug}/fossil/api/status",
c588255… ragelink 226 data=json.dumps({"checkin": "abc123", "context": "ci/tests", "state": "bogus"}),
c588255… ragelink 227 content_type="application/json",
c588255… ragelink 228 HTTP_AUTHORIZATION=f"Bearer {raw_token}",
c588255… ragelink 229 )
c588255… ragelink 230 assert response.status_code == 400
c588255… ragelink 231 assert "state" in response.json()["error"]
c588255… ragelink 232
c588255… ragelink 233 def test_post_invalid_json_returns_400(self, client, sample_project, fossil_repo_obj, api_token):
c588255… ragelink 234 _, raw_token = api_token
c588255… ragelink 235 response = client.post(
c588255… ragelink 236 f"/projects/{sample_project.slug}/fossil/api/status",
c588255… ragelink 237 data="not json",
c588255… ragelink 238 content_type="application/json",
c588255… ragelink 239 HTTP_AUTHORIZATION=f"Bearer {raw_token}",
c588255… ragelink 240 )
c588255… ragelink 241 assert response.status_code == 400
c588255… ragelink 242
c588255… ragelink 243 def test_post_expired_token_returns_401(self, client, sample_project, fossil_repo_obj, admin_user):
c588255… ragelink 244 from datetime import timedelta
c588255… ragelink 245
c588255… ragelink 246 from django.utils import timezone
c588255… ragelink 247
c588255… ragelink 248 raw, token_hash, prefix = APIToken.generate()
c588255… ragelink 249 APIToken.objects.create(
c588255… ragelink 250 repository=fossil_repo_obj,
c588255… ragelink 251 name="Expired Token",
c588255… ragelink 252 token_hash=token_hash,
c588255… ragelink 253 token_prefix=prefix,
c588255… ragelink 254 expires_at=timezone.now() - timedelta(days=1),
c588255… ragelink 255 created_by=admin_user,
c588255… ragelink 256 )
c588255… ragelink 257 response = client.post(
c588255… ragelink 258 f"/projects/{sample_project.slug}/fossil/api/status",
c588255… ragelink 259 data=json.dumps({"checkin": "abc", "context": "ci/tests", "state": "success"}),
c588255… ragelink 260 content_type="application/json",
c588255… ragelink 261 HTTP_AUTHORIZATION=f"Bearer {raw}",
c588255… ragelink 262 )
c588255… ragelink 263 assert response.status_code == 401
c588255… ragelink 264
c588255… ragelink 265 def test_post_token_without_status_write_returns_403(self, client, sample_project, fossil_repo_obj, admin_user):
c588255… ragelink 266 raw, token_hash, prefix = APIToken.generate()
c588255… ragelink 267 APIToken.objects.create(
c588255… ragelink 268 repository=fossil_repo_obj,
c588255… ragelink 269 name="Read Only Token",
c588255… ragelink 270 token_hash=token_hash,
c588255… ragelink 271 token_prefix=prefix,
c588255… ragelink 272 permissions="status:read",
c588255… ragelink 273 created_by=admin_user,
c588255… ragelink 274 )
c588255… ragelink 275 response = client.post(
c588255… ragelink 276 f"/projects/{sample_project.slug}/fossil/api/status",
c588255… ragelink 277 data=json.dumps({"checkin": "abc", "context": "ci/tests", "state": "success"}),
c588255… ragelink 278 content_type="application/json",
c588255… ragelink 279 HTTP_AUTHORIZATION=f"Bearer {raw}",
c588255… ragelink 280 )
c588255… ragelink 281 assert response.status_code == 403
c588255… ragelink 282
c588255… ragelink 283 def test_method_not_allowed(self, client, sample_project, fossil_repo_obj):
c588255… ragelink 284 response = client.delete(f"/projects/{sample_project.slug}/fossil/api/status")
c588255… ragelink 285 assert response.status_code == 405
c588255… ragelink 286
c588255… ragelink 287
c588255… ragelink 288 # --- Status Check API GET Tests ---
c588255… ragelink 289
c588255… ragelink 290
c588255… ragelink 291 @pytest.mark.django_db
c588255… ragelink 292 class TestStatusCheckAPIGet:
c588255… ragelink 293 def test_get_checks_for_checkin(self, admin_client, sample_project, fossil_repo_obj, status_check):
c588255… ragelink 294 response = admin_client.get(f"/projects/{sample_project.slug}/fossil/api/status?checkin={status_check.checkin_uuid}")
c588255… ragelink 295 assert response.status_code == 200
c588255… ragelink 296 data = response.json()
c588255… ragelink 297 assert len(data["checks"]) == 1
c588255… ragelink 298 assert data["checks"][0]["context"] == "ci/tests"
c588255… ragelink 299 assert data["checks"][0]["state"] == "success"
c588255… ragelink 300
c588255… ragelink 301 def test_get_without_checkin_param_returns_400(self, admin_client, sample_project, fossil_repo_obj):
c588255… ragelink 302 response = admin_client.get(f"/projects/{sample_project.slug}/fossil/api/status")
c588255… ragelink 303 assert response.status_code == 400
c588255… ragelink 304
c588255… ragelink 305 def test_get_empty_results(self, admin_client, sample_project, fossil_repo_obj):
c588255… ragelink 306 response = admin_client.get(f"/projects/{sample_project.slug}/fossil/api/status?checkin=nonexistent")
c588255… ragelink 307 assert response.status_code == 200
c588255… ragelink 308 assert len(response.json()["checks"]) == 0
c588255… ragelink 309
c588255… ragelink 310
c588255… ragelink 311 # --- Status Badge Tests ---
c588255… ragelink 312
c588255… ragelink 313
c588255… ragelink 314 @pytest.mark.django_db
c588255… ragelink 315 class TestStatusBadge:
c588255… ragelink 316 def test_badge_unknown_no_checks(self, client, sample_project, fossil_repo_obj):
c588255… ragelink 317 response = client.get(f"/projects/{sample_project.slug}/fossil/api/status/abc123/badge.svg")
c588255… ragelink 318 assert response.status_code == 200
c588255… ragelink 319 assert response["Content-Type"] == "image/svg+xml"
c588255… ragelink 320 assert "unknown" in response.content.decode()
c588255… ragelink 321
c588255… ragelink 322 def test_badge_passing(self, client, sample_project, fossil_repo_obj):
c588255… ragelink 323 StatusCheck.objects.create(repository=fossil_repo_obj, checkin_uuid="pass123", context="ci/tests", state="success")
c588255… ragelink 324 response = client.get(f"/projects/{sample_project.slug}/fossil/api/status/pass123/badge.svg")
c588255… ragelink 325 assert response.status_code == 200
c588255… ragelink 326 assert "passing" in response.content.decode()
c588255… ragelink 327
c588255… ragelink 328 def test_badge_failing(self, client, sample_project, fossil_repo_obj):
c588255… ragelink 329 StatusCheck.objects.create(repository=fossil_repo_obj, checkin_uuid="fail123", context="ci/tests", state="failure")
c588255… ragelink 330 response = client.get(f"/projects/{sample_project.slug}/fossil/api/status/fail123/badge.svg")
c588255… ragelink 331 assert response.status_code == 200
c588255… ragelink 332 assert "failing" in response.content.decode()
c588255… ragelink 333
c588255… ragelink 334 def test_badge_pending(self, client, sample_project, fossil_repo_obj):
c588255… ragelink 335 StatusCheck.objects.create(repository=fossil_repo_obj, checkin_uuid="pend123", context="ci/tests", state="pending")
c588255… ragelink 336 response = client.get(f"/projects/{sample_project.slug}/fossil/api/status/pend123/badge.svg")
c588255… ragelink 337 assert response.status_code == 200
c588255… ragelink 338 assert "pending" in response.content.decode()
c588255… ragelink 339
c588255… ragelink 340 def test_badge_mixed_failing_wins(self, client, sample_project, fossil_repo_obj):
c588255… ragelink 341 """If any check is failing, the aggregate badge should say 'failing'."""
c588255… ragelink 342 StatusCheck.objects.create(repository=fossil_repo_obj, checkin_uuid="mixed123", context="ci/tests", state="success")
c588255… ragelink 343 StatusCheck.objects.create(repository=fossil_repo_obj, checkin_uuid="mixed123", context="ci/lint", state="failure")
c588255… ragelink 344 response = client.get(f"/projects/{sample_project.slug}/fossil/api/status/mixed123/badge.svg")
c588255… ragelink 345 assert "failing" in response.content.decode()
c588255… ragelink 346
c588255… ragelink 347 def test_badge_nonexistent_project_returns_404(self, client):
c588255… ragelink 348 response = client.get("/projects/nonexistent-project/fossil/api/status/abc/badge.svg")
c588255… ragelink 349 assert response.status_code == 404

Keyboard Shortcuts

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