File: conftest.py

package info (click to toggle)
python-ansible-compat 25.12.0-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 964 kB
  • sloc: python: 2,638; sh: 26; makefile: 24
file content (132 lines) | stat: -rw-r--r-- 3,859 bytes parent folder | download
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
"""Pytest fixtures."""

import importlib.metadata
import json
import pathlib
import subprocess
import sys
from collections.abc import Callable, Generator
from pathlib import Path

import pytest

from ansible_compat.runtime import Runtime


@pytest.fixture
def runtime(request: pytest.FixtureRequest) -> Generator[Runtime, None, None]:
    """Isolated runtime fixture with configurable parameters.

    Args:
        request: Pytest fixture request object containing test parameters.
    """
    provided_params = getattr(request, "param", {}) if hasattr(request, "param") else {}
    use_params = provided_params or {"isolated": True}
    instance = Runtime(**use_params)
    yield instance
    instance.clean()


@pytest.fixture
# pylint: disable=unused-argument
def runtime_tmp(
    tmp_path: pathlib.Path,
    scope: str = "session",  # noqa: ARG001
) -> Generator[Runtime, None, None]:
    """Isolated runtime fixture using a temp directory."""
    instance = Runtime(project_dir=tmp_path, isolated=True)
    yield instance
    instance.clean()


def query_pkg_version(pkg: str) -> str:
    """Get the version of a current installed package.

    :param pkg: Package name
    :return: Package version
    """
    return importlib.metadata.version(pkg)


@pytest.fixture
def pkg_version() -> Callable[[str], str]:
    """Get the version of a current installed package.

    :return: Callable function to get package version
    """
    return query_pkg_version


class VirtualEnvironment:
    """Virtualenv wrapper."""

    def __init__(self, path: Path) -> None:
        """Initialize.

        :param path: Path to virtualenv
        """
        self.project = path
        self.venv_path = self.project / "venv"
        self.venv_bin_path = self.venv_path / "bin"
        self.venv_python_path = self.venv_bin_path / "python"

    def create(self) -> None:
        """Create virtualenv."""
        cmd = [str(sys.executable), "-m", "venv", str(self.venv_path)]
        subprocess.check_call(args=cmd)
        # Install this package into the virtual environment
        self.install(f"{__file__}/../..")

    def install(self, *packages: str) -> None:
        """Install packages in virtualenv.

        :param packages: Packages to install
        """
        cmd = [str(self.venv_python_path), "-m", "pip", "install", *packages]
        subprocess.check_call(args=cmd)

    def python_script_run(self, script: str) -> subprocess.CompletedProcess[str]:
        """Run command in project dir using venv.

        :param args: Command to run
        """
        proc = subprocess.run(
            args=[self.venv_python_path, "-c", script],
            capture_output=True,
            cwd=self.project,
            check=False,
            text=True,
        )
        return proc

    def site_package_dirs(self) -> list[Path]:
        """Get site packages.

        :return: List of site packages dirs
        """
        script = "import json, site; print(json.dumps(site.getsitepackages()))"
        proc = subprocess.run(
            args=[self.venv_python_path, "-c", script],
            capture_output=True,
            check=False,
            text=True,
        )
        dirs = json.loads(proc.stdout)
        if not isinstance(dirs, list):
            msg = "Expected list of site packages"
            raise TypeError(msg)
        sanitized = list({Path(d).resolve() for d in dirs})
        return sanitized


@pytest.fixture(scope="module")
def venv_module(tmp_path_factory: pytest.TempPathFactory) -> VirtualEnvironment:
    """Create a virtualenv in a temporary directory.

    :param tmp_path: pytest fixture for temp path
    :return: VirtualEnvironment instance
    """
    test_project = tmp_path_factory.mktemp(basename="test_project-", numbered=True)
    venv_ = VirtualEnvironment(test_project)
    venv_.create()
    return venv_