Mercurial > repos > guerler > springsuite

comparison planemo/lib/python3.7/site-packages/prov/tests/test_extras.py @ 1:56ad4e20f292 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:32:28 -0400
parents
children
comparison
equal deleted inserted replaced
0:d30785e31577 1:56ad4e20f292
1 from __future__ import (absolute_import, division, print_function,
2 unicode_literals)
3
4 import io
5 import unittest
6
7 from prov.model import *
8 from prov.dot import prov_to_dot
9 from prov.serializers import Registry
10 from prov.tests.examples import primer_example, primer_example_alternate
11
12
13 EX_NS = Namespace('ex', 'http://example.org/')
14 EX2_NS = Namespace('ex2', 'http://example2.org/')
15 EX_OTHER_NS = Namespace('other', 'http://exceptions.example.org/')
16
17
18 def add_label(record):
19 record.add_attributes(
20 [('prov:label', Literal("hello"))]
21 )
22
23
24 def add_labels(record):
25 record.add_attributes([
26 ('prov:label', Literal("hello")),
27 ('prov:label', Literal("bye", langtag="en")),
28 ('prov:label', Literal("bonjour", langtag="fr"))
29 ])
30
31
32 def add_types(record):
33 record.add_attributes([
34 ('prov:type', 'a'),
35 ('prov:type', 1),
36 ('prov:type', 1.0),
37 ('prov:type', True),
38 ('prov:type', EX_NS['abc']),
39 ('prov:type', datetime.datetime.now()),
40 ('prov:type', Literal('http://boiled-egg.example.com',
41 datatype=XSD_ANYURI)),
42 ])
43
44
45 def add_locations(record):
46 record.add_attributes([
47 ('prov:Location', "Southampton"),
48 ('prov:Location', 1),
49 ('prov:Location', 1.0),
50 ('prov:Location', True),
51 ('prov:Location', EX_NS['london']),
52 ('prov:Location', datetime.datetime.now()),
53 ('prov:Location', EX_NS.uri + "london"),
54 ('prov:Location', Literal(2002, datatype=XSD['gYear'])),
55 ])
56
57
58 def add_value(record):
59 record.add_attributes([
60 ('prov:value', EX_NS['avalue'])
61 ])
62
63
64 def add_further_attributes(record):
65 record.add_attributes([
66 (EX_NS['tag1'], "hello"),
67 (EX_NS['tag2'], "bye"),
68 (EX2_NS['tag3'], "hi"),
69 (EX_NS['tag1'], "hello\nover\nmore\nlines"),
70 ])
71
72
73 def add_further_attributes0(record):
74 record.add_attributes([
75 (EX_NS['tag1'], "hello"),
76 (EX_NS['tag2'], "bye"),
77 (EX_NS['tag2'], Literal("hola", langtag="es")),
78 (EX2_NS['tag3'], "hi"),
79 (EX_NS['tag'], 1),
80 # long on python 2, int on python 3
81 (EX_NS['tag'], six.integer_types[-1](1)),
82 (EX_NS['tag'], Literal(1, datatype=XSD_SHORT)),
83 (EX_NS['tag'], Literal(1, datatype=XSD_DOUBLE)),
84 (EX_NS['tag'], 1.0),
85 (EX_NS['tag'], True),
86 (EX_NS['tag'], EX_NS.uri + "southampton"),
87 ])
88
89 add_further_attributes_with_qnames(record)
90
91
92 def add_further_attributes_with_qnames(record):
93 record.add_attributes([
94 (EX_NS['tag'], EX2_NS['newyork']),
95 (EX_NS['tag'], EX_NS['london']),
96 ])
97
98
99 class TestExtras(unittest.TestCase):
100 def test_dot(self):
101 # This is naive, since we can't programatically check the output is
102 # correct
103 document = ProvDocument()
104
105 bundle1 = ProvBundle(identifier=EX_NS['bundle1'])
106 bundle1.usage(
107 activity=EX_NS['a1'], entity=EX_NS['e1'], identifier=EX_NS['use1']
108 )
109 bundle1.entity(
110 identifier=EX_NS['e1'], other_attributes={PROV_ROLE: "sausage"}
111 )
112 bundle1.activity(identifier=EX_NS['a1'])
113 document.activity(EX_NS['a2'])
114
115 bundle2 = ProvBundle(identifier=EX_NS['bundle2'])
116 bundle2.usage(
117 activity=EX_NS['aa1'], entity=EX_NS['ee1'],
118 identifier=EX_NS['use2']
119 )
120 bundle2.entity(identifier=EX_NS['ee1'])
121 bundle2.activity(identifier=EX_NS['aa1'])
122
123 document.add_bundle(bundle1)
124 document.add_bundle(bundle2)
125 prov_to_dot(document)
126
127 def test_extra_attributes(self):
128
129 document = ProvDocument()
130
131 inf = document.influence(
132 EX_NS['a2'], EX_NS['a1'], identifier=EX_NS['inf7']
133 )
134 add_labels(inf)
135 add_types(inf)
136 add_further_attributes(inf)
137
138 self.assertEqual(
139 len(inf.attributes),
140 len(list(inf.formal_attributes) + inf.extra_attributes)
141 )
142
143 def test_serialize_to_path(self):
144 document = ProvDocument()
145 document.serialize("output.json")
146 os.remove('output.json')
147
148 document.serialize("http://netloc/outputmyprov/submit.php")
149
150 def test_bundle_no_id(self):
151 document = ProvDocument()
152
153 def test():
154 bundle = ProvBundle()
155 document.add_bundle(bundle)
156
157 self.assertRaises(ProvException, test)
158
159 def test_use_set_time_helpers(self):
160 dt = datetime.datetime.now()
161 document1 = ProvDocument()
162 document1.activity(EX_NS['a8'], startTime=dt, endTime=dt)
163
164 document2 = ProvDocument()
165 a = document2.activity(EX_NS['a8'])
166 a.set_time(startTime=dt, endTime=dt)
167
168 self.assertEqual(document1, document2)
169 self.assertEqual(a.get_startTime(), dt)
170 self.assertEqual(a.get_endTime(), dt)
171
172 def test_bundle_add_garbage(self):
173 document = ProvDocument()
174
175 def test():
176 document.add_bundle(
177 document.entity(EX_NS['entity_trying_to_be_a_bundle'])
178 )
179
180 self.assertRaises(ProvException, test)
181
182 def test():
183 bundle = ProvBundle()
184 document.add_bundle(bundle)
185
186 self.assertRaises(ProvException, test)
187
188 def test_bundle_equality_garbage(self):
189 document = ProvBundle()
190 self.assertNotEqual(document, 1)
191
192 def test_bundle_is_bundle(self):
193 document = ProvBundle()
194 self.assertTrue(document.is_bundle())
195
196 def test_bundle_get_record_by_id(self):
197 document = ProvDocument()
198 self.assertEqual(document.get_record(None), None)
199
200 # record = document.entity(identifier=EX_NS['e1'])
201 # self.assertEqual(document.get_record(EX_NS['e1']), record)
202 #
203 # bundle = document.bundle(EX_NS['b'])
204 # self.assertEqual(bundle.get_record(EX_NS['e1']), record)
205
206 def test_bundle_get_records(self):
207 document = ProvDocument()
208
209 document.entity(identifier=EX_NS['e1'])
210 document.agent(identifier=EX_NS['e1'])
211 self.assertEqual(len(list(document.get_records(ProvAgent))), 1)
212 self.assertEqual(len(document.get_records()), 2)
213
214 def test_bundle_name_clash(self):
215 document = ProvDocument()
216
217 def test():
218 document.bundle(EX_NS['indistinct'])
219 document.bundle(EX_NS['indistinct'])
220
221 self.assertRaises(ProvException, test)
222
223 document = ProvDocument()
224
225 def test():
226 document.bundle(EX_NS['indistinct'])
227 bundle = ProvBundle(identifier=EX_NS['indistinct'])
228 document.add_bundle(bundle)
229
230 self.assertRaises(ProvException, test)
231
232 def test_document_helper_methods(self):
233 document = ProvDocument()
234 self.assertFalse(document.is_bundle())
235 self.assertFalse(document.has_bundles())
236 document.bundle(EX_NS['b'])
237 self.assertTrue(document.has_bundles())
238 self.assertEqual(u'<ProvDocument>', str(document))
239
240 def test_reading_and_writing_to_file_like_objects(self):
241 """
242 Tests reading and writing to and from file like objects.
243 """
244 # Create some random document.
245 document = ProvDocument()
246 document.entity(EX2_NS["test"])
247
248 objects = [io.BytesIO, io.StringIO]
249
250 Registry.load_serializers()
251 formats = Registry.serializers.keys()
252
253 for obj in objects:
254 for format in formats:
255 try:
256 buf = obj()
257 document.serialize(destination=buf, format=format)
258 buf.seek(0, 0)
259 new_document = ProvDocument.deserialize(source=buf,
260 format=format)
261 self.assertEqual(document, new_document)
262 except NotImplementedError:
263 # Some serializers might not implement serialize or
264 # deserialize method
265 pass # and this is fine in the context of this test
266 finally:
267 buf.close()
268
269 # def test_document_unification(self):
270 # # TODO: Improve testing of this...
271 # document = ProvDocument()
272 # bundle = document.bundle(identifier=EX_NS['b'])
273 # e1 = bundle.entity(EX_NS['e'])
274 # e2 = bundle.entity(EX_NS['e'])
275 # unified = document.unified()
276 #
277 # self.assertEqual(len(unified._bundles[0]._records), 1)
278
279 def test_primer_alternate(self):
280 g1 = primer_example()
281 g2 = primer_example_alternate()
282 self.assertEqual(g1, g2)
283
284
285 if __name__ == '__main__':
286 unittest.main()