Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/prov/tests/test_extras.py @ 0:26e78fe6e8c4 draft
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author | shellac |
---|---|
date | Sat, 02 May 2020 07:14:21 -0400 |
parents | |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.7/site-packages/prov/tests/test_extras.py Sat May 02 07:14:21 2020 -0400 @@ -0,0 +1,286 @@ +from __future__ import (absolute_import, division, print_function, + unicode_literals) + +import io +import unittest + +from prov.model import * +from prov.dot import prov_to_dot +from prov.serializers import Registry +from prov.tests.examples import primer_example, primer_example_alternate + + +EX_NS = Namespace('ex', 'http://example.org/') +EX2_NS = Namespace('ex2', 'http://example2.org/') +EX_OTHER_NS = Namespace('other', 'http://exceptions.example.org/') + + +def add_label(record): + record.add_attributes( + [('prov:label', Literal("hello"))] + ) + + +def add_labels(record): + record.add_attributes([ + ('prov:label', Literal("hello")), + ('prov:label', Literal("bye", langtag="en")), + ('prov:label', Literal("bonjour", langtag="fr")) + ]) + + +def add_types(record): + record.add_attributes([ + ('prov:type', 'a'), + ('prov:type', 1), + ('prov:type', 1.0), + ('prov:type', True), + ('prov:type', EX_NS['abc']), + ('prov:type', datetime.datetime.now()), + ('prov:type', Literal('http://boiled-egg.example.com', + datatype=XSD_ANYURI)), + ]) + + +def add_locations(record): + record.add_attributes([ + ('prov:Location', "Southampton"), + ('prov:Location', 1), + ('prov:Location', 1.0), + ('prov:Location', True), + ('prov:Location', EX_NS['london']), + ('prov:Location', datetime.datetime.now()), + ('prov:Location', EX_NS.uri + "london"), + ('prov:Location', Literal(2002, datatype=XSD['gYear'])), + ]) + + +def add_value(record): + record.add_attributes([ + ('prov:value', EX_NS['avalue']) + ]) + + +def add_further_attributes(record): + record.add_attributes([ + (EX_NS['tag1'], "hello"), + (EX_NS['tag2'], "bye"), + (EX2_NS['tag3'], "hi"), + (EX_NS['tag1'], "hello\nover\nmore\nlines"), + ]) + + +def add_further_attributes0(record): + record.add_attributes([ + (EX_NS['tag1'], "hello"), + (EX_NS['tag2'], "bye"), + (EX_NS['tag2'], Literal("hola", langtag="es")), + (EX2_NS['tag3'], "hi"), + (EX_NS['tag'], 1), + # long on python 2, int on python 3 + (EX_NS['tag'], six.integer_types[-1](1)), + (EX_NS['tag'], Literal(1, datatype=XSD_SHORT)), + (EX_NS['tag'], Literal(1, datatype=XSD_DOUBLE)), + (EX_NS['tag'], 1.0), + (EX_NS['tag'], True), + (EX_NS['tag'], EX_NS.uri + "southampton"), + ]) + + add_further_attributes_with_qnames(record) + + +def add_further_attributes_with_qnames(record): + record.add_attributes([ + (EX_NS['tag'], EX2_NS['newyork']), + (EX_NS['tag'], EX_NS['london']), + ]) + + +class TestExtras(unittest.TestCase): + def test_dot(self): + # This is naive, since we can't programatically check the output is + # correct + document = ProvDocument() + + bundle1 = ProvBundle(identifier=EX_NS['bundle1']) + bundle1.usage( + activity=EX_NS['a1'], entity=EX_NS['e1'], identifier=EX_NS['use1'] + ) + bundle1.entity( + identifier=EX_NS['e1'], other_attributes={PROV_ROLE: "sausage"} + ) + bundle1.activity(identifier=EX_NS['a1']) + document.activity(EX_NS['a2']) + + bundle2 = ProvBundle(identifier=EX_NS['bundle2']) + bundle2.usage( + activity=EX_NS['aa1'], entity=EX_NS['ee1'], + identifier=EX_NS['use2'] + ) + bundle2.entity(identifier=EX_NS['ee1']) + bundle2.activity(identifier=EX_NS['aa1']) + + document.add_bundle(bundle1) + document.add_bundle(bundle2) + prov_to_dot(document) + + def test_extra_attributes(self): + + document = ProvDocument() + + inf = document.influence( + EX_NS['a2'], EX_NS['a1'], identifier=EX_NS['inf7'] + ) + add_labels(inf) + add_types(inf) + add_further_attributes(inf) + + self.assertEqual( + len(inf.attributes), + len(list(inf.formal_attributes) + inf.extra_attributes) + ) + + def test_serialize_to_path(self): + document = ProvDocument() + document.serialize("output.json") + os.remove('output.json') + + document.serialize("http://netloc/outputmyprov/submit.php") + + def test_bundle_no_id(self): + document = ProvDocument() + + def test(): + bundle = ProvBundle() + document.add_bundle(bundle) + + self.assertRaises(ProvException, test) + + def test_use_set_time_helpers(self): + dt = datetime.datetime.now() + document1 = ProvDocument() + document1.activity(EX_NS['a8'], startTime=dt, endTime=dt) + + document2 = ProvDocument() + a = document2.activity(EX_NS['a8']) + a.set_time(startTime=dt, endTime=dt) + + self.assertEqual(document1, document2) + self.assertEqual(a.get_startTime(), dt) + self.assertEqual(a.get_endTime(), dt) + + def test_bundle_add_garbage(self): + document = ProvDocument() + + def test(): + document.add_bundle( + document.entity(EX_NS['entity_trying_to_be_a_bundle']) + ) + + self.assertRaises(ProvException, test) + + def test(): + bundle = ProvBundle() + document.add_bundle(bundle) + + self.assertRaises(ProvException, test) + + def test_bundle_equality_garbage(self): + document = ProvBundle() + self.assertNotEqual(document, 1) + + def test_bundle_is_bundle(self): + document = ProvBundle() + self.assertTrue(document.is_bundle()) + + def test_bundle_get_record_by_id(self): + document = ProvDocument() + self.assertEqual(document.get_record(None), None) + + # record = document.entity(identifier=EX_NS['e1']) + # self.assertEqual(document.get_record(EX_NS['e1']), record) + # + # bundle = document.bundle(EX_NS['b']) + # self.assertEqual(bundle.get_record(EX_NS['e1']), record) + + def test_bundle_get_records(self): + document = ProvDocument() + + document.entity(identifier=EX_NS['e1']) + document.agent(identifier=EX_NS['e1']) + self.assertEqual(len(list(document.get_records(ProvAgent))), 1) + self.assertEqual(len(document.get_records()), 2) + + def test_bundle_name_clash(self): + document = ProvDocument() + + def test(): + document.bundle(EX_NS['indistinct']) + document.bundle(EX_NS['indistinct']) + + self.assertRaises(ProvException, test) + + document = ProvDocument() + + def test(): + document.bundle(EX_NS['indistinct']) + bundle = ProvBundle(identifier=EX_NS['indistinct']) + document.add_bundle(bundle) + + self.assertRaises(ProvException, test) + + def test_document_helper_methods(self): + document = ProvDocument() + self.assertFalse(document.is_bundle()) + self.assertFalse(document.has_bundles()) + document.bundle(EX_NS['b']) + self.assertTrue(document.has_bundles()) + self.assertEqual(u'<ProvDocument>', str(document)) + + def test_reading_and_writing_to_file_like_objects(self): + """ + Tests reading and writing to and from file like objects. + """ + # Create some random document. + document = ProvDocument() + document.entity(EX2_NS["test"]) + + objects = [io.BytesIO, io.StringIO] + + Registry.load_serializers() + formats = Registry.serializers.keys() + + for obj in objects: + for format in formats: + try: + buf = obj() + document.serialize(destination=buf, format=format) + buf.seek(0, 0) + new_document = ProvDocument.deserialize(source=buf, + format=format) + self.assertEqual(document, new_document) + except NotImplementedError: + # Some serializers might not implement serialize or + # deserialize method + pass # and this is fine in the context of this test + finally: + buf.close() + + # def test_document_unification(self): + # # TODO: Improve testing of this... + # document = ProvDocument() + # bundle = document.bundle(identifier=EX_NS['b']) + # e1 = bundle.entity(EX_NS['e']) + # e2 = bundle.entity(EX_NS['e']) + # unified = document.unified() + # + # self.assertEqual(len(unified._bundles[0]._records), 1) + + def test_primer_alternate(self): + g1 = primer_example() + g2 = primer_example_alternate() + self.assertEqual(g1, g2) + + +if __name__ == '__main__': + unittest.main()