Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/prov/tests/test_model.py @ 0:26e78fe6e8c4 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:26e78fe6e8c4
1 """
2 Created on Jan 25, 2012
3
4 @author: Trung Dong Huynh
5 """
6 from __future__ import (absolute_import, division, print_function,
7 unicode_literals)
8
9 import unittest
10 import logging
11 import os
12
13 from prov.model import ProvDocument, ProvBundle, ProvException, first, Literal
14 from prov.tests import examples
15 from prov.tests.attributes import TestAttributesBase
16 from prov.tests.qnames import TestQualifiedNamesBase
17 from prov.tests.statements import TestStatementsBase
18 from prov.tests.utility import RoundTripTestCase
19
20 logger = logging.getLogger(__name__)
21
22
23 EX_URI = 'http://www.example.org/'
24 EX2_URI = 'http://www.example2.org/'
25
26
27 class TestExamplesBase(object):
28 """This is the base class for testing support for all the examples provided
29 in prov.tests.examples.
30 It is not runnable and needs to be included in a subclass of
31 RoundTripTestCase.
32 """
33 def test_all_examples(self):
34 counter = 0
35 for name, graph in examples.tests:
36 counter += 1
37 logger.info('%d. Testing the %s example', counter, name)
38 g = graph()
39 self.do_tests(g)
40
41
42 class TestLoadingProvToolboxJSON(unittest.TestCase):
43 def setUp(self):
44 self.json_path = os.path.dirname(os.path.abspath(__file__)) + '/json/'
45 filenames = os.listdir(self.json_path)
46 self.fails = []
47 for filename in filenames:
48 if filename.endswith('.json'):
49 with open(self.json_path + filename) as json_file:
50 try:
51 g1 = ProvDocument.deserialize(json_file)
52 json_str = g1.serialize(indent=4)
53 g2 = ProvDocument.deserialize(content=json_str)
54 self.assertEqual(
55 g1, g2,
56 'Round-trip JSON encoding/decoding failed: %s.'
57 % filename
58 )
59 except:
60 self.fails.append(filename)
61
62 def test_loading_all_json(self):
63 # self.assertFalse(fails, 'Failed to load/round-trip %d JSON files (%s)' % (len(fails), ', '.join(fails)))
64
65 # Code for debugging the failed tests
66 for filename in self.fails:
67 # Reload the failed files
68 filepath = self.json_path + filename
69 # os.rename(json_path + filename, json_path + filename + '-fail')
70 with open(filepath) as json_file:
71 logger.info("Loading %s...", filepath)
72 g1 = ProvDocument.deserialize(json_file)
73 json_str = g1.serialize(indent=4)
74 g2 = ProvDocument.deserialize(content=json_str)
75 self.assertEqual(
76 g1, g2,
77 'Round-trip JSON encoding/decoding failed: %s.'
78 % filename
79 )
80
81
82 class TestFlattening(unittest.TestCase):
83 def test_flattening(self):
84 for name, graph in examples.tests:
85 logger.info('Testing flattening of the %s example', name)
86 document = graph()
87 flattened = document.flattened()
88 flattened_records = set(flattened.get_records())
89 # counting all the records:
90 n_records = 0
91 for record in document.get_records():
92 n_records += 1
93 self.assertIn(record, flattened_records)
94 for bundle in document.bundles:
95 for record in bundle.get_records():
96 n_records += 1
97 self.assertIn(record, flattened_records)
98 self.assertEqual(n_records, len(flattened.get_records()))
99
100
101 class TestUnification(unittest.TestCase):
102 def test_unifying(self):
103 # This is a very trivial test just to exercise the unified() function
104 # TODO: Create a proper unification test
105 json_path = os.path.dirname(os.path.abspath(__file__)) + '/unification/'
106 filenames = os.listdir(json_path)
107 for filename in filenames:
108 if not filename.endswith('.json'):
109 continue
110 filepath = json_path + filename
111 with open(filepath) as json_file:
112 logger.info('Testing unifying: %s', filename)
113 logger.debug("Loading %s...", filepath)
114 document = ProvDocument.deserialize(json_file)
115 flattened = document.flattened()
116 unified = flattened.unified()
117 self.assertLess(
118 len(unified.get_records()),
119 len(flattened.get_records())
120 )
121
122
123 class TestBundleUpdate(unittest.TestCase):
124 def test_bundle_update_simple(self):
125 doc = ProvDocument()
126 doc.set_default_namespace(EX_URI)
127
128 b1 = doc.bundle('b1')
129 b1.entity('e')
130
131 b2 = doc.bundle('b2')
132 b2.entity('e')
133
134 self.assertRaises(ProvException, lambda: b1.update(1))
135 self.assertRaises(ProvException, lambda: b1.update(doc))
136
137 b1.update(b2)
138 self.assertEqual(len(b1.get_records()), 2)
139
140 def test_document_update_simple(self):
141 d1 = ProvDocument()
142 d1.set_default_namespace(EX_URI)
143 d1.entity('e')
144
145 b1 = d1.bundle('b1')
146 b1.entity('e')
147
148 d2 = ProvDocument()
149 d2.set_default_namespace(EX_URI)
150 d2.entity('e')
151
152 b1 = d2.bundle('b1')
153 b1.entity('e')
154 b2 = d2.bundle('b2')
155 b2.entity('e')
156
157 self.assertRaises(ProvException, lambda: d1.update(1))
158
159 d1.update(d2)
160 self.assertEqual(len(d1.get_records()), 2)
161 self.assertEqual(len(d1.bundles), 2)
162
163
164 class TestAddBundle(unittest.TestCase):
165 def document_1(self):
166 d1 = ProvDocument()
167 ns_ex = d1.add_namespace('ex', EX_URI)
168 d1.entity(ns_ex['e1'])
169 return d1
170
171 def document_2(self):
172 d2 = ProvDocument()
173 ns_ex = d2.add_namespace('ex', EX2_URI)
174 d2.activity(ns_ex['a1'])
175 return d2
176
177 def bundle_0(self):
178 b = ProvBundle(namespaces={'ex': EX2_URI})
179 return b
180
181 def test_add_bundle_simple(self):
182 d1 = self.document_1()
183 b0 = self.bundle_0()
184
185 def sub_test_1():
186 d1.add_bundle(b0)
187 self.assertRaises(ProvException, sub_test_1)
188 self.assertFalse(d1.has_bundles())
189
190 d1.add_bundle(b0, 'ex:b0')
191 self.assertTrue(d1.has_bundles())
192 self.assertIn(b0, d1.bundles)
193
194 def sub_test_2():
195 ex2_b0 = b0.identifier
196 d1.add_bundle(ProvBundle(identifier=ex2_b0))
197 self.assertRaises(ProvException, sub_test_2)
198
199 d1.add_bundle(ProvBundle(), 'ex:b0')
200 self.assertEqual(len(d1.bundles), 2)
201
202 def test_add_bundle_document(self):
203 d1 = self.document_1()
204 d2 = self.document_2()
205
206 def sub_test_1():
207 d1.add_bundle(d2)
208 self.assertRaises(ProvException, sub_test_1)
209
210 ex2_b2 = d2.valid_qualified_name('ex:b2')
211 d1.add_bundle(d2, 'ex:b2')
212 self.assertEqual(ex2_b2, first(d1.bundles).identifier)
213 self.assertNotIn(d2, d1.bundles)
214 b2 = ProvBundle()
215 b2.update(d2)
216 self.assertIn(b2, d1.bundles)
217
218 class TestLiteralRepresentation(unittest.TestCase):
219 def test_literal_provn_with_single_quotes(self):
220 l = Literal('{"foo": "bar"}')
221 string_rep = l.provn_representation()
222 self.assertTrue('{\\"f' in string_rep)
223
224 def test_literal_provn_with_triple_quotes(self):
225 l = Literal('"""foo\\nbar"""')
226 string_rep = l.provn_representation()
227 self.assertTrue('\\"\\"\\"f' in string_rep)
228
229 class AllTestsBase(TestExamplesBase, TestStatementsBase,
230 TestAttributesBase, TestQualifiedNamesBase):
231 """This is a test to include all available tests.
232 """
233 pass
234
235
236 class RoundTripModelTest(RoundTripTestCase, AllTestsBase):
237 def assertRoundTripEquivalence(self, prov_doc, msg=None):
238 """Exercises prov.model without the actual serialization and PROV-N
239 generation.
240 """
241 provn_content = prov_doc.get_provn()
242 # Checking for self-equality
243 self.assertEqual(
244 prov_doc, prov_doc,
245 'The document is not self-equal:\n' + provn_content)
246
247
248 if __name__ == "__main__":
249 unittest.main()