Add detail_context and PulpMasterContext

[noissue]

Author: mdellweg

CHANGES/pulp-glue/+cast.feature

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

Makefile

@@ -34,7 +34,7 @@ tests/cli.toml:
 	@echo "In order to configure the tests to talk to your test server, you might need to edit $@ ."
 
 test: | tests/cli.toml
-	pytest -v tests
+	pytest -v tests pulp-glue/tests
 
 servedocs:
 	mkdocs serve -w docs_theme -w pulp-glue/pulp_glue -w pulpcore/cli/common/generic.py

pulp-glue/pulp_glue/common/context.py

@@ -1067,7 +1067,36 @@ def needs_capability(self, capability: str) -> None:
             )
 
 
-class PulpRemoteContext(PulpEntityContext):
+class PulpMasterContext(PulpEntityContext):
+    TYPE_REGISTRY: 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.
     """
@@ -1093,27 +1122,15 @@ class PulpRemoteContext(PulpEntityContext):
         "sock_read_timeout",
         "rate_limit",
     }
-    TYPE_REGISTRY: 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.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:
@@ -1123,20 +1140,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.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):
@@ -1198,7 +1209,7 @@ def repair(self, href: t.Optional[str] = None) -> t.Any:
         return self.call("repair", parameters={self.HREF: href or self.pulp_href}, body={})
 
 
-class PulpRepositoryContext(PulpEntityContext):
+class PulpRepositoryContext(PulpMasterContext):
     """Base class for repository contexts."""
 
     ENTITY = _("repository")
@@ -1207,12 +1218,6 @@ class PulpRepositoryContext(PulpEntityContext):
     ID_PREFIX = "repositories"
     VERSION_CONTEXT: t.ClassVar[t.Type[PulpRepositoryVersionContext]] = PulpRepositoryVersionContext
     NULLABLES = {"description", "retain_repo_versions"}
-    TYPE_REGISTRY: 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,
@@ -1324,18 +1329,13 @@ 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"
-    TYPE_REGISTRY: 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 upload(
         self,
@@ -1376,38 +1376,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.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.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/conftest.py

@@ -0,0 +1,26 @@
+import typing as t
+
+import pytest
+
+from pulp_glue.common.context import PulpContext
+from pulp_glue.common.openapi import BasicAuthProvider
+
+
+class PulpTestContext(PulpContext):
+    # TODO check if we can just make the base class ignore echo.
+    def echo(*args, **kwargs) -> None:
+        return
+
+
+@pytest.fixture
+def pulp_ctx(pulp_cli_settings: t.Dict[str, t.Dict[str, t.Any]]) -> PulpContext:
+    settings = pulp_cli_settings["cli"]
+    return PulpTestContext(
+        api_kwargs={
+            "base_url": settings["base_url"],
+            "auth_provider": BasicAuthProvider(settings.get("username"), settings.get("password")),
+        },
+        api_root=settings.get("api_root", "pulp/"),
+        background_tasks=False,
+        timeout=settings.get("timeout", 120),
+    )

tests/test_api_quirks.py renamed to pulp-glue/tests/test_api_quirks.py

@@ -1,27 +1,18 @@
-import typing as t
 from copy import deepcopy
 
 import pytest
+
 from pulp_glue.common.context import _REGISTERED_API_QUIRKS, PulpContext
 
+pytestmark = pytest.mark.glue
+
 
-@pytest.mark.glue
-def test_api_quirks_idempotent(
-    pulp_cli_settings: t.Tuple[t.Any, t.Dict[str, t.Dict[str, t.Any]]]
-) -> None:
+def test_api_quirks_idempotent(pulp_ctx: PulpContext) -> None:
     """
     Test, that the applied api quirks can be applied twice without failing.
 
     This can let us hope they will not fail once the api is fixed upstream.
     """
-    settings = pulp_cli_settings[1]["cli"]
-    pulp_ctx = PulpContext(
-        api_kwargs={"base_url": settings["base_url"]},
-        api_root=settings.get("api_root", "pulp/"),
-        background_tasks=False,
-        timeout=settings.get("timeout", 120),
-    )
-
     assert {
         "patch_content_in_query_filters",
         "patch_field_select_filters",

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"]

pulpcore/cli/common/context.py

@@ -1,3 +1,5 @@
+import warnings
+
 from pulp_glue.common.context import EntityDefinition  # noqa: F401
 from pulp_glue.common.context import (
     EntityFieldDefinition,
@@ -15,6 +17,10 @@
     registered_repository_contexts,
 )
 
+warnings.warn(
+    DeprecationWarning("This module is deprecated. Import from pulp_glue.common.context instead.")
+)
+
 __all__ = [
     "EntityDefinition",
     "EntityFieldDefinition",

pytest_pulp_cli/__init__.py

@@ -82,7 +82,7 @@ def pulp_cli_vars() -> t.Dict[str, str]:
 
 
 @pytest.fixture(scope="session")
-def pulp_cli_settings(tmp_path_factory: pytest.TempPathFactory) -> t.Tuple[pathlib.Path, t.Any]:
+def pulp_cli_settings() -> t.Dict[str, t.Dict[str, t.Any]]:
     """
     This fixture will setup the config file once per session only.
     It is most likely not useful to be included standalone.
@@ -92,11 +92,18 @@ def pulp_cli_settings(tmp_path_factory: pytest.TempPathFactory) -> t.Tuple[pathl
     if os.environ.get("PULP_API_ROOT"):
         for key in settings:
             settings[key]["api_root"] = os.environ["PULP_API_ROOT"]
+    return settings
+
+
+@pytest.fixture(scope="session")
+def pulp_cli_settings_path(
+    tmp_path_factory: pytest.TempPathFactory, pulp_cli_settings: t.Dict[str, t.Dict[str, t.Any]]
+) -> pathlib.Path:
     settings_path = tmp_path_factory.mktemp("config", numbered=False)
     (settings_path / "pulp").mkdir(parents=True)
     with open(settings_path / "pulp" / "cli.toml", "w") as settings_file:
-        toml.dump(settings, settings_file)
-    return settings_path, settings
+        toml.dump(pulp_cli_settings, settings_file)
+    return settings_path
 
 
 @pytest.fixture(scope="session")
@@ -122,11 +129,12 @@ def pulp_cli_gnupghome(tmp_path_factory: pytest.TempPathFactory) -> pathlib.Path
 
 @pytest.fixture
 def pulp_cli_env(
-    pulp_cli_settings: t.Tuple[pathlib.Path, t.Dict[str, t.Any]],
+    pulp_cli_settings: t.Dict[str, t.Dict[str, t.Any]],
+    pulp_cli_settings_path: pathlib.Path,
     pulp_cli_vars: t.Dict[str, str],
     pulp_cli_gnupghome: pathlib.Path,
     monkeypatch: pytest.MonkeyPatch,
-) -> t.Dict[str, t.Any]:
+) -> None:
     """
     This fixture will set up the environment for cli commands by:
 
@@ -135,16 +143,15 @@ def pulp_cli_env(
     * pointing XDG_CONFIG_HOME accordingly
     * supplying other useful environment vars
     """
-    settings_path, settings = pulp_cli_settings
-    monkeypatch.setenv("XDG_CONFIG_HOME", str(settings_path))
-    monkeypatch.setenv("PULP_BASE_URL", settings["cli"]["base_url"])
-    monkeypatch.setenv("VERIFY_SSL", str(settings["cli"].get("verify_ssl", True)).lower())
+    monkeypatch.setenv("XDG_CONFIG_HOME", str(pulp_cli_settings_path))
+    monkeypatch.setenv("PULP_BASE_URL", pulp_cli_settings["cli"]["base_url"])
+    monkeypatch.setenv("VERIFY_SSL", str(pulp_cli_settings["cli"].get("verify_ssl", True)).lower())
     monkeypatch.setenv("GNUPGHOME", str(pulp_cli_gnupghome))
 
     for key, value in pulp_cli_vars.items():
         monkeypatch.setenv(key, value)
 
-    return settings
+    return None
 
 
 if "PULP_LOGGING" in os.environ: