|
c588255…
|
ragelink
|
1 |
from unittest.mock import MagicMock, patch |
|
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.models import FossilRepository |
|
c588255…
|
ragelink
|
8 |
from organization.models import Team |
|
c588255…
|
ragelink
|
9 |
from projects.models import ProjectTeam |
|
c588255…
|
ragelink
|
10 |
|
|
c588255…
|
ragelink
|
11 |
|
|
c588255…
|
ragelink
|
12 |
@pytest.fixture |
|
c588255…
|
ragelink
|
13 |
def fossil_repo_obj(sample_project): |
|
c588255…
|
ragelink
|
14 |
"""Return the auto-created FossilRepository for sample_project.""" |
|
c588255…
|
ragelink
|
15 |
return FossilRepository.objects.get(project=sample_project, deleted_at__isnull=True) |
|
c588255…
|
ragelink
|
16 |
|
|
c588255…
|
ragelink
|
17 |
|
|
c588255…
|
ragelink
|
18 |
@pytest.fixture |
|
c588255…
|
ragelink
|
19 |
def writer_user(db, admin_user, sample_project): |
|
c588255…
|
ragelink
|
20 |
"""User with write access but not admin.""" |
|
c588255…
|
ragelink
|
21 |
writer = User.objects.create_user(username="writer", password="testpass123") |
|
c588255…
|
ragelink
|
22 |
team = Team.objects.create(name="Writers", organization=sample_project.organization, created_by=admin_user) |
|
c588255…
|
ragelink
|
23 |
team.members.add(writer) |
|
c588255…
|
ragelink
|
24 |
ProjectTeam.objects.create(project=sample_project, team=team, role="write", created_by=admin_user) |
|
c588255…
|
ragelink
|
25 |
return writer |
|
c588255…
|
ragelink
|
26 |
|
|
c588255…
|
ragelink
|
27 |
|
|
c588255…
|
ragelink
|
28 |
@pytest.fixture |
|
c588255…
|
ragelink
|
29 |
def writer_client(writer_user): |
|
c588255…
|
ragelink
|
30 |
client = Client() |
|
c588255…
|
ragelink
|
31 |
client.login(username="writer", password="testpass123") |
|
c588255…
|
ragelink
|
32 |
return client |
|
c588255…
|
ragelink
|
33 |
|
|
c588255…
|
ragelink
|
34 |
|
|
c588255…
|
ragelink
|
35 |
# --- Shun List View Tests --- |
|
c588255…
|
ragelink
|
36 |
|
|
c588255…
|
ragelink
|
37 |
|
|
c588255…
|
ragelink
|
38 |
@pytest.mark.django_db |
|
c588255…
|
ragelink
|
39 |
class TestShunListView: |
|
c588255…
|
ragelink
|
40 |
def test_list_shunned_as_admin(self, admin_client, sample_project, fossil_repo_obj): |
|
c588255…
|
ragelink
|
41 |
with patch("fossil.cli.FossilCLI") as mock_cli_cls: |
|
c588255…
|
ragelink
|
42 |
cli_instance = MagicMock() |
|
c588255…
|
ragelink
|
43 |
cli_instance.is_available.return_value = True |
|
c588255…
|
ragelink
|
44 |
cli_instance.shun_list.return_value = ["abc123def456", "789012345678"] |
|
c588255…
|
ragelink
|
45 |
mock_cli_cls.return_value = cli_instance |
|
c588255…
|
ragelink
|
46 |
# Patch exists_on_disk |
|
c588255…
|
ragelink
|
47 |
with patch.object(type(fossil_repo_obj), "exists_on_disk", new_callable=lambda: property(lambda self: True)): |
|
c588255…
|
ragelink
|
48 |
response = admin_client.get(f"/projects/{sample_project.slug}/fossil/admin/shun/") |
|
c588255…
|
ragelink
|
49 |
assert response.status_code == 200 |
|
c588255…
|
ragelink
|
50 |
content = response.content.decode() |
|
c588255…
|
ragelink
|
51 |
assert "Shunned Artifacts" in content |
|
c588255…
|
ragelink
|
52 |
|
|
c588255…
|
ragelink
|
53 |
def test_list_empty(self, admin_client, sample_project, fossil_repo_obj): |
|
c588255…
|
ragelink
|
54 |
with patch("fossil.cli.FossilCLI") as mock_cli_cls: |
|
c588255…
|
ragelink
|
55 |
cli_instance = MagicMock() |
|
c588255…
|
ragelink
|
56 |
cli_instance.is_available.return_value = True |
|
c588255…
|
ragelink
|
57 |
cli_instance.shun_list.return_value = [] |
|
c588255…
|
ragelink
|
58 |
mock_cli_cls.return_value = cli_instance |
|
c588255…
|
ragelink
|
59 |
with patch.object(type(fossil_repo_obj), "exists_on_disk", new_callable=lambda: property(lambda self: True)): |
|
c588255…
|
ragelink
|
60 |
response = admin_client.get(f"/projects/{sample_project.slug}/fossil/admin/shun/") |
|
c588255…
|
ragelink
|
61 |
assert response.status_code == 200 |
|
c588255…
|
ragelink
|
62 |
assert "No artifacts have been shunned" in response.content.decode() |
|
c588255…
|
ragelink
|
63 |
|
|
c588255…
|
ragelink
|
64 |
def test_list_denied_for_writer(self, writer_client, sample_project, fossil_repo_obj): |
|
c588255…
|
ragelink
|
65 |
response = writer_client.get(f"/projects/{sample_project.slug}/fossil/admin/shun/") |
|
c588255…
|
ragelink
|
66 |
assert response.status_code == 403 |
|
c588255…
|
ragelink
|
67 |
|
|
c588255…
|
ragelink
|
68 |
def test_list_denied_for_no_perm(self, no_perm_client, sample_project): |
|
c588255…
|
ragelink
|
69 |
response = no_perm_client.get(f"/projects/{sample_project.slug}/fossil/admin/shun/") |
|
c588255…
|
ragelink
|
70 |
assert response.status_code == 403 |
|
c588255…
|
ragelink
|
71 |
|
|
c588255…
|
ragelink
|
72 |
def test_list_denied_for_anon(self, client, sample_project): |
|
c588255…
|
ragelink
|
73 |
response = client.get(f"/projects/{sample_project.slug}/fossil/admin/shun/") |
|
c588255…
|
ragelink
|
74 |
assert response.status_code == 302 # redirect to login |
|
c588255…
|
ragelink
|
75 |
|
|
c588255…
|
ragelink
|
76 |
|
|
c588255…
|
ragelink
|
77 |
# --- Shun Artifact View Tests --- |
|
c588255…
|
ragelink
|
78 |
|
|
c588255…
|
ragelink
|
79 |
|
|
c588255…
|
ragelink
|
80 |
@pytest.mark.django_db |
|
c588255…
|
ragelink
|
81 |
class TestShunArtifactView: |
|
c588255…
|
ragelink
|
82 |
def test_shun_artifact_success(self, admin_client, sample_project, fossil_repo_obj): |
|
c588255…
|
ragelink
|
83 |
with patch("fossil.cli.FossilCLI") as mock_cli_cls: |
|
c588255…
|
ragelink
|
84 |
cli_instance = MagicMock() |
|
c588255…
|
ragelink
|
85 |
cli_instance.is_available.return_value = True |
|
c588255…
|
ragelink
|
86 |
cli_instance.shun.return_value = {"success": True, "message": "Artifact shunned"} |
|
c588255…
|
ragelink
|
87 |
mock_cli_cls.return_value = cli_instance |
|
c588255…
|
ragelink
|
88 |
with patch.object(type(fossil_repo_obj), "exists_on_disk", new_callable=lambda: property(lambda self: True)): |
|
c588255…
|
ragelink
|
89 |
response = admin_client.post( |
|
c588255…
|
ragelink
|
90 |
f"/projects/{sample_project.slug}/fossil/admin/shun/add/", |
|
c588255…
|
ragelink
|
91 |
{ |
|
c588255…
|
ragelink
|
92 |
"artifact_uuid": "a1b2c3d4e5f67890", |
|
c588255…
|
ragelink
|
93 |
"confirmation": "a1b2c3d4", |
|
c588255…
|
ragelink
|
94 |
"reason": "Leaked secret", |
|
c588255…
|
ragelink
|
95 |
}, |
|
c588255…
|
ragelink
|
96 |
) |
|
c588255…
|
ragelink
|
97 |
assert response.status_code == 302 |
|
c588255…
|
ragelink
|
98 |
assert response.url == f"/projects/{sample_project.slug}/fossil/admin/shun/" |
|
c588255…
|
ragelink
|
99 |
|
|
c588255…
|
ragelink
|
100 |
def test_shun_requires_confirmation(self, admin_client, sample_project, fossil_repo_obj): |
|
c588255…
|
ragelink
|
101 |
response = admin_client.post( |
|
c588255…
|
ragelink
|
102 |
f"/projects/{sample_project.slug}/fossil/admin/shun/add/", |
|
c588255…
|
ragelink
|
103 |
{ |
|
c588255…
|
ragelink
|
104 |
"artifact_uuid": "a1b2c3d4e5f67890", |
|
c588255…
|
ragelink
|
105 |
"confirmation": "wrong", |
|
c588255…
|
ragelink
|
106 |
"reason": "test", |
|
c588255…
|
ragelink
|
107 |
}, |
|
c588255…
|
ragelink
|
108 |
) |
|
c588255…
|
ragelink
|
109 |
assert response.status_code == 302 # redirects with error message |
|
c588255…
|
ragelink
|
110 |
|
|
c588255…
|
ragelink
|
111 |
def test_shun_empty_uuid_rejected(self, admin_client, sample_project, fossil_repo_obj): |
|
c588255…
|
ragelink
|
112 |
response = admin_client.post( |
|
c588255…
|
ragelink
|
113 |
f"/projects/{sample_project.slug}/fossil/admin/shun/add/", |
|
c588255…
|
ragelink
|
114 |
{ |
|
c588255…
|
ragelink
|
115 |
"artifact_uuid": "", |
|
c588255…
|
ragelink
|
116 |
"confirmation": "", |
|
c588255…
|
ragelink
|
117 |
}, |
|
c588255…
|
ragelink
|
118 |
) |
|
c588255…
|
ragelink
|
119 |
assert response.status_code == 302 |
|
c588255…
|
ragelink
|
120 |
|
|
c588255…
|
ragelink
|
121 |
def test_shun_invalid_uuid_format(self, admin_client, sample_project, fossil_repo_obj): |
|
c588255…
|
ragelink
|
122 |
response = admin_client.post( |
|
c588255…
|
ragelink
|
123 |
f"/projects/{sample_project.slug}/fossil/admin/shun/add/", |
|
c588255…
|
ragelink
|
124 |
{ |
|
c588255…
|
ragelink
|
125 |
"artifact_uuid": "not-a-hex-hash!!!", |
|
c588255…
|
ragelink
|
126 |
"confirmation": "not-a-he", |
|
c588255…
|
ragelink
|
127 |
}, |
|
c588255…
|
ragelink
|
128 |
) |
|
c588255…
|
ragelink
|
129 |
assert response.status_code == 302 |
|
c588255…
|
ragelink
|
130 |
|
|
c588255…
|
ragelink
|
131 |
def test_shun_get_redirects(self, admin_client, sample_project, fossil_repo_obj): |
|
c588255…
|
ragelink
|
132 |
response = admin_client.get(f"/projects/{sample_project.slug}/fossil/admin/shun/add/") |
|
c588255…
|
ragelink
|
133 |
assert response.status_code == 302 |
|
c588255…
|
ragelink
|
134 |
|
|
c588255…
|
ragelink
|
135 |
def test_shun_denied_for_writer(self, writer_client, sample_project, fossil_repo_obj): |
|
c588255…
|
ragelink
|
136 |
response = writer_client.post( |
|
c588255…
|
ragelink
|
137 |
f"/projects/{sample_project.slug}/fossil/admin/shun/add/", |
|
c588255…
|
ragelink
|
138 |
{ |
|
c588255…
|
ragelink
|
139 |
"artifact_uuid": "a1b2c3d4e5f67890", |
|
c588255…
|
ragelink
|
140 |
"confirmation": "a1b2c3d4", |
|
c588255…
|
ragelink
|
141 |
}, |
|
c588255…
|
ragelink
|
142 |
) |
|
c588255…
|
ragelink
|
143 |
assert response.status_code == 403 |
|
c588255…
|
ragelink
|
144 |
|
|
c588255…
|
ragelink
|
145 |
def test_shun_denied_for_anon(self, client, sample_project): |
|
c588255…
|
ragelink
|
146 |
response = client.post( |
|
c588255…
|
ragelink
|
147 |
f"/projects/{sample_project.slug}/fossil/admin/shun/add/", |
|
c588255…
|
ragelink
|
148 |
{ |
|
c588255…
|
ragelink
|
149 |
"artifact_uuid": "a1b2c3d4e5f67890", |
|
c588255…
|
ragelink
|
150 |
"confirmation": "a1b2c3d4", |
|
c588255…
|
ragelink
|
151 |
}, |
|
c588255…
|
ragelink
|
152 |
) |
|
c588255…
|
ragelink
|
153 |
assert response.status_code == 302 # redirect to login |
|
c588255…
|
ragelink
|
154 |
|
|
c588255…
|
ragelink
|
155 |
def test_shun_cli_failure(self, admin_client, sample_project, fossil_repo_obj): |
|
c588255…
|
ragelink
|
156 |
with patch("fossil.cli.FossilCLI") as mock_cli_cls: |
|
c588255…
|
ragelink
|
157 |
cli_instance = MagicMock() |
|
c588255…
|
ragelink
|
158 |
cli_instance.is_available.return_value = True |
|
c588255…
|
ragelink
|
159 |
cli_instance.shun.return_value = {"success": False, "message": "Unknown artifact"} |
|
c588255…
|
ragelink
|
160 |
mock_cli_cls.return_value = cli_instance |
|
c588255…
|
ragelink
|
161 |
with patch.object(type(fossil_repo_obj), "exists_on_disk", new_callable=lambda: property(lambda self: True)): |
|
c588255…
|
ragelink
|
162 |
response = admin_client.post( |
|
c588255…
|
ragelink
|
163 |
f"/projects/{sample_project.slug}/fossil/admin/shun/add/", |
|
c588255…
|
ragelink
|
164 |
{ |
|
c588255…
|
ragelink
|
165 |
"artifact_uuid": "a1b2c3d4e5f67890", |
|
c588255…
|
ragelink
|
166 |
"confirmation": "a1b2c3d4", |
|
c588255…
|
ragelink
|
167 |
"reason": "test", |
|
c588255…
|
ragelink
|
168 |
}, |
|
c588255…
|
ragelink
|
169 |
) |
|
c588255…
|
ragelink
|
170 |
assert response.status_code == 302 |
|
c588255…
|
ragelink
|
171 |
|
|
c588255…
|
ragelink
|
172 |
|
|
c588255…
|
ragelink
|
173 |
# --- CLI Shun Method Tests --- |
|
c588255…
|
ragelink
|
174 |
|
|
c588255…
|
ragelink
|
175 |
|
|
c588255…
|
ragelink
|
176 |
@pytest.mark.django_db |
|
c588255…
|
ragelink
|
177 |
class TestFossilCLIShun: |
|
c588255…
|
ragelink
|
178 |
def test_shun_calls_fossil_binary(self): |
|
c588255…
|
ragelink
|
179 |
from fossil.cli import FossilCLI |
|
c588255…
|
ragelink
|
180 |
|
|
c588255…
|
ragelink
|
181 |
cli = FossilCLI(binary="/usr/bin/fossil") |
|
c588255…
|
ragelink
|
182 |
with patch("subprocess.run") as mock_run: |
|
c588255…
|
ragelink
|
183 |
mock_run.return_value = MagicMock(returncode=0, stdout="Artifact shunned\n", stderr="") |
|
c588255…
|
ragelink
|
184 |
result = cli.shun("/tmp/test.fossil", "abc123def456", reason="test") |
|
c588255…
|
ragelink
|
185 |
assert result["success"] is True |
|
c588255…
|
ragelink
|
186 |
assert "Artifact shunned" in result["message"] |
|
c588255…
|
ragelink
|
187 |
call_args = mock_run.call_args[0][0] |
|
c588255…
|
ragelink
|
188 |
assert "shun" in call_args |
|
c588255…
|
ragelink
|
189 |
assert "abc123def456" in call_args |
|
c588255…
|
ragelink
|
190 |
|
|
c588255…
|
ragelink
|
191 |
def test_shun_returns_failure(self): |
|
c588255…
|
ragelink
|
192 |
from fossil.cli import FossilCLI |
|
c588255…
|
ragelink
|
193 |
|
|
c588255…
|
ragelink
|
194 |
cli = FossilCLI(binary="/usr/bin/fossil") |
|
c588255…
|
ragelink
|
195 |
with patch("subprocess.run") as mock_run: |
|
c588255…
|
ragelink
|
196 |
mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="Not found") |
|
c588255…
|
ragelink
|
197 |
result = cli.shun("/tmp/test.fossil", "nonexistent") |
|
c588255…
|
ragelink
|
198 |
assert result["success"] is False |
|
c588255…
|
ragelink
|
199 |
assert "Not found" in result["message"] |
|
c588255…
|
ragelink
|
200 |
|
|
c588255…
|
ragelink
|
201 |
def test_shun_list_returns_uuids(self): |
|
c588255…
|
ragelink
|
202 |
from fossil.cli import FossilCLI |
|
c588255…
|
ragelink
|
203 |
|
|
c588255…
|
ragelink
|
204 |
cli = FossilCLI(binary="/usr/bin/fossil") |
|
c588255…
|
ragelink
|
205 |
with patch("subprocess.run") as mock_run: |
|
c588255…
|
ragelink
|
206 |
mock_run.return_value = MagicMock(returncode=0, stdout="abc123\ndef456\n", stderr="") |
|
c588255…
|
ragelink
|
207 |
result = cli.shun_list("/tmp/test.fossil") |
|
c588255…
|
ragelink
|
208 |
assert result == ["abc123", "def456"] |
|
c588255…
|
ragelink
|
209 |
|
|
c588255…
|
ragelink
|
210 |
def test_shun_list_empty(self): |
|
c588255…
|
ragelink
|
211 |
from fossil.cli import FossilCLI |
|
c588255…
|
ragelink
|
212 |
|
|
c588255…
|
ragelink
|
213 |
cli = FossilCLI(binary="/usr/bin/fossil") |
|
c588255…
|
ragelink
|
214 |
with patch("subprocess.run") as mock_run: |
|
c588255…
|
ragelink
|
215 |
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="") |
|
c588255…
|
ragelink
|
216 |
result = cli.shun_list("/tmp/test.fossil") |
|
c588255…
|
ragelink
|
217 |
assert result == [] |
|
c588255…
|
ragelink
|
218 |
|
|
c588255…
|
ragelink
|
219 |
def test_shun_list_failure_returns_empty(self): |
|
c588255…
|
ragelink
|
220 |
from fossil.cli import FossilCLI |
|
c588255…
|
ragelink
|
221 |
|
|
c588255…
|
ragelink
|
222 |
cli = FossilCLI(binary="/usr/bin/fossil") |
|
c588255…
|
ragelink
|
223 |
with patch("subprocess.run") as mock_run: |
|
c588255…
|
ragelink
|
224 |
mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="error") |
|
c588255…
|
ragelink
|
225 |
result = cli.shun_list("/tmp/test.fossil") |
|
c588255…
|
ragelink
|
226 |
assert result == [] |