First unit tests

.github/workflows/pull_request.yaml

@@ -1,6 +1,6 @@
 name: Pull Request
 on:
-  - pull_request
+  pull_request: {}
 
 jobs:
   test:
@@ -26,3 +26,6 @@ jobs:
 
       - name: Pytest
         run: poetry run pytest
+
+      #- name: MyPy
+      #  run: mypy -m mkdocs_deploy

.gitignore

@@ -8,3 +8,4 @@ dist
 .pytest_cache
 .idea
 .coverage
+htmlcov

pyproject.toml

@@ -43,6 +43,7 @@ mypy = "^1.8.0"
 
 [tool.poetry.group.dev.dependencies]
 build = "^0.10.0"
+coverage = "^7.4.0"
 
 [tool.poetry-dynamic-versioning]
 enable = true
@@ -60,4 +61,9 @@ built_site = "dist/mkdocs_site"
 [tool.pytest.ini_options]
 testpaths = [
     "source/tests",
-]
+]
+
+[tool.coverage.run]
+source = [
+    "source/mkdocs_deploy",
+]

source/mkdocs_deploy/abstract.py

@@ -1,6 +1,7 @@
 import urllib.parse
 from abc import abstractmethod
-from typing import Callable, IO, Iterable, Optional, Protocol, Union
+from enum import Enum
+from typing import Callable, IO, Iterable, Optional, Protocol
 
 from .versions import DeploymentAlias, DeploymentSpec
 
@@ -13,6 +14,16 @@ class RedirectMechanismNotFound(Exception):
     pass
 
 
+class _DefaultVersionType(Enum):
+    DEFAULT_VERSION = 0
+
+
+DEFAULT_VERSION = _DefaultVersionType.DEFAULT_VERSION
+
+
+Version = str | _DefaultVersionType
+
+
 class Source(Protocol):
     """
     Source is where a site is loaded from.
@@ -39,6 +50,7 @@ def close(self) -> None:
         """
         Close any underlying resource handles
         """
+        return None
 
     def __enter__(self):
         """
@@ -78,7 +90,7 @@ def delete_version(self, version_id: str) -> None:
         """
 
     @abstractmethod
-    def upload_file(self, version_id: Union[str, type(...)], filename: str, file_obj: IO[bytes]) -> None:
+    def upload_file(self, version_id: Version, filename: str, file_obj: IO[bytes]) -> None:
         """
         Upload a file to the target
 
@@ -89,7 +101,7 @@ def upload_file(self, version_id: Union[str, type(...)], filename: str, file_obj
         """
 
     @abstractmethod
-    def download_file(self, version_id: Union[str, type(...)], filename: str) -> IO[bytes]:
+    def download_file(self, version_id: Version, filename: str) -> IO[bytes]:
         """
         Open a file handle to read content of a file
 
@@ -104,7 +116,7 @@ def download_file(self, version_id: Union[str, type(...)], filename: str) -> IO[
 
 
     @abstractmethod
-    def delete_file(self, version_id: Union[str, type(...)], filename: str) -> None:
+    def delete_file(self, version_id: Version, filename: str) -> None:
         """
         Delete a file, or mark it for deletion on close.
         :param version_id: The version to delete from
@@ -132,7 +144,7 @@ def close(self, success: bool = False) -> None:
         """
 
     @abstractmethod
-    def set_alias(self, alias_id: Union[str, type(...)], alias: Optional[DeploymentAlias]) -> None:
+    def set_alias(self, alias_id: Version, alias: Optional[DeploymentAlias]) -> None:
         """
         Create or delete an alias.
 
@@ -184,26 +196,26 @@ class RedirectMechanism(Protocol):
     """
 
     @abstractmethod
-    def create_redirect(self, session: TargetSession, alias: Union[str, type(...)], version_id: str) -> None:
+    def create_redirect(self, session: TargetSession, alias: Version, version_id: str) -> None:
         """
         Create a redirect
 
         :param session: The TargetSession to apply changes to
-        :param alias: The new alias to create.  If ``...`` is passed, then a redirect from the root is created.  IE: ""
+        :param alias: The new alias to create.  If ``None`` is passed, then a redirect from the root is created.  IE: ""
         defines what the default version is.
         :param version_id: The version to redirect to.
         """
 
     @abstractmethod
-    def delete_redirect(self, session: TargetSession, alias: Union[str, type(...)]) -> None:
+    def delete_redirect(self, session: TargetSession, alias:Version) -> None:
         """
         Delete the named redirect.
 
         :param session: The TargetSession to apply changes to
         :param alias: The alias to delete. ``...`` is the default redirect.
         """
 
-    def refresh_redirect(self, session: TargetSession, alias: Union[str, type(...)], version_id: str) -> None:
+    def refresh_redirect(self, session: TargetSession, alias: Version, version_id: str) -> None:
         """
         Called to ensure all redirects still work after a version has been altered.
 

source/mkdocs_deploy/actions.py

@@ -8,9 +8,9 @@
 """
 import importlib.metadata
 import logging
-from typing import Iterable, Optional, Union
+from typing import Iterable, Optional
 
-from .abstract import Source, TargetSession, VersionNotFound
+from .abstract import DEFAULT_VERSION, Source, TargetSession, Version, VersionNotFound
 from .versions import DeploymentAlias
 
 _logger = logging.getLogger(__name__)
@@ -33,7 +33,7 @@ def load_plugins() -> None:
 
 
 
-def upload(source: Source, target: TargetSession, version_id: str, title: Optional[str]) -> None:
+def upload(source: Source, target: TargetSession, version_id: str, title: str | None) -> None:
     """
     Upload a file (to s3)
     :param source: The site to upload.  This may be a directory, or it may be zipped
@@ -82,7 +82,7 @@ def delete_version(target: TargetSession, version_id: str) -> None:
 
 
 def create_alias(
-    target: TargetSession, alias_id: Union[str, type(...)], version: str,  mechanisms: Optional[Iterable[str]] = None
+    target: TargetSession, alias_id: Version, version: str,  mechanisms: Iterable[str] | None = None
 ) -> None:
     """
     Create a new alias for a version.
@@ -103,7 +103,7 @@ def create_alias(
                 raise ValueError(f"LocalFileTreeTarget does not support redirect mechanism: {mechanism}")
 
     # Check if the alias already exists ...
-    # If mechanisms wasn't spefied use whatever is on the existing one.
+    # If mechanisms wasn't specified use whatever is on the existing one.
     deployment_spec = target.deployment_spec
     if alias_id in deployment_spec.versions:
         raise ValueError(f"Cannot create an alias with the same name as an existing version. "
@@ -152,9 +152,7 @@ def create_alias(
     target.set_alias(alias_id, alias)
 
 
-def delete_alias(
-    target: TargetSession, alias_id: Union[str, type(...)], mechanisms: Optional[Iterable[str]] = None
-) -> None:
+def delete_alias(target: TargetSession, alias_id: Version, mechanisms: Iterable[str] | None = None) -> None:
     """
     Delete an alias.
 
@@ -166,7 +164,7 @@ def delete_alias(
     :param mechanisms: Optional iterable of mechanisms to remove.
     """
     _logger.info("Deleting alias %s mechanism %s", alias_id, "default" if mechanisms is None else list(mechanisms))
-    if alias_id is ...:
+    if alias_id is DEFAULT_VERSION:
         alias = target.deployment_spec.default_version
         if alias is None:
             _logger.debug("Default alias not set")
@@ -198,9 +196,7 @@ def delete_alias(
         target.set_alias(alias_id, None)
 
 
-def refresh_alias(
-    target: TargetSession, alias_id: Union[str, type(...)], mechanisms: Optional[Iterable[str]] = None
-) -> None:
+def refresh_alias(target: TargetSession, alias_id: Version, mechanisms: Iterable[str] | None = None) -> None:
     """
     Refresh redirects.
 
@@ -211,14 +207,15 @@ def refresh_alias(
     :param mechanisms: Optional list of mechanisms to refresh.  If None (default) all will be refreshed.
     """
     _logger.info("Refreshing alias %s mechanisms %s", alias_id, "all" if mechanisms is None else list(mechanisms))
-    if alias_id is ...:
+    if alias_id is DEFAULT_VERSION:
         alias = target.deployment_spec.default_version
     else:
         alias = target.deployment_spec.aliases.get(alias_id, None)
     if alias is None:
         _logger.warning("Cannot refresh alias %s, it doesn't exist", alias_id)
+        return
     if mechanisms is not None:
-        to_refresh = [mechanism for mechanism in mechanisms if mechanism in alias.redirect_mechanisms]
+        to_refresh = {mechanism for mechanism in mechanisms if mechanism in alias.redirect_mechanisms}
     else:
         to_refresh = alias.redirect_mechanisms
     available_mechanisms = target.available_redirect_mechanisms

source/mkdocs_deploy/plugins/local_filesystem.py

@@ -55,14 +55,17 @@ def __init__(self, file_path: Union[Path, IO[bytes]], prefix: str = "site/"):
             self._tar_file = tarfile.open(name=file_path, mode="r")
         else:
             self._tar_file = tarfile.open(fileobj=file_path, mode="r")
+        self._file_path = file_path
 
     def iter_files(self) -> Iterable[str]:
         for file in self._tar_file.getmembers():
             if file.isreg() and file.name.startswith(self._prefix):
                 yield file.name[len(self._prefix):]
 
     def open_file_for_read(self, filename: str) -> IO[bytes]:
-        return self._tar_file.extractfile(self._prefix + filename)
+        result = self._tar_file.extractfile(self._prefix + filename)
+        if result is None:
+            raise RuntimeError(f"Requested file is not a regular file: {filename} in {self._file_path}")
 
     def close(self):
         self._tar_file.close()
@@ -131,10 +134,9 @@ def open_file_obj_source(file: IO[bytes]) -> abstract.Source:
 
 
 def _path_from_url(url: str) -> Path:
-    url = urllib.parse.urlparse(url)
-    if url.path.startswith("/"):
-        return Path(url.path[1:])
-    return Path(url.path)
+    if "://" in url:
+        return Path((url.path or "")[1:])
+    return Path(url)
 
 
 class LocalFileTreeTargetSession(abstract.TargetSession):
@@ -179,7 +181,8 @@ def close(self, success: bool = False) -> None:
             for file_name, content in shared_implementations.generate_meta_data(self._deployment_spec).items():
                 with open(self._path_for_file(..., file_name), "wb") as file:
                     file.write(content)
-
+# PosixPath('/Users/philip/Documents/Development/MkdocsDeploy/private/var/folders/nb/9f9993hs2yg3gjs966_ltd8c0000gn/T/pytest-of-philip/pytest-16/test_upload0/mock_target')
+# PosixPath('/Users/philip/Documents/Development/MkdocsDeploy/private/var/folders/nb/9f9993hs2yg3gjs966_ltd8c0000gn/T/pytest-of-philip/pytest-16/test_upload0/mock_target/deployments.json')
     def iter_files(self, version_id: str) -> Iterable[str]:
         def _iter_files(file_path: Path):
             try:
@@ -257,8 +260,9 @@ def _path_for_file(self, version_id: Union[str, type(...)], filename: str = "")
         elif version_id not in self._deployment_spec.versions and version_id not in self._deployment_spec.aliases:
             raise abstract.VersionNotFound(version_id)
         result = Path(self._target_path, *filename.split("/"))
-        if result.relative_to(self._target_path).parts[0] == "..":
-            raise ValueError(f"Refusing to operate on the site: {result} not in {self._target_path}")
+        # Raise a ValueError if the result is above the base path
+        result.relative_to(self._target_path)
+        return result
 
     def _check_version_exists(self, version_id: Union[str, type(...)]) -> None:
         if version_id is ...:

source/mkdocs_deploy/versions.py

@@ -28,7 +28,7 @@ class DeploymentVersion(pydantic.BaseModel):
 
     @pydantic.root_validator()
     def _default_title(cls, values: dict):
-        if values["title"] is None:
+        if values.get("title", None) is None:
             values["title"] = values["version_id"]
         return values