Coverage for test/add_sources_from_rdf_test.py: 95%

1import unittest 

2import os 

3from oc_meta.run.fixer.prov.add_sources_from_rdf import extract_and_process_json 

4import shutil 

5from rdflib import ConjunctiveGraph, URIRef 

6from rdflib.compare import isomorphic, graph_diff 

7import zipfile 

8import json 

9 

10class TestExtractAndProcessJson(unittest.TestCase): 

11 def setUp(self): 

12 # Percorsi per i file zip di sorgente e di destinazione 

13 self.source_zip_folder = os.path.join('test', 'fixer_tests', 'data', 'source') + os.sep 

14 self.source_zip_path = os.path.join('test', 'fixer_tests', 'data', 'source', 'br', '0610', '10000', '1000', 'prov', 'se.zip') 

15 self.destination_zip_path = os.path.join('test', 'fixer_tests', 'data', 'destination', 'br', '0610', '10000', '1000', 'prov', 'se.zip') 

16 

17 # Percorso temporaneo per evitare modifiche al file originale 

18 self.temp_destination_zip_path = os.path.join('test', 'fixer_tests', 'data', 'tmp', 'br', '0610', '10000', '1000', 'prov', 'temp_destination.zip') 

19 

20 # Assicurati che i file sorgente e destinazione esistano 

21 assert os.path.exists(self.source_zip_path) 

22 assert os.path.exists(self.destination_zip_path) 

23 

24 # Copia il contenuto del file di destinazione in un file temporaneo 

25 if os.path.exists(self.temp_destination_zip_path): 

26 os.remove(self.temp_destination_zip_path) 

27 shutil.copyfile(self.destination_zip_path, self.temp_destination_zip_path) 

28 

29 def tearDown(self): 

30 # Pulizia dopo ogni test 

31 if os.path.exists(self.temp_destination_zip_path): 

32 os.remove(self.temp_destination_zip_path) 

33 

34 def test_extraction_and_modification(self): 

35 original_graph = self.load_rdf_from_zip(self.source_zip_path) 

36 extract_and_process_json(self.temp_destination_zip_path, self.source_zip_folder) 

37 modified_graph = self.load_rdf_from_zip(self.temp_destination_zip_path) 

38 original_graph.remove((URIRef('https://w3id.org/oc/meta/br/0610897/prov/se/1'), URIRef('http://www.w3.org/ns/prov#hadPrimarySource'), URIRef('https://api.crossref.org/'))) 

39 original_graph.add((URIRef('https://w3id.org/oc/meta/br/0610897/prov/se/1'), URIRef('http://www.w3.org/ns/prov#hadPrimarySource'), URIRef('https://api.crossref.org/snapshots/monthly/2022/12/all.json.tar.gz'))) 

40 self.assertTrue(self.graphs_equal(original_graph, modified_graph)) 

41 

42 def load_rdf_from_zip(self, zip_path): 

43 with zipfile.ZipFile(zip_path, 'r') as zf: 

44 with zf.open(zf.namelist()[0], 'r') as file: 

45 data = json.load(file) 

46 graph = ConjunctiveGraph() 

47 graph.parse(data=json.dumps(data), format='json-ld') 

48 return graph 

49 

50 def graphs_equal(self, graph1, graph2): 

51 return isomorphic(graph1, graph2) 

52 

53if __name__ == '__main__': 

54 unittest.main()