Actions fully unit tested

Author: couling

pyproject.toml

@@ -40,6 +40,7 @@ mypy = "^1.8.0"
 [tool.poetry.plugins."mkdocs_deploy.plugins"]
 "local" = "mkdocs_deploy.plugins.local_filesystem:enable_plugin"
 "s3" = "mkdocs_deploy.plugins.aws_s3:enable_plugin"
+"html-redirect" = "mkdocs_deploy.plugins.html_redirect:enable_plugin"
 
 [tool.poetry.group.dev.dependencies]
 build = "^0.10.0"

source/mkdocs_deploy/abstract.py

@@ -17,6 +17,9 @@ class RedirectMechanismNotFound(Exception):
 class _DefaultVersionType(Enum):
     DEFAULT_VERSION = 0
 
+    def __repr__(self) -> str:
+        return self.name
+
 
 DEFAULT_VERSION = _DefaultVersionType.DEFAULT_VERSION
 
@@ -268,4 +271,28 @@ def target_for_url(target_url: str) -> Target:
     :return:
     """
     handler = _TARGETS[urllib.parse.urlparse(target_url).scheme]
-    return handler(target_url)
+    return handler(target_url)
+
+
+_SHARED_REDIRECT_MECHANISMS: dict[str, RedirectMechanism] = {}
+
+
+def get_redirect_mechanisms(session: TargetSession) -> dict[str, RedirectMechanism]:
+    """Get all available redirect mechanisms for a target session
+
+    Unlike the property returned by the target session itself, this will also include shared redirect mechanisms.
+    """
+    result = _SHARED_REDIRECT_MECHANISMS.copy()
+    result.update(session.available_redirect_mechanisms)
+    return result
+
+
+def register_shared_redirect_mechanism(mechanism_key: str, mechanism: RedirectMechanism) -> None:
+    """Register a redirect mechanism which can work with any target type from any plugin.
+
+    DO NOT use this to simply add a mechanism to your own plugin.  The target session should return mechanisms that
+    only work on that target.
+
+    There are believed to be very few of these: html is the only inbuilt mechanism.  This can work with any target
+    because it only generates html files and all target types support uploading html files."""
+    _SHARED_REDIRECT_MECHANISMS[mechanism_key] = mechanism

source/mkdocs_deploy/actions.py

@@ -8,9 +8,9 @@
 """
 import importlib.metadata
 import logging
-from typing import Iterable, Optional
+from typing import Collection
 
-from .abstract import DEFAULT_VERSION, Source, TargetSession, Version, VersionNotFound
+from .abstract import DEFAULT_VERSION, Source, TargetSession, Version, VersionNotFound, get_redirect_mechanisms
 from .versions import DeploymentAlias
 
 _logger = logging.getLogger(__name__)
@@ -71,6 +71,8 @@ def delete_version(target: TargetSession, version_id: str) -> None:
     :param version_id:
     """
     deployment_spec = target.deployment_spec
+    if deployment_spec.default_version is not None and deployment_spec.default_version.version_id == version_id:
+        delete_alias(target, DEFAULT_VERSION)
     for alias_id in deployment_spec.aliases_for_version(version_id):
         delete_alias(target, alias_id)
     _logger.info("Deleting version %s", version_id)
@@ -82,7 +84,7 @@ def delete_version(target: TargetSession, version_id: str) -> None:
 
 
 def create_alias(
-    target: TargetSession, alias_id: Version, version: str,  mechanisms: Iterable[str] | None = None
+    target: TargetSession, alias_id: Version, version: str,  mechanisms: Collection[str] | None = None
 ) -> None:
     """
     Create a new alias for a version.
@@ -93,66 +95,68 @@ def create_alias(
     :param target: The target session to create the alias on
     :param alias_id: The new alias id
     :param version: The version_id to point to
-    :param mechanisms: The named mechanisms to use.  If None then 'html' target will choose the mechanism.
+    :param mechanisms: The named mechanisms to use.  If None then all available mechanisms will be used.
     """
     # Check if the given mechanisms can be implemented by this target
-    available_redirect_mechanisms = target.available_redirect_mechanisms
+    available_redirect_mechanisms = get_redirect_mechanisms(target)
     if mechanisms is not None:
         for mechanism in mechanisms:
             if mechanism not in available_redirect_mechanisms:
-                raise ValueError(f"LocalFileTreeTarget does not support redirect mechanism: {mechanism}")
+                raise ValueError(f"{type(TargetSession).__name__} does not support redirect mechanism: {mechanism}")
 
     # Check if the alias already exists ...
     # 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. "
                          f"Delete the version first! Alias name: {alias_id}")
-    if alias_id is ... and deployment_spec.default_version is not None:
+    if alias_id is DEFAULT_VERSION and deployment_spec.default_version is not None:
         # This is the "default" alias
         alias = deployment_spec.default_version
-        if mechanisms is None:
-            mechanisms = alias.redirect_mechanisms
-    if alias_id in deployment_spec.aliases:
+    elif alias_id in deployment_spec.aliases:
         alias = deployment_spec.aliases[alias_id]
-        if mechanisms is None:
-            mechanisms = alias.redirect_mechanisms
     else:
         # No existing alias was found. Make a new one.
         alias = DeploymentAlias(version_id=version, redirect_mechanisms=set())
         target.set_alias(alias_id, alias)
-        # Must set the alias first or creating the mechanism will fail.
-        if mechanisms is None:
-            mechanisms = ["html"]
+
+    if mechanisms is None:
+        mechanisms = available_redirect_mechanisms.keys()
 
     _logger.info("Creating %s alias redirect %s to %s", ", ".join(mechanisms), alias_id, version)
     # Remove any redirect mechanisms to a different version that we are not going to replace
-    if alias.version_id != version:
-        for mechanism in alias.redirect_mechanisms.copy():
-            if mechanism not in mechanisms:
-                try:
-                    available_redirect_mechanisms[mechanism].delete_redirect(target, alias_id)
-                except KeyError:
-                    raise ValueError(f"LocalFileTreeTarget does not support redirect mechanism: {mechanism}.  "
-                                     f"Unable to remove redirect for {alias_id}-->{alias.version_id}")
-                alias.redirect_mechanisms.discard(mechanism)
-        alias.version_id = version
+    for mechanism in alias.redirect_mechanisms.copy():
+        if mechanism not in mechanisms:
+            try:
+                _logger.warning(
+                    "Implicitly deleting redirect %s mechanism %s to %s", alias_id, mechanism, alias.version_id
+                )
+                available_redirect_mechanisms[mechanism].delete_redirect(target, alias_id)
+            except KeyError:
+                raise ValueError(
+                    f"{type(TargetSession).__name__} does not support redirect mechanism: {mechanism}. "
+                    f"Unable to remove redirect for {alias_id}-->{alias.version_id}"
+                )
+            alias.redirect_mechanisms.discard(mechanism)
 
     # Create the redirects or refresh them to their new location.
     for mechanism in mechanisms:
         if mechanism in alias.redirect_mechanisms:
             if alias.version_id != version:
+                _logger.debug("Modifying %s mechanism %s from %s to %s", alias_id, mechanism, alias.version_id, version)
                 available_redirect_mechanisms[mechanism].refresh_redirect(target, alias_id, version)
             else:
                 _logger.debug("mechanism %s already in place, skipping", mechanism)
         else:
+            _logger.debug("Creating %s mechanism %s to %s", alias_id, mechanism, version)
             available_redirect_mechanisms[mechanism].create_redirect(target, alias_id, version)
             alias.redirect_mechanisms.add(mechanism)
 
+    alias.version_id = version
     target.set_alias(alias_id, alias)
 
 
-def delete_alias(target: TargetSession, alias_id: Version, mechanisms: Iterable[str] | None = None) -> None:
+def delete_alias(target: TargetSession, alias_id: Version, mechanisms: Collection[str] | None = None) -> None:
     """
     Delete an alias.
 
@@ -167,20 +171,20 @@ def delete_alias(target: TargetSession, alias_id: Version, mechanisms: Iterable[
     if alias_id is DEFAULT_VERSION:
         alias = target.deployment_spec.default_version
         if alias is None:
-            _logger.debug("Default alias not set")
+            _logger.warning("Cannot delete default alias as it is not set")
             return
     else:
         try:
             alias = target.deployment_spec.aliases[alias_id]
         except KeyError:
-            _logger.debug("Alias %s not set, skipping", alias_id)
+            _logger.warning("Cannot delete alias %s not set, it has not been set", alias_id)
             return
 
     if mechanisms is not None:
-        to_delete = [mechanism for mechanism in mechanisms if mechanism in alias.redirect_mechanisms]
+        to_delete: list | set = [mechanism for mechanism in mechanisms if mechanism in alias.redirect_mechanisms]
     else:
         to_delete = alias.redirect_mechanisms.copy()
-    available_mechanisms = target.available_redirect_mechanisms
+    available_mechanisms = get_redirect_mechanisms(target)
     for mechanism in to_delete:
         try:
             available_mechanisms[mechanism].delete_redirect(
@@ -196,7 +200,7 @@ def delete_alias(target: TargetSession, alias_id: Version, mechanisms: Iterable[
         target.set_alias(alias_id, None)
 
 
-def refresh_alias(target: TargetSession, alias_id: Version, mechanisms: Iterable[str] | None = None) -> None:
+def refresh_alias(target: TargetSession, alias_id: Version, mechanisms: Collection[str] | None = None) -> None:
     """
     Refresh redirects.
 
@@ -218,7 +222,7 @@ def refresh_alias(target: TargetSession, alias_id: Version, mechanisms: Iterable
         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
+    available_mechanisms = get_redirect_mechanisms(target)
     for mechanism in to_refresh:
         available_mechanisms[mechanism].refresh_redirect(
             session=target,

source/mkdocs_deploy/plugins/aws_s3.py

@@ -18,10 +18,6 @@
 logging.getLogger("s3transfer").setLevel("INFO")
 
 
-AWS_S3_REDIRECT_MECHANISMS = shared_implementations.SHARED_REDIRECT_MECHANISMS.copy()
-# TODO add S3 redirect
-
-
 def enable_plugin() -> None:
     """
     Enables the plugin.
@@ -179,7 +175,9 @@ def set_alias(self, alias_id: Union[str, type(...)], alias: Optional[versions.De
 
     @property
     def available_redirect_mechanisms(self) -> dict[str, abstract.RedirectMechanism]:
-        return AWS_S3_REDIRECT_MECHANISMS.copy()
+        # TODO add S3 redirect
+        # TODO cloud front redirect rules
+        return {}
 
     @property
     def deployment_spec(self) -> versions.DeploymentSpec:

source/mkdocs_deploy/plugins/html_redirect.py

@@ -0,0 +1,74 @@
+from io import BytesIO
+
+from ..abstract import DEFAULT_VERSION, RedirectMechanism, TargetSession, Version, register_shared_redirect_mechanism
+
+
+def enable_plugin() -> None:
+    """Enables the plugin.
+
+    Registers HTML redirect mechanism"""
+    register_shared_redirect_mechanism("html", HtmlRedirect())
+
+
+class HtmlRedirect(RedirectMechanism):
+
+    def create_redirect(self, session: TargetSession, alias: Version, version_id: str) -> None:
+        if alias is DEFAULT_VERSION:
+            session.upload_file(
+                version_id=DEFAULT_VERSION,
+                filename="index.html",
+                file_obj=BytesIO(_HTML_REDIRECT_PATTERN.format(url=version_id+"/").encode("utf-8"))
+            )
+        else:
+            files_created = set()
+            for filename in session.iter_files(version_id):
+                if filename.endswith(".html") or filename.endswith(".htm"):
+                    if filename == "404.html" or filename.endswith("/404.htm"):
+                        session.upload_file(
+                            version_id=alias,
+                            filename=filename,
+                            file_obj=session.download_file(version_id=version_id, filename=filename)
+                        )
+                    else:
+                        parts = filename.split("/")
+                        depth = len(parts)
+                        url = ("../" * depth + version_id + "/" + "/".join(parts[:-1]))
+                        session.upload_file(
+                            version_id=alias, # Yes that's correct!
+                            filename=filename,
+                            file_obj=BytesIO(_HTML_REDIRECT_PATTERN.format(url=url).encode("utf-8"))
+                        )
+                    files_created.add(filename)
+            for filename in session.iter_files(alias):
+                if filename not in files_created and (filename.endswith("html") or filename.endswith("htm")):
+                    session.delete_file(alias, filename)
+
+    def refresh_redirect(self, session: TargetSession, alias: Version, version_id: str) -> None:
+        # create_redirect already cleans up so no need to explicitly delete the old one
+        self.create_redirect(session, alias, version_id)
+
+    def delete_redirect(self, session: TargetSession, alias: Version) -> None:
+        if alias is ...:
+            session.delete_file(version_id=..., filename="index.html")
+        else:
+            for filename in session.iter_files(alias):
+                if filename.endswith("html") or filename.endswith("htm"):
+                    session.delete_file(version_id=alias, filename=filename)
+
+_HTML_REDIRECT_PATTERN="""<!DOCTYPE html>
+<html>
+<head>
+  <meta charset="utf-8">
+  <title>Redirecting</title>
+  <noscript>
+    <meta http-equiv="refresh" content="1; url={url}" />
+  </noscript>
+  <script>
+    window.location.replace("{url}" + window.location.hash);
+  </script>
+</head>
+<body>
+  Redirecting to <a href="{url}">{url}</a>...
+</body>
+</html>
+"""

source/mkdocs_deploy/plugins/local_filesystem.py

@@ -10,8 +10,6 @@
 from .. import abstract, shared_implementations
 from ..versions import DeploymentAlias, DeploymentSpec, DeploymentVersion
 
-LOCAL_FILE_REDIRECT_MECHANISMS = shared_implementations.SHARED_REDIRECT_MECHANISMS.copy()
-
 _logger = logging.getLogger(__name__)
 
 
@@ -66,6 +64,7 @@ def open_file_for_read(self, filename: str) -> IO[bytes]:
         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}")
+        return result
 
     def close(self):
         self._tar_file.close()
@@ -233,8 +232,7 @@ def set_alias(self, alias_id: Union[str, type(...)], alias: Optional[DeploymentA
 
     @property
     def available_redirect_mechanisms(self) -> dict[str, abstract.RedirectMechanism]:
-        return LOCAL_FILE_REDIRECT_MECHANISMS.copy()
-
+        return {}
 
     @property
     def deployment_spec(self) -> DeploymentSpec: