Coverage for product_risk_suite/scraper/tests.py: 99%

1from unittest.mock import Mock, patch 

2 

3import requests 

4from django.contrib.auth.models import User 

5from django.core.exceptions import ValidationError 

6from django.test import TestCase 

7 

8from risk_assessment.models import Evidence, Status 

9 

10from .models import AssigneeMapping, RemoteCredential, Scraper, ScraperStatusMapping 

11from .pathutils import resolve_html_value, resolve_json_path 

12from .services import run_scraper_for_evidence 

13 

14 

15class PathUtilsTest(TestCase): 

16 def test_resolve_json_path_simple(self): 

17 data = {"assignee": {"name": "Jane Doe"}} 

18 self.assertEqual(resolve_json_path(data, "assignee.name"), "Jane Doe") 

19 

20 def test_resolve_json_path_with_index(self): 

21 data = {"fields": [{"assignee": "Jane Doe"}, {"assignee": "John Roe"}]} 

22 self.assertEqual(resolve_json_path(data, "fields[1].assignee"), "John Roe") 

23 

24 def test_resolve_json_path_missing_returns_none(self): 

25 data = {"assignee": {"name": "Jane Doe"}} 

26 self.assertIsNone(resolve_json_path(data, "assignee.email")) 

27 self.assertIsNone(resolve_json_path(data, "fields[0].assignee")) 

28 

29 def test_resolve_html_value_text(self): 

30 from bs4 import BeautifulSoup 

31 soup = BeautifulSoup('<div class="assignee">Jane Doe</div>', "lxml") 

32 self.assertEqual(resolve_html_value(soup, ".assignee"), "Jane Doe") 

33 

34 def test_resolve_html_value_attr(self): 

35 from bs4 import BeautifulSoup 

36 soup = BeautifulSoup('<div class="assignee" data-user="jdoe">Jane Doe</div>', "lxml") 

37 self.assertEqual(resolve_html_value(soup, ".assignee::attr(data-user)"), "jdoe") 

38 

39 def test_resolve_html_value_missing_returns_none(self): 

40 from bs4 import BeautifulSoup 

41 soup = BeautifulSoup('<div class="assignee">Jane Doe</div>', "lxml") 

42 self.assertIsNone(resolve_html_value(soup, ".status")) 

43 

44 

45class RemoteCredentialTest(TestCase): 

46 def test_password_round_trip_and_not_stored_plain(self): 

47 cred = RemoteCredential(name="c1", url="https://example.com", auth_type=RemoteCredential.BASIC, username="u") 

48 cred.password = "unsecure-password!91" 

49 cred.save() 

50 

51 self.assertNotIn("unsecure-password!91", cred._password_enc) 

52 self.assertEqual(RemoteCredential.objects.get(pk=cred.pk).password, "unsecure-password!91") 

53 

54 def test_api_token_round_trip(self): 

55 cred = RemoteCredential(name="c2", url="https://example.com", auth_type=RemoteCredential.TOKEN) 

56 cred.api_token = "tok_abc123" 

57 cred.save() 

58 

59 self.assertEqual(RemoteCredential.objects.get(pk=cred.pk).api_token, "tok_abc123") 

60 

61 def test_clean_requires_password_for_basic(self): 

62 cred = RemoteCredential(name="c3", url="https://example.com", auth_type=RemoteCredential.BASIC, username="u") 

63 with self.assertRaises(ValidationError): 

64 cred.clean() 

65 

66 def test_clean_requires_api_token_for_token(self): 

67 cred = RemoteCredential(name="c4", url="https://example.com", auth_type=RemoteCredential.TOKEN) 

68 with self.assertRaises(ValidationError): 

69 cred.clean() 

70 

71 

72class RunScraperForEvidenceTest(TestCase): 

73 def setUp(self): 

74 self.user = User.objects.create(username="jdoe", email="jdoe@example.com") 

75 self.status_open = Status.objects.create(status="Open") 

76 self.status_going = Status.objects.create(status="On going") 

77 self.credential = RemoteCredential( 

78 name="openproject", url="https://tracker.example.com", auth_type=RemoteCredential.BASIC, username="u", 

79 ) 

80 self.credential.password = "pw" 

81 self.credential.save() 

82 

83 self.json_scraper = Scraper.objects.create( 

84 name="json-scraper", credential=self.credential, response_format=Scraper.JSON, 

85 assignee_path="assignee.email", status_path="status.name", 

86 ) 

87 ScraperStatusMapping.objects.create(scraper=self.json_scraper, remote_value="In progress", status=self.status_going) 

88 

89 self.evidence = Evidence.objects.create( 

90 due_date="2026-04-23", status=self.status_open, scraper=self.json_scraper, 

91 evidence_link="https://tracker.example.com/tickets/1", 

92 ) 

93 

94 def _mock_response(self, json_data=None, text=None, status_code=200): 

95 response = Mock() 

96 response.status_code = status_code 

97 response.raise_for_status = Mock() 

98 if status_code >= 400: 

99 response.raise_for_status.side_effect = requests.HTTPError("boom") 

100 if json_data is not None: 

101 response.json = Mock(return_value=json_data) 

102 if text is not None: 

103 response.text = text 

104 return response 

105 

106 @patch("scraper.services.requests.get") 

107 def test_success_updates_status_and_responsible(self, mock_get): 

108 mock_get.return_value = self._mock_response( 

109 json_data={"assignee": {"email": "jdoe@example.com"}, "status": {"name": "In progress"}} 

110 ) 

111 

112 result = run_scraper_for_evidence(self.evidence) 

113 

114 self.evidence.refresh_from_db() 

115 self.assertTrue(result) 

116 self.assertFalse(self.evidence.last_scrape_failed) 

117 self.assertIsNotNone(self.evidence.last_successful_scrape_at) 

118 self.assertEqual(self.evidence.status, self.status_going) 

119 self.assertEqual(self.evidence.responsible, self.user) 

120 

121 @patch("scraper.services.requests.get") 

122 def test_unmapped_status_is_a_failure(self, mock_get): 

123 mock_get.return_value = self._mock_response( 

124 json_data={"assignee": {"email": "jdoe@example.com"}, "status": {"name": "Unknown remote status"}} 

125 ) 

126 

127 result = run_scraper_for_evidence(self.evidence) 

128 

129 self.evidence.refresh_from_db() 

130 self.assertFalse(result) 

131 self.assertTrue(self.evidence.last_scrape_failed) 

132 self.assertIsNone(self.evidence.last_successful_scrape_at) 

133 self.assertEqual(self.evidence.status, self.status_open) 

134 

135 @patch("scraper.services.requests.get") 

136 def test_unmatched_assignee_still_succeeds(self, mock_get): 

137 mock_get.return_value = self._mock_response( 

138 json_data={"assignee": {"email": "nobody@example.com"}, "status": {"name": "In progress"}} 

139 ) 

140 

141 result = run_scraper_for_evidence(self.evidence) 

142 

143 self.evidence.refresh_from_db() 

144 self.assertTrue(result) 

145 self.assertEqual(self.evidence.status, self.status_going) 

146 self.assertIsNone(self.evidence.responsible) 

147 

148 @patch("scraper.services.requests.get") 

149 def test_unmatched_assignee_unassigns_a_previously_set_responsible(self, mock_get): 

150 self.evidence.responsible = self.user 

151 self.evidence.save() 

152 mock_get.return_value = self._mock_response( 

153 json_data={"assignee": {"email": "nobody@example.com"}, "status": {"name": "In progress"}} 

154 ) 

155 

156 result = run_scraper_for_evidence(self.evidence) 

157 

158 self.evidence.refresh_from_db() 

159 self.assertTrue(result) 

160 self.assertIsNone(self.evidence.responsible) 

161 

162 @patch("scraper.services.requests.get") 

163 def test_network_error_is_a_failure(self, mock_get): 

164 mock_get.side_effect = requests.ConnectionError("no route to host") 

165 

166 result = run_scraper_for_evidence(self.evidence) 

167 

168 self.evidence.refresh_from_db() 

169 self.assertFalse(result) 

170 self.assertTrue(self.evidence.last_scrape_failed) 

171 

172 def test_url_outside_credential_domain_is_rejected(self): 

173 self.evidence.evidence_link = "https://evil.example.com/steal" 

174 self.evidence.save() 

175 

176 result = run_scraper_for_evidence(self.evidence) 

177 

178 self.evidence.refresh_from_db() 

179 self.assertFalse(result) 

180 self.assertTrue(self.evidence.last_scrape_failed) 

181 

182 def test_no_scraper_is_a_noop(self): 

183 self.evidence.scraper = None 

184 self.evidence.save() 

185 

186 self.assertIsNone(run_scraper_for_evidence(self.evidence)) 

187 

188 @patch("scraper.services.requests.get") 

189 def test_public_scraper_without_credential(self, mock_get): 

190 public_scraper = Scraper.objects.create( 

191 name="public-scraper", credential=None, response_format=Scraper.JSON, 

192 assignee_path="assignee.email", status_path="status.name", 

193 ) 

194 ScraperStatusMapping.objects.create(scraper=public_scraper, remote_value="In progress", status=self.status_going) 

195 self.evidence.scraper = public_scraper 

196 # a public scraper isn't restricted to any credential's domain 

197 self.evidence.scrape_url = "https://public.example.com/tickets/1" 

198 self.evidence.save() 

199 

200 mock_get.return_value = self._mock_response( 

201 json_data={"assignee": {"email": "jdoe@example.com"}, "status": {"name": "In progress"}} 

202 ) 

203 

204 result = run_scraper_for_evidence(self.evidence) 

205 

206 self.evidence.refresh_from_db() 

207 self.assertTrue(result) 

208 self.assertEqual(self.evidence.status, self.status_going) 

209 self.assertEqual(self.evidence.responsible, self.user) 

210 

211 _, call_kwargs = mock_get.call_args 

212 self.assertIsNone(call_kwargs["auth"]) 

213 self.assertNotIn("Authorization", call_kwargs["headers"]) 

214 

215 @patch("scraper.services.requests.get") 

216 def test_html_response_format(self, mock_get): 

217 html_scraper = Scraper.objects.create( 

218 name="html-scraper", credential=self.credential, response_format=Scraper.HTML, 

219 assignee_path=".assignee::attr(data-email)", status_path=".status", 

220 ) 

221 ScraperStatusMapping.objects.create(scraper=html_scraper, remote_value="In progress", status=self.status_going) 

222 self.evidence.scraper = html_scraper 

223 self.evidence.save() 

224 

225 mock_get.return_value = self._mock_response( 

226 text='<div class="assignee" data-email="jdoe@example.com"></div><div class="status">In progress</div>' 

227 ) 

228 

229 result = run_scraper_for_evidence(self.evidence) 

230 

231 self.evidence.refresh_from_db() 

232 self.assertTrue(result) 

233 self.assertEqual(self.evidence.status, self.status_going) 

234 self.assertEqual(self.evidence.responsible, self.user) 

235 

236 

237class AssigneeMappingResolutionTest(TestCase): 

238 def setUp(self): 

239 self.status_open = Status.objects.create(status="Open") 

240 self.status_going = Status.objects.create(status="On going") 

241 self.credential = RemoteCredential( 

242 name="codeberg", url="https://codeberg.org", auth_type=RemoteCredential.TOKEN, 

243 ) 

244 self.credential.api_token = "tok" 

245 self.credential.save() 

246 

247 self.scraper = Scraper.objects.create( 

248 name="codeberg-scraper", credential=self.credential, response_format=Scraper.JSON, 

249 assignee_path="assignee.login", status_path="status.name", 

250 ) 

251 ScraperStatusMapping.objects.create(scraper=self.scraper, remote_value="In progress", status=self.status_going) 

252 

253 self.evidence = Evidence.objects.create( 

254 due_date="2026-04-23", status=self.status_open, scraper=self.scraper, 

255 evidence_link="https://codeberg.org/some/repo/issues/1", 

256 ) 

257 

258 def _mock_response(self, assignee_login): 

259 response = Mock() 

260 response.status_code = 200 

261 response.raise_for_status = Mock() 

262 response.json = Mock(return_value={"assignee": {"login": assignee_login}, "status": {"name": "In progress"}}) 

263 return response 

264 

265 @patch("scraper.services.requests.get") 

266 def test_mapped_assignee_uses_mapped_local_user(self, mock_get): 

267 local_user = User.objects.create(username="schaefer.simon") 

268 AssigneeMapping.objects.create(base_url="https://codeberg.org", remote_username="CMon", user=local_user) 

269 mock_get.return_value = self._mock_response("CMon") 

270 

271 result = run_scraper_for_evidence(self.evidence) 

272 

273 self.evidence.refresh_from_db() 

274 self.assertTrue(result) 

275 self.assertEqual(self.evidence.responsible, local_user) 

276 

277 @patch("scraper.services.requests.get") 

278 def test_mapping_for_a_different_host_is_ignored(self, mock_get): 

279 other_user = User.objects.create(username="other") 

280 AssigneeMapping.objects.create(base_url="https://gitea.example.com", remote_username="CMon", user=other_user) 

281 same_name_user = User.objects.create(username="CMon") 

282 mock_get.return_value = self._mock_response("CMon") 

283 

284 result = run_scraper_for_evidence(self.evidence) 

285 

286 self.evidence.refresh_from_db() 

287 self.assertTrue(result) 

288 # no mapping matches this host, so it falls back to an exact local username match 

289 self.assertEqual(self.evidence.responsible, same_name_user) 

290 

291 @patch("scraper.services.requests.get") 

292 def test_unmapped_and_unknown_username_is_unassigned(self, mock_get): 

293 mock_get.return_value = self._mock_response("CMon") 

294 

295 result = run_scraper_for_evidence(self.evidence) 

296 

297 self.evidence.refresh_from_db() 

298 self.assertTrue(result) 

299 self.assertIsNone(self.evidence.responsible)