reintroduce_fs_models_test: reintroduce (#217)

Author: DanielElisenberg

pytest.ini

@@ -0,0 +1,66 @@
+import pytest
+from pydantic import ValidationError
+
+from job_executor.adapter.fs.models.datastore_versions import (
+    DataStructureUpdate,
+)
+from job_executor.common.exceptions import ReleaseStatusException
+
+DATA_STRUCTURE_ADD_UPDATE = {
+    "name": "KJOENN",
+    "description": "Første publisering",
+    "operation": "ADD",
+    "releaseStatus": "DRAFT",
+}
+DATA_STRUCTURE_REMOVE_UPDATE = {
+    "name": "KJOENN",
+    "description": "Fjernet variabel",
+    "operation": "REMOVE",
+    "releaseStatus": "PENDING_DELETE",
+}
+DATA_STRUCTURE_UPDATE_MISSING_FIELD = {
+    "name": "KJOENN",
+    "description": "Første publisering",
+    "operation": "ADD",
+}
+
+
+def test_data_structure_update():
+    data_structure_update = DataStructureUpdate(**DATA_STRUCTURE_ADD_UPDATE)
+    assert (
+        data_structure_update.model_dump(by_alias=True, exclude_none=True)
+        == DATA_STRUCTURE_ADD_UPDATE
+    )
+
+
+def test_invalid_data_structure_update():
+    with pytest.raises(ValidationError):
+        DataStructureUpdate(**DATA_STRUCTURE_UPDATE_MISSING_FIELD)
+
+
+def test_set_release_status():
+    data_structure_update = DataStructureUpdate(**DATA_STRUCTURE_ADD_UPDATE)
+    data_structure_update.set_release_status("PENDING_RELEASE")
+    assert data_structure_update.release_status == "PENDING_RELEASE"
+
+    data_structure_update.set_release_status("DRAFT")
+    assert data_structure_update.release_status == "DRAFT"
+
+    with pytest.raises(ReleaseStatusException) as e:
+        data_structure_update.set_release_status("PENDING_DELETE")
+    assert "Can't set release status: PENDING_DELETE" in str(e)
+    with pytest.raises(ReleaseStatusException) as e:
+        data_structure_update.set_release_status("NO_SUCH_RELEASE_STATUS")
+    assert "Invalid release status: NO_SUCH_RELEASE_STATUS" in str(e)
+
+    data_structure_update = DataStructureUpdate(**DATA_STRUCTURE_REMOVE_UPDATE)
+    data_structure_update.set_release_status("PENDING_DELETE")
+    assert data_structure_update.release_status == "PENDING_DELETE"
+
+    with pytest.raises(ReleaseStatusException) as e:
+        data_structure_update.set_release_status("DRAFT")
+    assert "Can't set release status: DRAFT" in str(e)
+
+    with pytest.raises(ReleaseStatusException) as e:
+        data_structure_update.set_release_status("PENDING_RELEASE")
+    assert "Can't set release status: PENDING_RELEASE" in str(e)

tests/unit/adapter/fs/model/test_data_structure_update.py

@@ -0,0 +1,166 @@
+import json
+import os
+import shutil
+
+import pytest
+
+from job_executor.adapter.fs.models.datastore_versions import (
+    DatastoreVersion,
+    DataStructureUpdate,
+    DraftVersion,
+)
+from job_executor.common.exceptions import (
+    BumpException,
+    ExistingDraftException,
+    NoSuchDraftException,
+)
+
+
+def load_json(file_path):
+    return json.load(open(file_path, encoding="utf"))
+
+
+TEST_DIR = "tests/unit/resources/adapter/fs/model/datastore_version"
+DRAFT_VERSION_IDENTICAL = load_json(f"{TEST_DIR}/draft_version_identical.json")
+DRAFT_VERSION_ONLY_PENDING = load_json(
+    f"{TEST_DIR}/draft_version_only_pending.json"
+)
+DRAFT_VERSION_ADDED_PENDING = load_json(
+    f"{TEST_DIR}/draft_version_added_pending.json"
+)
+DATASTORE_VERSION = {
+    "version": "0.1.0.1635299291",
+    "description": "Første release",
+    "releaseTime": 1635299291,
+    "languageCode": "no",
+    "dataStructureUpdates": [
+        {
+            "name": "INNTEKT",
+            "description": "Første publisering",
+            "operation": "ADD",
+            "releaseStatus": "RELEASED",
+        },
+        {
+            "name": "SIVSTAND",
+            "description": "Første publisering",
+            "operation": "PATCH_METADATA",
+            "releaseStatus": "RELEASED",
+        },
+    ],
+    "updateType": "MINOR",
+}
+
+
+def setup_function():
+    if os.path.isdir("tests/resources_backup"):
+        shutil.rmtree("tests/resources_backup")
+
+    shutil.copytree("tests/resources", "tests/resources_backup")
+
+
+def teardown_function():
+    shutil.rmtree("tests/resources")
+    shutil.move("tests/resources_backup", "tests/resources")
+
+
+def test_get_dataset_release_status():
+    datastore_version = DatastoreVersion.model_validate(DATASTORE_VERSION)
+
+    release_status = datastore_version.get_dataset_release_status("INNTEKT")
+    assert release_status == "RELEASED"
+
+    release_status = datastore_version.get_dataset_release_status(
+        "DOES_NOT_EXIST"
+    )
+    assert release_status is None
+
+
+def test_draft_version_delete_draft():
+    draft_version = DraftVersion.model_validate(DRAFT_VERSION_IDENTICAL)
+    release_time = draft_version.release_time
+    version = draft_version.version
+    draft_version.delete_draft("UTDANNING")
+    update_names = [
+        update.name for update in draft_version.data_structure_updates
+    ]
+    assert "UTDANNING" not in update_names
+    assert len(update_names) == 1
+    assert release_time != draft_version.release_time
+    assert version != draft_version.version
+
+    with pytest.raises(NoSuchDraftException) as e:
+        draft_version.delete_draft("NO_SUCH_DATASET")
+    assert "Can't delete draft for NO_SUCH_DATASET" in str(e)
+
+
+def test_add_draft_version_already_existing_dataset():
+    draft_version = DraftVersion.model_validate(DRAFT_VERSION_IDENTICAL)
+    with pytest.raises(ExistingDraftException) as e:
+        draft_version.add(
+            DataStructureUpdate(
+                name="UTDANNING",
+                description="",
+                operation="ADD",
+                release_status="DRAFT",
+            )
+        )
+    assert "Draft for UTDANNING already exists" in str(e)
+
+
+def test_draft_version_validate_bump_manifesto():
+    draft_version = DraftVersion.model_validate(DRAFT_VERSION_IDENTICAL)
+
+    bump_manifesto = DatastoreVersion.model_validate(DRAFT_VERSION_IDENTICAL)
+    assert draft_version.validate_bump_manifesto(bump_manifesto)
+
+    bump_manifesto_only_pending = DatastoreVersion.model_validate(
+        DRAFT_VERSION_ONLY_PENDING
+    )
+    assert draft_version.validate_bump_manifesto(bump_manifesto_only_pending)
+
+    bump_manifesto_added_pending = DatastoreVersion.model_validate(
+        DRAFT_VERSION_ADDED_PENDING
+    )
+    assert not draft_version.validate_bump_manifesto(
+        bump_manifesto_added_pending
+    )
+
+
+def test_draft_version_release_pending():
+    draft_version = DraftVersion.model_validate(DRAFT_VERSION_IDENTICAL)
+    release_time = draft_version.release_time
+    version = draft_version.version
+    updates, update_type = draft_version.release_pending()
+    assert update_type == "MINOR"
+    assert [
+        update.model_dump(by_alias=True, exclude_none=True)
+        for update in updates
+    ] == [
+        {
+            "description": "Første publisering",
+            "name": "BRUTTO_INNTEKT",
+            "operation": "ADD",
+            "releaseStatus": "PENDING_RELEASE",
+        }
+    ]
+    assert release_time != draft_version.release_time
+    assert version != draft_version.version
+    with pytest.raises(BumpException):
+        draft_version.release_pending()
+
+
+def test_set_draft_release_status():
+    draft_version = DraftVersion.model_validate(DRAFT_VERSION_IDENTICAL)
+    release_time = draft_version.release_time
+    version = draft_version.version
+    draft_version.set_draft_release_status("UTDANNING", "PENDING_RELEASE")
+    for update in draft_version:
+        if update.name == "UTDANNING":
+            assert update.release_status == "PENDING_RELEASE"
+    assert release_time != draft_version.release_time
+    assert version != draft_version.version
+
+    draft_version.set_draft_release_status("UTDANNING", "DRAFT")
+    for update in draft_version:
+        if update.name == "UTDANNING":
+            assert update.release_status == "DRAFT"

tests/unit/adapter/fs/model/test_datastore_version.py

@@ -0,0 +1,73 @@
+import json
+import os
+import shutil
+from pathlib import Path
+
+from job_executor.adapter.fs import LocalStorageAdapter
+from job_executor.adapter.fs.models.datastore_versions import (
+    DatastoreVersions,
+    DataStructureUpdate,
+)
+
+
+def load_json(file_path):
+    return json.load(open(file_path, encoding="utf"))
+
+
+DATASTORE_DIR = "tests/unit/resources/adapter/fs/TEST_DATASTORE"
+METADATA_DIR = f"{DATASTORE_DIR}/datastore"
+local_storage = LocalStorageAdapter(Path(DATASTORE_DIR))
+DATASTORE_VERSIONS_PATH = f"{METADATA_DIR}/datastore_versions.json"
+
+
+def setup_function():
+    if os.path.isdir("tests/resources_backup"):
+        shutil.rmtree("tests/resources_backup")
+
+    shutil.copytree("tests/resources", "tests/resources_backup")
+
+
+def teardown_function():
+    shutil.rmtree("tests/resources")
+    shutil.move("tests/resources_backup", "tests/resources")
+
+
+def test_datastore_versions():
+    datastore_versions = DatastoreVersions.model_validate(
+        local_storage.datastore_dir.get_datastore_versions()
+    )
+    assert datastore_versions.model_dump(
+        by_alias=True, exclude_none=True
+    ) == load_json(DATASTORE_VERSIONS_PATH)
+
+
+def test_add_new_release_version():
+    datastore_versions = DatastoreVersions.model_validate(
+        local_storage.datastore_dir.get_datastore_versions()
+    )
+    datastore_versions.add_new_release_version(
+        [
+            DataStructureUpdate(
+                name="NEW_DATASET",
+                description="Første publisering",
+                operation="ADD",
+                release_status="PENDING_RELEASE",
+            )
+        ],
+        "new datastore version",
+        "MAJOR",
+    )
+    assert len(datastore_versions.versions) == 3
+
+
+def test_get_dataset_release_status():
+    datastore_versions = DatastoreVersions.model_validate(
+        local_storage.datastore_dir.get_datastore_versions()
+    )
+    assert (
+        datastore_versions.get_dataset_release_status("SIVSTAND") == "RELEASED"
+    )
+    assert datastore_versions.get_dataset_release_status("INNTEKT") == "DELETED"
+    assert (
+        datastore_versions.get_dataset_release_status("DOES_NOT_EXIST") is None
+    )