Coverage for test/duplicated_ids

1# #!python

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.

16# import json

17# import os

18# import unittest

19# from shutil import rmtree

20# from subprocess import call

21# from sys import executable

22# from test.curator_test import reset_server

24# from oc_ocdm import Storer

25# from oc_ocdm.graph import GraphSet

26# from oc_ocdm.prov import ProvSet

27# from oc_ocdm.reader import Reader

28# from rdflib import URIRef

30# BASE = os.path.join('test', 'fixer', 'duplicated_ids')

31# CONFIG = os.path.join(BASE, 'meta_config.yaml')

33# class test_duplicated_ids(unittest.TestCase):

34# def test_find_duplicated_ids_in_entity_type(self):

35# reset_server()

36# for stuff in os.listdir(BASE):

37# if os.path.isdir(os.path.join(BASE, stuff)) and stuff not in {'input'}:

38# rmtree(os.path.join(BASE, stuff))

39# elif os.path.isfile(os.path.join(BASE, stuff)) and stuff not in {'meta_config.yaml'}:

40# os.remove(os.path.join(BASE, stuff))

41# call([executable, '-m', 'oc_meta.run.meta_process', '-c', CONFIG])

42# base_iri = 'https://w3id.org/oc/meta/'

43# info_dir = os.path.join(BASE, 'info_dir', 'creator')

44# g_set = GraphSet(base_iri, info_dir, supplier_prefix='060', wanted_label=False)

45# endpoint = 'http://127.0.0.1:8805/sparql'

46# resp_agent = 'https://orcid.org/0000-0002-8420-0696'

47# rdf = os.path.join(BASE, 'rdf') + os.sep

48# reader = Reader()

49# reader.import_entity_from_triplestore(g_set, endpoint, URIRef('https://w3id.org/oc/meta/ra/0605'), resp_agent, enable_validation=False)

50# ieee = g_set.get_entity(URIRef('https://w3id.org/oc/meta/ra/0605'))

51# duplicated_id = g_set.add_id(resp_agent)

52# triplicated_id = g_set.add_id(resp_agent)

53# duplicated_id.create_crossref('263')

54# triplicated_id.create_crossref('263')

55# ieee.has_identifier(duplicated_id)

56# ieee.has_identifier(triplicated_id)

57# provset = ProvSet(g_set, base_iri, info_dir, wanted_label=False, supplier_prefix='060')

58# provset.generate_provenance()

59# graph_storer = Storer(g_set, dir_split=10000, n_file_item=1000, zip_output=False)

60# prov_storer = Storer(provset, dir_split=10000, n_file_item=1000, zip_output=False)

61# graph_storer.store_all(rdf, base_iri)

62# prov_storer.store_all(rdf, base_iri)

63# graph_storer.upload_all(endpoint)

64# call([executable, '-m', 'oc_meta.run.fixer.duplicated_ids', '-e', 'ra', '-c', os.path.join(BASE, 'meta_config.yaml'), '-r', 'https://orcid.org/0000-0002-8420-0696', '-ca', os.path.join(BASE, 'cache.json')])

65# for filepath in [

66# os.path.join(BASE, 'rdf', 'ra', '060', '10000', '1000.json'),

67# os.path.join(BASE, 'rdf', 'ra', '060', '10000', '1000', 'prov', 'se.json')

68# ]:

69# with open(filepath, 'r', encoding='utf8') as f:

70# data = json.load(f)

71# for graph in data:

72# graph_data = graph['@graph']

73# for entity in graph_data:

74# if entity['@id'] == 'https://w3id.org/oc/meta/ra/0605':

75# identifiers = {identifier['@id'] for identifier in entity['http://purl.org/spar/datacite/hasIdentifier']}

76# self.assertTrue(identifiers == {'https://w3id.org/oc/meta/id/0604', 'https://w3id.org/oc/meta/id/0603'})

77# elif entity['@id'] == 'https://w3id.org/oc/meta/ra/0605/prov/se/3':

78# update_query = entity['https://w3id.org/oc/ontology/hasUpdateQuery'][0]['@value']

79# self.assertTrue(update_query, 'DELETE DATA { GRAPH <https://w3id.org/oc/meta/ra/> { <https://w3id.org/oc/meta/ra/0605> <http://purl.org/spar/datacite/hasIdentifier> <https://w3id.org/oc/meta/id/0604> . } }')

80# for stuff in os.listdir(BASE):

81# if os.path.isdir(os.path.join(BASE, stuff)) and stuff not in {'input'}:

82# rmtree(os.path.join(BASE, stuff))

83# elif os.path.isfile(os.path.join(BASE, stuff)) and stuff not in {'meta_config.yaml'}:

84# os.remove(os.path.join(BASE, stuff))

Coverage for test/duplicated_ids_test.py: 100%

0 statements