remove absolute path dependencies in tests [#50]

src/fidesops/core/config.py

@@ -3,13 +3,10 @@
 import hashlib
 import logging
 import os
-
-from typing import Dict, List, Optional, Union, Tuple, MutableMapping, Any
-
+from typing import Dict, List, Optional, Union, Tuple, Any, MutableMapping
 
 import bcrypt
 import toml
-
 from pydantic import (
     AnyHttpUrl,
     BaseSettings,
@@ -180,63 +177,71 @@ class Config:  # pylint: disable=C0115
         case_sensitive = True
 
 
-def load_toml(
-    file_name: str, config_path: str = ""
-) -> Optional[MutableMapping[str, Any]]:
-    """Load a (raw) toml file and return a dictionary of settings."""
-    possible_config_locations = [
-        config_path,
-        os.path.join(os.curdir, file_name),
-        os.path.join(os.pardir, file_name),
-        os.path.join(os.path.expanduser("~"), file_name),
+def load_file(file_name: str) -> str:
+    """Load a file and from the first matching location.
+
+    In order, will check:
+    - A path set at ENV variable FIDESOPS_CONFIG_PATH
+    - The current directory
+    - The parent directory
+    - users home (~) directory
+
+    raises FileNotFound if none is found
+    """
+
+    possible_directories = [
+        os.getenv("FIDESOPS_CONFIG_PATH"),
+        os.curdir,
+        os.pardir,
+        os.path.expanduser("~"),
     ]
 
-    # if FIDESOPS_CONFIG_PATH is set that should be used first:
-    fides_ops_config_path = os.getenv("FIDESOPS_CONFIG_PATH")
-    if fides_ops_config_path:
-        possible_config_locations.insert(
-            0, os.path.join(fides_ops_config_path, file_name)
-        )
-    for file_location in possible_config_locations:
-        if file_location != "" and os.path.isfile(file_location):
-            try:
-                settings = toml.load(file_location)
-                logger.info(f"Config loaded from {file_location}")
-                return settings
-            except IOError:
-                logger.info(f"Error reading config file from {file_location}")
-            break
+    directories: List[str] = [d for d in possible_directories if d]
+
+    for dir_str in directories:
+        possible_location = os.path.join(dir_str, file_name)
+        if possible_location and os.path.isfile(possible_location):
+            logger.info("Loading file %s from %s", file_name, dir_str)
+            return possible_location
+        logger.debug("%s not found at %s", file_name, dir_str)
+    raise FileNotFoundError
 
-    return None
 
+def load_toml(file_name: str) -> MutableMapping[str, Any]:
+    """
+    Load toml file from possible locations specified in load_file.
 
-def get_config(config_path: str = "") -> FidesopsConfig:
+    Will raise FileNotFoundError or ValidationError on missing or
+    bad file
+    """
+    return toml.load(load_file(file_name))
+
+
+def get_config() -> FidesopsConfig:
     """
     Attempt to read config file from:
-    a) passed in configuration, if it exists
-    b) env var FIDESOPS_CONFIG_PATH
-    c) local directory
+    a) env var FIDESOPS_CONFIG_PATH
+    b) local directory
+    c) parent directory
     d) home directory
     This will fail on the first encountered bad conf file.
     """
-    settings = load_toml("fidesops.toml", config_path)
-    if settings is not None:
-        the_config = FidesopsConfig.parse_obj(settings)
-    else:
+    try:
+        return FidesopsConfig.parse_obj(load_toml("fidesops.toml"))
+    except (FileNotFoundError, ValidationError) as e:
+        logger.warning("fidesops.toml could not be loaded: %s", e)
         # If no path is specified Pydantic will attempt to read settings from
         # the environment. Default values will still be used if the matching
         # environment variable is not set.
         try:
-            the_config = FidesopsConfig()
+            return FidesopsConfig()
         except ValidationError as exc:
             logger.error(exc)
             # If FidesopsConfig is missing any required values Pydantic will throw
             # an ImportError. This means the config has not been correctly specified
             # so we can throw the missing config error.
             raise MissingConfig(exc.args[0])
 
-    return the_config
-
 
 CONFIG_KEY_ALLOWLIST = {
     "database": [

tests/api/v1/endpoints/test_privacy_request_endpoints.py

@@ -13,7 +13,9 @@
 import pytest
 from starlette.testclient import TestClient
 
-from fidesops.api.v1.endpoints.privacy_request_endpoints import EMBEDDED_EXECUTION_LOG_LIMIT
+from fidesops.api.v1.endpoints.privacy_request_endpoints import (
+    EMBEDDED_EXECUTION_LOG_LIMIT,
+)
 from fidesops.api.v1.urn_registry import (
     PRIVACY_REQUESTS,
     V1_URL_PREFIX,
@@ -29,7 +31,11 @@
     get_db_session,
 )
 from fidesops.models.client import ClientDetail
-from fidesops.models.privacy_request import PrivacyRequest, ExecutionLog, ExecutionLogStatus
+from fidesops.models.privacy_request import (
+    PrivacyRequest,
+    ExecutionLog,
+    ExecutionLogStatus,
+)
 from fidesops.models.policy import DataCategory, ActionType
 from fidesops.schemas.dataset import DryRunDatasetResponse
 from fidesops.util.cache import get_identity_cache_key
@@ -255,7 +261,9 @@ def test_create_and_process_access_request(
             assert results[key] is not None
             assert results[key] != {}
 
-        result_key_prefix = f"EN_{pr.id}__access_request__postgres_example_test_dataset:"
+        result_key_prefix = (
+            f"EN_{pr.id}__access_request__postgres_example_test_dataset:"
+        )
         customer_key = result_key_prefix + "customer"
         assert results[customer_key][0]["email"] == customer_email
 
@@ -750,12 +758,12 @@ def test_verbose_privacy_requests(
         assert resp == expected_resp
 
     def test_verbose_privacy_request_embed_limit(
-            self,
-            db,
-            api_client: TestClient,
-            generate_auth_header,
-            privacy_request: PrivacyRequest,
-            url,
+        self,
+        db,
+        api_client: TestClient,
+        generate_auth_header,
+        privacy_request: PrivacyRequest,
+        url,
     ):
         for i in range(0, EMBEDDED_EXECUTION_LOG_LIMIT + 10):
             ExecutionLog.create(
@@ -775,8 +783,13 @@ def test_verbose_privacy_request_embed_limit(
         assert 200 == response.status_code
 
         resp = response.json()
-        assert len(resp["items"][0]["results"]["my-postgres-db"]) == EMBEDDED_EXECUTION_LOG_LIMIT
-        db.query(ExecutionLog).filter(ExecutionLog.privacy_request_id==privacy_request.id).delete()
+        assert (
+            len(resp["items"][0]["results"]["my-postgres-db"])
+            == EMBEDDED_EXECUTION_LOG_LIMIT
+        )
+        db.query(ExecutionLog).filter(
+            ExecutionLog.privacy_request_id == privacy_request.id
+        ).delete()
 
 
 class TestGetExecutionLogs:

tests/fixtures.py

@@ -11,6 +11,7 @@
 import logging
 from faker import Faker
 
+from fidesops.core.config import load_file
 from fidesops.models.client import ClientDetail
 from fidesops.models.connectionconfig import (
     ConnectionConfig,
@@ -606,7 +607,8 @@ def example_datasets() -> List[Dict]:
         "data/dataset/mongo_example_test_dataset.yml",
     ]
     for filename in example_filenames:
-        with open(filename, "r") as file:
+        yaml_file = load_file(filename)
+        with open(yaml_file, "r") as file:
             example_datasets += yaml.safe_load(file).get("dataset", [])
     return example_datasets
 

tests/integration_tests/test_sql_task.py

@@ -155,7 +155,9 @@ def test_sql_erasure_task(db, postgres_inserts, integration_postgres_config):
 @pytest.mark.integration
 def test_sql_access_request_task(db, policy, integration_postgres_config) -> None:
 
-    privacy_request = PrivacyRequest(id=f"test_sql_access_request_task_{random.randint(0, 1000)}")
+    privacy_request = PrivacyRequest(
+        id=f"test_sql_access_request_task_{random.randint(0, 1000)}"
+    )
 
     v = graph_task.run_access_request(
         privacy_request,

tests/service/masking/strategy/test_masking_strategy_null.py

@@ -1,6 +1,7 @@
 from fidesops.schemas.masking.masking_configuration import NullMaskingConfiguration
-from fidesops.service.masking.strategy.masking_strategy_nullify import NullMaskingStrategy
-
+from fidesops.service.masking.strategy.masking_strategy_nullify import (
+    NullMaskingStrategy,
+)
 
 
 def test_mask_with_value():