Skip to content

Commit

Permalink
more tests
Browse files Browse the repository at this point in the history
  • Loading branch information
alex75 committed Sep 23, 2024
1 parent 3b3c741 commit 61aabb4
Showing 1 changed file with 161 additions and 0 deletions.
161 changes: 161 additions & 0 deletions tests/test_20_licence_manager.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import operator
import os.path
import uuid
from typing import Any

import pytest_mock
import sqlalchemy as sa
Expand All @@ -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:
Expand Down Expand Up @@ -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

0 comments on commit 61aabb4

Please sign in to comment.