Skip to content
Snippets Groups Projects
Commit 9fe12872 authored by Christian Boettcher's avatar Christian Boettcher
Browse files

better test config and cleanup

exclude hooks and operators from unittests (need to be tested in airflow)
parent 71bae68a
Branches
Tags
No related merge requests found
Pipeline #87449 passed
# .coveragerc to control coverage.py
[run]
branch = true
omit =
*/hooks.py
*/operators.py
[report]
# Regexes for lines to exclude from consideration
exclude_lines =
# Have to re-enable the standard pragma
pragma: no cover
# Don't complain about missing debug-only code:
def __repr__
if self\.debug
# Don't complain if tests don't hit defensive assertion code:
raise AssertionError
raise NotImplementedError
# Don't complain if non-runnable code isn't run:
if 0:
if __name__ == .__main__.:
# Don't complain about abstract methods, they aren't run:
@(abc\.)?abstractmethod
ignore_errors = True
[html]
directory = coverage_html_report
\ No newline at end of file
...@@ -9,7 +9,6 @@ from datacat_integration.connection import DataCatConnection, DataCatalogEntry, ...@@ -9,7 +9,6 @@ from datacat_integration.connection import DataCatConnection, DataCatalogEntry,
from datacat_integration.auth import BearerAuth from datacat_integration.auth import BearerAuth
class DataCatalogHook(HttpHook): class DataCatalogHook(HttpHook):
connection: DataCatConnection = None connection: DataCatConnection = None
......
import os
from urllib.parse import urljoin
from dotenv import load_dotenv
import requests
from datacat_integration.connection import DataCatConnection
def delete_entries():
name = "foo"
conn_type = "storage_target"
load_dotenv("tests/testing-authentication.env", verbose=False) # does nothing if file is not present
url = os.getenv('DATACAT_URL', "https://zam10036.zam.kfa-juelich.de")
user = os.getenv('DATACAT_LOGIN', "dls-testing")
password = os.getenv('DATACAT_PASSWORD')
connection = DataCatConnection(url, user, password)
token = connection._auth_token
headers = {
'accept' : 'application/json',
'Authorization' : f'Bearer {token}'
}
for entry in connection.list_type(conn_type):
if entry[0] == name:
requests.delete(f"{url}/{conn_type}/{entry[1]}", headers=headers)
...@@ -3,6 +3,7 @@ import os, random, json ...@@ -3,6 +3,7 @@ import os, random, json
from dotenv import load_dotenv from dotenv import load_dotenv
from datacat_integration.connection import DataCatalogEntry, DataCatConnection, get_connection_from_entry from datacat_integration.connection import DataCatalogEntry, DataCatConnection, get_connection_from_entry
from .delete_created_entries import delete_entries
class GetConnectionTest(TestCase): class GetConnectionTest(TestCase):
...@@ -36,6 +37,9 @@ class EntryTest(TestCase): ...@@ -36,6 +37,9 @@ class EntryTest(TestCase):
self.json_string = '{ "name" : "foo", "url" : "bar", "metadata" : { "key1" : "val1", "key2" : "val2" } }' self.json_string = '{ "name" : "foo", "url" : "bar", "metadata" : { "key1" : "val1", "key2" : "val2" } }'
self.entry : DataCatalogEntry = DataCatalogEntry("foo", "bar", {"key1" : "val1", "key2" : "val2"}) self.entry : DataCatalogEntry = DataCatalogEntry("foo", "bar", {"key1" : "val1", "key2" : "val2"})
def tearDown(self) -> None:
delete_entries()
def test_create_entry_from_json(self): def test_create_entry_from_json(self):
entry_from_json = DataCatalogEntry.from_json(self.json_string) entry_from_json = DataCatalogEntry.from_json(self.json_string)
self.assertDictEqual(self.entry.metadata, entry_from_json.metadata) self.assertDictEqual(self.entry.metadata, entry_from_json.metadata)
...@@ -58,6 +62,9 @@ class ConnectionTest(TestCase): ...@@ -58,6 +62,9 @@ class ConnectionTest(TestCase):
self.assertIsNotNone(self.user) self.assertIsNotNone(self.user)
self.assertIsNotNone(self.password) self.assertIsNotNone(self.password)
def tearDown(self) -> None:
delete_entries()
def test_create_token(self): def test_create_token(self):
connection = DataCatConnection(self.url, self.user, self.password) # creates token on __init__ connection = DataCatConnection(self.url, self.user, self.password) # creates token on __init__
self.assertIsNotNone(connection._auth_token) self.assertIsNotNone(connection._auth_token)
...@@ -94,8 +101,8 @@ class ConnectionTest(TestCase): ...@@ -94,8 +101,8 @@ class ConnectionTest(TestCase):
def test_list_entries(self): def test_list_entries(self):
connection = DataCatConnection(self.url, self.user, self.password) connection = DataCatConnection(self.url, self.user, self.password)
entries = connection.list_type("storage_target") entries = connection.list_type("storage_target")
self.assertIn("7aa3877e-2860-4c65-8d48-3e080ceedca2", [e[1] for e in entries]) self.assertIn("6642e0be-b642-499d-a997-d76b8e350387", [e[1] for e in entries])
self.assertIn("foo", [e[0] for e in entries]) self.assertIn("DLS-Testing-Connection-Object 1", [e[0] for e in entries])
self.assertGreaterEqual(len(entries), 5) self.assertGreaterEqual(len(entries), 5)
def test_failed_auth(self): def test_failed_auth(self):
......
from unittest import TestCase from unittest import TestCase
import os
from dotenv import load_dotenv
from datacat_integration.secrets import DataCatConnectionWithSecrets, get_connection_with_secrets_from_entry, DatacatSecretsBackend from datacat_integration.secrets import DataCatConnectionWithSecrets, get_connection_with_secrets_from_entry, DatacatSecretsBackend
class TestSecretsBackenbd(TestCase): class TestSecretsBackenbd(TestCase):
def setUp(self): def setUp(self):
load_dotenv("tests/testing-authentication.env", verbose=False) # does nothing if file is not present
self.backend = DatacatSecretsBackend() self.backend = DatacatSecretsBackend()
self.backend.url = "https://zam10036.zam.kfa-juelich.de"
self.backend.user = "dls-testing" self.backend.url = os.getenv('DATACAT_URL', "https://zam10036.zam.kfa-juelich.de")
self.backend.password = "dls-testing-pass" self.backend.user = os.getenv('DATACAT_LOGIN', "dls-testing")
self.backend.password = os.getenv('DATACAT_PASSWORD')
self.assertIsNotNone(self.backend.url)
self.assertIsNotNone(self.backend.user)
self.assertIsNotNone(self.backend.password)
def test_get_connection_from_oid(self): def test_get_connection_from_oid(self):
conn = self.backend.get_connection("860355e9-975f-4253-9421-1815e20c879b") conn = self.backend.get_connection("860355e9-975f-4253-9421-1815e20c879b")
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please to comment