Coverage for test/sparql_analyser_test.py: 87%

1#!/usr/bin/python 

2# Copyright 2025, Arcangelo Massari <arcangelo.massari@unibo.it> 

3# 

4# Permission to use, copy, modify, and/or distribute this software for any purpose 

5# with or without fee is hereby granted, provided that the above copyright notice 

6# and this permission notice appear in all copies. 

7# 

8# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH 

9# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND 

10# FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, 

11# OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, 

12# DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS 

13# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS 

14# SOFTWARE. 

15 

16import sys 

17import unittest 

18from unittest.mock import MagicMock, patch 

19 

20from oc_meta.run.analyser.sparql_analyser import OCMetaSPARQLAnalyser 

21 

22 

23class TestOCMetaSPARQLAnalyser(unittest.TestCase): 

24 """Test cases for OCMetaSPARQLAnalyser class.""" 

25 

26 def setUp(self): 

27 """Set up test fixtures before each test method.""" 

28 self.test_endpoint = "http://test.example.com/sparql" 

29 self.analyser = OCMetaSPARQLAnalyser(sparql_endpoint=self.test_endpoint) 

30 

31 def test_initialization_with_endpoint(self): 

32 """Test that analyser initializes correctly with provided endpoint.""" 

33 self.assertEqual(self.analyser.sparql_endpoint, self.test_endpoint) 

34 self.assertIsNotNone(self.analyser.sparql) 

35 

36 def test_initialization_requires_endpoint(self): 

37 """Test that analyser requires an endpoint parameter.""" 

38 with self.assertRaises(TypeError): 

39 OCMetaSPARQLAnalyser() 

40 

41 def test_execute_sparql_query_success(self): 

42 """Test successful SPARQL query execution.""" 

43 mock_results = { 

44 "results": { 

45 "bindings": [ 

46 {"count": {"value": "100"}} 

47 ] 

48 } 

49 } 

50 

51 with patch.object(self.analyser.sparql, 'queryAndConvert', return_value=mock_results): 

52 result = self.analyser._execute_sparql_query("SELECT * WHERE { ?s ?p ?o }") 

53 self.assertEqual(result, mock_results) 

54 

55 def test_execute_sparql_query_failure(self): 

56 """Test SPARQL query execution failure.""" 

57 self.analyser.max_retries = 1 

58 with patch.object(self.analyser.sparql, 'queryAndConvert', side_effect=Exception("Query failed")): 

59 with self.assertRaises(Exception) as context: 

60 self.analyser._execute_sparql_query("INVALID QUERY") 

61 self.assertIn("SPARQL query failed after multiple retries", str(context.exception)) 

62 

63 @patch('time.sleep', return_value=None) 

64 def test_execute_sparql_query_retry_and_succeed(self, mock_sleep): 

65 """Test that query succeeds after a retry.""" 

66 self.analyser.max_retries = 2 

67 self.analyser.retry_delay = 1 

68 

69 mock_results = {"results": {"bindings": []}} 

70 side_effects = [Exception("Connection failed"), mock_results] 

71 

72 with patch.object(self.analyser.sparql, 'queryAndConvert', side_effect=side_effects) as mock_query: 

73 result = self.analyser._execute_sparql_query("ANY QUERY") 

74 

75 self.assertEqual(result, mock_results) 

76 self.assertEqual(mock_query.call_count, 2) 

77 mock_sleep.assert_called_once_with(1) 

78 

79 @patch('time.sleep', return_value=None) 

80 def test_execute_sparql_query_retry_and_fail(self, mock_sleep): 

81 """Test that query fails after all retries.""" 

82 self.analyser.max_retries = 3 

83 self.analyser.retry_delay = 1 

84 

85 side_effects = [Exception("Error 1"), Exception("Error 2"), Exception("Error 3")] 

86 

87 with patch.object(self.analyser.sparql, 'queryAndConvert', side_effect=side_effects) as mock_query: 

88 with self.assertRaises(Exception) as context: 

89 self.analyser._execute_sparql_query("ANY QUERY") 

90 

91 self.assertIn("SPARQL query failed after multiple retries", str(context.exception)) 

92 self.assertEqual(mock_query.call_count, 3) 

93 self.assertEqual(mock_sleep.call_count, 2) 

94 

95 def test_count_expressions(self): 

96 """Test counting fabio:Expression entities.""" 

97 mock_results = { 

98 "results": { 

99 "bindings": [ 

100 {"count": {"value": "1500"}} 

101 ] 

102 } 

103 } 

104 

105 with patch.object(self.analyser, '_execute_sparql_query', return_value=mock_results): 

106 count = self.analyser.count_expressions() 

107 self.assertEqual(count, 1500) 

108 

109 def test_count_role_entities(self): 

110 """Test counting pro role entities.""" 

111 mock_results = { 

112 "results": { 

113 "bindings": [ 

114 {"role": {"value": "http://purl.org/spar/pro/author"}, "count": {"value": "800"}}, 

115 {"role": {"value": "http://purl.org/spar/pro/publisher"}, "count": {"value": "200"}}, 

116 {"role": {"value": "http://purl.org/spar/pro/editor"}, "count": {"value": "100"}} 

117 ] 

118 } 

119 } 

120 

121 with patch.object(self.analyser, '_execute_sparql_query', return_value=mock_results): 

122 counts = self.analyser.count_role_entities() 

123 expected = { 

124 'pro:author': 800, 

125 'pro:publisher': 200, 

126 'pro:editor': 100 

127 } 

128 self.assertEqual(counts, expected) 

129 

130 def test_count_role_entities_empty_results(self): 

131 """Test counting pro role entities with empty results.""" 

132 mock_results = { 

133 "results": { 

134 "bindings": [] 

135 } 

136 } 

137 

138 with patch.object(self.analyser, '_execute_sparql_query', return_value=mock_results): 

139 counts = self.analyser.count_role_entities() 

140 expected = { 

141 'pro:author': 0, 

142 'pro:publisher': 0, 

143 'pro:editor': 0 

144 } 

145 self.assertEqual(counts, expected) 

146 

147 def test_count_venues(self): 

148 """Test counting distinct venues (simple count).""" 

149 mock_results = { 

150 "results": { 

151 "bindings": [ 

152 {"count": {"value": "350"}} 

153 ] 

154 } 

155 } 

156 

157 with patch.object(self.analyser, '_execute_sparql_query', return_value=mock_results): 

158 count = self.analyser.count_venues() 

159 self.assertEqual(count, 350) 

160 

161 def test_count_venues_disambiguated(self): 

162 """Test counting disambiguated venues.""" 

163 mock_results = { 

164 "results": { 

165 "bindings": [ 

166 { 

167 "venue": {"value": "http://venue1"}, 

168 "venueName": {"value": "Nature"}, 

169 "has_identifier": {"value": "true"} 

170 }, 

171 { 

172 "venue": {"value": "http://venue2"}, 

173 "venueName": {"value": "Science"}, 

174 "has_identifier": {"value": "false"} 

175 }, 

176 { 

177 "venue": {"value": "http://venue3"}, 

178 "venueName": {"value": "Cell"}, 

179 "has_identifier": {"value": "true"} 

180 } 

181 ] 

182 } 

183 } 

184 

185 with patch.object(self.analyser, '_execute_sparql_query', return_value=mock_results): 

186 count = self.analyser.count_venues_disambiguated() 

187 # venue1 and venue3 have external IDs (count by OMID = 2) 

188 # venue2 has no external IDs (count by name = 1) 

189 # Total = 3 distinct venues 

190 self.assertEqual(count, 3) 

191 

192 @patch('builtins.print') 

193 def test_run_all_analyses_success(self, mock_print): 

194 """Test running all analyses successfully.""" 

195 with patch.object(self.analyser, 'count_expressions', return_value=1500), \ 

196 patch.object(self.analyser, 'count_role_entities', return_value={'pro:author': 800, 'pro:publisher': 200, 'pro:editor': 100}), \ 

197 patch.object(self.analyser, 'count_venues_disambiguated', return_value=350), \ 

198 patch.object(self.analyser, 'count_venues', return_value=400): 

199 

200 results = self.analyser.run_all_analyses() 

201 

202 expected = { 

203 'fabio_expressions': 1500, 

204 'roles': {'pro:author': 800, 'pro:publisher': 200, 'pro:editor': 100}, 

205 'venues_disambiguated': 350, 

206 'venues_simple': 400 

207 } 

208 self.assertEqual(results, expected) 

209 

210 @patch('builtins.print') 

211 def test_run_selected_analyses_br_only(self, mock_print): 

212 """Test running only bibliographic resources analysis.""" 

213 with patch.object(self.analyser, 'count_expressions', return_value=1500): 

214 results = self.analyser.run_selected_analyses(analyze_br=True, analyze_ar=False, analyze_venues=False) 

215 

216 expected = { 

217 'fabio_expressions': 1500 

218 } 

219 self.assertEqual(results, expected) 

220 

221 @patch('builtins.print') 

222 def test_run_selected_analyses_venues_only(self, mock_print): 

223 """Test running only venues analysis.""" 

224 with patch.object(self.analyser, 'count_venues_disambiguated', return_value=350), \ 

225 patch.object(self.analyser, 'count_venues', return_value=400): 

226 

227 results = self.analyser.run_selected_analyses(analyze_br=False, analyze_ar=False, analyze_venues=True) 

228 

229 expected = { 

230 'venues_disambiguated': 350, 

231 'venues_simple': 400 

232 } 

233 self.assertEqual(results, expected) 

234 

235 @patch('builtins.print') 

236 def test_run_all_analyses_with_errors(self, mock_print): 

237 """Test running all analyses with some errors.""" 

238 with patch.object(self.analyser, 'count_expressions', side_effect=Exception("Connection error")), \ 

239 patch.object(self.analyser, 'count_role_entities', return_value={'pro:author': 800, 'pro:publisher': 200, 'pro:editor': 100}), \ 

240 patch.object(self.analyser, 'count_venues_disambiguated', return_value=350), \ 

241 patch.object(self.analyser, 'count_venues', return_value=400): 

242 

243 results = self.analyser.run_all_analyses() 

244 

245 expected = { 

246 'fabio_expressions': None, 

247 'roles': {'pro:author': 800, 'pro:publisher': 200, 'pro:editor': 100}, 

248 'venues_disambiguated': 350, 

249 'venues_simple': 400 

250 } 

251 self.assertEqual(results, expected) 

252 

253 

254class TestMainFunction(unittest.TestCase): 

255 """Test cases for the main function.""" 

256 

257 @patch('sys.argv', ['sparql_analyser.py', 'http://test.example.com/sparql']) 

258 @patch('oc_meta.run.analyser.sparql_analyser.OCMetaSPARQLAnalyser') 

259 @patch('builtins.print') 

260 def test_main_success(self, mock_print, mock_analyser_class): 

261 """Test main function with successful analysis.""" 

262 mock_analyser = MagicMock() 

263 mock_results = { 

264 'fabio_expressions': 1500, 

265 'roles': {'pro:author': 800, 'pro:publisher': 200, 'pro:editor': 100}, 

266 'venues_disambiguated': 350, 

267 'venues_simple': 400 

268 } 

269 mock_analyser.run_selected_analyses.return_value = mock_results 

270 mock_analyser_class.return_value = mock_analyser 

271 

272 from oc_meta.run.analyser.sparql_analyser import main 

273 result = main() 

274 

275 self.assertEqual(result, mock_results) 

276 mock_analyser_class.assert_called_once_with('http://test.example.com/sparql') 

277 mock_analyser.run_selected_analyses.assert_called_once_with(True, True, True) 

278 

279 @patch('sys.argv', ['sparql_analyser.py', 'http://test.example.com/sparql']) 

280 @patch('oc_meta.run.analyser.sparql_analyser.OCMetaSPARQLAnalyser') 

281 @patch('builtins.print') 

282 def test_main_failure(self, mock_print, mock_analyser_class): 

283 """Test main function with analysis failure.""" 

284 mock_analyser_class.side_effect = Exception("Initialization failed") 

285 

286 from oc_meta.run.analyser.sparql_analyser import main 

287 result = main() 

288 

289 self.assertIsNone(result) 

290 

291 @patch('sys.argv', ['sparql_analyser.py']) 

292 def test_main_missing_argument(self): 

293 """Test main function with missing endpoint argument.""" 

294 from oc_meta.run.analyser.sparql_analyser import main 

295 

296 with self.assertRaises(SystemExit): 

297 main() 

298 

299 @patch('sys.argv', ['sparql_analyser.py', 'http://test.example.com/sparql', '--br']) 

300 @patch('oc_meta.run.analyser.sparql_analyser.OCMetaSPARQLAnalyser') 

301 @patch('builtins.print') 

302 def test_main_br_only(self, mock_print, mock_analyser_class): 

303 """Test main function with --br option only.""" 

304 mock_analyser = MagicMock() 

305 mock_results = {'fabio_expressions': 1500} 

306 mock_analyser.run_selected_analyses.return_value = mock_results 

307 mock_analyser_class.return_value = mock_analyser 

308 

309 from oc_meta.run.analyser.sparql_analyser import main 

310 result = main() 

311 

312 self.assertEqual(result, mock_results) 

313 mock_analyser.run_selected_analyses.assert_called_once_with(True, False, False) 

314 

315 @patch('sys.argv', ['sparql_analyser.py', 'http://test.example.com/sparql', '--venues']) 

316 @patch('oc_meta.run.analyser.sparql_analyser.OCMetaSPARQLAnalyser') 

317 @patch('builtins.print') 

318 def test_main_venues_only(self, mock_print, mock_analyser_class): 

319 """Test main function with --venues option only.""" 

320 mock_analyser = MagicMock() 

321 mock_results = {'venues_disambiguated': 350, 'venues_simple': 400} 

322 mock_analyser.run_selected_analyses.return_value = mock_results 

323 mock_analyser_class.return_value = mock_analyser 

324 

325 from oc_meta.run.analyser.sparql_analyser import main 

326 result = main() 

327 

328 self.assertEqual(result, mock_results) 

329 mock_analyser.run_selected_analyses.assert_called_once_with(False, False, True) 

330 

331 

332if __name__ == '__main__': 

333 """ 

334 Run the test suite. 

335  

336 This will execute all test cases and provide a summary of results. 

337 To run specific tests, use: 

338 python sparql_analyser_test.py TestOCMetaSPARQLAnalyser.test_count_expressions 

339 """ 

340 print("Starting OCMetaSPARQLAnalyser test suite...") 

341 print("="*60) 

342 

343 # Create test suite 

344 suite = unittest.TestLoader().loadTestsFromModule(sys.modules[__name__]) 

345 

346 # Run tests with detailed output 

347 runner = unittest.TextTestRunner(verbosity=2) 

348 result = runner.run(suite) 

349 

350 print("\n" + "="*60) 

351 print("TEST SUMMARY") 

352 print("="*60) 

353 print(f"Tests run: {result.testsRun}") 

354 print(f"Failures: {len(result.failures)}") 

355 print(f"Errors: {len(result.errors)}") 

356 

357 if result.failures: 

358 print("\nFailures:") 

359 for test, failure in result.failures: 

360 print(f" - {test}: {failure}") 

361 

362 if result.errors: 

363 print("\nErrors:") 

364 for test, error in result.errors: 

365 print(f" - {test}: {error}") 

366 

367 if result.wasSuccessful(): 

368 print("\nAll tests passed successfully! ✅") 

369 else: 

370 print("\nSome tests failed. ❌") 

371 

372 # Exit with appropriate code 

373 sys.exit(0 if result.wasSuccessful() else 1)