From 61aabb44d846fce55d875f67186b25ba04140955 Mon Sep 17 00:00:00 2001 From: Alessio Siniscalchi Date: Mon, 23 Sep 2024 10:49:04 +0200 Subject: [PATCH] more tests --- tests/test_20_licence_manager.py | 161 +++++++++++++++++++++++++++++++ 1 file changed, 161 insertions(+) diff --git a/tests/test_20_licence_manager.py b/tests/test_20_licence_manager.py index cd598f6..30976ce 100644 --- a/tests/test_20_licence_manager.py +++ b/tests/test_20_licence_manager.py @@ -1,5 +1,7 @@ import operator import os.path +import uuid +from typing import Any import pytest_mock import sqlalchemy as sa @@ -10,6 +12,41 @@ TESTDATA_PATH = os.path.join(THIS_PATH, "data") +def mock_dataset( + resource_uid: str | None = None, + abstract: str = "abstract", + description: dict[str, Any] | None = None, + **kwargs, +) -> database.Resource: + dataset = database.Resource( + resource_uid=resource_uid or str(uuid.uuid4()), + abstract=abstract, + description=description or dict(), + type="dataset", + **kwargs, + ) + return dataset + + +def mock_licence( + licence_uid: str | None = None, + revision: int = 1, + title: str = "title", + download_filename: str = "licence.pdf", + md_filename: str = "licence.md", + **kwargs, +) -> database.Licence: + licence = database.Licence( + licence_uid=licence_uid or str(uuid.uuid4()), + revision=revision, + title=title, + download_filename=download_filename, + md_filename=md_filename, + **kwargs, + ) + return licence + + def test_licence_sync( session_obj: sa.orm.sessionmaker, mocker: pytest_mock.MockerFixture ) -> None: @@ -156,3 +193,127 @@ def test_load_licences_from_folder() -> None: ) assert licences == expected_licences + + +def test_update_catalogue_licences( + session_obj: sa.orm.sessionmaker, mocker: pytest_mock.MockerFixture +) -> None: + # load and add some licences into the db + licences_folder_path = os.path.join(TESTDATA_PATH, "cads-licences") + my_settings_dict = { + "object_storage_url": "object/storage/url", + "storage_admin": "admin1", + "storage_password": "secret1", + "catalogue_bucket": "mycatalogue_bucket", + "document_storage_url": "my/url", + } + storage_settings = config.ObjectStorageSettings(**my_settings_dict) + _ = mocker.patch( + "cads_catalogue.object_storage.store_file", + return_value="an url", + ) + with session_obj() as session: + licence_attrs = licence_manager.update_catalogue_licences( + session, licences_folder_path, storage_settings + ) + assert sorted(licence_attrs) == [ + ("CCI-data-policy-for-satellite-surface-radiation-budget", 4), + ("data-protection-privacy-statement", 24), + ("eumetsat-cm-saf", 1), + ("licence-to-use-copernicus-products", 12), + ] + session.commit() + for licence_uid, revision in licence_attrs: + licence_obj = session.scalars( + sa.select(database.Licence).filter_by( + licence_uid=licence_uid, revision=revision + ) + ).all() + assert len(licence_obj) == 1 + + +def test_remove_orphan_licences(session_obj: sa.orm.sessionmaker) -> None: + resource_uids = ["datasetA", "datasetB", "datasetC"] + licence_uids = [("licence_1", 1), ("licence_2", 1), ("licence_3", 1)] + dataset_objs = dict() + licence_objs = dict() + with session_obj() as session: + # add some datasets + for resource_uid in resource_uids: + dataset_obj = mock_dataset(resource_uid=resource_uid) + session.add(dataset_obj) + dataset_objs[resource_uid] = dataset_obj + session.commit() + # add some licences + for licence_uid, revision in licence_uids: + licence_obj = mock_licence(licence_uid=licence_uid, revision=revision) + session.add(licence_obj) + licence_objs[licence_uid] = licence_obj + # add some relationships + licence_objs["licence_1"].resources = [dataset_objs["datasetA"]] + licence_objs["licence_2"].resources = [ + dataset_objs["datasetA"], + dataset_objs["datasetB"], + ] + licence_objs["licence_3"].resources = [dataset_objs["datasetC"]] + session.commit() + + # case 1: do not remove anything, all licences are to keep + keep_licences = licence_uids + licence_manager.remove_orphan_licences(session, keep_licences, resource_uids) + session.commit() + for licence_uid, revision in licence_uids: + query_licences = session.scalars( + sa.select(database.Licence).filter_by( + licence_uid=licence_uid, revision=revision + ) + ).all() + assert len(query_licences) == 1 + + # case 2: do not remove anything, not to keep but they all have datasets + keep_licences = [] + licence_manager.remove_orphan_licences(session, keep_licences, resource_uids) + session.commit() + for licence_uid, revision in licence_uids: + query_licences = session.scalars( + sa.select(database.Licence).filter_by( + licence_uid=licence_uid, revision=revision + ) + ).all() + assert len(query_licences) == 1 + + # case 3: remove a licence, not to keep and unrelated to any dataset + keep_licences = [ + ("licence_1", 1), + ("licence_2", 1), + ] + licence_objs["licence_3"].resources = [] + licence_manager.remove_orphan_licences(session, keep_licences, resource_uids) + session.commit() + for licence_uid, revision in keep_licences: + query_licences = session.scalars( + sa.select(database.Licence).filter_by( + licence_uid=licence_uid, revision=revision + ) + ).all() + assert len(query_licences) == 1 + query_licences = session.scalars( + sa.select(database.Licence).filter_by( + licence_uid="licence_3", revision=1 + ) + ).all() + assert len(query_licences) == 0 + + # case 4: remove a licence, not to keep and related to dataset not to keep + keep_licences = [] + resource_uids = ["datasetB"] + licence_manager.remove_orphan_licences(session, keep_licences, resource_uids) + session.commit() + query_licences = session.scalars( + sa.select(database.Licence).filter_by(licence_uid="licence_2", revision=1) + ).all() + assert len(query_licences) == 1 + query_licences = session.scalars( + sa.select(database.Licence).filter_by(licence_uid="licence_1", revision=1) + ).all() + assert len(query_licences) == 0