First version

Signed-off-by: Bernát Gábor <[email protected]>

.gitignore

@@ -8,4 +8,4 @@ __pycache__
 **.pyc
 build
 dist
-src/pytest_devpi/version.py
+src/devpi_process/version.py

CHANGELOG.md

@@ -1,59 +1,5 @@
 # Changelog
 
-## [Unreleased](https://github.com/gaborbernat/pytest-devpi/tree/HEAD)
-
-## [0.1.0](https://github.com/gaborbernat/pytest-devpi/tree/0.1.0) (2021-06-17)
-
-[Full Changelog](https://github.com/gaborbernat/pytest-devpi/compare/0.2.1...0.3.0)
-
-**Merged pull requests:**
-
-- Drop Python 2 support [\#6](https://github.com/gaborbernat/pytest-devpi/issues/7)
-
-## [0.2.1](https://github.com/gaborbernat/pytest-devpi/tree/0.2.1) (2020-10-23)
-
-[Full Changelog](https://github.com/gaborbernat/pytest-devpi/compare/0.2.0...0.2.1)
-
-**Implemented enhancements:**
-
-- add session level support [\#6](https://github.com/gaborbernat/pytest-devpi/issues/6)
-
-## [0.2.0](https://github.com/gaborbernat/pytest-devpi/tree/0.2.0) (2020-08-04)
-
-[Full Changelog](https://github.com/gaborbernat/pytest-devpi/compare/0.1.3...0.2.0)
-
-## [0.1.3](https://github.com/gaborbernat/pytest-devpi/tree/0.1.3) (2019-09-03)
-
-[Full Changelog](https://github.com/gaborbernat/pytest-devpi/compare/0.1.2...0.1.3)
-
-**Implemented enhancements:**
-
-- allow force on, document flags [\#3](https://github.com/gaborbernat/pytest-devpi/pull/3) ([gaborbernat](https://github.com/gaborbernat))
-
-**Merged pull requests:**
-
-- Remove PyPy special cases [\#4](https://github.com/gaborbernat/pytest-devpi/pull/4) ([vtbassmatt](https://github.com/vtbassmatt))
-
-## [0.1.2](https://github.com/gaborbernat/pytest-devpi/tree/0.1.2) (2018-11-29)
-
-[Full Changelog](https://github.com/gaborbernat/pytest-devpi/compare/0.1.1...0.1.2)
-
-## [0.1.1](https://github.com/gaborbernat/pytest-devpi/tree/0.1.1) (2018-11-15)
-
-[Full Changelog](https://github.com/gaborbernat/pytest-devpi/compare/0.1.0...0.1.1)
-
-**Closed issues:**
-
-- How to use “pytest\_print” [\#1](https://github.com/gaborbernat/pytest-devpi/issues/1)
-
-**Merged pull requests:**
-
-- Update setup.py [\#2](https://github.com/gaborbernat/pytest-devpi/pull/2) ([shashanksingh28](https://github.com/shashanksingh28))
-
-## [0.1.0](https://github.com/gaborbernat/pytest-devpi/tree/0.1.0) (2018-04-14)
-
-[Full Changelog](https://github.com/gaborbernat/pytest-devpi/compare/727896d18cab117ad84010086cbc4c9a16d9e8f7...0.1.0)
-
 
 
 \* *This Changelog was automatically generated by [github_changelog_generator](https://github.com/github-changelog-generator/github-changelog-generator)*

README.md

@@ -1,18 +1,39 @@
-# pytest-devpi
-
-[![PyPI](https://img.shields.io/pypi/v/pytest-devpi?style=flat-square)](https://pypi.org/project/pytest-devpi)
-[![PyPI - Implementation](https://img.shields.io/pypi/implementation/pytest-devpi?style=flat-square)](https://pypi.org/project/pytest-devpi)
-[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/pytest-devpi?style=flat-square)](https://pypi.org/project/pytest-devpi)
-[![PyPI - Downloads](https://img.shields.io/pypi/dm/pytest-devpi?style=flat-square)](https://pypistats.org/packages/pytest-devpi)
-[![PyPI - License](https://img.shields.io/pypi/l/pytest-devpi?style=flat-square)](https://opensource.org/licenses/MIT)
-[![check](https://github.com/gaborbernat/pytest-devpi/workflows/check/badge.svg)](https://github.com/gaborbernat/pytest-devpi/actions?query=workflow%3Acheck)
+# devpi-process
+
+[![PyPI](https://img.shields.io/pypi/v/devpi-process?style=flat-square)](https://pypi.org/project/devpi-process)
+[![PyPI - Implementation](https://img.shields.io/pypi/implementation/devpi-process?style=flat-square)](https://pypi.org/project/devpi-process)
+[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/devpi-process?style=flat-square)](https://pypi.org/project/devpi-process)
+[![PyPI - Downloads](https://img.shields.io/pypi/dm/devpi-process?style=flat-square)](https://pypistats.org/packages/devpi-process)
+[![PyPI - License](https://img.shields.io/pypi/l/devpi-process?style=flat-square)](https://opensource.org/licenses/MIT)
+[![check](https://github.com/gaborbernat/devpi-process/workflows/check/badge.svg)](https://github.com/gaborbernat/devpi-process/actions?query=workflow%3Acheck)
 [![Code style:
 black](https://img.shields.io/badge/code%20style-black-000000.svg?style=flat-square)](https://github.com/psf/black)
 
-Create a devpi instance for your pytest suite.
+Allows you to create [devpi](https://devpi.net/docs/devpi/devpi/stable/+d/index.html) server process with indexes, and
+upload artifacts to that programmatically.
 
 ## install
 
 ```sh
-pip install pytest-devpi
+pip install devpi-process
+```
+
+## use
+
+```python
+from pathlib import Path
+
+from devpi_process import Index, IndexServer
+
+with IndexServer(Path("server-dir")) as server:
+    # create an index mirroring an Artifactory instance
+    magic_index_url = "https://magic.com/artifactory/api/pypi/magic-pypi/simple"
+    base_name = "magic"
+    server.create_index(base_name, "type=mirror", f"mirror_url={magic_index_url}")
+
+    # create a dev index server that bases of magic PyPI, and upload a wheel to it
+    dev: Index = server.create_index("dev", f"bases={server.user}/{base_name}")
+    dev.upload("magic-2.24.0-py3-none-any.whl")
+
+    assert dev.url  # point the tool consuming the index server to this
 ```

pyproject.toml

@@ -7,10 +7,10 @@ line-length = 120
 
 [tool.isort]
 profile = "black"
-known_first_party = ["pytest_devpi"]
+known_first_party = ["devpi_process"]
 
 [tool.setuptools_scm]
-write_to = "src/pytest_devpi/version.py"
+write_to = "src/devpi_process/version.py"
 write_to_template = """
 \"\"\" Version information \"\"\"
 

setup.cfg

@@ -1,18 +1,16 @@
 [metadata]
-name = pytest_devpi
-description = pytest-devpi adds a fixture to create devpi instances within your tests
+name = devpi_process
+description = devpi process provides a programmatic API to create and use a devpi server process
 long_description = file: README.md
 long_description_content_type = text/markdown
-url = https://github.com/gaborbernat/pytest-devpi#pytest-devpi
+url = https://github.com/gaborbernat/devpi_process
 maintainer = Bernat Gabor
 maintainer_email = [email protected]
 license = MIT
 license_file = LICENSE.txt
 platforms = any
 classifiers =
     Development Status :: 5 - Production/Stable
-    Environment :: Plugins
-    Framework :: Pytest
     Intended Audience :: Developers
     License :: OSI Approved :: MIT License
     Operating System :: MacOS :: MacOS X
@@ -26,19 +24,17 @@ classifiers =
     Programming Language :: Python :: 3.9
     Programming Language :: Python :: 3.10
     Topic :: Software Development :: Libraries
-    Topic :: Software Development :: Testing
     Topic :: Utilities
-keywords = pytest, print, debug
+keywords = devpi, programmatic
 project_urls =
-    Source=https://github.com/gaborbernat/pytest-devpi
-    Tracker=https://github.com/gaborbernat/pytest-devpi/issues
+    Source=https://github.com/gaborbernat/devpi-process
+    Tracker=https://github.com/gaborbernat/devpi-process
 
 [options]
 packages = find:
 install_requires =
     devpi-client>=5.2
     devpi-server>=6
-    pytest>=6
 python_requires = >=3.6
 include_package_data = True
 package_dir =
@@ -48,22 +44,18 @@ zip_safe = True
 [options.packages.find]
 where = src
 
-[options.entry_points]
-pytest11 = pytest_devpi = pytest_devpi
-
 [options.extras_require]
 test =
     coverage>=5
+    httpx>=0.18
+    pytest>=6
 
 [options.package_data]
-pytest_devpi = py.typed
+devpi_process = py.typed
 
 [sdist]
 formats = gztar
 
-[bdist_wheel]
-universal = true
-
 [flake8]
 max-line-length = 120
 ignore = F401, H301, E203
@@ -80,8 +72,6 @@ dynamic_context = test_function
 fail_under = 100
 skip_covered = true
 show_missing = true
-omit =
-    tests/example.py
 
 [coverage:html]
 show_contexts = True
@@ -98,9 +88,8 @@ source =
     *\src
 
 [tool:pytest]
-addopts = -ra --showlocals -vv
+addopts = -ra --showlocals
 testpaths = tests
-xfail_strict = True
 junit_family = xunit2
 
 [mypy]
@@ -121,3 +110,6 @@ implicit_reexport = False
 strict_equality = True
 warn_unused_configs = True
 pretty = True
+
+[mypy-httpx.*]
+ignore_missing_imports = True

src/devpi_process/__init__.py

@@ -0,0 +1,154 @@
+import random
+import socket
+import string
+import sys
+import sysconfig
+from contextlib import closing
+from pathlib import Path
+from subprocess import PIPE, Popen, check_call
+from threading import Thread
+from types import TracebackType
+from typing import IO, Dict, Iterator, List, Optional, Sequence, Type, cast
+
+from .version import __version__
+
+
+class Index:
+    def __init__(self, base_url: str, name: str, user: str, client_cmd_base: List[str]) -> None:
+        self._client_cmd_base = client_cmd_base
+        self._server_url = base_url
+        self.name = name
+        self.user = user
+
+    @property
+    def url(self) -> str:
+        return f"{self._server_url}/{self.name}/+simple"
+
+    def use(self) -> None:
+        check_call(self._client_cmd_base + ["use", f"{self.user}/{self.name}"], stdout=PIPE, stderr=PIPE)
+
+    def upload(self, *files: Path) -> None:
+        cmd = self._client_cmd_base + ["upload", "--index", self.name] + [str(i) for i in files]
+        check_call(cmd)
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}(url={self.url})"
+
+
+class IndexServer:
+    def __init__(self, path: Path, with_root_pypi: bool = False, start_args: Optional[Sequence[str]] = None) -> None:
+        self.path = path
+        self._with_root_pypi = with_root_pypi
+        self._start_args: Sequence[str] = [] if start_args is None else start_args
+
+        self.host, self.port = "localhost", _find_free_port()
+        self._passwd = "".join(random.choices(string.ascii_letters, k=8))
+
+        scripts_dir = sysconfig.get_path("scripts")
+        if scripts_dir is None:
+            raise RuntimeError("could not get scripts folder of host interpreter")  # pragma: no cover
+
+        def _exe(name: str) -> str:
+            return str(Path(cast(str, scripts_dir)) / f"{name}{'.exe' if sys.platform == 'win32' else ''}")
+
+        self._init: str = _exe("devpi-init")
+        self._server: str = _exe("devpi-server")
+        self._client: str = _exe("devpi")
+
+        self._server_dir = self.path / "server"
+        self._client_dir = self.path / "client"
+        self._indexes: Dict[str, Index] = {}
+        self._process: Optional["Popen[str]"] = None
+        self._has_use = False
+        self._stdout_drain: Optional[Thread] = None
+
+    @property
+    def user(self) -> str:
+        return "root"
+
+    def __enter__(self) -> "IndexServer":
+        self._create_and_start_server()
+        self._setup_client()
+        return self
+
+    def _create_and_start_server(self) -> None:
+        self._server_dir.mkdir(exist_ok=True)
+        server_at = str(self._server_dir)
+        # 1. create the server
+        cmd = [self._init, "--serverdir", server_at]
+        cmd.extend(("--role", "standalone", "--root-passwd", self._passwd))
+        if self._with_root_pypi is False:
+            cmd.append("--no-root-pypi")
+        check_call(cmd, stdout=PIPE, stderr=PIPE)
+        # 2. start the server
+        cmd = [self._server, "--serverdir", server_at, "--port", str(self.port)]
+        cmd.extend(self._start_args)
+        self._process = Popen(cmd, stdout=PIPE, universal_newlines=True)
+        stdout = self._drain_stdout()
+        for line in stdout:  # pragma: no branch # will always loop at least once
+            if "serving at url" in line:
+
+                def _keep_draining() -> None:
+                    for _ in stdout:
+                        pass
+
+                # important to keep draining the stdout, otherwise once the buffer is full Windows blocks the process
+                self._stdout_drain = Thread(target=_keep_draining, name="tox-test-stdout-drain")
+                self._stdout_drain.start()
+                break
+
+    def _drain_stdout(self) -> Iterator[str]:
+        process = cast("Popen[str]", self._process)
+        stdout = cast(IO[str], process.stdout)
+        while True:
+            if process.poll() is not None:  # pragma: no cover
+                print(f"devpi server with pid {process.pid} at {self._server_dir} died")
+                break
+            yield stdout.readline()
+
+    def _setup_client(self) -> None:
+        """create a user on the server and authenticate it"""
+        self._client_dir.mkdir(exist_ok=True)
+        base = ["--clientdir", str(self._client_dir)]
+        check_call([self._client, "use"] + base + [self.url], stdout=PIPE, stderr=PIPE)
+        check_call([self._client, "login"] + base + [self.user, "--password", self._passwd], stdout=PIPE, stderr=PIPE)
+
+    def create_index(self, name: str, *args: str) -> Index:
+        if name in self._indexes:  # pragma: no cover
+            raise ValueError(f"index {name} already exists")
+        base = [self._client, "--clientdir", str(self._client_dir)]
+        check_call(base + ["index", "-c", name, *args], stdout=PIPE, stderr=PIPE)
+        index = Index(f"{self.url}/{self.user}", name, self.user, base)
+        self._indexes[name] = index
+        return index
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],  # noqa: U100
+        exc_val: Optional[BaseException],  # noqa: U100
+        exc_tb: Optional[TracebackType],  # noqa: U100
+    ) -> None:
+        if self._process is not None:  # pragma: no cover # defend against devpi startup fail
+            self._process.terminate()
+        if self._stdout_drain is not None and self._stdout_drain.is_alive():  # pragma: no cover # devpi startup fail
+            self._stdout_drain.join()
+
+    @property
+    def url(self) -> str:
+        return f"http://{self.host}:{self.port}"
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}(url={self.url}, indexes={list(self._indexes)})"
+
+
+def _find_free_port() -> int:
+    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as socket_handler:
+        socket_handler.bind(("", 0))
+        return cast(int, socket_handler.getsockname()[1])
+
+
+__all__ = [
+    "__version__",
+    "Index",
+    "IndexServer",
+]