Coverage for test/csv_generator_lite_test.py: 99%

1#!/usr/bin/python 

2# -*- coding: utf-8 -*- 

3# Copyright (c) 2024 Arcangelo Massari <arcangelo.massari@unibo.it> 

4# 

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

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

7# and this permission notice appear in all copies. 

8# 

9# THE SOFTWARE IS PROVIDED 'AS IS' AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH 

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

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

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

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

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

15# SOFTWARE. 

16 

17import csv 

18import json 

19import os 

20import unittest 

21from shutil import rmtree 

22from zipfile import ZipFile 

23 

24import redis 

25from oc_meta.lib.file_manager import get_csv_data 

26from oc_meta.plugins.csv_generator_lite.csv_generator_lite import ( 

27 generate_csv, 

28 init_redis_connection, 

29 is_omid_processed, 

30 load_processed_omids_to_redis, 

31) 

32 

33 

34class TestCSVGeneratorLite(unittest.TestCase): 

35 def setUp(self): 

36 self.base_dir = os.path.join("test", "csv_generator_lite") 

37 self.input_dir = os.path.join(self.base_dir, "input") 

38 self.output_dir = os.path.join(self.base_dir, "output") 

39 

40 os.makedirs(self.input_dir, exist_ok=True) 

41 os.makedirs(self.output_dir, exist_ok=True) 

42 

43 self.rdf_dir = os.path.join(self.input_dir, "rdf") 

44 self.br_dir = os.path.join(self.rdf_dir, "br") 

45 os.makedirs(self.br_dir, exist_ok=True) 

46 

47 self.redis_client = init_redis_connection(port=6381, db=5) 

48 self.redis_client.flushdb() # Clear test database 

49 

50 def tearDown(self): 

51 if os.path.exists(self.base_dir): 

52 rmtree(self.base_dir) 

53 self.redis_client.flushdb() 

54 

55 def _write_test_data(self, data): 

56 os.makedirs(os.path.join(self.br_dir, "060", "10000"), exist_ok=True) 

57 test_data = [ 

58 { 

59 "@graph": [ 

60 { 

61 "@id": f"https://w3id.org/oc/meta/{item['id'].replace('omid:', '')}", 

62 "@type": [ 

63 "http://purl.org/spar/fabio/Expression", 

64 "http://purl.org/spar/fabio/JournalArticle", 

65 ], 

66 "http://purl.org/dc/terms/title": [{"@value": item["title"]}], 

67 } 

68 for item in data 

69 ] 

70 } 

71 ] 

72 with ZipFile( 

73 os.path.join(self.br_dir, "060", "10000", "1000.zip"), "w" 

74 ) as zip_file: 

75 zip_file.writestr("1000.json", json.dumps(test_data)) 

76 

77 def test_redis_connection_and_caching(self): 

78 redis_client = init_redis_connection(port=6381, db=5) 

79 self.assertIsInstance(redis_client, redis.Redis) 

80 

81 # Create a test CSV file with some OMIDs 

82 test_data = [ 

83 {"id": "omid:br/0601", "title": "Test 1"}, 

84 {"id": "omid:br/0602", "title": "Test 2"}, 

85 {"id": "omid:br/0603 issn:456", "title": "Test 3"}, 

86 ] 

87 os.makedirs(self.output_dir, exist_ok=True) 

88 with open( 

89 os.path.join(self.output_dir, "test.csv"), "w", newline="", encoding="utf-8" 

90 ) as f: 

91 writer = csv.DictWriter(f, fieldnames=["id", "title"]) 

92 writer.writeheader() 

93 writer.writerows(test_data) 

94 

95 count = load_processed_omids_to_redis(self.output_dir, redis_client) 

96 self.assertEqual(count, 3) 

97 

98 self.assertTrue(is_omid_processed("omid:br/0601", redis_client)) 

99 self.assertTrue(is_omid_processed("omid:br/0602", redis_client)) 

100 self.assertTrue(is_omid_processed("omid:br/0603", redis_client)) 

101 self.assertFalse(is_omid_processed("omid:br/0604", redis_client)) 

102 

103 def test_redis_cache_persistence(self): 

104 # Create initial test data 

105 test_data = [ 

106 { 

107 "@graph": [ 

108 { 

109 "@id": "https://w3id.org/oc/meta/br/0601", 

110 "@type": [ 

111 "http://purl.org/spar/fabio/Expression", 

112 "http://purl.org/spar/fabio/JournalArticle", 

113 ], 

114 "http://purl.org/dc/terms/title": [{"@value": "First Run"}], 

115 } 

116 ] 

117 } 

118 ] 

119 

120 os.makedirs(os.path.join(self.br_dir, "060", "10000"), exist_ok=True) 

121 with ZipFile( 

122 os.path.join(self.br_dir, "060", "10000", "1000.zip"), "w" 

123 ) as zip_file: 

124 zip_file.writestr("1000.json", json.dumps(test_data)) 

125 

126 generate_csv( 

127 input_dir=self.rdf_dir, 

128 output_dir=self.output_dir, 

129 dir_split_number=10000, 

130 items_per_file=1000, 

131 redis_port=6381, 

132 redis_db=5, 

133 ) 

134 

135 self.assertFalse(is_omid_processed("omid:br/0601", self.redis_client)) 

136 

137 # Create new test data 

138 test_data_2 = [ 

139 { 

140 "@graph": [ 

141 { 

142 "@id": "https://w3id.org/oc/meta/br/0601", # Same OMID as before 

143 "@type": [ 

144 "http://purl.org/spar/fabio/Expression", 

145 "http://purl.org/spar/fabio/JournalArticle", 

146 ], 

147 "http://purl.org/dc/terms/title": [ 

148 {"@value": "Should Be Skipped"} 

149 ], 

150 }, 

151 { 

152 "@id": "https://w3id.org/oc/meta/br/0602", # New OMID 

153 "@type": [ 

154 "http://purl.org/spar/fabio/Expression", 

155 "http://purl.org/spar/fabio/JournalArticle", 

156 ], 

157 "http://purl.org/dc/terms/title": [ 

158 {"@value": "Should Be Processed"} 

159 ], 

160 }, 

161 ] 

162 } 

163 ] 

164 

165 with ZipFile( 

166 os.path.join(self.br_dir, "060", "10000", "1000.zip"), "w" 

167 ) as zip_file: 

168 zip_file.writestr("1000.json", json.dumps(test_data_2)) 

169 

170 generate_csv( 

171 input_dir=self.rdf_dir, 

172 output_dir=self.output_dir, 

173 dir_split_number=10000, 

174 items_per_file=1000, 

175 redis_port=6381, 

176 redis_db=5, 

177 ) 

178 

179 output_data = [] 

180 for filename in os.listdir(self.output_dir): 

181 if filename.endswith(".csv"): 

182 output_data.extend( 

183 get_csv_data(os.path.join(self.output_dir, filename)) 

184 ) 

185 

186 self.assertEqual(len(output_data), 2) 

187 

188 first_run_entry = next( 

189 item for item in output_data if item["title"] == "First Run" 

190 ) 

191 second_run_entry = next( 

192 item for item in output_data if item["title"] == "Should Be Processed" 

193 ) 

194 

195 self.assertEqual(first_run_entry["title"], "First Run") 

196 self.assertEqual(first_run_entry["id"], "omid:br/0601") 

197 

198 self.assertEqual(second_run_entry["title"], "Should Be Processed") 

199 self.assertEqual(second_run_entry["id"], "omid:br/0602") 

200 

201 self.assertFalse(is_omid_processed("omid:br/0601", self.redis_client)) 

202 self.assertFalse(is_omid_processed("omid:br/0602", self.redis_client)) 

203 

204 def test_redis_cache_cleanup(self): 

205 input_data = [{"id": "omid:br/0601", "title": "First Entry"}] 

206 self._write_test_data(input_data) 

207 

208 generate_csv( 

209 input_dir=self.rdf_dir, 

210 output_dir=self.output_dir, 

211 dir_split_number=10000, 

212 items_per_file=1000, 

213 redis_port=6381, 

214 redis_db=5, 

215 ) 

216 

217 self.assertFalse(is_omid_processed("omid:br/0601", self.redis_client)) 

218 

219 load_processed_omids_to_redis(self.output_dir, self.redis_client) 

220 

221 self.assertTrue(is_omid_processed("omid:br/0601", self.redis_client)) 

222 

223 generate_csv( 

224 input_dir="/nonexistent/dir", 

225 output_dir=self.output_dir, 

226 dir_split_number=10000, 

227 items_per_file=1000, 

228 redis_port=6381, 

229 redis_db=5, 

230 ) 

231 

232 self.assertTrue( 

233 is_omid_processed("omid:br/0601", self.redis_client), 

234 "Redis cache should be retained after a failed run", 

235 ) 

236 

237 def test_redis_error_handling(self): 

238 with self.assertRaises(redis.ConnectionError): 

239 init_redis_connection(port=9999) # Invalid port 

240 

241 count = load_processed_omids_to_redis("/nonexistent/dir", self.redis_client) 

242 self.assertEqual(count, 0) 

243 

244 def test_concurrent_processing_with_redis(self): 

245 # Create multiple test files 

246 test_data = [] 

247 for i in range(100): # Create 100 test entries 

248 test_data.append( 

249 { 

250 "@id": f"https://w3id.org/oc/meta/br/06{i:02d}", 

251 "@type": [ 

252 "http://purl.org/spar/fabio/Expression", 

253 "http://purl.org/spar/fabio/JournalArticle", 

254 ], 

255 "http://purl.org/dc/terms/title": [{"@value": f"Article {i}"}], 

256 } 

257 ) 

258 

259 os.makedirs(os.path.join(self.br_dir, "060", "10000"), exist_ok=True) 

260 for i in range(0, 100, 10): # Create 10 files with 10 entries each 

261 file_data = [{"@graph": test_data[i : i + 10]}] 

262 with ZipFile( 

263 os.path.join(self.br_dir, "060", "10000", f"{i+1000}.zip"), "w" 

264 ) as zip_file: 

265 zip_file.writestr(f"{i+1000}.json", json.dumps(file_data)) 

266 

267 generate_csv( 

268 input_dir=self.rdf_dir, 

269 output_dir=self.output_dir, 

270 dir_split_number=10000, 

271 items_per_file=1000, 

272 redis_port=6381, 

273 redis_db=5, 

274 ) 

275 

276 # Create more test entries 

277 more_test_data = [] 

278 for i in range(100, 200): # Create 100 more test entries 

279 more_test_data.append( 

280 { 

281 "@id": f"https://w3id.org/oc/meta/br/06{i:02d}", 

282 "@type": [ 

283 "http://purl.org/spar/fabio/Expression", 

284 "http://purl.org/spar/fabio/JournalArticle", 

285 ], 

286 "http://purl.org/dc/terms/title": [{"@value": f"Article {i}"}], 

287 } 

288 ) 

289 

290 for i in range(0, 100, 10): 

291 file_data = [{"@graph": more_test_data[i : i + 10]}] 

292 with ZipFile( 

293 os.path.join(self.br_dir, "060", "10000", f"{i+2000}.zip"), "w" 

294 ) as zip_file: 

295 zip_file.writestr(f"{i+2000}.json", json.dumps(file_data)) 

296 

297 generate_csv( 

298 input_dir=self.rdf_dir, 

299 output_dir=self.output_dir, 

300 dir_split_number=10000, 

301 items_per_file=1000, 

302 redis_port=6381, 

303 redis_db=5, 

304 ) 

305 

306 all_output_data = [] 

307 for filename in os.listdir(self.output_dir): 

308 if filename.endswith(".csv"): 

309 all_output_data.extend( 

310 get_csv_data(os.path.join(self.output_dir, filename)) 

311 ) 

312 

313 self.assertEqual(len(all_output_data), 200) 

314 

315 processed_ids = {row["id"] for row in all_output_data} 

316 self.assertEqual(len(processed_ids), 200) 

317 

318 def test_basic_br_processing(self): 

319 test_data = [ 

320 { 

321 "@graph": [ 

322 { 

323 "@id": "https://w3id.org/oc/meta/br/0601", 

324 "@type": [ 

325 "http://purl.org/spar/fabio/Expression", 

326 "http://purl.org/spar/fabio/JournalArticle", 

327 ], 

328 "http://purl.org/dc/terms/title": [{"@value": "Test Article"}], 

329 "http://prismstandard.org/namespaces/basic/2.0/publicationDate": [ 

330 {"@value": "2024-01-01"} 

331 ], 

332 "http://purl.org/spar/datacite/hasIdentifier": [ 

333 {"@id": "https://w3id.org/oc/meta/id/0601"} 

334 ], 

335 } 

336 ], 

337 "@id": "https://w3id.org/oc/meta/br/", 

338 } 

339 ] 

340 

341 os.makedirs(os.path.join(self.br_dir, "060", "10000"), exist_ok=True) 

342 with ZipFile( 

343 os.path.join(self.br_dir, "060", "10000", "1000.zip"), "w" 

344 ) as zip_file: 

345 zip_file.writestr("1000.json", json.dumps(test_data)) 

346 

347 generate_csv( 

348 input_dir=self.rdf_dir, 

349 output_dir=self.output_dir, 

350 dir_split_number=10000, 

351 items_per_file=1000, 

352 redis_port=6381, 

353 redis_db=5, 

354 ) 

355 

356 output_files = os.listdir(self.output_dir) 

357 self.assertEqual(len(output_files), 1) 

358 

359 output_data = get_csv_data(os.path.join(self.output_dir, output_files[0])) 

360 self.assertEqual(len(output_data), 1) 

361 self.assertEqual(output_data[0]["title"], "Test Article") 

362 self.assertEqual(output_data[0]["pub_date"], "2024-01-01") 

363 self.assertEqual(output_data[0]["type"], "journal article") 

364 self.assertEqual(output_data[0]["id"], "omid:br/0601") 

365 

366 def test_complex_br_with_related_entities(self): 

367 # Create directory structure for each entity type 

368 supplier_prefix = "060" 

369 for entity_type in ["br", "ra", "ar", "id"]: 

370 os.makedirs( 

371 os.path.join(self.rdf_dir, entity_type, supplier_prefix, "10000"), 

372 exist_ok=True, 

373 ) 

374 

375 # BR data including both the article and the venue 

376 br_data = [ 

377 { 

378 "@graph": [ 

379 { 

380 "@id": f"https://w3id.org/oc/meta/br/{supplier_prefix}2", 

381 "@type": [ 

382 "http://purl.org/spar/fabio/Expression", 

383 "http://purl.org/spar/fabio/JournalArticle", 

384 ], 

385 "http://purl.org/dc/terms/title": [ 

386 {"@value": "Complex Article"} 

387 ], 

388 "http://prismstandard.org/namespaces/basic/2.0/publicationDate": [ 

389 {"@value": "2024-02-01"} 

390 ], 

391 "http://purl.org/spar/pro/isDocumentContextFor": [ 

392 {"@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}1"} 

393 ], 

394 "http://purl.org/vocab/frbr/core#partOf": [ 

395 {"@id": f"https://w3id.org/oc/meta/br/{supplier_prefix}3"} 

396 ], 

397 }, 

398 { 

399 "@id": f"https://w3id.org/oc/meta/br/{supplier_prefix}3", 

400 "@type": [ 

401 "http://purl.org/spar/fabio/Expression", 

402 "http://purl.org/spar/fabio/Journal", 

403 ], 

404 "http://purl.org/dc/terms/title": [{"@value": "Test Journal"}], 

405 }, 

406 ], 

407 "@id": "https://w3id.org/oc/meta/br/", 

408 } 

409 ] 

410 

411 ar_data = [ 

412 { 

413 "@graph": [ 

414 { 

415 "@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}1", 

416 "http://purl.org/spar/pro/withRole": [ 

417 {"@id": "http://purl.org/spar/pro/author"} 

418 ], 

419 "http://purl.org/spar/pro/isHeldBy": [ 

420 {"@id": f"https://w3id.org/oc/meta/ra/{supplier_prefix}1"} 

421 ], 

422 } 

423 ], 

424 "@id": "https://w3id.org/oc/meta/ar/", 

425 } 

426 ] 

427 

428 ra_data = [ 

429 { 

430 "@graph": [ 

431 { 

432 "@id": f"https://w3id.org/oc/meta/ra/{supplier_prefix}1", 

433 "http://xmlns.com/foaf/0.1/name": [{"@value": "Test Author"}], 

434 } 

435 ], 

436 "@id": "https://w3id.org/oc/meta/ra/", 

437 } 

438 ] 

439 

440 data_files = {"br": br_data, "ra": ra_data, "ar": ar_data} 

441 

442 for entity_type, data in data_files.items(): 

443 zip_path = os.path.join( 

444 self.rdf_dir, entity_type, supplier_prefix, "10000", "1000.zip" 

445 ) 

446 with ZipFile(zip_path, "w") as zip_file: 

447 zip_file.writestr("1000.json", json.dumps(data)) 

448 

449 generate_csv( 

450 input_dir=self.rdf_dir, 

451 output_dir=self.output_dir, 

452 dir_split_number=10000, 

453 items_per_file=1000, 

454 redis_port=6381, 

455 redis_db=5, 

456 ) 

457 

458 output_data = get_csv_data(os.path.join(self.output_dir, "output_0.csv")) 

459 self.assertEqual(len(output_data), 2) # Should have 2 rows: article and journal 

460 

461 article = next( 

462 (item for item in output_data if item["type"] == "journal article"), None 

463 ) 

464 journal = next( 

465 (item for item in output_data if item["type"] == "journal"), None 

466 ) 

467 

468 self.assertIsNotNone(article) 

469 self.assertEqual(article["title"], "Complex Article") 

470 self.assertEqual(article["venue"], f"Test Journal [omid:br/{supplier_prefix}3]") 

471 self.assertEqual(article["author"], "Test Author [omid:ra/0601]") 

472 self.assertEqual(article["id"], f"omid:br/{supplier_prefix}2") 

473 

474 self.assertIsNotNone(journal) 

475 self.assertEqual(journal["title"], "Test Journal") 

476 self.assertEqual(journal["type"], "journal") 

477 self.assertEqual(journal["id"], f"omid:br/{supplier_prefix}3") 

478 

479 def test_empty_input_directory(self): 

480 generate_csv( 

481 input_dir=self.rdf_dir, 

482 output_dir=self.output_dir, 

483 dir_split_number=10000, 

484 items_per_file=1000, 

485 redis_port=6381, 

486 redis_db=5, 

487 ) 

488 

489 self.assertEqual(len(os.listdir(self.output_dir)), 0) 

490 

491 def test_br_with_multiple_authors_and_editors(self): 

492 supplier_prefix = "060" 

493 br_data = [ 

494 { 

495 "@graph": [ 

496 { 

497 "@id": f"https://w3id.org/oc/meta/br/{supplier_prefix}1", 

498 "@type": [ 

499 "http://purl.org/spar/fabio/Expression", 

500 "http://purl.org/spar/fabio/Book", 

501 ], 

502 "http://purl.org/dc/terms/title": [ 

503 {"@value": "Multi-Author Book"} 

504 ], 

505 "http://purl.org/spar/pro/isDocumentContextFor": [ 

506 { 

507 "@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}1" 

508 }, # First author 

509 { 

510 "@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}2" 

511 }, # Second author 

512 { 

513 "@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}3" 

514 }, # First editor 

515 { 

516 "@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}4" 

517 }, # Second editor 

518 ], 

519 } 

520 ] 

521 } 

522 ] 

523 

524 # Setup agent roles for authors and editors with hasNext relations 

525 ar_data = [ 

526 { 

527 "@graph": [ 

528 { 

529 "@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}1", 

530 "http://purl.org/spar/pro/withRole": [ 

531 {"@id": "http://purl.org/spar/pro/author"} 

532 ], 

533 "http://purl.org/spar/pro/isHeldBy": [ 

534 {"@id": f"https://w3id.org/oc/meta/ra/{supplier_prefix}1"} 

535 ], 

536 "https://w3id.org/oc/ontology/hasNext": [ 

537 {"@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}2"} 

538 ], 

539 }, 

540 { 

541 "@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}2", 

542 "http://purl.org/spar/pro/withRole": [ 

543 {"@id": "http://purl.org/spar/pro/author"} 

544 ], 

545 "http://purl.org/spar/pro/isHeldBy": [ 

546 {"@id": f"https://w3id.org/oc/meta/ra/{supplier_prefix}2"} 

547 ], 

548 "https://w3id.org/oc/ontology/hasNext": [ 

549 {"@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}3"} 

550 ], 

551 }, 

552 { 

553 "@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}3", 

554 "http://purl.org/spar/pro/withRole": [ 

555 {"@id": "http://purl.org/spar/pro/editor"} 

556 ], 

557 "http://purl.org/spar/pro/isHeldBy": [ 

558 {"@id": f"https://w3id.org/oc/meta/ra/{supplier_prefix}3"} 

559 ], 

560 "https://w3id.org/oc/ontology/hasNext": [ 

561 {"@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}4"} 

562 ], 

563 }, 

564 { 

565 "@id": f"https://w3id.org/oc/meta/ar/{supplier_prefix}4", 

566 "http://purl.org/spar/pro/withRole": [ 

567 {"@id": "http://purl.org/spar/pro/editor"} 

568 ], 

569 "http://purl.org/spar/pro/isHeldBy": [ 

570 {"@id": f"https://w3id.org/oc/meta/ra/{supplier_prefix}4"} 

571 ], 

572 }, 

573 ] 

574 } 

575 ] 

576 

577 # Setup responsible agents 

578 ra_data = [ 

579 { 

580 "@graph": [ 

581 { 

582 "@id": f"https://w3id.org/oc/meta/ra/{supplier_prefix}1", 

583 "http://xmlns.com/foaf/0.1/familyName": [{"@value": "Smith"}], 

584 "http://xmlns.com/foaf/0.1/givenName": [{"@value": "John"}], 

585 }, 

586 { 

587 "@id": f"https://w3id.org/oc/meta/ra/{supplier_prefix}2", 

588 "http://xmlns.com/foaf/0.1/familyName": [{"@value": "Doe"}], 

589 "http://xmlns.com/foaf/0.1/givenName": [{"@value": "Jane"}], 

590 }, 

591 { 

592 "@id": f"https://w3id.org/oc/meta/ra/{supplier_prefix}3", 

593 "http://xmlns.com/foaf/0.1/familyName": [{"@value": "Brown"}], 

594 "http://xmlns.com/foaf/0.1/givenName": [{"@value": "Bob"}], 

595 }, 

596 { 

597 "@id": f"https://w3id.org/oc/meta/ra/{supplier_prefix}4", 

598 "http://xmlns.com/foaf/0.1/familyName": [{"@value": "Wilson"}], 

599 "http://xmlns.com/foaf/0.1/givenName": [{"@value": "Alice"}], 

600 }, 

601 ] 

602 } 

603 ] 

604 

605 data_files = {"br": br_data, "ra": ra_data, "ar": ar_data} 

606 

607 for entity_type, data in data_files.items(): 

608 dir_path = os.path.join(self.rdf_dir, entity_type, supplier_prefix, "10000") 

609 os.makedirs(dir_path, exist_ok=True) 

610 

611 zip_path = os.path.join(dir_path, "1000.zip") 

612 with ZipFile(zip_path, "w") as zip_file: 

613 zip_file.writestr("1000.json", json.dumps(data)) 

614 

615 generate_csv( 

616 input_dir=self.rdf_dir, 

617 output_dir=self.output_dir, 

618 dir_split_number=10000, 

619 items_per_file=1000, 

620 redis_port=6381, 

621 redis_db=5, 

622 ) 

623 

624 output_data = get_csv_data(os.path.join(self.output_dir, "output_0.csv")) 

625 self.assertEqual(len(output_data), 1) 

626 

627 expected_authors = ( 

628 f"Smith, John [omid:ra/{supplier_prefix}1]; " 

629 f"Doe, Jane [omid:ra/{supplier_prefix}2]" 

630 ) 

631 expected_editors = ( 

632 f"Brown, Bob [omid:ra/{supplier_prefix}3]; " 

633 f"Wilson, Alice [omid:ra/{supplier_prefix}4]" 

634 ) 

635 

636 self.assertEqual(output_data[0]["author"], expected_authors) 

637 self.assertEqual(output_data[0]["editor"], expected_editors) 

638 

639 def test_br_with_identifiers(self): 

640 supplier_prefix = "060" 

641 br_data = [ 

642 { 

643 "@graph": [ 

644 { 

645 "@id": f"https://w3id.org/oc/meta/br/{supplier_prefix}1", 

646 "@type": [ 

647 "http://purl.org/spar/fabio/Expression", 

648 "http://purl.org/spar/fabio/JournalArticle", 

649 ], 

650 "http://purl.org/dc/terms/title": [ 

651 {"@value": "Article With DOI"} 

652 ], 

653 "http://purl.org/spar/datacite/hasIdentifier": [ 

654 {"@id": f"https://w3id.org/oc/meta/id/{supplier_prefix}1"}, 

655 {"@id": f"https://w3id.org/oc/meta/id/{supplier_prefix}2"}, 

656 ], 

657 } 

658 ] 

659 } 

660 ] 

661 

662 id_data = [ 

663 { 

664 "@graph": [ 

665 { 

666 "@id": f"https://w3id.org/oc/meta/id/{supplier_prefix}1", 

667 "http://purl.org/spar/datacite/usesIdentifierScheme": [ 

668 {"@id": "http://purl.org/spar/datacite/doi"} 

669 ], 

670 "http://www.essepuntato.it/2010/06/literalreification/hasLiteralValue": [ 

671 {"@value": "10.1234/test.123"} 

672 ], 

673 }, 

674 { 

675 "@id": f"https://w3id.org/oc/meta/id/{supplier_prefix}2", 

676 "http://purl.org/spar/datacite/usesIdentifierScheme": [ 

677 {"@id": "http://purl.org/spar/datacite/isbn"} 

678 ], 

679 "http://www.essepuntato.it/2010/06/literalreification/hasLiteralValue": [ 

680 {"@value": "978-0-123456-47-2"} 

681 ], 

682 }, 

683 ] 

684 } 

685 ] 

686 

687 data_files = {"br": br_data, "id": id_data} 

688 

689 for entity_type, data in data_files.items(): 

690 # Create all necessary directories 

691 dir_path = os.path.join(self.rdf_dir, entity_type, supplier_prefix, "10000") 

692 os.makedirs(dir_path, exist_ok=True) 

693 

694 zip_path = os.path.join(dir_path, "1000.zip") 

695 with ZipFile(zip_path, "w") as zip_file: 

696 zip_file.writestr("1000.json", json.dumps(data)) 

697 

698 generate_csv( 

699 input_dir=self.rdf_dir, 

700 output_dir=self.output_dir, 

701 dir_split_number=10000, 

702 items_per_file=1000, 

703 redis_port=6381, 

704 redis_db=5, 

705 ) 

706 

707 output_data = get_csv_data(os.path.join(self.output_dir, "output_0.csv")) 

708 self.assertEqual(len(output_data), 1) 

709 

710 expected_ids = ( 

711 f"omid:br/{supplier_prefix}1 doi:10.1234/test.123 isbn:978-0-123456-47-2" 

712 ) 

713 self.assertEqual(output_data[0]["id"], expected_ids) 

714 

715 def test_br_with_page_numbers(self): 

716 supplier_prefix = "060" 

717 br_data = [ 

718 { 

719 "@graph": [ 

720 { 

721 "@id": f"https://w3id.org/oc/meta/br/{supplier_prefix}1", 

722 "@type": [ 

723 "http://purl.org/spar/fabio/Expression", 

724 "http://purl.org/spar/fabio/JournalArticle", 

725 ], 

726 "http://purl.org/dc/terms/title": [{"@value": "Paged Article"}], 

727 "http://purl.org/vocab/frbr/core#embodiment": [ 

728 {"@id": f"https://w3id.org/oc/meta/re/{supplier_prefix}1"} 

729 ], 

730 } 

731 ] 

732 } 

733 ] 

734 

735 re_data = [ 

736 { 

737 "@graph": [ 

738 { 

739 "@id": f"https://w3id.org/oc/meta/re/{supplier_prefix}1", 

740 "http://prismstandard.org/namespaces/basic/2.0/startingPage": [ 

741 {"@value": "100"} 

742 ], 

743 "http://prismstandard.org/namespaces/basic/2.0/endingPage": [ 

744 {"@value": "120"} 

745 ], 

746 } 

747 ] 

748 } 

749 ] 

750 

751 data_files = {"br": br_data, "re": re_data} 

752 

753 for entity_type, data in data_files.items(): 

754 # Create all necessary directories 

755 dir_path = os.path.join(self.rdf_dir, entity_type, supplier_prefix, "10000") 

756 os.makedirs(dir_path, exist_ok=True) 

757 

758 zip_path = os.path.join(dir_path, "1000.zip") 

759 with ZipFile(zip_path, "w") as zip_file: 

760 zip_file.writestr("1000.json", json.dumps(data)) 

761 

762 generate_csv( 

763 input_dir=self.rdf_dir, 

764 output_dir=self.output_dir, 

765 dir_split_number=10000, 

766 items_per_file=1000, 

767 redis_port=6381, 

768 redis_db=5, 

769 ) 

770 

771 output_data = get_csv_data(os.path.join(self.output_dir, "output_0.csv")) 

772 self.assertEqual(len(output_data), 1) 

773 self.assertEqual(output_data[0]["page"], "100-120") 

774 

775 def test_malformed_data_handling(self): 

776 supplier_prefix = "060" 

777 br_data = [ 

778 { 

779 "@graph": [ 

780 { 

781 "@id": f"https://w3id.org/oc/meta/br/{supplier_prefix}1", 

782 "@type": [ 

783 "http://purl.org/spar/fabio/Expression", 

784 "http://purl.org/spar/fabio/JournalArticle", 

785 ], 

786 # Missing title 

787 "http://purl.org/spar/pro/isDocumentContextFor": [ 

788 {"@id": "invalid_uri"}, # Invalid URI 

789 ], 

790 "http://purl.org/vocab/frbr/core#partOf": [ 

791 {"@id": "non_existent_venue"} # Non-existent venue 

792 ], 

793 } 

794 ] 

795 } 

796 ] 

797 

798 data_files = {"br": br_data} 

799 

800 for entity_type, data in data_files.items(): 

801 # Create all necessary directories 

802 dir_path = os.path.join(self.rdf_dir, entity_type, supplier_prefix, "10000") 

803 os.makedirs(dir_path, exist_ok=True) 

804 

805 zip_path = os.path.join(dir_path, "1000.zip") 

806 with ZipFile(zip_path, "w") as zip_file: 

807 zip_file.writestr("1000.json", json.dumps(data)) 

808 

809 generate_csv( 

810 input_dir=self.rdf_dir, 

811 output_dir=self.output_dir, 

812 dir_split_number=10000, 

813 items_per_file=1000, 

814 redis_port=6381, 

815 redis_db=5, 

816 ) 

817 

818 output_data = get_csv_data(os.path.join(self.output_dir, "output_0.csv")) 

819 self.assertEqual(len(output_data), 1) 

820 self.assertEqual(output_data[0]["title"], "") 

821 self.assertEqual(output_data[0]["author"], "") 

822 self.assertEqual(output_data[0]["venue"], "") 

823 

824 def test_br_with_hierarchical_venue_structures(self): 

825 supplier_prefix = "060" 

826 

827 br_data = [ 

828 { 

829 "@graph": [ 

830 # Article in issue->volume->journal structure 

831 { 

832 "@id": f"https://w3id.org/oc/meta/br/{supplier_prefix}1", 

833 "@type": [ 

834 "http://purl.org/spar/fabio/Expression", 

835 "http://purl.org/spar/fabio/JournalArticle", 

836 ], 

837 "http://purl.org/dc/terms/title": [ 

838 {"@value": "Article in Full Hierarchy"} 

839 ], 

840 "http://purl.org/vocab/frbr/core#partOf": [ 

841 { 

842 "@id": f"https://w3id.org/oc/meta/br/{supplier_prefix}2" 

843 } # Issue 

844 ], 

845 }, 

846 # Article in issue->journal structure (no volume) 

847 { 

848 "@id": f"https://w3id.org/oc/meta/br/{supplier_prefix}5", 

849 "@type": [ 

850 "http://purl.org/spar/fabio/Expression", 

851 "http://purl.org/spar/fabio/JournalArticle", 

852 ], 

853 "http://purl.org/dc/terms/title": [