tests/functional2: Make Command Environment actually declarative
So far, the environment used by `command` was completely leaky and the one used by `nix` was very leaky despite it trying to be a "hermetic" environment. This commit moves the hermaticity to `command` and changes its implementation to be not leak anything. To achieve this, the following changes were also nessecary: - the `files` and `snapshot` fixture now use the folder `test-home` within the tmp_path directory by default, as the `HOME` environment variable is set to there. (extraction not possible due to dependencies of command etc also using this directory) Fixes: #847, #848 Change-Id: I55f86ee0e1615e73fcf442ee2f28f3b89893bbb4
This commit is contained in:
@@ -407,6 +407,7 @@ if enable_embedded_sandbox_shell
|
||||
endif
|
||||
|
||||
sandbox_shell = get_option('sandbox-shell')
|
||||
build_test_shell = get_option('build-test-shell')
|
||||
# Consider it required if we're on Linux and the user explicitly specified a non-default value.
|
||||
sandbox_shell_required = sandbox_shell != 'busybox' and host_machine.system() == 'linux'
|
||||
# NOTE(Qyriad): package.nix puts busybox in buildInputs for Linux.
|
||||
|
||||
@@ -24,6 +24,10 @@ option('sandbox-shell', type : 'string', value : 'busybox',
|
||||
description : 'path to a statically-linked shell to use as /bin/sh in sandboxes (usually busybox)',
|
||||
)
|
||||
|
||||
option('build-test-shell', type: 'string',
|
||||
description : 'path to a statically-linked shell which also offeres coreutils functionality'
|
||||
)
|
||||
|
||||
option('pasta-path', type : 'string', value : 'pasta',
|
||||
description : 'path to the location of pasta (provided by passt)',
|
||||
)
|
||||
|
||||
+21
-12
@@ -1,5 +1,6 @@
|
||||
{
|
||||
pkgs,
|
||||
pkgsStatic,
|
||||
lib,
|
||||
stdenv,
|
||||
aws-sdk-cpp,
|
||||
@@ -254,6 +255,7 @@ stdenv.mkDerivation (finalAttrs: {
|
||||
# which don't actually get added to PATH. And buildInputs is correct over
|
||||
# nativeBuildInputs since this should be a busybox executable on the host.
|
||||
"-Dsandbox-shell=${lib.getExe' busybox-sandbox-shell "busybox"}"
|
||||
"-Dbuild-test-shell=${pkgsStatic.busybox}/bin"
|
||||
"-Dpasta-path=${lib.getExe' passt-lix "pasta"}"
|
||||
]
|
||||
++ lib.optional hostPlatform.isStatic "-Denable-embedded-sandbox-shell=true"
|
||||
@@ -345,7 +347,6 @@ stdenv.mkDerivation (finalAttrs: {
|
||||
++ lib.optionals stdenv.hostPlatform.isStatic [ llvmPackages.libunwind ]
|
||||
++ lib.optionals hostPlatform.isLinux [
|
||||
libseccomp
|
||||
busybox-sandbox-shell
|
||||
passt-lix
|
||||
]
|
||||
++ lib.optional internalApiDocs rapidcheck
|
||||
@@ -357,10 +358,14 @@ stdenv.mkDerivation (finalAttrs: {
|
||||
# configure, but we don't actually want to *run* the checks here.
|
||||
++ lib.optionals lintInsteadOfBuild finalAttrs.checkInputs;
|
||||
|
||||
checkInputs = [
|
||||
gtest
|
||||
rapidcheck
|
||||
];
|
||||
checkInputs =
|
||||
[
|
||||
gtest
|
||||
rapidcheck
|
||||
]
|
||||
++ lib.optionals hostPlatform.isLinux [
|
||||
pkgsStatic.busybox
|
||||
];
|
||||
|
||||
propagatedBuildInputs = lib.optionals (!finalAttrs.dontBuild) maybePropagatedInputs;
|
||||
|
||||
@@ -370,14 +375,18 @@ stdenv.mkDerivation (finalAttrs: {
|
||||
finalAttrs.lixPythonForBuild
|
||||
];
|
||||
|
||||
env = {
|
||||
# Meson allows referencing a /usr/share/cargo/registry shaped thing for subproject sources.
|
||||
# Turns out the Nix-generated Cargo dependencies are named the same as they
|
||||
# would be in a Cargo registry cache.
|
||||
MESON_PACKAGE_CACHE_DIR = finalAttrs.cargoDeps;
|
||||
env =
|
||||
{
|
||||
# Meson allows referencing a /usr/share/cargo/registry shaped thing for subproject sources.
|
||||
# Turns out the Nix-generated Cargo dependencies are named the same as they
|
||||
# would be in a Cargo registry cache.
|
||||
MESON_PACKAGE_CACHE_DIR = finalAttrs.cargoDeps;
|
||||
|
||||
VERSION_SUFFIX = versionSuffix;
|
||||
};
|
||||
VERSION_SUFFIX = versionSuffix;
|
||||
}
|
||||
// lib.optionalAttrs hostPlatform.isLinux {
|
||||
BUILD_TEST_SHELL = "${pkgsStatic.busybox}/bin";
|
||||
};
|
||||
|
||||
cargoDeps = rustPlatform.importCargoLock { lockFile = ./Cargo.lock; };
|
||||
|
||||
|
||||
@@ -155,6 +155,22 @@ For a full intro on what exactly fixtures are, see the Pytest documentation ["Ab
|
||||
For an exhaustive documentation on how fixtures work and how to use them check the Pytest documentation ["How-to use fixtures"](https://docs.pytest.org/en/stable/how-to/fixtures.html).
|
||||
For a complete list of fixtures and their documentation, run ```just test-functional2 --fixtures```.
|
||||
|
||||
#### [`env`](./testlib/fixtures/command.py)
|
||||
|
||||
Declarative Environment used by `Command` to execute sub commands (often `nix`)
|
||||
without leaking any global configuration.
|
||||
Provides rich access to setting and unsetting variables
|
||||
as well as modifying `PATH`.
|
||||
|
||||
#### [`command`](./testlib/fixtures/command.py)
|
||||
|
||||
Get a function to create Commands with the given arguments and stdin.
|
||||
|
||||
This fixture pre-applies the [declarative environment](#env).
|
||||
If one is already requesting the `env` fixture,
|
||||
one can alternatively instantiate the `Command` class directly, passing in `env` to the `_env` argument
|
||||
|
||||
|
||||
#### [`files`](./testlib/fixtures/file_helper.py)
|
||||
|
||||
Pass in file resources into the test's temporary runtime directory.
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import os
|
||||
import stat
|
||||
import sys
|
||||
|
||||
@@ -7,6 +6,8 @@ import pytest
|
||||
from pathlib import Path
|
||||
|
||||
from textwrap import dedent
|
||||
|
||||
from functional2.testlib.fixtures.env import ManagedEnv
|
||||
from functional2.testlib.fixtures.nix import Nix
|
||||
|
||||
COMMANDS = ["copy-closure", "collect-garbage"]
|
||||
@@ -25,21 +26,21 @@ def failing_sub_command_path(tmp_path_factory: pytest.TempPathFactory) -> Path:
|
||||
@pytest.fixture(scope="module", params=COMMANDS)
|
||||
def custom_sub_command(request: pytest.FixtureRequest, custom_sub_command_path: Path) -> str:
|
||||
# Create an external command that is an alias of an existing one
|
||||
command = request.param
|
||||
lix_cmd = request.param
|
||||
|
||||
executable = custom_sub_command_path / f"lix-{command}"
|
||||
executable = custom_sub_command_path / f"lix-{lix_cmd}"
|
||||
executable.write_text(
|
||||
dedent(f"""\
|
||||
#!{sys.executable}
|
||||
import os, sys
|
||||
# Start with args[0] set to the actual nix command used for testing
|
||||
# as we are not making Lix variants of those.
|
||||
os.execvp("nix-{command}", [ "nix-{command}" ] + sys.argv[1:])
|
||||
os.execvp("nix-{lix_cmd}", [ "nix-{lix_cmd}" ] + sys.argv[1:])
|
||||
""")
|
||||
)
|
||||
executable.chmod(stat.S_IXUSR | stat.S_IRUSR | stat.S_IWUSR)
|
||||
|
||||
return command
|
||||
return lix_cmd
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
@@ -59,48 +60,48 @@ def failing_sub_command(failing_sub_command_path: Path, custom_sub_command: str)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def path(custom_sub_command_path: Path) -> str:
|
||||
# Provide a modified search path
|
||||
search_path = os.environ.get("PATH").split(":")
|
||||
search_path += ["/some/incorrect", "/location/for/fun", str(custom_sub_command_path)]
|
||||
|
||||
return ":".join(search_path)
|
||||
def path(custom_sub_command_path: Path, env: ManagedEnv):
|
||||
"""Provide a modified search path"""
|
||||
env.path.append("/some/incorrect")
|
||||
env.path.append("/location/for/fun")
|
||||
env.path.append(str(custom_sub_command_path))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def path_with_failure(
|
||||
request: pytest.FixtureRequest, custom_sub_command_path: Path, failing_sub_command_path: Path
|
||||
) -> str:
|
||||
# Provide a search path with failing binaries inserted in the specified position
|
||||
request: pytest.FixtureRequest,
|
||||
custom_sub_command_path: Path,
|
||||
failing_sub_command_path: Path,
|
||||
env: ManagedEnv,
|
||||
):
|
||||
"""
|
||||
Provide a search path with failing binaries inserted in the specified position
|
||||
"""
|
||||
(first, fail) = request.param
|
||||
search_path = os.environ.get("PATH").split(":")
|
||||
|
||||
command_path = [str(custom_sub_command_path), str(failing_sub_command_path)]
|
||||
if fail:
|
||||
command_path.reverse()
|
||||
|
||||
if first:
|
||||
search_path = command_path + search_path
|
||||
env.path.prepend(command_path[1])
|
||||
env.path.prepend(command_path[0])
|
||||
else:
|
||||
search_path += command_path
|
||||
|
||||
return ":".join(search_path)
|
||||
env.path.append(command_path[0])
|
||||
env.path.append(command_path[1])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("nix_exe", "flag", "expected"),
|
||||
[("nix", False, 1), ("lix", False, 1), ("nix", True, 1), ("lix", True, 0)],
|
||||
)
|
||||
def test_sub_commands(
|
||||
nix: Nix, path: str, custom_sub_command: str, nix_exe: str, flag: bool, expected: int
|
||||
):
|
||||
# Test custom sub commands in various configurations
|
||||
nix_command = nix.nix([custom_sub_command, "--version"], nix_exe=nix_exe)
|
||||
nix_command.update_env(PATH=path)
|
||||
@pytest.mark.usefixtures("path")
|
||||
def test_sub_commands(nix: Nix, custom_sub_command: str, nix_exe: str, flag: bool, expected: int):
|
||||
if flag:
|
||||
nix_command.settings.feature("lix-custom-sub-commands")
|
||||
nix.settings.feature("lix-custom-sub-commands")
|
||||
|
||||
nix_command.run().expect(expected)
|
||||
# Test custom sub commands in various configurations
|
||||
nix.nix([custom_sub_command, "--version"], nix_exe=nix_exe).run().expect(expected)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@@ -108,16 +109,12 @@ def test_sub_commands(
|
||||
[((True, True), 42), ((True, False), 0), ((False, True), 42), ((False, False), 0)],
|
||||
indirect=["path_with_failure"],
|
||||
)
|
||||
def test_sub_command_path_order(
|
||||
nix: Nix, path_with_failure: str, failing_sub_command: str, expected: int
|
||||
):
|
||||
@pytest.mark.usefixtures("path_with_failure")
|
||||
def test_sub_command_path_order(nix: Nix, failing_sub_command: str, expected: int):
|
||||
# Test handling of the order of the path for custom sub commands
|
||||
# Incidentally also tests passing through exit codes
|
||||
nix_command = nix.nix([failing_sub_command, "--version"], nix_exe="lix")
|
||||
nix_command.update_env(PATH=path_with_failure)
|
||||
nix_command.settings.feature("lix-custom-sub-commands")
|
||||
|
||||
nix_command.run().expect(expected)
|
||||
nix.settings.feature("lix-custom-sub-commands")
|
||||
nix.nix([failing_sub_command, "--version"], nix_exe="lix").run().expect(expected)
|
||||
|
||||
|
||||
@pytest.mark.skip(
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
pytest_plugins = (
|
||||
"functional2.testlib.fixtures.env",
|
||||
"functional2.testlib.fixtures.formatter",
|
||||
"functional2.testlib.fixtures.file_helper",
|
||||
"functional2.testlib.fixtures.logger",
|
||||
"functional2.testlib.fixtures.command",
|
||||
"functional2.testlib.fixtures.nix",
|
||||
"functional2.testlib.fixtures.snapshot",
|
||||
"functional2.testlib.fixtures.pytest_command",
|
||||
|
||||
@@ -4,7 +4,7 @@ from functional2.testlib.fixtures.nix import Nix
|
||||
# ruff: noqa: N802
|
||||
def test_fetchTree_presence(nix: Nix):
|
||||
"""Ensures that fetchTree is actually absent if flakes are disabled"""
|
||||
settings = nix.settings().feature("nix-command")
|
||||
settings = nix.settings.feature("nix-command")
|
||||
assert nix.eval("builtins ? fetchTree", settings).json() is False
|
||||
|
||||
settings.feature("flakes")
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
import string
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
from textwrap import dedent
|
||||
|
||||
import pytest
|
||||
from functional2.testlib.commands import Command
|
||||
|
||||
from functional2.testlib.fixtures.command import Command
|
||||
from functional2.testlib.fixtures.file_helper import File, AssetSymlink
|
||||
from functional2.testlib.fixtures.snapshot import Snapshot
|
||||
from functional2.testlib.utils import get_functional2_lang_files
|
||||
@@ -572,11 +574,16 @@ def test_generic_throws_unused_files(pytest_command: Command):
|
||||
indirect=True,
|
||||
)
|
||||
@pytest.mark.usefixtures("files")
|
||||
def test_toml_throws_unused_files(pytest_command: Command):
|
||||
def test_toml_throws_unused_files(
|
||||
pytest_command: Command, balanced_templater: type[string.Template]
|
||||
):
|
||||
res = pytest_command.run().expect(1)
|
||||
out = res.stdout_plain
|
||||
assert "test_invalid_configuration[toml_unused-reasons0]" in out
|
||||
assert "the following files weren't referenced: {'in-1.nix', 'eval-fail.err.exp'}" in out
|
||||
pattern = balanced_templater("the following files weren't referenced: {'@A@', '@B@'}")
|
||||
assert pattern.substitute(A="eval-fail.err.exp", B="in-1.nix") in out or pattern.substitute(
|
||||
B="eval-fail.err.exp", A="in-1.nix"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
||||
@@ -12,7 +12,11 @@ xdist_opts = [
|
||||
# surprisingly, this actually works even if PATH is set to something before
|
||||
# meson gets hold of it. neat!
|
||||
functional2_env = environment()
|
||||
functional2_env.prepend('PATH', bindir)
|
||||
# set the bin dir for us to know where to find the lix binaries for our declarative path
|
||||
# without leaking all of PATH or env
|
||||
functional2_env.set('NIX_BIN_DIR', bindir)
|
||||
functional2_env.set('BUILD_TEST_SHELL', build_test_shell)
|
||||
functional2_env.set('system', host_system)
|
||||
|
||||
test(
|
||||
'functional2',
|
||||
@@ -32,4 +36,8 @@ test(
|
||||
# protocol : 'tap',
|
||||
suite : 'installcheck',
|
||||
timeout : 300,
|
||||
# Do not cut output of this runner on failure!
|
||||
# As TAP is currently not supported (see above), we do need all lines to actually know
|
||||
# why and how the failing tests actually failed
|
||||
verbose: true,
|
||||
)
|
||||
|
||||
@@ -101,7 +101,7 @@ EVIL_NARS: list[tuple[str, NarItem]] = [
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("name", "nar"), EVIL_NARS)
|
||||
@pytest.mark.parametrize(("name", "nar"), EVIL_NARS, ids=next(zip(*EVIL_NARS)))
|
||||
def test_evil_nar(nix: Nix, name: str, nar: NarItem, logger: Logger):
|
||||
bio = BytesIO()
|
||||
|
||||
|
||||
@@ -108,17 +108,15 @@ def test_http_simple(nix: Nix, store: HTTPStore, files: Path):
|
||||
store_path = result.stdout_plain
|
||||
hash_part, _ = Path(store_path).stem.split("-", 1)
|
||||
|
||||
nar_info_cache = nix.test_root / "test-home" / ".cache" / "nix" / "binary-cache-v6.sqlite"
|
||||
nar_info_cache = nix.env.dirs.xdg_cache_home / "nix" / "binary-cache-v6.sqlite"
|
||||
|
||||
app = start_server(store)
|
||||
with http_server(app) as httpd:
|
||||
url = f"http://localhost:{httpd.port}?compression=none"
|
||||
url = f"http://localhost:{httpd.port}?compression=none&store=/nix/store"
|
||||
nix.nix(cmd=["store", "ping", "--store", url], flake=True).run().ok()
|
||||
|
||||
# Narinfo shouldn't exist yet
|
||||
nix.nix(
|
||||
cmd=["path-info", "--store", f"http://localhost:{httpd.port}", store_path], flake=True
|
||||
).run().expect(1)
|
||||
nix.nix(cmd=["path-info", "--store", url, store_path], flake=True).run().expect(1)
|
||||
cache_entries = nars_from_narinfo_cache(nar_info_cache)
|
||||
|
||||
assert len(cache_entries) == 1
|
||||
@@ -127,17 +125,16 @@ def test_http_simple(nix: Nix, store: HTTPStore, files: Path):
|
||||
|
||||
# Successful upload
|
||||
nix.nix(
|
||||
cmd=["copy", "--from", nix.test_root / "store", "--to", url, store_path], flake=True
|
||||
cmd=["copy", "--from", nix.settings.store, "--to", url, store_path], flake=True
|
||||
).run().ok()
|
||||
assert hash_part in store.uploaded_nars
|
||||
|
||||
# Make sure the negative entry got removed
|
||||
assert not nars_from_narinfo_cache(nar_info_cache)
|
||||
cache_entries = nars_from_narinfo_cache(nar_info_cache)
|
||||
assert len(cache_entries) == 0
|
||||
|
||||
# Ensure that the narinfo can be found now.
|
||||
nix.nix(
|
||||
cmd=["path-info", "--store", f"http://localhost:{httpd.port}", store_path], flake=True
|
||||
).run().ok()
|
||||
nix.nix(cmd=["path-info", "--store", url, store_path], flake=True).run().ok()
|
||||
|
||||
# Ensure local narinfo cache is up-to-date.
|
||||
nar_entries = nars_from_narinfo_cache(nar_info_cache)
|
||||
|
||||
+30
-44
@@ -3,10 +3,15 @@ import json
|
||||
import logging
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from functional2.testlib.fixtures.env import ManagedEnv
|
||||
from functional2.testlib.terminal_code_eater import eat_terminal_codes
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -26,6 +31,7 @@ class CommandResult:
|
||||
assumes a return code of 0
|
||||
:raises CalledProcessError: if the return code wasn't 0 and logs the processes stdout and stderr
|
||||
"""
|
||||
__tracebackhide__ = True
|
||||
return self.expect(0)
|
||||
|
||||
def expect(self, rc: int) -> "CommandResult":
|
||||
@@ -34,12 +40,14 @@ class CommandResult:
|
||||
:param rc: The expected return code
|
||||
:raises CalledProcessError: if the return code wasn't `rc` and logs the processes stdout and stderr
|
||||
"""
|
||||
__tracebackhide__ = True
|
||||
if self.rc != rc:
|
||||
logger.error("stdout: %s", self.stdout_s)
|
||||
logger.error("stderr: %s", self.stderr_s)
|
||||
raise subprocess.CalledProcessError(
|
||||
exc = subprocess.CalledProcessError(
|
||||
returncode=self.rc, cmd=self.cmd, stderr=self.stderr, output=self.stdout
|
||||
)
|
||||
raise exc
|
||||
return self
|
||||
|
||||
@property
|
||||
@@ -67,75 +75,53 @@ class CommandResult:
|
||||
Assumes an ok() result and returns the Commands stdout parsed as json
|
||||
:return: A parsed json object
|
||||
"""
|
||||
__tracebackhide__ = True
|
||||
self.ok()
|
||||
return json.loads(self.stdout)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Command:
|
||||
"""
|
||||
Provides a way of configuring a shell command and then running it
|
||||
calls Popen internally
|
||||
"""
|
||||
|
||||
argv: list[str]
|
||||
"""
|
||||
Arguments of the Process; argv[0] is the name of the binary
|
||||
"""
|
||||
env: dict[str, str] = dataclasses.field(default_factory=dict)
|
||||
"""
|
||||
environment variables; Note that `$PATH` is not added by default.
|
||||
Use `.with_env(**os.environ.copy())` to add `$PATH`
|
||||
"""
|
||||
_env: ManagedEnv
|
||||
stdin: bytes | None = None
|
||||
"""
|
||||
Things to pipe into stdin of the process
|
||||
"""
|
||||
cwd: Path | None = None
|
||||
"""
|
||||
current-working-directory fo the process
|
||||
"""
|
||||
cwd: Path = dataclasses.field(default=None)
|
||||
_logger: logging.Logger = dataclasses.field(default=logger, init=False)
|
||||
|
||||
def with_env(self, **kwargs) -> "Command":
|
||||
"""
|
||||
sets the env to the given environment variables
|
||||
:param kwargs: keyword arguments containing the environment
|
||||
:return: self, command is chainable
|
||||
"""
|
||||
self.env = kwargs
|
||||
return self
|
||||
|
||||
def update_env(self, **kwargs) -> "Command":
|
||||
"""
|
||||
updates the current environment with the given dict of variables
|
||||
:param kwargs: new or updated environment variables
|
||||
:return: self, command is chainable
|
||||
"""
|
||||
self.env.update(kwargs)
|
||||
return self
|
||||
def __post_init__(self):
|
||||
if self.cwd is None:
|
||||
self.cwd = self._env.dirs.home
|
||||
|
||||
def with_stdin(self, stdin: bytes) -> "Command":
|
||||
"""
|
||||
sets the input provided to stdin of the commands
|
||||
:param stdin: data to pipe into stdin
|
||||
:return: self, command is chainable
|
||||
"""
|
||||
self.stdin = stdin
|
||||
return self
|
||||
|
||||
def set_args(self, *argv: str) -> "Command":
|
||||
self.argv = list(argv)
|
||||
return self
|
||||
|
||||
def run(self) -> CommandResult:
|
||||
"""
|
||||
Runs the configured command
|
||||
:return: Information about the Result of the execution
|
||||
"""
|
||||
self._logger.debug("Running Command with args: %s", self.argv)
|
||||
proc = subprocess.Popen(
|
||||
self.argv,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
stdin=subprocess.PIPE if self.stdin else subprocess.DEVNULL,
|
||||
cwd=self.cwd,
|
||||
env=self.env,
|
||||
env=self._env.to_env(),
|
||||
)
|
||||
(stdout, stderr) = proc.communicate(input=self.stdin)
|
||||
rc = proc.returncode
|
||||
return CommandResult(cmd=self.argv, rc=rc, stdout=stdout, stderr=stderr)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def command(env: ManagedEnv) -> Callable[..., Command]:
|
||||
def wrapper(*args, **kwargs) -> Command:
|
||||
return Command(_env=env, *args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
@@ -0,0 +1,258 @@
|
||||
import dataclasses
|
||||
import logging
|
||||
import os
|
||||
import platform
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
SLASHES_IN_STORE_PATH_UNTIL_PACKAGE = "/nix/store/hash-program_name/".count("/")
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class _ManagedPath:
|
||||
"""
|
||||
Wrapper class to handle building the `PATH` environment variable
|
||||
"""
|
||||
|
||||
build_shell: dataclasses.InitVar[str]
|
||||
"""statically linked shell to use within builds which provides coreutils functionality"""
|
||||
_path: list[str] = dataclasses.field(default_factory=list)
|
||||
|
||||
def __post_init__(self, build_shell: str):
|
||||
self.prepend(build_shell)
|
||||
|
||||
def to_path(self) -> str:
|
||||
"""
|
||||
:return: string to be put into the `PATH` environment variable containing all added paths/programs
|
||||
"""
|
||||
return ":".join(self._path)
|
||||
|
||||
def prepend(self, exec_path: str | Path) -> "_ManagedPath":
|
||||
"""
|
||||
Adds the given file or folder at the FRONT of the path variable
|
||||
:param exec_path: executable or folder containing executables to be added
|
||||
:return: self, to allow for chaining
|
||||
"""
|
||||
self._path.insert(0, str(exec_path))
|
||||
return self
|
||||
|
||||
def append(self, exec_path: str | Path) -> "_ManagedPath":
|
||||
"""
|
||||
Adds the given file or folder at the END of the path variable
|
||||
:param exec_path: executable or folder containing executables to be added
|
||||
:return: self, to allow for chaining
|
||||
"""
|
||||
self._path.append(str(exec_path))
|
||||
return self
|
||||
|
||||
def insert_at(self, exec_path: str | Path, index: int) -> "_ManagedPath":
|
||||
"""
|
||||
Adds the given file or folder at the GIVEN INDEX of the path variable
|
||||
:param exec_path: executable or folder containing executable to be added
|
||||
:param index: where to insert the path
|
||||
:return: self, to allow for chaining
|
||||
"""
|
||||
self._path.insert(index, str(exec_path))
|
||||
return self
|
||||
|
||||
def add_program(self, program_name: str, all_associated: bool = True) -> "_ManagedPath":
|
||||
"""
|
||||
Adds the given program to the path by name.
|
||||
:param program_name: executable/program to add
|
||||
:param all_associated: if True, the folder containing the executable will be added instead. Otherwise, only the provided executable will be added
|
||||
:raises ValueError: if the program could not be found
|
||||
:return: self, to allow for chaining
|
||||
"""
|
||||
path = shutil.which(program_name)
|
||||
if path is None:
|
||||
msg = f"Couldn't find program {program_name!r}"
|
||||
raise ValueError(msg)
|
||||
# Convert to path object for better checking and operations
|
||||
path = Path(path)
|
||||
if all_associated and not path.is_dir():
|
||||
path = path.parent
|
||||
# handle as string within the data structure
|
||||
path = str(path)
|
||||
if path not in self._path:
|
||||
self._path.append(path)
|
||||
return self
|
||||
|
||||
def remove_path(self, exec_path: str | Path) -> "_ManagedPath":
|
||||
"""
|
||||
Removes the given file or folder from the path
|
||||
:param exec_path: file or folder to remove
|
||||
:raises ValueError: if the file or folder could not be found
|
||||
:return: self, to allow for chaining
|
||||
"""
|
||||
self._path.remove(str(exec_path))
|
||||
return self
|
||||
|
||||
def remove_program(self, program_name: str) -> "_ManagedPath":
|
||||
"""
|
||||
Removes the given executable/program from the path by name
|
||||
:param program_name: executable/program to remove
|
||||
:raises ValueError: if the program was not found or isn't present in path
|
||||
:return: self, to allow for chaining
|
||||
"""
|
||||
path = shutil.which(program_name)
|
||||
if path is None:
|
||||
msg = f"Couldn't find program {program_name!r}"
|
||||
raise ValueError(msg)
|
||||
if path in self._path or (path := str(Path(path).parent)) in self._path:
|
||||
self.remove_path(path)
|
||||
else:
|
||||
# Mirror behavior of `remove_path`
|
||||
msg = f"path.remove({program_name}): {program_name} not in path"
|
||||
raise ValueError(msg)
|
||||
return self
|
||||
|
||||
def to_sandbox_paths(self) -> list[str]:
|
||||
"""
|
||||
:return: list of strings to put into the `sandbox_paths` nix setting
|
||||
"""
|
||||
ret = []
|
||||
for p in self._path:
|
||||
if p.startswith("/nix/store/"):
|
||||
# adds the entire package to the sandbox,
|
||||
# to ensure that dependencies and libraries from within the package are also present
|
||||
ret.append("/".join(p.split("/")[:SLASHES_IN_STORE_PATH_UNTIL_PACKAGE]))
|
||||
else:
|
||||
ret.append(p)
|
||||
return ret
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class _Dirs:
|
||||
test_root: Path | None
|
||||
home: Path | None
|
||||
nix_log_dir: Path | None
|
||||
nix_state_dir: Path | None
|
||||
nix_conf_dir: Path | None
|
||||
nix_bin_dir: Path | None
|
||||
nix_store_dir: Path | None
|
||||
cache_dir: Path | None
|
||||
xdg_cache_home: Path | None
|
||||
"""used for nar caching"""
|
||||
|
||||
def get_env_keys(self) -> set[str]:
|
||||
return {f.name.upper() for f in dataclasses.fields(self)}
|
||||
|
||||
def to_env_vars(self) -> dict[str, str]:
|
||||
return {k.upper(): v for k, v in dataclasses.asdict(self).items() if v is not None}
|
||||
|
||||
|
||||
class ManagedEnv:
|
||||
def __init__(self, tmp_path: Path):
|
||||
# Things fetched from the global env
|
||||
build_shell = os.environ.get("BUILD_TEST_SHELL")
|
||||
global_path = os.environ.get("PATH")
|
||||
# `NIX_BIN_DIR` either propagated from us or set by meson
|
||||
# Set to the codebase internal output if started standalone
|
||||
# This is where the current lix binaries are located.
|
||||
# local import to avoid cyclic dependencies
|
||||
from functional2.testlib.utils import lix_base_folder # noqa: PLC0415
|
||||
|
||||
lix_bin = Path(os.environ.get("NIX_BIN_DIR", lix_base_folder / "outputs/out/bin"))
|
||||
|
||||
self._env = {}
|
||||
self.path = _ManagedPath(build_shell)
|
||||
self._tmp_path = tmp_path
|
||||
self.shell_dir = build_shell or "/bin"
|
||||
|
||||
self.dirs = _Dirs(
|
||||
test_root=self._get_dir(""),
|
||||
home=self._get_dir("test-home"),
|
||||
nix_log_dir=self._get_dir("var/log/nix"),
|
||||
nix_state_dir=self._get_dir("var/nix"),
|
||||
nix_conf_dir=self._get_dir("etc/nix"),
|
||||
nix_bin_dir=lix_bin,
|
||||
nix_store_dir=self._get_dir("nix/store"),
|
||||
cache_dir=self._get_dir("binary-cache"),
|
||||
xdg_cache_home=self._get_dir("test-home/.cache"),
|
||||
)
|
||||
self.path.prepend(self.dirs.nix_bin_dir)
|
||||
self.init_defaults(global_path)
|
||||
|
||||
def _get_dir(self, sub_path: str) -> Path:
|
||||
p = self._tmp_path / sub_path
|
||||
p.mkdir(parents=True, exist_ok=True)
|
||||
return p
|
||||
|
||||
def init_defaults(self, global_path: str):
|
||||
self._env = {
|
||||
# Do not use the system-wide or local config for git, but *none* instead
|
||||
"GIT_CONFIG_SYSTEM": "/dev/null",
|
||||
# Shell to use, required by lix to run sub processes / commands
|
||||
"SHELL": f"{self.shell_dir}/sh",
|
||||
# when writing things to the terminal (esp with man pages) use cat, to print the full output to stdout
|
||||
"PAGER": "cat",
|
||||
"BUILD_TEST_SHELL": self.shell_dir,
|
||||
}
|
||||
if platform.system() == "Darwin":
|
||||
# Darwin / Apple behaves differently and requires _NIX_TEST_NO_SANDBOX to be set for whatever reason
|
||||
self._env |= {"_NIX_TEST_NO_SANDBOX": "1"}
|
||||
# copy global path to maintain features usually provided by busybox
|
||||
[self.path.append(p) for p in global_path.split(":")]
|
||||
|
||||
def set_env(self, name: str, value: str):
|
||||
if name in self.dirs.get_env_keys():
|
||||
msg = f"Overriding paths should be done using the `env.dirs` attribute, use `env.dirs.{name.lower()}` instead."
|
||||
raise ValueError(msg)
|
||||
if name == "PATH":
|
||||
msg = "Setting of path not supported. use `env.path` instead"
|
||||
raise ValueError(msg)
|
||||
if value is None:
|
||||
msg = "setting to `None` is not allowed. did you mean to use `env.unset_env`?"
|
||||
raise ValueError(msg)
|
||||
self._env[name] = value
|
||||
|
||||
def __setitem__(self, key: str, value: str) -> None:
|
||||
return self.set_env(key, value)
|
||||
|
||||
def get_env(self, name: str, default: str | Path | None = None) -> str | Path | None:
|
||||
if name in self.dirs.get_env_keys():
|
||||
return getattr(self.dirs, name.lower())
|
||||
if name == "PATH":
|
||||
msg = "getting of path not supported, use `env.path` instead"
|
||||
raise ValueError(msg)
|
||||
return self._env.get(name, default)
|
||||
|
||||
def __getitem__(self, item: str) -> str | Path | None:
|
||||
itm = self.get_env(item)
|
||||
if itm is None:
|
||||
msg = f"{itm} is not set"
|
||||
raise KeyError(msg)
|
||||
return itm
|
||||
|
||||
def unset_env(self, name: str) -> str | None:
|
||||
if name in self.dirs.get_env_keys():
|
||||
msg = f"Overriding paths should be done using the `env.dirs` attribute, use `env.dirs.{name.lower()}` instead."
|
||||
raise ValueError(msg)
|
||||
return self._env.pop(name, None)
|
||||
|
||||
def __delitem__(self, key: str) -> str | None:
|
||||
itm = self.unset_env(key)
|
||||
if itm is None:
|
||||
msg = f"{itm} is not set"
|
||||
raise KeyError(msg)
|
||||
return itm
|
||||
|
||||
def to_env(self) -> dict[str, str]:
|
||||
ret = self.dirs.to_env_vars()
|
||||
ret["PATH"] = self.path.to_path()
|
||||
for k, v in self._env.copy().items():
|
||||
if v is not None:
|
||||
ret[k] = v
|
||||
else:
|
||||
logger.warning("environment variable '%s' is none", k)
|
||||
return ret
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def env(tmp_path: Path) -> ManagedEnv:
|
||||
return ManagedEnv(tmp_path)
|
||||
@@ -6,6 +6,7 @@ from collections.abc import Callable, Iterable
|
||||
|
||||
import pytest
|
||||
|
||||
from functional2.testlib.fixtures.env import ManagedEnv
|
||||
from functional2.testlib.fixtures.formatter import BalancedTemplater
|
||||
|
||||
|
||||
@@ -177,7 +178,7 @@ def _init_files(files: FileDeclaration, tmp_path: Path, request: pytest.FixtureR
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def files(tmp_path: Path, request: pytest.FixtureRequest) -> Path:
|
||||
def files(env: ManagedEnv, request: pytest.FixtureRequest) -> Path:
|
||||
"""
|
||||
Initializes the given files into the TempDir of the test.
|
||||
This ensures all necessary files and only those are present
|
||||
@@ -185,9 +186,10 @@ def files(tmp_path: Path, request: pytest.FixtureRequest) -> Path:
|
||||
The test is run once for each of the sets of files provided as the second argument.
|
||||
Each Set of files should be of the :py:type:`FileDeclaration` type
|
||||
|
||||
:param tmp_path: TempDir of the test
|
||||
:param env: environment to get the HOME path from
|
||||
:param request: Fixture information provided by pytest, used to parametrize the files
|
||||
:return: Path to where the files were created
|
||||
"""
|
||||
_init_files(request.param, tmp_path, request)
|
||||
return tmp_path
|
||||
home = env.dirs.home
|
||||
_init_files(request.param, home, request)
|
||||
return home
|
||||
|
||||
@@ -1,13 +1,14 @@
|
||||
import dataclasses
|
||||
import os
|
||||
from functools import partialmethod
|
||||
from pathlib import Path
|
||||
from typing import Any, AnyStr
|
||||
from textwrap import dedent
|
||||
from typing import Any
|
||||
from collections.abc import Callable, Generator
|
||||
|
||||
import pytest
|
||||
|
||||
from functional2.testlib.commands import CommandResult, Command
|
||||
from functional2.testlib.fixtures.command import CommandResult, Command
|
||||
from functional2.testlib.fixtures.env import ManagedEnv
|
||||
from functional2.testlib.utils import is_value_of_type
|
||||
|
||||
|
||||
@@ -18,7 +19,7 @@ class NixSettings:
|
||||
experimental_features: set[str] | None = None
|
||||
store: str | None = None
|
||||
"""
|
||||
The store to operate on (may be a path or other thing, see nix help-stores).
|
||||
The store to operate on (may be a path or other thing, see `nix help-stores`).
|
||||
|
||||
Note that this can be set to the test's store directory if you want to use
|
||||
/nix/store paths inside that test rather than NIX_STORE_DIR renaming
|
||||
@@ -40,8 +41,14 @@ class NixSettings:
|
||||
self.experimental_features = (self.experimental_features or set()) | set(names)
|
||||
return self
|
||||
|
||||
def to_config(self) -> str:
|
||||
config = ""
|
||||
def to_config(self, env: ManagedEnv) -> str:
|
||||
config = dedent(f"""
|
||||
show-trace = true
|
||||
sandbox = true
|
||||
extra-sandbox-paths = {" ".join(env.path.to_sandbox_paths())}
|
||||
""")
|
||||
# Note: newline at the end is required due to nix being nix;
|
||||
# FIXME(Jade): #953 this is annoying in the CLI too, we should fix it!
|
||||
|
||||
def serialise(value: Any) -> str:
|
||||
# TODO(Commentator2.0): why exactly are ints supported?
|
||||
@@ -65,83 +72,42 @@ class NixSettings:
|
||||
)
|
||||
return config
|
||||
|
||||
def to_env_overlay(self) -> dict[str, str]:
|
||||
ret = {"NIX_CONFIG": self.to_config()}
|
||||
def to_env_overlay(self, env: ManagedEnv) -> None:
|
||||
cfg = self.to_config(env)
|
||||
(env.dirs.nix_conf_dir / "nix.conf").write_text(cfg)
|
||||
env.set_env("NIX_CONFIG", cfg)
|
||||
if self.nix_store_dir:
|
||||
ret["NIX_STORE_DIR"] = str(self.nix_store_dir)
|
||||
return ret
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class NixCommand(Command):
|
||||
"""
|
||||
Custom Command class which applies the given NixSettings before the command is run
|
||||
"""
|
||||
|
||||
settings: NixSettings = dataclasses.field(default_factory=NixSettings)
|
||||
|
||||
def apply_nix_config(self):
|
||||
self.env.update(self.settings.to_env_overlay())
|
||||
|
||||
def run(self) -> CommandResult:
|
||||
self.apply_nix_config()
|
||||
return super().run()
|
||||
env.dirs.nix_store_dir = str(self.nix_store_dir)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Nix:
|
||||
test_root: Path
|
||||
env: ManagedEnv
|
||||
_settings: NixSettings | None = dataclasses.field(init=False, default=None)
|
||||
|
||||
def hermetic_env(self) -> dict[str, Path]:
|
||||
# mirroring vars-and-functions.sh
|
||||
home = self.test_root / "test-home"
|
||||
home.mkdir(parents=True, exist_ok=True)
|
||||
return {
|
||||
"NIX_LOCALSTATE_DIR": self.test_root / "var",
|
||||
"NIX_LOG_DIR": self.test_root / "var/log/nix",
|
||||
"NIX_STATE_DIR": self.test_root / "var/nix",
|
||||
"NIX_CONF_DIR": self.test_root / "etc",
|
||||
"NIX_DAEMON_SOCKET_PATH": self.test_root / "daemon-socket",
|
||||
"NIX_USER_CONF_FILES": "",
|
||||
"HOME": home,
|
||||
"XDG_CACHE_HOME": home / ".cache",
|
||||
}
|
||||
|
||||
def make_env(self) -> dict[AnyStr, AnyStr]:
|
||||
# We conservatively assume that people might want to successfully get
|
||||
# some env through to the subprocess, so we override whatever is in the
|
||||
# global env.
|
||||
d = os.environ.copy()
|
||||
d.update(self.hermetic_env())
|
||||
return d
|
||||
|
||||
def cmd(self, argv: list[str]) -> Command:
|
||||
return Command(argv=argv, cwd=self.test_root, env=self.make_env())
|
||||
|
||||
def settings(self, allow_builds: bool = False) -> NixSettings:
|
||||
@property
|
||||
def settings(self) -> NixSettings:
|
||||
"""
|
||||
Parameters:
|
||||
- allow_builds: relocate the Nix store so that builds work (however, makes store paths non-reproducible across test runs!)
|
||||
:return: the settings for the nix instance
|
||||
"""
|
||||
settings = NixSettings()
|
||||
store_path = self.test_root / "store"
|
||||
if allow_builds:
|
||||
settings.nix_store_dir = store_path
|
||||
else:
|
||||
settings.store = str(store_path)
|
||||
return settings
|
||||
if self._settings is None:
|
||||
self._settings = NixSettings()
|
||||
self._settings.store = f"local?root={self.env.dirs.test_root}&store=/nix/store"
|
||||
|
||||
def nix_cmd(self, argv: list[str], flake: bool = False) -> NixCommand:
|
||||
return self._settings
|
||||
|
||||
def nix_cmd(self, argv: list[str], flake: bool = False) -> Command:
|
||||
"""
|
||||
Constructs a NixCommand with the appropriate settings.
|
||||
"""
|
||||
settings = self.settings()
|
||||
# Create a copy of settings to not have a writing side effect
|
||||
settings = dataclasses.replace(self.settings)
|
||||
if flake:
|
||||
settings.feature("nix-command", "flakes")
|
||||
settings.to_env_overlay(self.env)
|
||||
return Command(argv=argv, _env=self.env)
|
||||
|
||||
return NixCommand(argv=argv, cwd=self.test_root, env=self.make_env(), settings=settings)
|
||||
|
||||
def nix(self, cmd: list[str], nix_exe: str = "nix", flake: bool = False) -> NixCommand:
|
||||
def nix(self, cmd: list[str], nix_exe: str = "nix", flake: bool = False) -> Command:
|
||||
return self.nix_cmd([nix_exe, *cmd], flake=flake)
|
||||
|
||||
# Mark each of these as correct as they are not ClassVars, but we also don't want to turn off RUF045
|
||||
@@ -154,25 +120,32 @@ class Nix:
|
||||
nix_prefetch_url = partialmethod(nix, nix_exe="nix-prefetch-url") # noqa: RUF045
|
||||
|
||||
def eval(self, expr: str, settings: NixSettings | None = None) -> CommandResult:
|
||||
if settings is None:
|
||||
settings = self.settings()
|
||||
# clone due to reference-shenanigans
|
||||
settings = dataclasses.replace(settings).feature("nix-command")
|
||||
"""
|
||||
calls `nix eval --json --expr {expr}` using the given expression
|
||||
:param expr: what to evaluate
|
||||
:param settings: if none, the global settings will be used, otherwise the given one
|
||||
:return: result of the evaluation
|
||||
"""
|
||||
orig = dataclasses.replace(self.settings)
|
||||
self._settings = settings or self.settings
|
||||
self.settings.feature("nix-command")
|
||||
|
||||
cmd = self.nix(["eval", "--json", "--expr", expr])
|
||||
cmd.settings = settings
|
||||
# restore previous settings
|
||||
self._settings = orig
|
||||
return cmd.run()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def nix(tmp_path: Path) -> Generator[Nix, Any, None]:
|
||||
def nix(tmp_path: Path, env: ManagedEnv) -> Generator[Nix, Any, None]:
|
||||
"""
|
||||
Provides a rich way of calling `nix`.
|
||||
For pre-applied commands use `nix.nix_instantiate`, `nix.nix_build` etc.
|
||||
After configuring the command, use `.run()` to run it
|
||||
"""
|
||||
yield Nix(tmp_path)
|
||||
yield Nix(env)
|
||||
# when things are done using the nix store, the permissions for the store are read only
|
||||
# after the test was executed, we set the permissions to rwx (write being the important part)
|
||||
# for pytest to be able to delete the files during cleanup
|
||||
Command(argv=["chmod", "-R", "+w", str(tmp_path.absolute())], env=os.environ.copy()).run().ok()
|
||||
cmd = Command(argv=["chmod", "-R", "+w", str(tmp_path.absolute())], _env=env)
|
||||
cmd.run().ok()
|
||||
|
||||
@@ -1,14 +1,12 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from _pytest.fixtures import FixtureRequest
|
||||
|
||||
from functional2.testlib.commands import Command
|
||||
from functional2.testlib.fixtures.command import Command
|
||||
from functional2.testlib.fixtures.env import ManagedEnv
|
||||
|
||||
|
||||
@pytest.fixture(name="pytest_command")
|
||||
def _pytest_command(request: FixtureRequest, tmp_path: Path, do_snapshot_update: bool) -> Command:
|
||||
def _pytest_command(env: ManagedEnv, request: FixtureRequest, do_snapshot_update: bool) -> Command:
|
||||
"""
|
||||
returns a preconfigured pytest command.
|
||||
The following things must be passed into the parametrization:
|
||||
@@ -25,12 +23,9 @@ def _pytest_command(request: FixtureRequest, tmp_path: Path, do_snapshot_update:
|
||||
else:
|
||||
flags = params
|
||||
propagate_update = True
|
||||
|
||||
env = os.environ.copy()
|
||||
env.path.add_program("pytest")
|
||||
cwd = env.get_env("HOME") / "functional2"
|
||||
cmd = Command(argv=["pytest", "--basetemp", "../pytest_files", *flags], _env=env, cwd=cwd)
|
||||
if propagate_update and do_snapshot_update:
|
||||
flags.append("--accept-tests")
|
||||
else:
|
||||
env.pop("_NIX_TEST_ACCEPT", None)
|
||||
return Command(
|
||||
argv=["pytest", "--basetemp", "../pytest_files", *flags], cwd=tmp_path / "functional2"
|
||||
).with_env(**env)
|
||||
return cmd
|
||||
|
||||
@@ -94,7 +94,8 @@ def snapshot(
|
||||
as otherwise the expected output won't be updated.
|
||||
:return: a snapshot object which one can use `==` on
|
||||
"""
|
||||
|
||||
tmp_path = tmp_path / "test-home"
|
||||
tmp_path.mkdir(parents=True, exist_ok=True)
|
||||
snaps: list[Snapshot] = []
|
||||
|
||||
def create_snapshot(expected_output_path: str) -> Snapshot:
|
||||
|
||||
+26
-22
@@ -1,42 +1,42 @@
|
||||
import logging
|
||||
import os
|
||||
import stat
|
||||
from pathlib import Path
|
||||
from collections.abc import Callable
|
||||
from subprocess import CalledProcessError
|
||||
|
||||
import pytest
|
||||
from _pytest.logging import LogCaptureFixture
|
||||
from functional2.testlib.commands import Command
|
||||
from functional2.testlib.fixtures.command import Command
|
||||
from functional2.testlib.fixtures.env import ManagedEnv
|
||||
from functional2.testlib.fixtures.file_helper import File
|
||||
|
||||
|
||||
def test_command_valid_runs():
|
||||
cmd = Command(["echo", "water"]).with_env(**os.environ.copy())
|
||||
def test_command_valid_runs(command: Callable[[list[str]], Command]):
|
||||
cmd = command(["echo", "water"])
|
||||
cmd.run().ok()
|
||||
|
||||
|
||||
def test_command_captures_stdout():
|
||||
cmd = Command(["echo", "fire"]).with_env(**os.environ.copy())
|
||||
def test_command_captures_stdout(command: Callable[[list[str]], Command]):
|
||||
cmd = command(["echo", "fire"])
|
||||
res = cmd.run().ok()
|
||||
assert res.stdout_s == "fire\n"
|
||||
|
||||
|
||||
def test_command_plain_strips():
|
||||
cmd = Command(["echo", " earth "]).with_env(**os.environ.copy())
|
||||
def test_command_plain_strips(command: Callable[[list[str]], Command]):
|
||||
cmd = command(["echo", " earth "])
|
||||
res = cmd.run().ok()
|
||||
assert res.stdout_plain == "earth"
|
||||
|
||||
|
||||
def test_command_stdin_passed_correctly():
|
||||
def test_command_stdin_passed_correctly(command: Callable[[list[str]], Command]):
|
||||
inp = b"air"
|
||||
cmd = Command(["cat", "/dev/stdin"]).with_stdin(inp).with_env(**os.environ.copy())
|
||||
cmd = command(["cat", "/dev/stdin"]).with_stdin(inp)
|
||||
res = cmd.run().ok()
|
||||
assert res.stdout == inp
|
||||
|
||||
|
||||
def test_command_expect_failure():
|
||||
cmd = Command(["grep", "xxx"]).with_env(**os.environ.copy()).with_stdin(b"")
|
||||
cmd.run().expect(1)
|
||||
def test_command_expect_failure(env: ManagedEnv):
|
||||
env.path.add_program("grep")
|
||||
Command(_env=env, argv=["grep", "xxx"], stdin=b"").run().expect(1)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@@ -50,12 +50,14 @@ def test_command_expect_failure():
|
||||
],
|
||||
indirect=True,
|
||||
)
|
||||
def test_command_ok_fails_on_bad_exit_code(files: Path, caplog: LogCaptureFixture):
|
||||
cmd = Command(["./script.sh"], cwd=files).with_env(**os.environ.copy())
|
||||
res = cmd.run()
|
||||
@pytest.mark.usefixtures("files")
|
||||
def test_command_ok_fails_on_bad_exit_code(
|
||||
command: Callable[[list[str]], Command], caplog: LogCaptureFixture
|
||||
):
|
||||
cmd = command(["./script.sh"])
|
||||
|
||||
with pytest.raises(CalledProcessError), caplog.at_level(logging.ERROR):
|
||||
res.ok()
|
||||
cmd.run().ok()
|
||||
msgs = caplog.messages
|
||||
assert len(msgs) == 2
|
||||
out_msg, err_msg = msgs
|
||||
@@ -74,12 +76,14 @@ def test_command_ok_fails_on_bad_exit_code(files: Path, caplog: LogCaptureFixtur
|
||||
],
|
||||
indirect=True,
|
||||
)
|
||||
def test_command_exec_fails_on_other_bad_exit_code(files: Path, caplog: LogCaptureFixture):
|
||||
cmd = Command(["./script.sh"], cwd=files).with_env(**os.environ.copy())
|
||||
res = cmd.run()
|
||||
@pytest.mark.usefixtures("files")
|
||||
def test_command_exec_fails_on_other_bad_exit_code(
|
||||
command: Callable[[list[str]], Command], caplog: LogCaptureFixture
|
||||
):
|
||||
cmd = command(["./script.sh"])
|
||||
|
||||
with pytest.raises(CalledProcessError), caplog.at_level(logging.ERROR):
|
||||
res.expect(1)
|
||||
cmd.run().expect(1)
|
||||
msgs = caplog.messages
|
||||
assert len(msgs) == 2
|
||||
out_msg, err_msg = msgs
|
||||
@@ -0,0 +1,219 @@
|
||||
import dataclasses
|
||||
import re
|
||||
import shutil
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from functional2.testlib.fixtures.env import ManagedEnv, _ManagedPath
|
||||
|
||||
|
||||
def test_env_inits_defaults(tmp_path: Path):
|
||||
env = ManagedEnv(tmp_path)
|
||||
assert env.get_env("GIT_CONFIG_SYSTEM") == "/dev/null"
|
||||
if sys.platform != "darwin":
|
||||
# Darwin doesn't like sandboxes so we don't have a sandbox shell here
|
||||
assert "busybox" in env.get_env("SHELL")
|
||||
else:
|
||||
assert "/bin/sh" in env.get_env("SHELL")
|
||||
assert env.get_env("PAGER") == "cat"
|
||||
|
||||
assert env.dirs.test_root == tmp_path
|
||||
assert env.dirs.home == tmp_path / "test-home"
|
||||
assert env.dirs.nix_store_dir == tmp_path / "nix/store"
|
||||
|
||||
|
||||
def test_env_unknown_none(env: ManagedEnv):
|
||||
assert env.get_env("SOME_UNKNOWN_STUFF") is None
|
||||
|
||||
|
||||
def test_env_unknown_default(env: ManagedEnv):
|
||||
assert env.get_env("QUESTION_OF_LIFE", "42") == "42"
|
||||
|
||||
|
||||
def test_env_sets(env: ManagedEnv):
|
||||
assert env.get_env("DRGN") is None
|
||||
env.set_env("DRGN", "cute")
|
||||
assert env.get_env("DRGN") == "cute"
|
||||
|
||||
|
||||
def test_env_setitem_success(env: ManagedEnv):
|
||||
env["HELLO"] = "world"
|
||||
assert env.get_env("HELLO") == "world"
|
||||
|
||||
|
||||
def test_env_getitem_known(env: ManagedEnv):
|
||||
env["HELLO"] = "world"
|
||||
assert env["HELLO"] == "world"
|
||||
|
||||
|
||||
def test_env_getitem_unknown(env: ManagedEnv):
|
||||
with pytest.raises(KeyError):
|
||||
_ = env["HELLO"]
|
||||
|
||||
|
||||
def test_env_overrides_custom(env: ManagedEnv):
|
||||
env.set_env("VALID", "you")
|
||||
assert env.get_env("VALID") == "you"
|
||||
env.set_env("VALID", "every creature")
|
||||
assert env.get_env("VALID") == "every creature"
|
||||
|
||||
|
||||
def test_env_overrides_defaults(env: ManagedEnv):
|
||||
assert env.get_env("PAGER") == "cat"
|
||||
env.set_env("PAGER", "bat")
|
||||
assert env.get_env("PAGER") == "bat"
|
||||
|
||||
|
||||
def test_env_unset_custom(env: ManagedEnv):
|
||||
env.set_env("FAILURE", "me")
|
||||
assert env.get_env("FAILURE") == "me"
|
||||
env.unset_env("FAILURE")
|
||||
assert env.get_env("FAILURE") is None
|
||||
|
||||
|
||||
def test_env_unset_default(env: ManagedEnv):
|
||||
assert env.get_env("PAGER") == "cat"
|
||||
env.unset_env("PAGER")
|
||||
assert env.get_env("PAGER") is None
|
||||
|
||||
|
||||
def test_env_set_none_errors(env: ManagedEnv):
|
||||
with pytest.raises(ValueError, match=r".+env.unset_env.+"):
|
||||
env.set_env("PAGER", None) # type: ignore
|
||||
|
||||
|
||||
def test_env_set_no_dirs(env: ManagedEnv):
|
||||
with pytest.raises(ValueError, match=r"Overriding paths should be done .+"):
|
||||
env.set_env("HOME", "/home/zelda")
|
||||
|
||||
|
||||
def test_env_get_dir_works(env: ManagedEnv):
|
||||
assert env.get_env("HOME") == env.dirs.home
|
||||
|
||||
|
||||
def test_env_dirs_created(env: ManagedEnv):
|
||||
fields = dataclasses.asdict(env.dirs).values()
|
||||
assert len(fields) > 1
|
||||
for field in fields:
|
||||
field: Path
|
||||
assert field.exists()
|
||||
|
||||
|
||||
def test_env_get_path_fails(env: ManagedEnv):
|
||||
with pytest.raises(ValueError, match=r".+env\.path.+"):
|
||||
_ = env.get_env("PATH")
|
||||
|
||||
|
||||
def test_env_set_path_fails(env: ManagedEnv):
|
||||
with pytest.raises(ValueError, match=r".+env\.path.+"):
|
||||
env.set_env("PATH", "a:b")
|
||||
|
||||
|
||||
def test_env_to_env(tmp_path: Path):
|
||||
env = ManagedEnv(tmp_path)
|
||||
assert set(env.to_env().keys()) == {
|
||||
"GIT_CONFIG_SYSTEM",
|
||||
"SHELL",
|
||||
"PAGER",
|
||||
"HOME",
|
||||
"TEST_ROOT",
|
||||
"NIX_LOG_DIR",
|
||||
"NIX_STATE_DIR",
|
||||
"NIX_CONF_DIR",
|
||||
"NIX_BIN_DIR",
|
||||
"NIX_STORE_DIR",
|
||||
"CACHE_DIR",
|
||||
"XDG_CACHE_HOME",
|
||||
"PATH",
|
||||
"BUILD_TEST_SHELL",
|
||||
} | ({"_NIX_TEST_NO_SANDBOX"} if sys.platform == "darwin" else set())
|
||||
|
||||
|
||||
def test_path_inits_build_shell():
|
||||
path = _ManagedPath("/path/to/build_shell")
|
||||
assert len(path._path) == 1
|
||||
assert path._path[0] == "/path/to/build_shell"
|
||||
|
||||
|
||||
def test_path_appends_at_end():
|
||||
path = _ManagedPath("/path/to/build_shell")
|
||||
path.append("other_path")
|
||||
assert len(path._path) == 2
|
||||
assert path._path[1] == "other_path"
|
||||
|
||||
|
||||
def test_path_prepends_at_front():
|
||||
path = _ManagedPath("/path/to/build_shell")
|
||||
path.prepend("other/path")
|
||||
assert len(path._path) == 2
|
||||
assert path._path[0] == "other/path"
|
||||
|
||||
|
||||
def test_remove_path_known():
|
||||
path = _ManagedPath("/path/to/build_shell")
|
||||
path.append("other/path")
|
||||
assert len(path._path) == 2
|
||||
path.remove_path("other/path")
|
||||
assert path._path == ["/path/to/build_shell"]
|
||||
|
||||
|
||||
def test_remove_path_unknown():
|
||||
path = _ManagedPath("/path/to/build_shell")
|
||||
path.append("other/path")
|
||||
assert len(path._path) == 2
|
||||
with pytest.raises(ValueError, match=r".+ x not in list"):
|
||||
path.remove_path("/home/nest/builder")
|
||||
assert len(path._path) == 2
|
||||
|
||||
|
||||
def test_path_add_program_resolves():
|
||||
path = _ManagedPath("/path/to/build_shell")
|
||||
pytest_path = shutil.which("pytest")
|
||||
path.add_program("pytest", False)
|
||||
assert len(path._path) == 2
|
||||
assert path._path[1] == pytest_path
|
||||
|
||||
|
||||
def test_path_add_program_entire_dir():
|
||||
path = _ManagedPath("/path/to/build_shell")
|
||||
pytest_path = shutil.which("pytest")
|
||||
path.add_program("pytest", True)
|
||||
assert len(path._path) == 2
|
||||
actual = path._path[1]
|
||||
assert actual != pytest_path
|
||||
assert actual == pytest_path.rpartition("/")[0]
|
||||
|
||||
|
||||
def test_path_remove_program_specific():
|
||||
path = _ManagedPath("/path/to/build_shell")
|
||||
path.add_program("pytest", False)
|
||||
assert len(path._path) == 2
|
||||
path.remove_program("pytest")
|
||||
assert len(path._path) == 1
|
||||
|
||||
|
||||
def test_path_remove_program_associated():
|
||||
path = _ManagedPath("/path/to/build_shell")
|
||||
path.add_program("pytest", True)
|
||||
assert len(path._path) == 2
|
||||
path.remove_program("pytest")
|
||||
assert len(path._path) == 1
|
||||
|
||||
|
||||
def test_path_to_path():
|
||||
path = _ManagedPath("/path/to/build_shell")
|
||||
path.prepend("pictures/latest/kate")
|
||||
path.prepend("out/bin/lix")
|
||||
path.append("fops/with/noms")
|
||||
assert path.to_path() == "out/bin/lix:pictures/latest/kate:/path/to/build_shell:fops/with/noms"
|
||||
|
||||
|
||||
def test_path_sandbox_entire_store_package():
|
||||
path = _ManagedPath("/path/to/build_shell")
|
||||
path.add_program(shutil.which("pytest"))
|
||||
sb_paths = path.to_sandbox_paths()
|
||||
assert len(sb_paths) == 2
|
||||
assert sb_paths[0] == "/path/to/build_shell"
|
||||
assert re.fullmatch(r"^/nix/store/\w{32}-python3-3\.\d{1,2}\.\d{1,2}-[^/]+$", sb_paths[1])
|
||||
@@ -1,72 +1,77 @@
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from functional2.testlib.fixtures.env import ManagedEnv
|
||||
from functional2.testlib.fixtures.nix import NixSettings
|
||||
|
||||
|
||||
def test_nix_settings_serializes_xf():
|
||||
def test_nix_settings_serializes_xf(env: ManagedEnv):
|
||||
settings = NixSettings(nix_store_dir=Path("/store/nix"))
|
||||
settings.feature("a", "b")
|
||||
|
||||
expected = "experimental-features = a b\n"
|
||||
assert settings.to_config() == expected
|
||||
expected = r"(.|\n)*experimental-features = (a|b) (a|b)\n(.|\n)*"
|
||||
assert re.fullmatch(expected, settings.to_config(env), re.MULTILINE)
|
||||
|
||||
|
||||
def test_nix_settings_serializes_store():
|
||||
def test_nix_settings_serializes_store(env: ManagedEnv):
|
||||
settings = NixSettings(nix_store_dir=Path("/store/nix"))
|
||||
settings.store = "some/path"
|
||||
|
||||
expected = "store = some/path\n"
|
||||
assert settings.to_config() == expected
|
||||
assert expected in settings.to_config(env)
|
||||
|
||||
|
||||
def test_nix_settings_serializes_both():
|
||||
def test_nix_settings_serializes_both(env: ManagedEnv):
|
||||
settings = NixSettings(nix_store_dir=Path("/store/nix"))
|
||||
settings.feature("a", "b")
|
||||
settings.store = "some/path"
|
||||
|
||||
expected = "experimental-features = a b\nstore = some/path\n"
|
||||
assert settings.to_config() == expected
|
||||
assert expected in settings.to_config(env)
|
||||
|
||||
|
||||
def test_nix_settings_ser_fails_bad_top_level_type():
|
||||
def test_nix_settings_ser_fails_bad_top_level_type(env: ManagedEnv):
|
||||
settings = NixSettings(nix_store_dir=Path("/store/nix"))
|
||||
settings.experimental_features = {"a": "b"} # type: ignore we are testing the types here
|
||||
|
||||
with pytest.raises(ValueError, match=r"Value is unsupported in nix config: {'a': 'b'}"):
|
||||
settings.to_config()
|
||||
settings.to_config(env)
|
||||
|
||||
|
||||
def test_nix_settings_ser_fails_bad_sub_type():
|
||||
def test_nix_settings_ser_fails_bad_sub_type(env: ManagedEnv):
|
||||
settings = NixSettings(nix_store_dir=Path("/store/nix"))
|
||||
settings.experimental_features = [["a", "b"], "c"] # type: ignore we are testing the types here
|
||||
|
||||
with pytest.raises(ValueError, match=r"Value is unsupported in nix config: .+"):
|
||||
settings.to_config()
|
||||
settings.to_config(env)
|
||||
|
||||
|
||||
def test_nix_settings_fails_without_store_and_store_dir():
|
||||
def test_nix_settings_fails_without_store_and_store_dir(env: ManagedEnv):
|
||||
settings = NixSettings()
|
||||
|
||||
with pytest.raises(
|
||||
AssertionError,
|
||||
match=r"Failing to set either nix_store_dir or store will cause accidental use of the system store.",
|
||||
):
|
||||
settings.to_config()
|
||||
settings.to_config(env)
|
||||
|
||||
|
||||
def test_nix_settings_to_env_overlay_no_store_dir():
|
||||
def test_nix_settings_to_env_overlay_no_store_dir(tmp_path: Path):
|
||||
env = ManagedEnv(tmp_path)
|
||||
settings = NixSettings()
|
||||
settings.store = "some/path"
|
||||
|
||||
expected = {"NIX_CONFIG": "store = some/path\n"}
|
||||
assert settings.to_env_overlay() == expected
|
||||
settings.to_env_overlay(env)
|
||||
assert "store = some/path\n" in env._env["NIX_CONFIG"]
|
||||
|
||||
|
||||
def test_nix_settings_to_env_overlay_store_dir():
|
||||
def test_nix_settings_to_env_overlay_store_dir(tmp_path: Path):
|
||||
env = ManagedEnv(tmp_path)
|
||||
settings = NixSettings()
|
||||
settings.nix_store_dir = Path("/some/path")
|
||||
|
||||
expected = {"NIX_CONFIG": "", "NIX_STORE_DIR": "/some/path"}
|
||||
assert settings.to_env_overlay() == expected
|
||||
settings.to_env_overlay(env)
|
||||
assert "store = " not in env._env["NIX_CONFIG"]
|
||||
assert env.dirs.nix_store_dir == "/some/path"
|
||||
|
||||
@@ -5,7 +5,8 @@ from textwrap import dedent
|
||||
|
||||
import pytest
|
||||
from _pytest.logging import LogCaptureFixture
|
||||
from functional2.testlib.commands import Command
|
||||
from functional2.testlib.fixtures.command import Command
|
||||
from functional2.testlib.fixtures.env import ManagedEnv
|
||||
from functional2.testlib.fixtures.file_helper import (
|
||||
CopyFile,
|
||||
File,
|
||||
@@ -116,11 +117,11 @@ _update_test_files = _get_f2_snapshot_files(
|
||||
)
|
||||
@pytest.mark.parametrize("set_env", [False, True])
|
||||
@pytest.mark.usefixtures("files")
|
||||
def test_do_update_true_when_any_set(pytest_command: Command, set_env: bool):
|
||||
def test_do_update_true_when_any_set(env: ManagedEnv, pytest_command: Command, set_env: bool):
|
||||
if not (set_env or "--accept-tests" in pytest_command.argv):
|
||||
pytest.skip("not this test case")
|
||||
if set_env:
|
||||
pytest_command.update_env(_NIX_TEST_ACCEPT="1")
|
||||
env.set_env("_NIX_TEST_ACCEPT", "1")
|
||||
pytest_command.run().ok()
|
||||
|
||||
|
||||
@@ -132,7 +133,7 @@ def _snapshot_test_files(content: str) -> FileDeclaration:
|
||||
"test_snapshot.py": File(
|
||||
dedent(f"""
|
||||
def test_snapshot(snapshot, tmp_path):
|
||||
(tmp_path / "out.exp").write_text("{content}")
|
||||
(tmp_path / "test-home" / "out.exp").write_text("{content}")
|
||||
assert snapshot("out.exp") == "plush plush"
|
||||
""")
|
||||
)
|
||||
@@ -168,7 +169,7 @@ def test_snapshot_fails_on_diff(pytest_command: Command):
|
||||
indirect=True,
|
||||
)
|
||||
def test_snapshot_updates_diff(files: Path, pytest_command: Command):
|
||||
output_file = files / "pytest_files/test_snapshot0/out.exp"
|
||||
output_file = files / "pytest_files/test_snapshot0/test-home/out.exp"
|
||||
pytest_command.run().ok()
|
||||
assert output_file.read_text() == "plush plush"
|
||||
|
||||
@@ -181,7 +182,7 @@ def test_snapshot_updates_diff(files: Path, pytest_command: Command):
|
||||
def test_snapshot_updates_shares_updated_location_without_symlink(
|
||||
files: Path, pytest_command: Command
|
||||
):
|
||||
expected_path = "pytest_files/test_snapshot0/out.exp"
|
||||
expected_path = "pytest_files/test_snapshot0/test-home/out.exp"
|
||||
output_file = files / expected_path
|
||||
res = pytest_command.run().ok()
|
||||
assert output_file.read_text() == "plush plush"
|
||||
@@ -195,7 +196,7 @@ def test_snapshot_updates_shares_updated_location_without_symlink(
|
||||
indirect=True,
|
||||
)
|
||||
def test_snapshot_marks_skip_after_update(files: Path, pytest_command: Command):
|
||||
expected_path = "pytest_files/test_snapshot0/out.exp"
|
||||
expected_path = "pytest_files/test_snapshot0/test-home/out.exp"
|
||||
output_file = files / expected_path
|
||||
res = pytest_command.run().ok()
|
||||
assert output_file.read_text() == "plush plush"
|
||||
@@ -213,8 +214,8 @@ def test_snapshot_marks_skip_after_update(files: Path, pytest_command: Command):
|
||||
"test_snapshot.py": File(
|
||||
dedent("""
|
||||
def test_snapshot(snapshot, tmp_path):
|
||||
(tmp_path / "out.exp").write_text("fops plush")
|
||||
(tmp_path / "err.exp").write_text("shork plush")
|
||||
(tmp_path / "test-home" / "out.exp").write_text("fops plush")
|
||||
(tmp_path / "test-home" / "err.exp").write_text("shork plush")
|
||||
assert snapshot("out.exp") == "plush plush"
|
||||
assert snapshot("err.exp") == "blobhaj"
|
||||
""")
|
||||
@@ -229,7 +230,7 @@ def test_snapshot_marks_skip_after_update(files: Path, pytest_command: Command):
|
||||
indirect=True,
|
||||
)
|
||||
def test_snapshot_updates_multiple(files: Path, pytest_command: Command):
|
||||
expected_path = "pytest_files/test_snapshot0"
|
||||
expected_path = "pytest_files/test_snapshot0/test-home"
|
||||
first_file = files / expected_path / "out.exp"
|
||||
second_file = files / expected_path / "err.exp"
|
||||
res = pytest_command.run().ok()
|
||||
@@ -249,8 +250,8 @@ def test_snapshot_updates_multiple(files: Path, pytest_command: Command):
|
||||
"test_snapshot.py": File(
|
||||
dedent("""
|
||||
def test_snapshot(snapshot, tmp_path):
|
||||
(tmp_path / "updated.txt").write_text("snek plush")
|
||||
(tmp_path / "out.exp").symlink_to("./updated.txt")
|
||||
(tmp_path / "test-home" / "updated.txt").write_text("snek plush")
|
||||
(tmp_path / "test-home" / "out.exp").symlink_to("./updated.txt")
|
||||
assert snapshot("out.exp") == "plush plush"
|
||||
""")
|
||||
)
|
||||
@@ -264,7 +265,7 @@ def test_snapshot_updates_multiple(files: Path, pytest_command: Command):
|
||||
indirect=True,
|
||||
)
|
||||
def test_snapshot_updates_no_location_when_symlink(files: Path, pytest_command: Command):
|
||||
expected_path = "pytest_files/test_snapshot0/updated.txt"
|
||||
expected_path = "pytest_files/test_snapshot0/test-home/updated.txt"
|
||||
output_file = files / expected_path
|
||||
res = pytest_command.run().ok()
|
||||
assert output_file.read_text() == "plush plush"
|
||||
|
||||
@@ -129,7 +129,9 @@ def write_with_export_header(nar: NarItem, name: bytes, out: NarListener):
|
||||
# Store::exportPath
|
||||
serialize_nar(nar, out)
|
||||
out.int_(export_magic)
|
||||
out.str_(b"/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-" + name)
|
||||
# Due to `nix` setting the store using chroots /nix/store is still the test-local store and not the global one
|
||||
store_path = b"/nix/store"
|
||||
out.str_(store_path + b"/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-" + name)
|
||||
# no references
|
||||
out.int_(0)
|
||||
# no deriver
|
||||
|
||||
Reference in New Issue
Block a user