chore: respect mypy strict

src/cfglock/docs_server.py

@@ -26,7 +26,7 @@
 waitForReload();
 </script>
 """
-# TODO: refactor this file at a later stage!
+# TODO: refactor this file at a later stage! This file might be removed to be used with something else at some later stage.
 
 
 class ReloadState:

src/cfglock/helper.py.save

@@ -0,0 +1,186 @@
+import filecmp
+import json
+import os
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import ClassVar, Any
+from collections.abc import Mapping
+
+import typer
+import yaml
+from dotenv import load_dotenv
+
+from cfglock.validator import (
+    ConfigLockError,
+    ValidationContext,
+    keys_to_ignore,
+    walk_yaml_in_order,
+    walk_yaml_with_no_order,
+)
+
+load_dotenv()
+CONFIG_LOG_FILE_PATH: str = os.environ.get("CONFIG_LOG_FILE_PATH", "config.lock.json")
+
+# TODO: remove typer.echo at some point!
+
+
+type MappingValue = dict[str, Any]
+
+
+class FileReader[T: Mapping[str, Any]](ABC):
+    @abstractmethod
+    def read(self, file_path: str) -> T:
+        """Reads the appropriate file."""
+        pass
+
+
+class YamlReader(FileReader[MappingValue]):
+    def read(self, file_path: str) -> MappingValue:
+        # TODO: i belive thsi can also return back a list with no dict actually...
+        with open(file_path, "r") as f:
+            data = yaml.safe_load(f) or {}
+
+            if not isinstance(data, dict):
+                raise TypeError(
+                    f"Expected YAML object/dict in {file_path}, got {type(data).__name__}"
+                )
+        typer.echo("Successfully read file")
+        return data
+
+
+class JsonReader(FileReader[MappingValue]):
+    def read(self, file_path: str) -> MappingValue:
+        with open(file_path, "r") as f:
+            data = json.load(f) or {}
+
+            if not isinstance(data, dict):
+                raise TypeError(
+                    f"Expected JSON in {file_path}, got {type(data).__name__}"
+                )
+
+        typer.echo("Successfully read file")
+        return data
+
+
+class FileReaderFactory:
+    reader: ClassVar[dict[str, FileReader[MappingValue]]] = {
+        ".yaml": YamlReader(),
+        ".yml": YamlReader(),
+        ".json": JsonReader(),
+    }
+
+    @classmethod
+    def load(cls, file_path: str) -> MappingValue:
+        """Loads the relevant filetype.
+        args
+            file_path(str): str representation of file path.
+        returns
+            a dictionary with the contents of the file
+        """
+
+        path = Path(file_path)
+        suffix = path.suffix.lower()
+
+        reader = cls.reader.get(suffix)
+
+        if not reader:
+            typer.echo(
+                f"File not suppported: {suffix}. Use .yaml, .yml, or .json.",
+                err=True,
+            )
+            raise ValueError("Error not able to read the file")
+
+        typer.echo(f"Reading {file_path}...")
+        return reader.read(file_path)
+
+
+def check_file_identicality(
+    file_path: str, config_file_path: str = CONFIG_LOG_FILE_PATH
+) -> bool:
+    """Checks if files are identical, if they are it returns True, False otherwise"""
+    try:
+        a = FileReaderFactory.load(file_path)
+        b = FileReaderFactory.load(config_file_path)
+
+        # metadata we do not care about
+        if isinstance(a, dict):
+            a = {k: v for k, v in a.items() if k not in keys_to_ignore}
+        if isinstance(b, dict):
+            b = {k: v for k, v in b.items() if k not in keys_to_ignore}
+
+        if a == b:
+            return True
+        filecmp.clear_cache()
+        res = filecmp.cmp(file_path, config_file_path, shallow=False)
+        return res
+    except OSError:
+        filecmp.clear_cache()
+        res = filecmp.cmp(file_path, config_file_path, shallow=False)
+        return res
+
+
+def check_file_exists(file_path: str | None = None) -> bool:
+    if file_path is None:
+        file_path = CONFIG_LOG_FILE_PATH
+
+    path_parent = Path(file_path).parent
+    path = path_parent / CONFIG_LOG_FILE_PATH
+    exists = path.exists()
+    if not exists:
+        typer.echo(f"The path does not exist: {path}")
+    return exists
+
+
+def write_json(data: MappingValue, file_path: str | None = None) -> None:
+    data.update({"version": 1})
+    if file_path is None:
+        file_path = CONFIG_LOG_FILE_PATH
+    parent_path = Path(file_path)
+    new_path = parent_path / CONFIG_LOG_FILE_PATH
+    try:
+        with open(new_path, "w") as json_file:
+            json.dump(data, json_file, indent=4)
+    except TypeError as exc:
+        raise TypeError(f"Data could not be serialized to JSON: {exc}") from exc
+    except OSError as exc:
+        raise OSError(f"Could not write JSON file at {file_path}: {exc}") from exc
+    else:
+        typer.echo("Successfully wrote file")
+
+
+def check_compatibility(
+    new_file_path: str,
+    curr_file_path: str | None = None,
+    *,
+    order_matters: bool = False,
+) -> None:
+    """ ""
+    Check compatiblity for two files given the file paths
+    1) Keys must be same as previous keys, and order_matters can determine if the order also matters
+    2) Values must have the same types as previously
+    3) Adding new entries is allowed
+    4) Deleting entries is not allowed
+    """
+    # the lock file
+    current_file_path = None
+    if curr_file_path is None:
+        current_file_path = Path(CONFIG_LOG_FILE_PATH)
+    else:
+        parent_new = Path(new_file_path).parent
+        current_file_path = parent_new / "config.lock.json"
+    context = ValidationContext(
+        new_path=new_file_path,
+        current_path=current_file_path,
+        order_matters=order_matters,
+    )
+    new_data = FileReaderFactory.load(new_file_path)
+
+    try:
+        current_data = FileReaderFactory.load(current_file_path)
+    except FileNotFoundError:
+        raise ConfigLockError("lock file was not found, please use init")
+
+    if order_matters:
+        walk_yaml_in_order(current_data, new_data, context)
+    else:
+        walk_yaml_with_no_order(current_data, new_data, context)

src/cfglock/validator.py

@@ -40,14 +40,14 @@ def __init__(
         actual_value,
         message="Validation Failed",
         order_matters=False,
-    ):
+    ) -> None:
         super().__init__(message, error_code=100)
         self.path = path
         self.expected_value = expected_value
         self.actual_value = actual_value
         self.order_matters = order_matters
 
-    def __str__(self):
+    def __str__(self) -> str:
         base_msg = f""" 
         In path: {self.path} 
         Expected: {self.expected_value}
@@ -131,13 +131,15 @@ def accept_new_keys(
         )
 
 
-def accept_new_value(current_value, new_value, context: ValidationContext) -> None:
+def accept_new_value[T](
+    current_value: T, new_value: T, context: ValidationContext
+) -> None:
     """
     General logic for accepting new values:
     1) the type of the new_value cannot be different than the type of the current_value
     """
 
-    def typed_value(value):
+    def typed_value(value) -> str:
         return f"<{type(value).__name__}> with value: {value!r}"
 
     if type(current_value) is not type(new_value):

tests/integration_test/test_init.py

@@ -5,7 +5,7 @@
 from cfglock import cli
 
 
-def test_init_correct(runner_setup, fixture_dir):
+def test_init_correct(runner_setup, fixture_dir) -> None:
     fixture = fixture_dir / "config.yaml"
 
     with open(fixture, "r", encoding="utf-8") as f:

tests/integration_test/test_lock.py

@@ -6,20 +6,20 @@
 from cfglock.validator import ValidationError
 
 
-def helper_ok(result, output):
+def helper_ok(result, output) -> None:
     print(result.exception)
     assert result.exit_code == 0
     assert output["version"] == 1
     # interesting case, false -> False (real boolean type)
     assert not output["app_settings"]["maintenance_mode"]
 
 
-def helper_fail(result, error):
+def helper_fail(result, error) -> None:
     assert result.exit_code == 1
     assert isinstance(result.exception, error)
 
 
-def test_lock_correct(runner_with_lockfile, fixture_dir):
+def test_lock_correct(runner_with_lockfile, fixture_dir) -> None:
     fixture = fixture_dir / "config.yaml"
 
     with open(fixture, "r", encoding="utf-8") as f:
@@ -40,7 +40,7 @@ def test_lock_correct(runner_with_lockfile, fixture_dir):
     assert "environment" in output["app_settings"]
 
 
-def test_lock_string_change_ok(runner_with_lock_file_setup, fixture_dir):
+def test_lock_string_change_ok(runner_with_lock_file_setup, fixture_dir) -> None:
     fixture = fixture_dir / "config_string_change.yaml"
 
     with open(fixture, "r", encoding="utf-8") as f:
@@ -58,7 +58,7 @@ def test_lock_string_change_ok(runner_with_lock_file_setup, fixture_dir):
     assert "production" in output["app_settings"]["environment"]
 
 
-def test_lock_add_on_change_ok(runner_with_lock_file_setup, fixture_dir):
+def test_lock_add_on_change_ok(runner_with_lock_file_setup, fixture_dir) -> None:
     fixture = fixture_dir / "config_add_on_change.yaml"
 
     with open(fixture, "r", encoding="utf-8") as f:
@@ -76,7 +76,7 @@ def test_lock_add_on_change_ok(runner_with_lock_file_setup, fixture_dir):
     assert "ok" in output["app_settings"]["new_entry"]
 
 
-def test_lock_fail_namechange(runner_with_lock_file_setup, fixture_dir):
+def test_lock_fail_namechange(runner_with_lock_file_setup, fixture_dir) -> None:
     fixture = fixture_dir / "configbreak_namechange.yaml"
 
     with open(fixture, "r", encoding="utf-8") as f:
@@ -93,7 +93,7 @@ def test_lock_fail_namechange(runner_with_lock_file_setup, fixture_dir):
     # assert "found: name_changed" in str(result.exception).lower() TODO: fix this at some point!!
 
 
-def test_lock_fail_typechange(runner_with_lock_file_setup, fixture_dir):
+def test_lock_fail_typechange(runner_with_lock_file_setup, fixture_dir) -> None:
     fixture = fixture_dir / "configbreak_typechange.yaml"
 
     with open(fixture, "r", encoding="utf-8") as f:
@@ -110,7 +110,7 @@ def test_lock_fail_typechange(runner_with_lock_file_setup, fixture_dir):
     assert "found: <str> with value: 'false'" in str(result.exception).lower()
 
 
-def test_lock_fail_orderchange(runner_with_lock_file_setup, fixture_dir):
+def test_lock_fail_orderchange(runner_with_lock_file_setup, fixture_dir) -> None:
     fixture = fixture_dir / "configbreak_orderchange.yaml"
 
     with open(fixture, "r", encoding="utf-8") as f:

tests/integration_test/test_sync.py

@@ -6,7 +6,7 @@
 from cfglock.validator import ConfigLockError
 
 
-def test_sync_correct(runner_with_lock_file_setup, fixture_dir):
+def test_sync_correct(runner_with_lock_file_setup, fixture_dir) -> None:
     fixture = fixture_dir / "config.yaml"
 
     with open(fixture, "r", encoding="utf-8") as f:
@@ -27,7 +27,7 @@ def test_sync_correct(runner_with_lock_file_setup, fixture_dir):
     assert not output["app_settings"]["maintenance_mode"]
 
 
-def test_sync_outdated_source_file(runner_with_lock_file_setup, fixture_dir):
+def test_sync_outdated_source_file(runner_with_lock_file_setup, fixture_dir) -> None:
     fixture = fixture_dir / "configbreak_namechange.yaml"
 
     with open(fixture, "r", encoding="utf-8") as f:
@@ -43,7 +43,7 @@ def test_sync_outdated_source_file(runner_with_lock_file_setup, fixture_dir):
     assert "lock file is outdated" in str(result.exception).lower()
 
 
-def test_sync_outdated_lock_file(runner_with_lock_file_setup, fixture_dir):
+def test_sync_outdated_lock_file(runner_with_lock_file_setup, fixture_dir) -> None:
     fixture = fixture_dir / "config.yaml"
 
     with open("config.lock.json", "w", encoding="utf-8") as f: