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