Coverage for test/sparql_analyser_test.py: 85%

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.client) 

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.client, 'query', 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 with patch.object(self.analyser.client, 'query', side_effect=Exception("Query failed")): 

58 with self.assertRaises(Exception) as context: 

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

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

61 

62 def test_count_expressions(self): 

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

64 mock_results = { 

65 "results": { 

66 "bindings": [ 

67 {"count": {"value": "1500"}} 

68 ] 

69 } 

70 } 

71 

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

73 count = self.analyser.count_expressions() 

74 self.assertEqual(count, 1500) 

75 

76 def test_count_role_entities(self): 

77 """Test counting pro role entities.""" 

78 mock_results = { 

79 "results": { 

80 "bindings": [ 

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

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

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

84 ] 

85 } 

86 } 

87 

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

89 counts = self.analyser.count_role_entities() 

90 expected = { 

91 'pro:author': 800, 

92 'pro:publisher': 200, 

93 'pro:editor': 100 

94 } 

95 self.assertEqual(counts, expected) 

96 

97 def test_count_role_entities_empty_results(self): 

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

99 mock_results = { 

100 "results": { 

101 "bindings": [] 

102 } 

103 } 

104 

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

106 counts = self.analyser.count_role_entities() 

107 expected = { 

108 'pro:author': 0, 

109 'pro:publisher': 0, 

110 'pro:editor': 0 

111 } 

112 self.assertEqual(counts, expected) 

113 

114 def test_count_venues(self): 

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

116 mock_results = { 

117 "results": { 

118 "bindings": [ 

119 {"count": {"value": "350"}} 

120 ] 

121 } 

122 } 

123 

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

125 count = self.analyser.count_venues() 

126 self.assertEqual(count, 350) 

127 

128 def test_count_venues_disambiguated(self): 

129 """Test counting disambiguated venues.""" 

130 mock_results = { 

131 "results": { 

132 "bindings": [ 

133 { 

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

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

136 "has_identifier": {"value": "true"} 

137 }, 

138 { 

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

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

141 "has_identifier": {"value": "false"} 

142 }, 

143 { 

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

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

146 "has_identifier": {"value": "true"} 

147 } 

148 ] 

149 } 

150 } 

151 

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

153 count = self.analyser.count_venues_disambiguated() 

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

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

156 # Total = 3 distinct venues 

157 self.assertEqual(count, 3) 

158 

159 @patch('builtins.print') 

160 def test_run_all_analyses_success(self, mock_print): 

161 """Test running all analyses successfully.""" 

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

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

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

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

166 

167 results = self.analyser.run_all_analyses() 

168 

169 expected = { 

170 'fabio_expressions': 1500, 

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

172 'venues_disambiguated': 350, 

173 'venues_simple': 400 

174 } 

175 self.assertEqual(results, expected) 

176 

177 @patch('builtins.print') 

178 def test_run_selected_analyses_br_only(self, mock_print): 

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

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

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

182 

183 expected = { 

184 'fabio_expressions': 1500 

185 } 

186 self.assertEqual(results, expected) 

187 

188 @patch('builtins.print') 

189 def test_run_selected_analyses_venues_only(self, mock_print): 

190 """Test running only venues analysis.""" 

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

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

193 

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

195 

196 expected = { 

197 'venues_disambiguated': 350, 

198 'venues_simple': 400 

199 } 

200 self.assertEqual(results, expected) 

201 

202 @patch('builtins.print') 

203 def test_run_all_analyses_with_errors(self, mock_print): 

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

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

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

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

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

209 

210 results = self.analyser.run_all_analyses() 

211 

212 expected = { 

213 'fabio_expressions': None, 

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

215 'venues_disambiguated': 350, 

216 'venues_simple': 400 

217 } 

218 self.assertEqual(results, expected) 

219 

220 

221class TestMainFunction(unittest.TestCase): 

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

223 

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

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

226 @patch('builtins.print') 

227 def test_main_success(self, mock_print, mock_analyser_class): 

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

229 mock_analyser = MagicMock() 

230 mock_results = { 

231 'fabio_expressions': 1500, 

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

233 'venues_disambiguated': 350, 

234 'venues_simple': 400 

235 } 

236 mock_analyser.run_selected_analyses.return_value = mock_results 

237 mock_analyser_class.return_value = mock_analyser 

238 mock_analyser_class.return_value.__enter__.return_value = mock_analyser 

239 

240 from oc_meta.run.analyser.sparql_analyser import main 

241 result = main() 

242 

243 self.assertEqual(result, mock_results) 

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

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

246 

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

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

249 @patch('builtins.print') 

250 def test_main_failure(self, mock_print, mock_analyser_class): 

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

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

253 

254 from oc_meta.run.analyser.sparql_analyser import main 

255 result = main() 

256 

257 self.assertIsNone(result) 

258 

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

260 def test_main_missing_argument(self): 

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

262 from oc_meta.run.analyser.sparql_analyser import main 

263 

264 with self.assertRaises(SystemExit): 

265 main() 

266 

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

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

269 @patch('builtins.print') 

270 def test_main_br_only(self, mock_print, mock_analyser_class): 

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

272 mock_analyser = MagicMock() 

273 mock_results = {'fabio_expressions': 1500} 

274 mock_analyser.run_selected_analyses.return_value = mock_results 

275 mock_analyser_class.return_value = mock_analyser 

276 mock_analyser_class.return_value.__enter__.return_value = mock_analyser 

277 

278 from oc_meta.run.analyser.sparql_analyser import main 

279 result = main() 

280 

281 self.assertEqual(result, mock_results) 

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

283 

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

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

286 @patch('builtins.print') 

287 def test_main_venues_only(self, mock_print, mock_analyser_class): 

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

289 mock_analyser = MagicMock() 

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

291 mock_analyser.run_selected_analyses.return_value = mock_results 

292 mock_analyser_class.return_value = mock_analyser 

293 mock_analyser_class.return_value.__enter__.return_value = mock_analyser 

294 

295 from oc_meta.run.analyser.sparql_analyser import main 

296 result = main() 

297 

298 self.assertEqual(result, mock_results) 

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

300 

301 

302if __name__ == '__main__': 

303 """ 

304 Run the test suite. 

305  

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

307 To run specific tests, use: 

308 python sparql_analyser_test.py TestOCMetaSPARQLAnalyser.test_count_expressions 

309 """ 

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

311 print("="*60) 

312 

313 # Create test suite 

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

315 

316 # Run tests with detailed output 

317 runner = unittest.TextTestRunner(verbosity=2) 

318 result = runner.run(suite) 

319 

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

321 print("TEST SUMMARY") 

322 print("="*60) 

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

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

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

326 

327 if result.failures: 

328 print("\nFailures:") 

329 for test, failure in result.failures: 

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

331 

332 if result.errors: 

333 print("\nErrors:") 

334 for test, error in result.errors: 

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

336 

337 if result.wasSuccessful(): 

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

339 else: 

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

341 

342 # Exit with appropriate code 

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