GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing github.com — pulp_python/pulp_python/pytest_plugin.py at main · decko/pulp_python Expand file tree/ Copy pathpytest_plugin.pyMore file actions 249 lines (191 loc) · 8.4 KB / Copy pathpytest_plugin.pyFile metadata and controls249 lines (191 loc) · 8.4 KB 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 import pytest import uuid import subprocess from pulpcore.tests.functional.utils import BindingsNamespace from pulp_python.tests.functional.constants import ( PYTHON_FIXTURE_URL, PYTHON_XS_PROJECT_SPECIFIER, PYTHON_EGG_FILENAME, PYTHON_URL, PYTHON_EGG_URL, PYTHON_WHEEL_URL, PYTHON_WHEEL_FILENAME, ) # Bindings API Fixtures @pytest.fixture(scope="session") def python_bindings(_api_client_set, bindings_cfg): """ A namespace providing preconfigured pulp_python api clients. e.g. `python_bindings.RepositoriesPythonApi.list()`. """ from pulpcore.client import pulp_python as python_bindings_module api_client = python_bindings_module.ApiClient(bindings_cfg) _api_client_set.add(api_client) yield BindingsNamespace(python_bindings_module, api_client) _api_client_set.remove(api_client) # Object Generation Fixtures @pytest.fixture def python_repo_factory(python_bindings, gen_object_with_cleanup): """A factory to generate a Python Repository with auto-cleanup.""" def _gen_python_repo(remote=None, pulp_domain=None, **body): body.setdefault("name", str(uuid.uuid4())) kwargs = {} if pulp_domain: kwargs["pulp_domain"] = pulp_domain if remote: body["remote"] = remote if isinstance(remote, str) else remote.pulp_href return gen_object_with_cleanup(python_bindings.RepositoriesPythonApi, body, **kwargs) return _gen_python_repo @pytest.fixture def python_repo(python_repo_factory): """Creates a Python Repository and deletes it at test cleanup time.""" return python_repo_factory() @pytest.fixture def python_distribution_factory(python_bindings, gen_object_with_cleanup): """A factory to generate a Python Distribution with auto-cleanup.""" def _gen_python_distribution( publication=None, repository=None, version=None, pulp_domain=None, **body ): name = str(uuid.uuid4()) body.setdefault("name", name) body.setdefault("base_path", name) if publication: body["publication"] = get_href(publication) elif repository: repo_href = get_href(repository) if version: if version.isnumeric(): ver_href = f"{repo_href}versions/{version}/" else: ver_href = get_href(version) body["repository_version"] = ver_href else: body["repository"] = repo_href kwargs = {} if pulp_domain: kwargs["pulp_domain"] = pulp_domain return gen_object_with_cleanup(python_bindings.DistributionsPypiApi, body, **kwargs) yield _gen_python_distribution @pytest.fixture def python_publication_factory(python_bindings, gen_object_with_cleanup): """A factory to generate a Python Publication with auto-cleanup.""" def _gen_python_publication(repository, version=None, pulp_domain=None): repo_href = get_href(repository) if version: if version.isnumeric(): ver_href = f"{repo_href}versions/{version}/" else: ver_href = get_href(version) body = {"repository_version": ver_href} else: body = {"repository": repo_href} kwargs = {} if pulp_domain: kwargs["pulp_domain"] = pulp_domain return gen_object_with_cleanup(python_bindings.PublicationsPypiApi, body, **kwargs) yield _gen_python_publication @pytest.fixture def python_remote_factory(python_bindings, gen_object_with_cleanup): """A factory to generate a Python Remote with auto-cleanup.""" def _gen_python_remote(url=PYTHON_FIXTURE_URL, includes=None, pulp_domain=None, **body): body.setdefault("name", str(uuid.uuid4())) body.setdefault("url", url) if includes is None: includes = PYTHON_XS_PROJECT_SPECIFIER body["includes"] = includes kwargs = {} if pulp_domain: kwargs["pulp_domain"] = pulp_domain return gen_object_with_cleanup(python_bindings.RemotesPythonApi, body, **kwargs) yield _gen_python_remote @pytest.fixture def python_repo_with_sync( python_bindings, python_repo_factory, python_remote_factory, monitor_task ): """A factory to generate a Python Repository synced with the passed in Remote.""" def _gen_python_repo_sync(remote=None, mirror=False, repository=None, **body): kwargs = {} if pulp_domain := body.get("pulp_domain"): kwargs["pulp_domain"] = pulp_domain remote = remote or python_remote_factory(**kwargs) repo = repository or python_repo_factory(**body) sync_body = {"mirror": mirror, "remote": remote.pulp_href} monitor_task(python_bindings.RepositoriesPythonApi.sync(repo.pulp_href, sync_body).task) return python_bindings.RepositoriesPythonApi.read(repo.pulp_href) yield _gen_python_repo_sync @pytest.fixture def download_python_file(tmp_path, http_get): """Download a Python file and return its path.""" def _download_python_file(relative_path, url): file_path = tmp_path / relative_path with open(file_path, mode="wb") as f: f.write(http_get(url)) return str(file_path) yield _download_python_file @pytest.fixture def python_file(download_python_file): """Get a default (shelf-reader.tar.gz) Python file.""" return download_python_file(PYTHON_EGG_FILENAME, PYTHON_URL) @pytest.fixture def python_content_factory(python_bindings, download_python_file, monitor_task): """A factory to create a Python Package Content.""" def _gen_python_content(relative_path=PYTHON_EGG_FILENAME, url=None, **body): body["relative_path"] = relative_path if url: body["file"] = download_python_file(relative_path, url) elif not any(x in body for x in ("artifact", "file", "upload")): body["file"] = download_python_file(PYTHON_EGG_FILENAME, PYTHON_URL) if repo := body.get("repository"): repo_href = repo if isinstance(repo, str) else repo.pulp_href body["repository"] = repo_href task = python_bindings.ContentPackagesApi.create(**body).task response = monitor_task(task) return python_bindings.ContentPackagesApi.read(response.created_resources[-1]) yield _gen_python_content @pytest.fixture def python_empty_repo_distro(python_repo_factory, python_distribution_factory): """Returns an empty repo with and distribution serving it.""" def _generate_empty_repo_distro(repo_body=None, distro_body=None): repo_body = repo_body or {} distro_body = distro_body or {} repo = python_repo_factory(**repo_body) distro = python_distribution_factory(repository=repo, **distro_body) return repo, distro yield _generate_empty_repo_distro # Utility fixtures @pytest.fixture def shelf_reader_cleanup(): """Take care of uninstalling shelf-reader before/after the test.""" cmd = ("pip", "uninstall", "shelf-reader", "-y") subprocess.run(cmd) yield subprocess.run(cmd) @pytest.fixture def python_content_summary(python_bindings): """Get a summary of the repository version's content.""" def _gen_summary(repository_version=None, repository=None, version=None): if repository_version is None: repo_href = get_href(repository) if version: repo_ver_href = f"{repo_href}versions/{version}/" else: repo_api = python_bindings.RepositoriesPythonApi repo_ver_href = repo_api.read(repo_href).latest_version_href else: repo_ver_href = get_href(repository_version) return python_bindings.RepositoriesPythonVersionsApi.read(repo_ver_href).content_summary yield _gen_summary def get_href(item): """Tries to get the href from the given item, whether it is a string or object.""" return item if isinstance(item, str) else item.pulp_href @pytest.fixture(scope="session") def python_package_dist_directory(tmp_path_factory, http_get): """Creates a temp dir to hold package distros for uploading.""" dist_dir = tmp_path_factory.mktemp("dist") egg_file = dist_dir / PYTHON_EGG_FILENAME wheel_file = dist_dir / PYTHON_WHEEL_FILENAME with open(egg_file, "wb") as f: f.write(http_get(PYTHON_EGG_URL)) with open(wheel_file, "wb") as f: f.write(http_get(PYTHON_WHEEL_URL)) yield dist_dir, egg_file, wheel_file |
| You are visitor 01402520 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |