diff --git a/datacat_integration/connection.py b/datacat_integration/connection.py index ddaa6bf3bd319c756dab3cd784171f5c4c7e3485..c7f4d2cbb493049cc2c473af0caca47dcc742341 100644 --- a/datacat_integration/connection.py +++ b/datacat_integration/connection.py @@ -76,13 +76,13 @@ class DataCatConnection: return self.refresh_token() - def get_object(self, datacat_type: str, oid: uuid): # GET /<type>/<oid> + def get_object(self, datacat_type: str, oid: uuid.UUID): # GET /<type>/<oid> """Returns a json of the given object from the server.""" headers = { 'accept' : 'application/json' } url = urljoin(self.url, "{}/{}".format(datacat_type, oid)) - return requests.get(url, headers=headers).json() + return json.dumps(requests.get(url, headers=headers).json()) def create_object(self, datacat_type: str, object: DataCatalogEntry): # POST /<type> """Creates a new object in the datacatalog. Returns the oid of successful.""" diff --git a/tests/test_connection.py b/tests/test_connection.py index 49e239a1616b3e748a23017e6653a8f050d6fc37..51c4fe3c9092b4c1b2b1aa7f4552dd3efaae8221 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -1,5 +1,6 @@ from unittest import TestCase -import os +import os, random +from dotenv import load_dotenv from datacat_integration.connection import DataCatalogEntry, DataCatConnection @@ -22,25 +23,60 @@ class EntryTest(TestCase): class ConnectionTest(TestCase): def setUp(self) -> None: - self.url = os.getenv('DATACAT_URL') - self.user = os.getenv('DATACAT_LOGIN') + load_dotenv("tests/testing-authentication.env", verbose=False) # does nothing if file is not present + self.url = os.getenv('DATACAT_URL', "https://zam10036.zam.kfa-juelich.de") + self.user = os.getenv('DATACAT_LOGIN', "dls-testing") self.password = os.getenv('DATACAT_PASSWORD') # if these are not set, connection can not be properly tested self.assertIsNotNone(self.url) self.assertIsNotNone(self.user) self.assertIsNotNone(self.password) - def test_create_token(self): - pass + def test_create_token(self): + connection = DataCatConnection(self.url, self.user, self.password) # creates token on __init__ + self.assertIsNotNone(connection._auth_token) def test_update_token(self): - pass + connection = DataCatConnection(self.url, self.user, self.password) + connection._auth_token = "fake and invalid token" + token = connection.get_token() + self.assertIsNotNone(token) + token2 = connection.get_token() + self.assertEqual(token, token2) - def test_get_object(self): - pass + + # may fail in future when the testing instance is moved or redeployed - test may need to be updated in that case, since the tested object may no longer exist + def test_get_entry(self): + connection = DataCatConnection(self.url, self.user, self.password) + entry_json = connection.get_object("storage_target", "7aa3877e-2860-4c65-8d48-3e080ceedca2") + entry = DataCatalogEntry.from_json(entry_json) + self.assertEqual(entry.name, "Yet Another Test Object") + self.assertEqual(entry.url, "1234") + self.assertEqual(entry.metadata['1'], '2') + self.assertEqual(entry.metadata['3'], '4') - def test_create_object(self): - pass # TODO + def test_create_entry(self): + connection = DataCatConnection(self.url, self.user, self.password) + entry = DataCatalogEntry("foo", "bar", {"1" : "2", "source" : "automated tests", "random" : str(random.getrandbits(32))}) + oid = connection.create_object("storage_target", entry) + entry_from_server = DataCatalogEntry.from_json(connection.get_object("storage_target", oid)) + self.assertDictEqual(entry.metadata, entry_from_server.metadata) + self.assertEqual(entry.name, entry_from_server.name) + self.assertEqual(entry.url, entry_from_server.url) - def test_list_objects(self): - pass \ No newline at end of file + # may fail in future when the testing instance is moved or redeployed - test may need to be updated in that case, since the tested objects may no longer exist + def test_list_entries(self): + connection = DataCatConnection(self.url, self.user, self.password) + entries = connection.list_type("storage_target") + self.assertIn("7aa3877e-2860-4c65-8d48-3e080ceedca2", [e[1] for e in entries]) + self.assertIn("foo", [e[0] for e in entries]) + self.assertGreaterEqual(len(entries), 5) + + def test_failed_auth(self): + connection = DataCatConnection(self.url, self.user, self.password) + connection.user = "foo" + connection._password = "bar" + self.assertRaises(ConnectionError, connection.refresh_token) + connection._auth_token = "foobar" + self.assertRaises(ConnectionError, connection.get_token) + self.assertRaises(ConnectionError, connection.create_object, "storage_target", DataCatalogEntry("foo", "bar", {})) \ No newline at end of file