test_manifest_traversal.py

#  Copyright 2020 Google LLC
#  Copyright 2020 EPAM Systems
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.

import json
import os
import sys

sys.path.append(f"{os.getenv('AIRFLOW_SRC_DIR')}/plugins")
sys.path.append(f"{os.getenv('AIRFLOW_SRC_DIR')}/dags")

import pytest
from file_paths import MANIFEST_WELLBORE_VALID_PATH, TRAVERSAL_WELLBORE_VALID_PATH, \
    MANIFEST_SEISMIC_TRACE_DATA_VALID_PATH, TRAVERSAL_SEISMIC_TRACE_DATA_VALID_PATH, MANIFEST_EMPTY_PATH, \
    TRAVERSAL_MANIFEST_EMPTY_PATH, MANIFEST_GENERIC_SCHEMA_PATH
from libs.exceptions import EmptyManifestError
from libs.traverse_manifest import ManifestTraversal


class TestManifestTraversal:
    @pytest.fixture
    def manifest_traversal(self, monkeypatch, manifest_file: str, manifest_schema_file: str):
        with open(manifest_file) as f:
            conf_manifest_file = json.load(f)
        with open(manifest_schema_file) as f:
            manifest_schema = json.load(f)
        traversal = ManifestTraversal(
            conf_manifest_file,
            manifest_schema
        )
        return traversal

    @pytest.mark.parametrize(
        "manifest_file,manifest_schema_file,traversal_manifest_file",
        [
            pytest.param(
                MANIFEST_WELLBORE_VALID_PATH,
                MANIFEST_GENERIC_SCHEMA_PATH,
                TRAVERSAL_WELLBORE_VALID_PATH,
                id="Valid manifest Wellore"),
            pytest.param(
                MANIFEST_SEISMIC_TRACE_DATA_VALID_PATH,
                MANIFEST_GENERIC_SCHEMA_PATH,
                TRAVERSAL_SEISMIC_TRACE_DATA_VALID_PATH,
                id="Valid manifest WPC"),
        ]
    )
    def test_traversal_manifest(self, monkeypatch, manifest_traversal, manifest_file: str,
                                manifest_schema_file: str, traversal_manifest_file: str):
        with open(traversal_manifest_file) as f:
            traversal_manifest = json.load(f)
        manifest_records = manifest_traversal.traverse_manifest()
        assert manifest_records == traversal_manifest

    @pytest.mark.parametrize(
        "manifest_file,manifest_schema_file,traversal_manifest_file",
        [
            pytest.param(
                MANIFEST_EMPTY_PATH,
                MANIFEST_GENERIC_SCHEMA_PATH,
                TRAVERSAL_MANIFEST_EMPTY_PATH,
                id="Empty manifest"),
        ]
    )
    def test_traversal_empty_manifest(self, monkeypatch,
                                      manifest_traversal,
                                      manifest_file: str,
                                      manifest_schema_file: str,
                                      traversal_manifest_file: str):
        with pytest.raises(EmptyManifestError):
            manifest_traversal.traverse_manifest()