Coverage for test/gen_rdf_from_dump_test.py: 100%

1# import unittest 

2# import os 

3# import sys 

4# import gzip 

5# import shutil 

6# import tempfile 

7# from rdflib import ConjunctiveGraph, URIRef, Literal 

8# from rdflib.namespace import RDF 

9# import subprocess 

10 

11 

12# class TestOCMetaRDFProcessing(unittest.TestCase): 

13# @classmethod 

14# def setUpClass(cls): 

15# output_dir = os.path.join(os.getcwd(), 'test', 'gen_rdf_from_dump_out') 

16# os.makedirs(output_dir, exist_ok=True) 

17 

18# cls.temp_dir = tempfile.mkdtemp(dir=output_dir) 

19 

20# cls.script_path = 'oc_meta/run/gen_rdf_from_export.py' 

21 

22# cls.input_dir = 'test/gen_rdf_from_dump/' 

23 

24# subprocess.run([sys.executable, cls.script_path,  

25# cls.input_dir,  

26# cls.temp_dir,  

27# '--base_iri', 'https://w3id.org/oc/meta/', 

28# '--input_format', 'nquads']) 

29 

30# @classmethod 

31# def tearDownClass(cls): 

32# shutil.rmtree(cls.temp_dir) 

33 

34# def test_specific_triples(self): 

35# output_graph = ConjunctiveGraph() 

36# for root, dirs, files in os.walk(self.temp_dir): 

37# for file in files: 

38# if file.endswith('.json'): 

39# with open(os.path.join(root, file), 'r', encoding='utf8') as f: 

40# output_graph.parse(f, format='json-ld') 

41 

42# self.assertIn(( 

43# URIRef('https://w3id.org/oc/meta/id/061904730419'), 

44# URIRef('http://purl.org/spar/datacite/usesIdentifierScheme'), 

45# URIRef('http://purl.org/spar/datacite/pmid'), 

46# URIRef('https://w3id.org/oc/meta/id/') 

47# ), output_graph) 

48 

49# self.assertIn(( 

50# URIRef('https://w3id.org/oc/meta/id/061904730419'), 

51# RDF.type, 

52# URIRef('http://purl.org/spar/datacite/Identifier'), 

53# URIRef('https://w3id.org/oc/meta/id/') 

54# ), output_graph) 

55 

56# self.assertIn(( 

57# URIRef('https://w3id.org/oc/meta/br/061401925407'), 

58# URIRef('http://purl.org/spar/datacite/hasIdentifier'), 

59# URIRef('https://w3id.org/oc/meta/id/061904730419'), 

60# URIRef('https://w3id.org/oc/meta/id/') 

61# ), output_graph) 

62 

63# self.assertIn(( 

64# URIRef('https://w3id.org/oc/meta/id/061904730419'), 

65# URIRef('http://www.essepuntato.it/2010/06/literalreification/hasLiteralValue'), 

66# Literal('33440179'), 

67# URIRef('https://w3id.org/oc/meta/id/') 

68# ), output_graph) 

69 

70# def test_quadruple_count(self): 

71# input_quad_count = 0 

72# for file in os.listdir(self.input_dir): 

73# if file.endswith('.nq.gz'): 

74# with gzip.open(os.path.join(self.input_dir, file), 'rt') as f: 

75# input_quad_count += sum(1 for line in f if line.strip()) 

76 

77# output_quad_count = 0 

78# for root, dirs, files in os.walk(self.temp_dir): 

79# for file in files: 

80# if file.endswith('.json'): 

81# with open(os.path.join(root, file), 'r') as f: 

82# graph = ConjunctiveGraph().parse(f, format='json-ld') 

83# output_quad_count += len(graph) 

84 

85# self.assertEqual(input_quad_count, output_quad_count) 

86 

87# if __name__ == '__main__': 

88# unittest.main()