Add detail_context and PulpMasterContext

Author: mdellweg

CHANGES/pulp-glue/+cast.feature

@@ -0,0 +1 @@
+Added `detail_context` to master-detail contexts.

pulp-glue/pulp_glue/common/context.py

@@ -1304,7 +1304,36 @@ def needs_capability(self, capability: str) -> None:
             )
 
 
-class PulpRemoteContext(PulpEntityContext):
+class PulpMasterContext(PulpEntityContext):
+    TYPE_REGISTRY: t.Final[t.ClassVar[t.Dict[str, t.Type["t.Self"]]]]
+
+    def __init_subclass__(cls, **kwargs: t.Any) -> None:
+        super().__init_subclass__(**kwargs)
+        if not hasattr(cls, "RESOURCE_TYPE"):
+            cls.TYPE_REGISTRY = {}
+        elif hasattr(cls, "PLUGIN"):
+            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
+
+    def detail_context(self, pulp_href: str) -> "t.Self":
+        """
+        Provide a detail context for a matching href.
+        """
+        m = re.search(self.HREF_PATTERN, pulp_href)
+        if m is None:
+            raise PulpException(f"'{pulp_href}' is not an href for {self.ENTITY}.")
+        plugin = m.group("plugin")
+        resource_type = m.group("resource_type")
+        try:
+            detail_class = self.TYPE_REGISTRY[f"{plugin}:{resource_type}"]
+        except KeyError:
+            raise PulpException(
+                f"{self.ENTITY} with plugin '{plugin}' and"
+                f"resource type '{resource_type}' is unknown."
+            )
+        return detail_class(self.pulp_ctx, pulp_href=pulp_href)
+
+
+class PulpRemoteContext(PulpMasterContext):
     """
     Base class for remote contexts.
     """
@@ -1330,27 +1359,15 @@ class PulpRemoteContext(PulpEntityContext):
         "sock_read_timeout",
         "rate_limit",
     }
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpRemoteContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
 
-class PulpPublicationContext(PulpEntityContext):
+class PulpPublicationContext(PulpMasterContext):
     """Base class for publication contexts."""
 
     ENTITY = _("publication")
     ENTITIES = _("publications")
     ID_PREFIX = "publications"
     HREF_PATTERN = r"publications/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpPublicationContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
     def list(self, limit: int, offset: int, parameters: t.Dict[str, t.Any]) -> t.List[t.Any]:
         if parameters.get("repository") is not None:
@@ -1360,20 +1377,14 @@ def list(self, limit: int, offset: int, parameters: t.Dict[str, t.Any]) -> t.Lis
         return super().list(limit, offset, parameters)
 
 
-class PulpDistributionContext(PulpEntityContext):
+class PulpDistributionContext(PulpMasterContext):
     """Base class for distribution contexts."""
 
     ENTITY = _("distribution")
     ENTITIES = _("distributions")
     ID_PREFIX = "distributions"
     HREF_PATTERN = r"distributions/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
     NULLABLES = {"content_guard", "publication", "remote", "repository", "repository_version"}
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpDistributionContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
 
 class PulpRepositoryVersionContext(PulpEntityContext):
@@ -1432,7 +1443,7 @@ def repair(self) -> t.Any:
         return self.call("repair", parameters={self.HREF: self.pulp_href}, body={})
 
 
-class PulpRepositoryContext(PulpEntityContext):
+class PulpRepositoryContext(PulpMasterContext):
     """Base class for repository contexts."""
 
     ENTITY = _("repository")
@@ -1441,12 +1452,6 @@ class PulpRepositoryContext(PulpEntityContext):
     ID_PREFIX = "repositories"
     VERSION_CONTEXT: t.ClassVar[t.Type[PulpRepositoryVersionContext]] = PulpRepositoryVersionContext
     NULLABLES = {"description", "retain_repo_versions"}
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpRepositoryContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
     def get_version_context(
         self,
@@ -1555,19 +1560,14 @@ def reclaim(
         return self.call("reclaim_space_reclaim", body=body)
 
 
-class PulpContentContext(PulpEntityContext):
+class PulpContentContext(PulpMasterContext):
     """Base class for content contexts."""
 
     ENTITY = _("content")
     ENTITIES = _("content")
+    HREF_PATTERN = r"content/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
     ID_PREFIX = "content"
     HREF_PATTERN = r"content/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpContentContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
     def __init__(
         self,
@@ -1669,38 +1669,26 @@ def upload(
         return self.create(body=body)
 
 
-class PulpACSContext(PulpEntityContext):
+class PulpACSContext(PulpMasterContext):
     """Base class for ACS contexts."""
 
     ENTITY = _("ACS")
     ENTITIES = _("ACSes")
     HREF_PATTERN = r"acs/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
     ID_PREFIX = "acs"
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpACSContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
     def refresh(self, href: t.Optional[str] = None) -> t.Any:
         return self.call("refresh", parameters={self.HREF: href or self.pulp_href})
 
 
-class PulpContentGuardContext(PulpEntityContext):
+class PulpContentGuardContext(PulpMasterContext):
     """Base class for content guard contexts."""
 
     ENTITY = "content guard"
     ENTITIES = "content guards"
     ID_PREFIX = "contentguards"
     HREF_PATTERN = r"contentguards/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
     NULLABLES = {"description"}
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpContentGuardContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
 
 EntityFieldDefinition = t.Union[None, str, PulpEntityContext]

pulp-glue/tests/test_entity_context.py

@@ -0,0 +1,25 @@
+import random
+import string
+import typing as t
+
+import pytest
+
+from pulp_glue.common.context import PulpContext, PulpRepositoryContext
+from pulp_glue.file.context import PulpFileRepositoryContext
+
+pytestmark = pytest.mark.glue
+
+
+@pytest.fixture
+def file_repository(pulp_ctx: PulpContext) -> t.Dict[str, t.Any]:
+    name = "".join(random.choices(string.ascii_letters, k=8))
+    file_repository_ctx = PulpFileRepositoryContext(pulp_ctx)
+    yield file_repository_ctx.create(body={"name": name})
+    file_repository_ctx.delete()
+
+
+def test_detail_context(pulp_ctx: PulpContext, file_repository: t.Dict[str, t.Any]) -> None:
+    master_ctx = PulpRepositoryContext(pulp_ctx)
+    detail_ctx = master_ctx.detail_context(pulp_href=file_repository["pulp_href"])
+    assert isinstance(detail_ctx, PulpFileRepositoryContext)
+    assert detail_ctx.entity["name"] == file_repository["name"]