From b48c99930bce86931ac0dde03208194c2352c679 Mon Sep 17 00:00:00 2001 From: Matthias Dellweg Date: Fri, 8 Mar 2024 14:25:23 +0100 Subject: [PATCH] Add detail_context and PulpMasterContext [noissue] --- CHANGES/pulp-glue/+cast.feature | 1 + Makefile | 2 +- pulp-glue/pulp_glue/common/context.py | 86 ++++++++----------- pulp-glue/tests/conftest.py | 26 ++++++ {tests => pulp-glue/tests}/test_api_quirks.py | 17 +--- pulp-glue/tests/test_entity_context.py | 25 ++++++ pytest_pulp_cli/__init__.py | 27 +++--- 7 files changed, 111 insertions(+), 73 deletions(-) create mode 100644 CHANGES/pulp-glue/+cast.feature create mode 100644 pulp-glue/tests/conftest.py rename {tests => pulp-glue/tests}/test_api_quirks.py (66%) create mode 100644 pulp-glue/tests/test_entity_context.py diff --git a/CHANGES/pulp-glue/+cast.feature b/CHANGES/pulp-glue/+cast.feature new file mode 100644 index 000000000..019b21b6a --- /dev/null +++ b/CHANGES/pulp-glue/+cast.feature @@ -0,0 +1 @@ +Added `detail_context` to master-detail contexts. diff --git a/Makefile b/Makefile index 4bc0116ba..1111056e8 100644 --- a/Makefile +++ b/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 docs: pulp-docs build diff --git a/pulp-glue/pulp_glue/common/context.py b/pulp-glue/pulp_glue/common/context.py index 76fa7389f..997b11f93 100644 --- a/pulp-glue/pulp_glue/common/context.py +++ b/pulp-glue/pulp_glue/common/context.py @@ -1052,7 +1052,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. """ @@ -1078,27 +1107,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[\w\-_]+)/(?P[\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: @@ -1108,7 +1125,7 @@ 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") @@ -1116,12 +1133,6 @@ class PulpDistributionContext(PulpEntityContext): ID_PREFIX = "distributions" HREF_PATTERN = r"distributions/(?P[\w\-_]+)/(?P[\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): @@ -1180,7 +1191,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") @@ -1189,12 +1200,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, @@ -1301,18 +1306,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[\w\-_]+)/(?P[\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, @@ -1353,25 +1353,19 @@ 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[\w\-_]+)/(?P[\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" @@ -1379,12 +1373,6 @@ class PulpContentGuardContext(PulpEntityContext): ID_PREFIX = "contentguards" HREF_PATTERN = r"contentguards/(?P[\w\-_]+)/(?P[\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] diff --git a/pulp-glue/tests/conftest.py b/pulp-glue/tests/conftest.py new file mode 100644 index 000000000..d8a61dcf1 --- /dev/null +++ b/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), + ) diff --git a/tests/test_api_quirks.py b/pulp-glue/tests/test_api_quirks.py similarity index 66% rename from tests/test_api_quirks.py rename to pulp-glue/tests/test_api_quirks.py index 3e99f586a..157b407bb 100644 --- a/tests/test_api_quirks.py +++ b/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", diff --git a/pulp-glue/tests/test_entity_context.py b/pulp-glue/tests/test_entity_context.py new file mode 100644 index 000000000..62491367c --- /dev/null +++ b/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"] diff --git a/pytest_pulp_cli/__init__.py b/pytest_pulp_cli/__init__.py index 1e529a001..4b5aa743c 100644 --- a/pytest_pulp_cli/__init__.py +++ b/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. @@ -95,11 +95,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") @@ -125,11 +132,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: @@ -138,16 +146,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: