Skip to content

Commit

Permalink
Add detail_context and PulpMasterContext
Browse files Browse the repository at this point in the history 
[noissue]
  • Loading branch information
@mdellweg
mdellweg committed Jul 19, 2024
1 parent b754909 commit 1cce481
Show file tree
Hide file tree
Showing 3 changed files with 63 additions and 49 deletions.
1 change: 1 addition & 0 deletions CHANGES/pulp-glue/+cast.feature
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Added `detail_context` to master-detail contexts.
86 changes: 37 additions & 49 deletions pulp-glue/pulp_glue/common/context.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1165,7 +1165,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.
"""
Expand All @@ -1191,27 +1220,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:
Expand All @@ -1221,20 +1238,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):
Expand Down Expand Up @@ -1293,7 +1304,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")
Expand All @@ -1302,12 +1313,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,
Expand Down Expand Up @@ -1414,18 +1419,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.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 upload(
self,
Expand Down Expand Up @@ -1466,38 +1466,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]
25 changes: 25 additions & 0 deletions pulp-glue/tests/test_entity_context.py
View file
Original file line number Diff line number Diff line change
@@ -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"]

0 comments on commit 1cce481

Please sign in to comment.