Merge changes I6830c2fc,Ib88565a1,I0b280587 into main
* changes: functional2: Added ruff formatter functional2: use loggers fix codestyle of functional2
This commit is contained in:
@@ -39,3 +39,5 @@ buildtime.bin
|
||||
|
||||
# Python compiled files from the code generators and test suite
|
||||
*.pyc
|
||||
|
||||
**/.idea
|
||||
|
||||
+4
-1
@@ -115,7 +115,10 @@ pre-commit-run {
|
||||
};
|
||||
treefmt = {
|
||||
enable = true;
|
||||
settings.formatters = [ pkgs.nixfmt-rfc-style ];
|
||||
settings.formatters = [
|
||||
pkgs.nixfmt-rfc-style
|
||||
pkgs.ruff
|
||||
];
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
@@ -197,6 +197,7 @@ let
|
||||
lixPythonForBuild = python3.pythonOnBuildForHost.withPackages (p: [
|
||||
p.pytest
|
||||
p.pytest-xdist
|
||||
p.ruff
|
||||
p.python-frontmatter
|
||||
p.aiohttp
|
||||
]);
|
||||
@@ -535,6 +536,7 @@ stdenv.mkDerivation (finalAttrs: {
|
||||
# wrapped python instead of build inputs for its python inputs
|
||||
p.pytest
|
||||
p.pytest-xdist
|
||||
p.ruff
|
||||
p.aiohttp
|
||||
p.python-frontmatter
|
||||
|
||||
|
||||
@@ -28,13 +28,15 @@ def custom_sub_command(request: pytest.FixtureRequest, custom_sub_command_path:
|
||||
command = request.param
|
||||
|
||||
executable = custom_sub_command_path / f"lix-{command}"
|
||||
executable.write_text(dedent(f"""\
|
||||
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:])
|
||||
"""))
|
||||
""")
|
||||
)
|
||||
executable.chmod(stat.S_IXUSR | stat.S_IRUSR | stat.S_IWUSR)
|
||||
|
||||
return command
|
||||
@@ -44,11 +46,13 @@ def custom_sub_command(request: pytest.FixtureRequest, custom_sub_command_path:
|
||||
def failing_sub_command(failing_sub_command_path: Path, custom_sub_command: str) -> str:
|
||||
# Create an external command that will intentionally cause an error
|
||||
executable = failing_sub_command_path / f"lix-{custom_sub_command}"
|
||||
executable.write_text(dedent(f"""\
|
||||
executable.write_text(
|
||||
dedent(f"""\
|
||||
#!{sys.executable}
|
||||
import sys
|
||||
sys.exit(42)
|
||||
"""))
|
||||
""")
|
||||
)
|
||||
executable.chmod(stat.S_IXUSR | stat.S_IRUSR | stat.S_IWUSR)
|
||||
|
||||
return custom_sub_command
|
||||
@@ -64,8 +68,9 @@ def path(custom_sub_command_path: Path) -> str:
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def path_with_failure(request: pytest.FixtureRequest, custom_sub_command_path: Path,
|
||||
failing_sub_command_path: Path) -> str:
|
||||
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
|
||||
(first, fail) = request.param
|
||||
search_path = os.environ.get("PATH").split(":")
|
||||
@@ -82,12 +87,13 @@ def path_with_failure(request: pytest.FixtureRequest, custom_sub_command_path: P
|
||||
return ":".join(search_path)
|
||||
|
||||
|
||||
@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):
|
||||
@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.with_env(PATH=path)
|
||||
@@ -97,13 +103,14 @@ def test_sub_commands(nix: Nix, path: str, custom_sub_command: str, nix_exe: str
|
||||
nix_command.run().expect(expected)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path_with_failure, expected", [((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.parametrize(
|
||||
("path_with_failure", "expected"),
|
||||
[((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
|
||||
):
|
||||
# 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")
|
||||
@@ -113,14 +120,17 @@ def test_sub_command_path_order(nix: Nix, path_with_failure: str, failing_sub_co
|
||||
nix_command.run().expect(expected)
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="TODO: we do not support auto completion for custom sub commands for now")
|
||||
def test_custom_sub_commands_auto_completion(nix: Nix, tmp_path: Path):
|
||||
pass
|
||||
@pytest.mark.skip(
|
||||
reason="TODO(raito): we do not support auto completion for custom sub commands for now"
|
||||
)
|
||||
def test_custom_sub_commands_auto_completion(nix: Nix, tmp_path: Path): ...
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="TODO: we do not test flag handling for custom sub commands for now")
|
||||
@pytest.mark.skip(
|
||||
reason="TODO(raito): we do not test flag handling for custom sub commands for now"
|
||||
)
|
||||
def test_custom_sub_command_flag_handling(nix: Nix, tmp_path: Path):
|
||||
# TODO: Short, long and multiple flags should be tested as well.
|
||||
# TODO: `--` special flag
|
||||
# TODO: test positional arguments, but only `nix-copy-closure` implements some and it's pesky to test here.
|
||||
pass
|
||||
# Short, long and multiple flags should be tested as well.
|
||||
# `--` special flag
|
||||
# test positional arguments, but only `nix-copy-closure` implements some, and it's pesky to test here.
|
||||
...
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
pytest_plugins = (
|
||||
"functional2.testlib.fixtures.file_helper",
|
||||
"functional2.testlib.fixtures.formatter",
|
||||
"functional2.testlib.fixtures.logger",
|
||||
"functional2.testlib.fixtures.nix",
|
||||
)
|
||||
|
||||
@@ -12,31 +12,33 @@ class ShouldError(NamedTuple):
|
||||
|
||||
ERR_CASES: list[ShouldError] = [
|
||||
# FIXME(jade): expect-test system for pytest that allows for updating these easily
|
||||
ShouldError('{}', '"x',
|
||||
"""error: missing closing quote in selection path '"x'"""),
|
||||
ShouldError("{}", '"x', """error: missing closing quote in selection path '"x'"""),
|
||||
ShouldError(
|
||||
'[]', 'x',
|
||||
"""error: the value being indexed in the selection path 'x' at '' should be a set but is a list: [ ]"""
|
||||
"[]",
|
||||
"x",
|
||||
"""error: the value being indexed in the selection path 'x' at '' should be a set but is a list: [ ]""",
|
||||
),
|
||||
ShouldError(
|
||||
'{}', '1',
|
||||
"""error: the expression selected by the selection path '1' should be a list but is a set: { }"""
|
||||
"{}",
|
||||
"1",
|
||||
"""error: the expression selected by the selection path '1' should be a list but is a set: { }""",
|
||||
),
|
||||
ShouldError('{}', '.',
|
||||
"""error: empty attribute name in selection path '.'"""),
|
||||
ShouldError('{ x."" = 2; }', 'x.""',
|
||||
"""error: empty attribute name in selection path 'x.""'"""),
|
||||
ShouldError('{ x."".y = 2; }', 'x."".y',
|
||||
"""error: empty attribute name in selection path 'x."".y'"""),
|
||||
ShouldError("{}", ".", """error: empty attribute name in selection path '.'"""),
|
||||
ShouldError(
|
||||
'[]', '1',
|
||||
"""error: list index 1 in selection path '1' is out of range for list [ ]"""
|
||||
'{ x."" = 2; }', 'x.""', """error: empty attribute name in selection path 'x.""'"""
|
||||
),
|
||||
ShouldError(
|
||||
'{ x.y = { z = 2; a = 3; }; }', 'x.y.c',
|
||||
'{ x."".y = 2; }', 'x."".y', """error: empty attribute name in selection path 'x."".y'"""
|
||||
),
|
||||
ShouldError(
|
||||
"[]", "1", """error: list index 1 in selection path '1' is out of range for list [ ]"""
|
||||
),
|
||||
ShouldError(
|
||||
"{ x.y = { z = 2; a = 3; }; }",
|
||||
"x.y.c",
|
||||
dedent("""\
|
||||
error: attribute 'c' in selection path 'x.y.c' not found inside path 'x.y', whose contents are: { a = 3; z = 2; }
|
||||
Did you mean one of a or z?""")
|
||||
Did you mean one of a or z?"""),
|
||||
),
|
||||
]
|
||||
|
||||
@@ -45,12 +47,13 @@ ERR_CASES: list[ShouldError] = [
|
||||
# to pass -A unconditionally and then allow a blank attribute to mean the whole
|
||||
# thing
|
||||
def test_attrpath_accepts_empty_attr_as_no_attr(nix: Nix):
|
||||
assert nix.nix_instantiate(['--eval', '--expr', '{}', '-A',
|
||||
'']).run().ok().stdout_plain == '{ }'
|
||||
assert (
|
||||
nix.nix_instantiate(["--eval", "--expr", "{}", "-A", ""]).run().ok().stdout_plain == "{ }"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(['expr', 'attr', 'error'], ERR_CASES)
|
||||
@pytest.mark.parametrize(("expr", "attr", "error"), ERR_CASES)
|
||||
def test_attrpath_error(nix: Nix, expr: str, attr: str, error: str):
|
||||
res = nix.nix_instantiate(['--eval', '--expr', expr, '-A', attr]).run()
|
||||
res = nix.nix_instantiate(["--eval", "--expr", expr, "-A", attr]).run()
|
||||
|
||||
assert res.expect(1).stderr_plain == error
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
from functional2.testlib.fixtures.nix import Nix
|
||||
|
||||
|
||||
def test_trivial_addition(nix: Nix):
|
||||
assert nix.eval('1 + 1').json() == 2
|
||||
assert nix.eval("1 + 1").json() == 2
|
||||
|
||||
@@ -1,10 +1,11 @@
|
||||
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")
|
||||
assert nix.eval("builtins ? fetchTree", settings).json() == False
|
||||
assert nix.eval("builtins ? fetchTree", settings).json() is False
|
||||
|
||||
settings.feature("flakes")
|
||||
assert nix.eval("builtins ? fetchTree", settings).json() == True
|
||||
assert nix.eval("builtins ? fetchTree", settings).json() is True
|
||||
|
||||
@@ -1,30 +1,36 @@
|
||||
from logging import Logger
|
||||
from pathlib import Path
|
||||
from textwrap import dedent
|
||||
from functional2.testlib.fixtures.nix import Nix
|
||||
import re
|
||||
|
||||
def test_invalid_flake_lock(nix: Nix, tmp_path: Path):
|
||||
flake_dir = tmp_path / 'flake'
|
||||
|
||||
def test_invalid_flake_lock(nix: Nix, tmp_path: Path, logger: Logger):
|
||||
flake_dir = tmp_path / "flake"
|
||||
flake_dir.mkdir()
|
||||
|
||||
(flake_dir / 'flake.nix').write_text(dedent("""
|
||||
(flake_dir / "flake.nix").write_text(
|
||||
dedent("""
|
||||
{
|
||||
inputs = {};
|
||||
outputs = inputs: {};
|
||||
}
|
||||
"""))
|
||||
(flake_dir / 'flake.lock').write_text(dedent("""
|
||||
""")
|
||||
)
|
||||
(flake_dir / "flake.lock").write_text(
|
||||
dedent("""
|
||||
{
|
||||
this flake.lock is obviously invalid
|
||||
}
|
||||
"""))
|
||||
""")
|
||||
)
|
||||
|
||||
cmd = nix.nix(["build"], flake=True)
|
||||
cmd.cwd = flake_dir
|
||||
res = cmd.run().expect(1)
|
||||
print(res.stderr_plain)
|
||||
logger.info(res.stderr_plain)
|
||||
|
||||
ERROR_RE1 = re.compile(fr"while updating the lock file of flake 'path:{flake_dir}.+'")
|
||||
ERROR_RE2 = re.compile(fr"while parsing the lock file at .+")
|
||||
assert ERROR_RE1.search(res.stderr_plain)
|
||||
assert ERROR_RE2.search(res.stderr_plain)
|
||||
error_re1 = re.compile(rf"while updating the lock file of flake 'path:{flake_dir}.+'")
|
||||
error_re2 = re.compile(r"while parsing the lock file at .+")
|
||||
assert error_re1.search(res.stderr_plain)
|
||||
assert error_re2.search(res.stderr_plain)
|
||||
|
||||
@@ -1,37 +1,39 @@
|
||||
from logging import Logger
|
||||
from pathlib import Path
|
||||
from textwrap import dedent
|
||||
from functional2.testlib.fixtures.nix import Nix
|
||||
import re
|
||||
|
||||
|
||||
def test_purity_traversal(nix: Nix, tmp_path: Path):
|
||||
ERROR_RE = re.compile(r"error: access to absolute path '.+' is forbidden in pure eval mode")
|
||||
def test_purity_traversal(nix: Nix, tmp_path: Path, logger: Logger):
|
||||
error_re = re.compile(r"error: access to absolute path '.+' is forbidden in pure eval mode")
|
||||
|
||||
flake_dir = tmp_path / 'flake'
|
||||
flake_dir = tmp_path / "flake"
|
||||
flake_dir.mkdir()
|
||||
|
||||
evilpath = tmp_path / 'sekrit.txt'
|
||||
evilpath.write_text('kitty kitty')
|
||||
evilpath = tmp_path / "sekrit.txt"
|
||||
evilpath.write_text("kitty kitty")
|
||||
|
||||
evilnix = tmp_path / 'default.nix'
|
||||
evilnix = tmp_path / "default.nix"
|
||||
evilnix.write_text('"woof"')
|
||||
|
||||
goodnix = flake_dir / 'good.nix'
|
||||
goodnix = flake_dir / "good.nix"
|
||||
goodnix.write_text("1")
|
||||
|
||||
nested_dir = flake_dir / 'nested' / 'nested2'
|
||||
nested_dir = flake_dir / "nested" / "nested2"
|
||||
nested_dir.mkdir(parents=True)
|
||||
(nested_dir / 'good.nix').write_text("1")
|
||||
(nested_dir / "good.nix").write_text("1")
|
||||
|
||||
(flake_dir / 'evil-link').symlink_to(evilpath)
|
||||
(flake_dir / 'evil-default-nix').symlink_to(tmp_path)
|
||||
(flake_dir / 'less-evil-link').symlink_to(tmp_path / 'link-to-flake')
|
||||
(tmp_path / 'link-to-flake').symlink_to(flake_dir / 'flake.nix')
|
||||
(flake_dir / "evil-link").symlink_to(evilpath)
|
||||
(flake_dir / "evil-default-nix").symlink_to(tmp_path)
|
||||
(flake_dir / "less-evil-link").symlink_to(tmp_path / "link-to-flake")
|
||||
(tmp_path / "link-to-flake").symlink_to(flake_dir / "flake.nix")
|
||||
|
||||
(flake_dir / 'nested-good.nix').symlink_to('nested/nested2/good.nix')
|
||||
(flake_dir / 'nested-bad.nix').symlink_to('nested/../../nested2/good.nix')
|
||||
(flake_dir / "nested-good.nix").symlink_to("nested/nested2/good.nix")
|
||||
(flake_dir / "nested-bad.nix").symlink_to("nested/../../nested2/good.nix")
|
||||
|
||||
(flake_dir / 'flake.nix').write_text(dedent("""
|
||||
(flake_dir / "flake.nix").write_text(
|
||||
dedent("""
|
||||
{
|
||||
inputs = {};
|
||||
outputs = inputs: {
|
||||
@@ -52,16 +54,17 @@ def test_purity_traversal(nix: Nix, tmp_path: Path):
|
||||
good5 = toString (import ./nested-good.nix);
|
||||
};
|
||||
}
|
||||
""").replace('@ABSPATH@', str(evilpath.absolute())))
|
||||
""").replace("@ABSPATH@", str(evilpath.absolute()))
|
||||
)
|
||||
|
||||
for idx in range(1, 10):
|
||||
cmd = nix.nix(['eval', f'.#bad{idx}'], flake=True)
|
||||
cmd = nix.nix(["eval", f".#bad{idx}"], flake=True)
|
||||
cmd.cwd = flake_dir
|
||||
res = cmd.run().expect(1)
|
||||
print(res.stderr_plain)
|
||||
assert ERROR_RE.search(res.stderr_plain)
|
||||
logger.info(res.stderr_plain)
|
||||
assert error_re.search(res.stderr_plain)
|
||||
for idx in range(1, 6):
|
||||
cmd = nix.nix(['eval', f'.#good{idx}'], flake=True)
|
||||
cmd = nix.nix(["eval", f".#good{idx}"], flake=True)
|
||||
cmd.cwd = flake_dir
|
||||
res = cmd.run().expect(0)
|
||||
assert res.stdout_plain == '"1"'
|
||||
|
||||
@@ -1,2 +1,184 @@
|
||||
[project]
|
||||
name = "functional2"
|
||||
version = "2"
|
||||
requires-python = ">=3.11"
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
addopts = "-p no:xonsh"
|
||||
log_cli = true
|
||||
log_cli_level = "INFO"
|
||||
# how the logs are being printed, default is `"%(filename)s %(lineno)d %(levelname)s %(message)s"`
|
||||
# Example log:
|
||||
# 2025-05-09T14:06:23Z [ INFO] [test_someting] This is a test message
|
||||
# 2025-05-09T14:06:25Z [ ERROR] [test_smt_else] Some error related log
|
||||
log_cli_format = "%(asctime)s [%(levelname)8s] [%(name)s] %(message)s"
|
||||
log_cli_date_format = "%Y-%m-%dT%H:%M:%SZ"
|
||||
|
||||
[tool.ruff]
|
||||
line-length = 100
|
||||
indent-width = 4
|
||||
# unless --fix or --no-fix is provided, automatically fix fixable violations
|
||||
fix = true
|
||||
# don't fix stuff that could break things
|
||||
unsafe-fixes = false
|
||||
|
||||
# we might want to switch it up for some integration stuff to "json" or "gitlab"
|
||||
output-format = "grouped"
|
||||
|
||||
# ignore files, which are ignored in the .gitignore file
|
||||
respect-gitignore = true
|
||||
|
||||
# show what violations have been fixed
|
||||
show-fixes = true
|
||||
|
||||
# need to test if "." or ".." is correct, as we assume functional2 to be the root directory for imports
|
||||
src = [".."]
|
||||
|
||||
[tool.ruff.lint]
|
||||
|
||||
preview = true
|
||||
|
||||
# do not allow characters like the minus sign, asterix oparator etc
|
||||
# which can be confused with dash (-) and star (*) respectively
|
||||
allowed-confusables = []
|
||||
|
||||
# ignore "unused variable" rules, for identifiers like `_`, `__`, `_var` etc
|
||||
# but NOT for `_var_` etc
|
||||
#
|
||||
# could be replaced by `^_$` if we only want to ignore `_` but nothing else
|
||||
dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
|
||||
|
||||
# we do not use custom logger objects for now
|
||||
logger-objects = []
|
||||
|
||||
# Comments to be ignored by commented-out code detection
|
||||
task-tags = ["TODO", "FIXME", "XXX"]
|
||||
# In order:
|
||||
# E: pycodestyle error violations
|
||||
# E4: import styles
|
||||
# E7: multi statements and semicolon uses as well as comparasion styles
|
||||
# E9: check for development setup issues (io errors while reading py files)
|
||||
# F: pyflakes violations; all enabled
|
||||
# F4: import style (unused; star imports)
|
||||
# F5: format style violations (% formatting, .format, fstrings)
|
||||
# F6: dictionary and variable unpacking
|
||||
# F7: loop and function keyword issues (return, yield, break, continue etc)
|
||||
# F8: undefined variables
|
||||
# F9: NotImplementedError stuff
|
||||
## Non-defaults:
|
||||
# ERA: commented-out-code
|
||||
# ASYNC: asyncio related things
|
||||
# ANN: Annotation stuff
|
||||
# ANN0: argument annotations
|
||||
# disabled: ANN002: annotation for *args
|
||||
# disabled: ANN002: annotation for **kwargs
|
||||
# removed: ANN1
|
||||
# ANN2: return type annotation
|
||||
# disabled: ANN4: no any type; not enabled as we support anys in multiple places
|
||||
# A: builtin shadowing
|
||||
# C4: list and generator comprehensions
|
||||
# EM: don't pass strings directly into exceptions, but use a variable; avoids duplicate printing of the message
|
||||
# ISC: implicit string concatination
|
||||
# INP: require __init__.py in all packages
|
||||
# LOG: creation of logger objects
|
||||
# G: style of logging messages
|
||||
# PIE: unnecessary providing of stuff (placeholders, definitions, duplicates etc)
|
||||
# T20: disallow prints in favor of using a logger
|
||||
# PT: pytest formatting things; fixtures, asserts, parametrizastions etc
|
||||
# Q: quote styling
|
||||
# RSE: check on raise statements
|
||||
# RET: return and continue conventions
|
||||
# SIM: code simplifications (double negation etc); all enabled
|
||||
# SIM1: collapsable if and bool things
|
||||
# SIM2: double negation in comparasions (`not a == b` instead of `a != b` etc)
|
||||
# SIM3: yoda conditions use `foo == "Foo"` instead of `"Foo" == foo`
|
||||
# SIM9: defaults for get and zip
|
||||
# TD: enforce TODO comment style
|
||||
# disabled TD001: allow FIXME and XXX comments
|
||||
# disabled TD003: todos don't require an explicit issue link for us
|
||||
# ARG: disallow unused arguments
|
||||
# PTH: use pathlib instead of os calls
|
||||
# N: enforce PEP-8 naming
|
||||
# PERF: performance things about lists and iterators
|
||||
# DOC: docstyling requirements
|
||||
# PL: general linting things
|
||||
# PLC: Conventions
|
||||
# PLC01: type mismatches
|
||||
# PLC02: iteration of dicts and sets
|
||||
# PLC04: minor import things
|
||||
# PLC1: bad compares
|
||||
# PLC2: non-ascii characters and dunder calls
|
||||
# PLC3: lambda things
|
||||
# PLE: Error
|
||||
# PLE01: non-local and init things
|
||||
# PLE03: invalid dunder returns
|
||||
# PLE06: index errors and __all__ things
|
||||
# disabled: PLE1: covered by other checks
|
||||
# PLE2: invalid escape sequences
|
||||
# disabled: PLR: Refactoring; covered by other rule sets
|
||||
# disabled: PLW: Warnings; covered by other rule stets
|
||||
# UP: use modern python features instead of by now depracated ones
|
||||
# RUF: Ambiguity and other general linting things
|
||||
#
|
||||
#
|
||||
#
|
||||
# Notincluded rule sets:
|
||||
# FAST: we don't use FAST Api
|
||||
# YTT: things about sys.version; not relevant here
|
||||
# S: conflicts with testing things and is used for production code, not test code
|
||||
# BLE: except without defining what to expect, covered by pytest
|
||||
# FBT: boolean arguments to functions
|
||||
# B: general codestyle things; covered by other rulesets
|
||||
# COM: trailing commas; covered by other rulesets
|
||||
# CPY: we don't need/have copyright notices in each file
|
||||
# DTZ: datetime things and formatting; not a usecase for us
|
||||
# T10: debugger things
|
||||
# DJ: we don't use django
|
||||
# EXE: we don't build an executable
|
||||
# FIX: we do carry around fixmes and other thigns to be done in separate commits
|
||||
# FA: things about future annotations
|
||||
# INT: we don't use gettext (translation interface)
|
||||
# ICN: import alias and import from banning
|
||||
# PYI: things for pyi files
|
||||
# SLF: accessing of private members
|
||||
# SLOT: subclassing of builtin-types (str, tuple, namedtuple) related issues
|
||||
# TID: banning of certain imports
|
||||
# TC: typechecking imports (typechecking imports required to be inside of a `if TYPECHECKING` block
|
||||
# FLY: string joins
|
||||
# I: import sorting; covered by other rulesets
|
||||
# C90: we don't use mccable
|
||||
# NPY: we don't use numpy
|
||||
# PD: we don't use pandas
|
||||
# D: doesn't support sphinx style docstyles as of 2025-05-01; see https://github.com/astral-sh/ruff/pull/13286
|
||||
# PGH: things about pygrep, we don't use
|
||||
# FURB: covered by other rule sets
|
||||
# TRY: try and raise related things, not helpful as we only do testing
|
||||
select = ["E4", "E7", "E9", "F", "ERA", "ASYNC", "ANN0", "ANN2", "A", "C4", "EM", "ISC", "INP", "LOG", "G", "PIE", "T20", "PT", "Q", "RSE", "RET", "SIM", "TD", "ARG", "PTH", "N", "PERF", "PLC", "PLE", "UP", "RUF"]
|
||||
ignore = ["ANN002", "ANN003", "TD001", "TD003", "PLE1", "RUF005"]
|
||||
|
||||
[tool.ruff.lint.per-file-ignores]
|
||||
# ignore open() and os.path.join() calls in test_evil_nars, as that file is working with raw bytes
|
||||
# which pathlib does not support
|
||||
"**/store/test_evil_nars.py" = ["PTH118", "PTH123"]
|
||||
|
||||
[tool.ruff.format]
|
||||
indent-style = "space"
|
||||
quote-style = "double"
|
||||
# allow for `def test(a,b):` style declarators and don't force arguments to all have a separate line
|
||||
skip-magic-trailing-comma = true
|
||||
|
||||
line-ending = "lf"
|
||||
|
||||
# if code within docstrings should be formatted too
|
||||
docstring-code-format = true
|
||||
# takes into acount indentation of code within docstrings
|
||||
docstring-code-line-length = "dynamic"
|
||||
|
||||
## Rule specific
|
||||
[tool.ruff.lint.flake8-annotations]
|
||||
# no return type requirement, if a function will only ever return `None`
|
||||
suppress-none-returning = true
|
||||
|
||||
[tool.ruff.lint.flake8-errmsg]
|
||||
# Allow raw strings to be up to 10 characters in error messages
|
||||
max-string-length = 10
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import os
|
||||
import unicodedata
|
||||
from io import BytesIO
|
||||
from logging import Logger
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
@@ -15,91 +16,111 @@ from ..testlib.nar import (
|
||||
write_with_export_header,
|
||||
)
|
||||
|
||||
meow_orig = 'méow'
|
||||
meow_nfc_ = unicodedata.normalize('NFC', meow_orig)
|
||||
meow_nfd_ = unicodedata.normalize('NFD', meow_orig)
|
||||
meow_nfc = meow_nfc_.encode('utf-8')
|
||||
meow_nfd = meow_nfd_.encode('utf-8')
|
||||
meow_orig = "méow"
|
||||
meow_nfc_ = unicodedata.normalize("NFC", meow_orig)
|
||||
meow_nfd_ = unicodedata.normalize("NFD", meow_orig)
|
||||
meow_nfc = meow_nfc_.encode("utf-8")
|
||||
meow_nfd = meow_nfd_.encode("utf-8")
|
||||
assert meow_nfc != meow_nfd
|
||||
|
||||
EVIL_NARS: list[tuple[str, NarItem]] = [
|
||||
('valid-dir-1', DirectoryUnordered([
|
||||
(b'a-nested', DirectoryUnordered([
|
||||
(b'loopy', Symlink(b'../abc-nested'))
|
||||
])),
|
||||
(b'b-file', Regular(False, b'meow kbity')),
|
||||
(b'c-exe', Regular(True, b'#!/usr/bin/env cat\nmeow kbity')),
|
||||
])),
|
||||
('invalid-slashes-1', DirectoryUnordered([
|
||||
(b'meow', Symlink(b'meowmeow')),
|
||||
(b'meow/nya', Regular(False, b'eepy')),
|
||||
])),
|
||||
('invalid-dot-1', DirectoryUnordered([
|
||||
(b'.', Symlink(b'meowmeow')),
|
||||
])),
|
||||
('invalid-dot-2', DirectoryUnordered([
|
||||
(b'..', Symlink(b'meowmeow')),
|
||||
])),
|
||||
('invalid-nul-1', DirectoryUnordered([
|
||||
(b'meow\0nya', Symlink(b'meowmeow')),
|
||||
])),
|
||||
('invalid-misorder-1', DirectoryUnordered([
|
||||
(b'zzz', Regular(False, b'eepy')),
|
||||
(b'kbity', Regular(False, b'meow')),
|
||||
])),
|
||||
('invalid-dupe-1', DirectoryUnordered([
|
||||
(b'zzz', Regular(False, b'eepy')),
|
||||
(b'zzz', Regular(False, b'meow')),
|
||||
])),
|
||||
('invalid-dupe-2', DirectoryUnordered([
|
||||
(b'zzz', DirectoryUnordered([
|
||||
(b'meow', Regular(False, b'kbity'))
|
||||
])),
|
||||
(b'zzz', Regular(False, b'meow')),
|
||||
])),
|
||||
('invalid-dupe-3', DirectoryUnordered([
|
||||
(b'zzz', DirectoryUnordered([
|
||||
(b'meow', Regular(False, b'kbity'))
|
||||
])),
|
||||
(b'zzz', DirectoryUnordered([
|
||||
(b'meow', Regular(False, b'kbityy'))
|
||||
])),
|
||||
])),
|
||||
('invalid-dupe-4', DirectoryUnordered([
|
||||
(b'zzz', Symlink(b'../kbity')),
|
||||
(b'zzz', DirectoryUnordered([
|
||||
(b'meow', Regular(False, b'kbityy'))
|
||||
])),
|
||||
])),
|
||||
('invalid-casehack-1', DirectoryUnordered([
|
||||
(b'ZZZ~nix~case~hack~2', Regular(False, b'meow')),
|
||||
(b'zzz~nix~case~hack~1', Regular(False, b'eepy')),
|
||||
])),
|
||||
('invalid-casehack-2', DirectoryUnordered([
|
||||
(b'ZZZ~nix~case~hack~1', Regular(False, b'meow')),
|
||||
(b'zzz~nix~case~hack~1', Regular(False, b'eepy')),
|
||||
])),
|
||||
(
|
||||
"valid-dir-1",
|
||||
DirectoryUnordered(
|
||||
[
|
||||
(b"a-nested", DirectoryUnordered([(b"loopy", Symlink(b"../abc-nested"))])),
|
||||
(b"b-file", Regular(False, b"meow kbity")),
|
||||
(b"c-exe", Regular(True, b"#!/usr/bin/env cat\nmeow kbity")),
|
||||
]
|
||||
),
|
||||
),
|
||||
(
|
||||
"invalid-slashes-1",
|
||||
DirectoryUnordered(
|
||||
[(b"meow", Symlink(b"meowmeow")), (b"meow/nya", Regular(False, b"eepy"))]
|
||||
),
|
||||
),
|
||||
("invalid-dot-1", DirectoryUnordered([(b".", Symlink(b"meowmeow"))])),
|
||||
("invalid-dot-2", DirectoryUnordered([(b"..", Symlink(b"meowmeow"))])),
|
||||
("invalid-nul-1", DirectoryUnordered([(b"meow\0nya", Symlink(b"meowmeow"))])),
|
||||
(
|
||||
"invalid-misorder-1",
|
||||
DirectoryUnordered(
|
||||
[(b"zzz", Regular(False, b"eepy")), (b"kbity", Regular(False, b"meow"))]
|
||||
),
|
||||
),
|
||||
(
|
||||
"invalid-dupe-1",
|
||||
DirectoryUnordered([(b"zzz", Regular(False, b"eepy")), (b"zzz", Regular(False, b"meow"))]),
|
||||
),
|
||||
(
|
||||
"invalid-dupe-2",
|
||||
DirectoryUnordered(
|
||||
[
|
||||
(b"zzz", DirectoryUnordered([(b"meow", Regular(False, b"kbity"))])),
|
||||
(b"zzz", Regular(False, b"meow")),
|
||||
]
|
||||
),
|
||||
),
|
||||
(
|
||||
"invalid-dupe-3",
|
||||
DirectoryUnordered(
|
||||
[
|
||||
(b"zzz", DirectoryUnordered([(b"meow", Regular(False, b"kbity"))])),
|
||||
(b"zzz", DirectoryUnordered([(b"meow", Regular(False, b"kbityy"))])),
|
||||
]
|
||||
),
|
||||
),
|
||||
(
|
||||
"invalid-dupe-4",
|
||||
DirectoryUnordered(
|
||||
[
|
||||
(b"zzz", Symlink(b"../kbity")),
|
||||
(b"zzz", DirectoryUnordered([(b"meow", Regular(False, b"kbityy"))])),
|
||||
]
|
||||
),
|
||||
),
|
||||
(
|
||||
"invalid-casehack-1",
|
||||
DirectoryUnordered(
|
||||
[
|
||||
(b"ZZZ~nix~case~hack~2", Regular(False, b"meow")),
|
||||
(b"zzz~nix~case~hack~1", Regular(False, b"eepy")),
|
||||
]
|
||||
),
|
||||
),
|
||||
(
|
||||
"invalid-casehack-2",
|
||||
DirectoryUnordered(
|
||||
[
|
||||
(b"ZZZ~nix~case~hack~1", Regular(False, b"meow")),
|
||||
(b"zzz~nix~case~hack~1", Regular(False, b"eepy")),
|
||||
]
|
||||
),
|
||||
),
|
||||
]
|
||||
|
||||
@pytest.mark.parametrize(['name', 'nar'], EVIL_NARS)
|
||||
def test_evil_nar(nix: Nix, name: str, nar: NarItem):
|
||||
|
||||
@pytest.mark.parametrize(("name", "nar"), EVIL_NARS)
|
||||
def test_evil_nar(nix: Nix, name: str, nar: NarItem, logger: Logger):
|
||||
bio = BytesIO()
|
||||
|
||||
listener = NarListener(bio)
|
||||
write_with_export_header(nar, name.encode(), listener)
|
||||
print(nar)
|
||||
logger.info(nar)
|
||||
|
||||
if name.startswith('valid-'):
|
||||
if name.startswith("valid-"):
|
||||
expected_rc = 0
|
||||
elif name.startswith('invalid-'):
|
||||
elif name.startswith("invalid-"):
|
||||
expected_rc = 1
|
||||
else:
|
||||
raise ValueError('bad name', name)
|
||||
raise ValueError("bad name", name)
|
||||
|
||||
res = nix.nix_store(['--import']).with_stdin(bio.getvalue()).run().expect(expected_rc)
|
||||
print(res)
|
||||
res = nix.nix_store(["--import"]).with_stdin(bio.getvalue()).run().expect(expected_rc)
|
||||
logger.info(res)
|
||||
|
||||
def test_unicode_evil_nar(nix: Nix, tmp_path: Path):
|
||||
|
||||
def test_unicode_evil_nar(nix: Nix, tmp_path: Path, logger: Logger):
|
||||
"""
|
||||
Depending on the filesystem in use, filenames that are equal modulo unicode
|
||||
normalization may hit the same file or not.
|
||||
@@ -107,23 +128,37 @@ def test_unicode_evil_nar(nix: Nix, tmp_path: Path):
|
||||
On macOS, such collisions will result in hitting the same file. We detect
|
||||
if the fs is like this before checking what Lix does.
|
||||
"""
|
||||
with open(os.path.join(bytes(tmp_path), meow_nfc), 'wb') as fh:
|
||||
fh.write(b'meow')
|
||||
with open(os.path.join(bytes(tmp_path), meow_nfc), "wb") as fh:
|
||||
fh.write(b"meow")
|
||||
|
||||
try:
|
||||
with open(os.path.join(bytes(tmp_path), meow_nfd), 'rb') as fh:
|
||||
assert fh.read() == b'meow'
|
||||
with open(os.path.join(bytes(tmp_path), meow_nfd), "rb") as fh:
|
||||
assert fh.read() == b"meow"
|
||||
except FileNotFoundError:
|
||||
# normalization is not applied to this system
|
||||
pytest.skip('filesystem does not use unicode normalization')
|
||||
pytest.skip("filesystem does not use unicode normalization")
|
||||
|
||||
test_evil_nar(nix, 'invalid-unicode-normalization-1', DirectoryUnordered([
|
||||
# méow
|
||||
(meow_nfd, Regular(False, b'eepy')),
|
||||
(meow_nfc, Symlink(b'meowmeow')),
|
||||
]))
|
||||
test_evil_nar(nix, 'invalid-unicode-normalization-2', DirectoryUnordered([
|
||||
# méow
|
||||
(meow_nfd, Symlink(b'meowmeow')),
|
||||
(meow_nfc, Regular(False, b'eepy')),
|
||||
]))
|
||||
test_evil_nar(
|
||||
nix,
|
||||
"invalid-unicode-normalization-1",
|
||||
DirectoryUnordered(
|
||||
[
|
||||
# méow
|
||||
(meow_nfd, Regular(False, b"eepy")),
|
||||
(meow_nfc, Symlink(b"meowmeow")),
|
||||
]
|
||||
),
|
||||
logger,
|
||||
)
|
||||
test_evil_nar(
|
||||
nix,
|
||||
"invalid-unicode-normalization-2",
|
||||
DirectoryUnordered(
|
||||
[
|
||||
# méow
|
||||
(meow_nfd, Symlink(b"meowmeow")),
|
||||
(meow_nfc, Regular(False, b"eepy")),
|
||||
]
|
||||
),
|
||||
logger,
|
||||
)
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
import shutil
|
||||
from abc import ABC, abstractmethod
|
||||
from enum import Enum
|
||||
from enum import StrEnum
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -21,7 +21,6 @@ class Fileish(ABC):
|
||||
:param path: TempDir for the test
|
||||
:param origin: Directory the tests originates in. Used to adjust relative paths
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class _ByContentFileish(Fileish, ABC):
|
||||
@@ -34,7 +33,6 @@ class _ByContentFileish(Fileish, ABC):
|
||||
Returns the content, which should be present in the current path
|
||||
:return: content as a string
|
||||
"""
|
||||
pass
|
||||
|
||||
def copy_to(self, path: Path, origin: Path) -> None:
|
||||
path.write_text(self.get_content(origin))
|
||||
@@ -81,7 +79,7 @@ class CopyTree(Fileish):
|
||||
|
||||
|
||||
class CopyTemplate(_ByContentFileish):
|
||||
def __init__(self, template: str, values: Dict[str, Any], mode: int | None = None):
|
||||
def __init__(self, template: str, values: dict[str, Any], mode: int | None = None):
|
||||
"""
|
||||
Declares a file as an initiated version of the given file template
|
||||
:param template: source template's file name. Parameters formatted as `{key_name}` are replaced by corresponding values
|
||||
@@ -103,7 +101,7 @@ class CopyTemplate(_ByContentFileish):
|
||||
return self.content
|
||||
|
||||
|
||||
class RelativeTo(str, Enum):
|
||||
class RelativeTo(StrEnum):
|
||||
TEST = "test"
|
||||
"""
|
||||
Base for the given path is the directory of the test
|
||||
@@ -154,7 +152,7 @@ class Symlink(Fileish):
|
||||
path.symlink_to(target_path)
|
||||
|
||||
|
||||
type FileDeclaration = Dict[str, Fileish | "FileDeclaration"]
|
||||
type FileDeclaration = dict[str, Fileish | "FileDeclaration"]
|
||||
|
||||
|
||||
def _init_files(files: FileDeclaration, tmp_path: Path, request: pytest.FixtureRequest) -> None:
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import string
|
||||
from string import Template
|
||||
from typing import Dict, Type
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -9,7 +8,7 @@ class BalancedTemplater(Template):
|
||||
delimiter = "@"
|
||||
pattern = r"@((?P<escaped>@)|(?P<named>\w+?)@|\{(?P<braced>\w+?)\}@|(?P<invalid>.*?))"
|
||||
|
||||
def substitute(self, mapping: Dict[str, object] | None = None, /, **kwargs) -> str:
|
||||
def substitute(self, mapping: dict[str, object] | None = None, /, **kwargs) -> str:
|
||||
if mapping is None:
|
||||
mapping = {}
|
||||
tmpl_idents = set(self.get_identifiers())
|
||||
@@ -25,5 +24,5 @@ class BalancedTemplater(Template):
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def balanced_templater() -> Type[string.Template]:
|
||||
def balanced_templater() -> type[string.Template]:
|
||||
return BalancedTemplater
|
||||
|
||||
@@ -1,24 +1,28 @@
|
||||
"""
|
||||
HTTP server fixture for tests which binds to an auto-assigned port on localhost.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import dataclasses
|
||||
import logging
|
||||
import time
|
||||
import socket
|
||||
import threading
|
||||
from typing import Tuple
|
||||
from queue import Queue
|
||||
import aiohttp.web as web
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class HttpServer:
|
||||
app: web.Application
|
||||
port: int
|
||||
|
||||
|
||||
class Event_ts(asyncio.Event):
|
||||
class EventTS(asyncio.Event):
|
||||
"""
|
||||
A thread safe version of the asyncio Event
|
||||
|
||||
@@ -27,10 +31,7 @@ class Event_ts(asyncio.Event):
|
||||
Taken from https://stackoverflow.com/a/33006667
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
*args,
|
||||
loop: asyncio.AbstractEventLoop | None = None,
|
||||
**kwargs):
|
||||
def __init__(self, *args, loop: asyncio.AbstractEventLoop | None = None, **kwargs):
|
||||
"""
|
||||
Creates a thread-safe event for the given loop (or the loop of the current thread).
|
||||
"""
|
||||
@@ -41,10 +42,10 @@ class Event_ts(asyncio.Event):
|
||||
self.target_loop.call_soon_threadsafe(super().set)
|
||||
|
||||
|
||||
def _make_localhost_socket() -> Tuple[socket.socket, int]:
|
||||
def _make_localhost_socket() -> tuple[socket.socket, int]:
|
||||
"""Creates a localhost-bound socket with an auto-assigned port."""
|
||||
sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
|
||||
sock.bind(('::1', 0))
|
||||
sock.bind(("::1", 0))
|
||||
# Shouldn't matter because we dynamically allocate ports, but this is generally preferred.
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
_, port = sock.getsockname()[:2]
|
||||
@@ -52,15 +53,13 @@ def _make_localhost_socket() -> Tuple[socket.socket, int]:
|
||||
return (sock, port)
|
||||
|
||||
|
||||
def _server_thread(app: web.Application, sock: socket.socket,
|
||||
shutdown_ev_q: Queue):
|
||||
|
||||
def _server_thread(app: web.Application, sock: socket.socket, shutdown_ev_q: Queue):
|
||||
async def async_main():
|
||||
nonlocal app, sock
|
||||
# Due to Reasons(tm) of event loop lifecycles and stuff of the sort,
|
||||
# it's far easier to just send the event object to the other thread
|
||||
# from inside the loop where it already knows which loop it is.
|
||||
shutdown_ev = Event_ts()
|
||||
shutdown_ev = EventTS()
|
||||
shutdown_ev_q.put(shutdown_ev)
|
||||
|
||||
runner = web.AppRunner(app, handle_signals=False)
|
||||
@@ -91,9 +90,11 @@ def http_server(app: web.Application):
|
||||
shutdown_ev = None
|
||||
try:
|
||||
sock, port = _make_localhost_socket()
|
||||
thr = threading.Thread(target=_server_thread,
|
||||
args=(app, sock, shutdown_ev_q),
|
||||
name=f'functional2 httpd [::1]:{port}')
|
||||
thr = threading.Thread(
|
||||
target=_server_thread,
|
||||
args=(app, sock, shutdown_ev_q),
|
||||
name=f"functional2 httpd [::1]:{port}",
|
||||
)
|
||||
thr.start()
|
||||
shutdown_ev = shutdown_ev_q.get()
|
||||
yield HttpServer(app=app, port=port)
|
||||
@@ -109,16 +110,18 @@ def http_server(app: web.Application):
|
||||
def dev_main():
|
||||
"""A little test server for poking at this manually"""
|
||||
|
||||
async def root(_req: web.Request):
|
||||
return web.Response(body='hello world')
|
||||
async def root(_req: web.Request) -> web.Response:
|
||||
# sadly required to make this function async
|
||||
await asyncio.sleep(0.01)
|
||||
return web.Response(body="hello world")
|
||||
|
||||
app = web.Application()
|
||||
app.add_routes([web.get('/', root)])
|
||||
app.add_routes([web.get("/", root)])
|
||||
|
||||
with http_server(app) as httpd:
|
||||
print(f'Listening on http://[::1]:{httpd.port}')
|
||||
logging.info("Listening on http://[::1]:%d", httpd.port)
|
||||
time.sleep(3600)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if __name__ == "__main__":
|
||||
dev_main()
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
import logging
|
||||
from logging import Logger
|
||||
|
||||
import pytest
|
||||
from _pytest.fixtures import FixtureRequest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def logger(request: FixtureRequest) -> Logger:
|
||||
"""
|
||||
Returns a logger object to use in a test function.
|
||||
:param request: provide by pytest, information about where this fixture was used
|
||||
:return: a logger object to use
|
||||
"""
|
||||
return logging.getLogger(request.function.__name__)
|
||||
@@ -5,13 +5,17 @@ import os
|
||||
import subprocess
|
||||
from functools import partialmethod
|
||||
from pathlib import Path
|
||||
from typing import Any, AnyStr, Callable, Dict
|
||||
from typing import Any, AnyStr
|
||||
from collections.abc import Callable
|
||||
|
||||
import pytest
|
||||
|
||||
from functional2.testlib.terminal_code_eater import eat_terminal_codes
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class CommandResult:
|
||||
cmd: list[str]
|
||||
@@ -24,8 +28,8 @@ class CommandResult:
|
||||
|
||||
def ok(self) -> "CommandResult":
|
||||
if self.rc != 0:
|
||||
print("stdout: %s", self.stderr_s)
|
||||
print("stderr: %s", self.stderr_s)
|
||||
logger.debug("stdout: %s", self.stderr_s)
|
||||
logger.debug("stderr: %s", self.stderr_s)
|
||||
raise subprocess.CalledProcessError(
|
||||
returncode=self.rc, cmd=self.cmd, stderr=self.stderr, output=self.stdout
|
||||
)
|
||||
@@ -33,8 +37,8 @@ class CommandResult:
|
||||
|
||||
def expect(self, rc: int) -> "CommandResult":
|
||||
if self.rc != rc:
|
||||
print("stdout: %s", self.stderr_s)
|
||||
print("stderr: %s", self.stderr_s)
|
||||
logger.debug("stdout: %s", self.stderr_s)
|
||||
logger.debug("stderr: %s", self.stderr_s)
|
||||
raise subprocess.CalledProcessError(
|
||||
returncode=self.rc, cmd=self.cmd, stderr=self.stderr, output=self.stdout
|
||||
)
|
||||
@@ -164,7 +168,7 @@ class NixCommand(Command):
|
||||
class Nix:
|
||||
test_root: Path
|
||||
|
||||
def hermetic_env(self) -> Dict[str, Path]:
|
||||
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)
|
||||
@@ -178,7 +182,7 @@ class Nix:
|
||||
"HOME": home,
|
||||
}
|
||||
|
||||
def make_env(self) -> Dict[AnyStr, AnyStr]:
|
||||
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.
|
||||
|
||||
@@ -1,30 +1,29 @@
|
||||
from string import Template
|
||||
from typing import Type
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def test_template_balanced_kwargs(balanced_templater: Type[Template]):
|
||||
def test_template_balanced_kwargs(balanced_templater: type[Template]):
|
||||
assert balanced_templater("@test_str@").substitute(test_str="valid") == "valid"
|
||||
|
||||
|
||||
def test_template_balanced_multi_use(balanced_templater: Type[Template]):
|
||||
def test_template_balanced_multi_use(balanced_templater: type[Template]):
|
||||
assert balanced_templater("@test_str@ @test_str@").substitute(test_str="valid") == "valid valid"
|
||||
|
||||
|
||||
def test_balanced_escapement(balanced_templater: Type[Template]):
|
||||
def test_balanced_escapement(balanced_templater: type[Template]):
|
||||
assert balanced_templater("@@ @@{test} @{test}@").substitute(test="t") == "@ @{test} t"
|
||||
|
||||
|
||||
# And that correct exceptions get thrown
|
||||
def test_template_balanced_extra_kwarg(balanced_templater: Type[Template]):
|
||||
def test_template_balanced_extra_kwarg(balanced_templater: type[Template]):
|
||||
with pytest.raises(ExceptionGroup) as excinfo:
|
||||
balanced_templater("{test_str}").substitute(test_str="valid", random_str="random")
|
||||
assert excinfo.group_contains(KeyError, match=r"random_str")
|
||||
assert not excinfo.group_contains(ValueError)
|
||||
|
||||
|
||||
def test_template_balanced_missing_arg(balanced_templater: Type[Template]):
|
||||
def test_template_balanced_missing_arg(balanced_templater: type[Template]):
|
||||
with pytest.raises(KeyError) as e:
|
||||
balanced_templater("@TEST@ @MISSING@").substitute(TEST="valid")
|
||||
assert e.match("MISSING")
|
||||
|
||||
@@ -13,8 +13,7 @@ from typing import Protocol
|
||||
class Writable(Protocol):
|
||||
"""Realistically could just be IOBase but this is more constrained"""
|
||||
|
||||
def write(self, data: bytes, /) -> int:
|
||||
...
|
||||
def write(self, data: bytes, /) -> int: ...
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
@@ -25,14 +24,14 @@ class NarListener:
|
||||
self.data.write(data)
|
||||
|
||||
def int_(self, v: int):
|
||||
self.literal(struct.pack('<Q', v))
|
||||
self.literal(struct.pack("<Q", v))
|
||||
|
||||
def add_pad(self, data_len: int):
|
||||
npad = 8 - data_len % 8
|
||||
if npad == 8:
|
||||
npad = 0
|
||||
# FIXME: implement nonzero padding
|
||||
self.literal(b'\0' * npad)
|
||||
# FIXME(Jade): implement nonzero padding
|
||||
self.literal(b"\0" * npad)
|
||||
|
||||
def str_(self, data: bytes):
|
||||
self.int_(len(data))
|
||||
@@ -44,7 +43,7 @@ class NarItem(metaclass=ABCMeta):
|
||||
type_: bytes
|
||||
|
||||
def serialize(self, out: NarListener):
|
||||
out.str_(b'type')
|
||||
out.str_(b"type")
|
||||
out.str_(self.type_)
|
||||
self.serialize_type(out)
|
||||
|
||||
@@ -57,13 +56,13 @@ class NarItem(metaclass=ABCMeta):
|
||||
class Regular(NarItem):
|
||||
executable: bool
|
||||
contents: bytes
|
||||
type_ = b'regular'
|
||||
type_ = b"regular"
|
||||
|
||||
def serialize_type(self, out: NarListener):
|
||||
if self.executable:
|
||||
out.str_(b'executable')
|
||||
out.str_(b'')
|
||||
out.str_(b'contents')
|
||||
out.str_(b"executable")
|
||||
out.str_(b"")
|
||||
out.str_(b"contents")
|
||||
out.str_(self.contents)
|
||||
|
||||
|
||||
@@ -71,23 +70,23 @@ class Regular(NarItem):
|
||||
class DirectoryUnordered(NarItem):
|
||||
entries: list[tuple[bytes, NarItem]]
|
||||
"""Entries in the directory, not required to be in order because this nar is evil"""
|
||||
type_ = b'directory'
|
||||
type_ = b"directory"
|
||||
|
||||
@staticmethod
|
||||
def entry(out: NarListener, name: bytes, item: 'NarItem'):
|
||||
def entry(out: NarListener, name: bytes, item: "NarItem"):
|
||||
# lol this format
|
||||
out.str_(b'entry')
|
||||
out.str_(b'(')
|
||||
out.str_(b'name')
|
||||
out.str_(b"entry")
|
||||
out.str_(b"(")
|
||||
out.str_(b"name")
|
||||
out.str_(name)
|
||||
out.str_(b'node')
|
||||
out.str_(b'(')
|
||||
out.str_(b"node")
|
||||
out.str_(b"(")
|
||||
item.serialize(out)
|
||||
out.str_(b')')
|
||||
out.str_(b')')
|
||||
out.str_(b")")
|
||||
out.str_(b")")
|
||||
|
||||
def serialize_type(self, out: NarListener):
|
||||
for (name, entry) in self.entries:
|
||||
for name, entry in self.entries:
|
||||
self.entry(out, name, entry)
|
||||
|
||||
|
||||
@@ -103,25 +102,25 @@ class Directory(NarItem):
|
||||
@dataclasses.dataclass
|
||||
class Symlink(NarItem):
|
||||
target: bytes
|
||||
type_ = b'symlink'
|
||||
type_ = b"symlink"
|
||||
|
||||
def serialize_type(self, out: NarListener):
|
||||
out.str_(b'target')
|
||||
out.str_(b"target")
|
||||
out.str_(self.target)
|
||||
|
||||
|
||||
def serialize_nar(toplevel: NarItem, out: NarListener):
|
||||
out.str_(b'nix-archive-1')
|
||||
out.str_(b'(')
|
||||
out.str_(b"nix-archive-1")
|
||||
out.str_(b"(")
|
||||
toplevel.serialize(out)
|
||||
out.str_(b')')
|
||||
out.str_(b")")
|
||||
|
||||
|
||||
def write_with_export_header(nar: NarItem, name: bytes, out: NarListener):
|
||||
# n.b. this is *not* actually a nar serialization, it just happens that nix
|
||||
# used exactly the same format for ints and strings in its protocol (and
|
||||
# nix-store --export) as it did in NARs lol
|
||||
EXPORT_MAGIC = 0x4558494e
|
||||
export_magic = 0x4558494E
|
||||
|
||||
# Store::exportPaths
|
||||
# For each path, put 1 then exportPath
|
||||
@@ -129,12 +128,12 @@ 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)
|
||||
out.int_(export_magic)
|
||||
out.str_(b"/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-" + name)
|
||||
# no references
|
||||
out.int_(0)
|
||||
# no deriver
|
||||
out.str_(b'')
|
||||
out.str_(b"")
|
||||
# end of path
|
||||
out.int_(0)
|
||||
|
||||
|
||||
@@ -18,34 +18,39 @@ class TerminalCodeEater:
|
||||
state: State = State.ExpectESC
|
||||
|
||||
def feed(self, data: bytes) -> bytes:
|
||||
is_param_char = lambda c: c >= 0x30 and c <= 0x3f
|
||||
is_intermediate_char = lambda c: c >= 0x20 and c <= 0x2f
|
||||
is_final_char = lambda c: c >= 0x40 and c <= 0x7e
|
||||
def is_param_char(char: int) -> bool:
|
||||
return 48 <= char <= 63
|
||||
|
||||
def is_intermediate_char(char: int) -> bool:
|
||||
return 32 <= char <= 47
|
||||
|
||||
def is_final_char(char: int) -> bool:
|
||||
return 64 <= char <= 126
|
||||
|
||||
ret = bytearray()
|
||||
for c in data:
|
||||
match self.state:
|
||||
case State.ExpectESC:
|
||||
match c:
|
||||
case 0x1b: # \e
|
||||
case 0x1B: # \e
|
||||
self._transition(State.ExpectESCSeq)
|
||||
continue
|
||||
case 0xd: # \r
|
||||
case 0xD: # \r
|
||||
continue
|
||||
ret.append(c)
|
||||
case State.ExpectESCSeq:
|
||||
match c:
|
||||
case 0x5b:
|
||||
# CSI ('[')
|
||||
case 0x5B:
|
||||
# corresponds to CSI ('[')
|
||||
self._transition(State.InCSIParams)
|
||||
continue
|
||||
case 0x5d:
|
||||
# OSC (']')
|
||||
case 0x5D:
|
||||
# corresponds to OSC (']')
|
||||
self._transition(State.InOSCParams)
|
||||
continue
|
||||
# FIXME(jade): whatever this was, we do not know how to
|
||||
# delimit it, so we just eat the next character and
|
||||
# keep going. Should we actually eat it?
|
||||
# delimit it, so we just eat the next character and
|
||||
# keep going. Should we actually eat it?
|
||||
case _:
|
||||
self._transition(State.ExpectESC)
|
||||
continue
|
||||
@@ -56,45 +61,44 @@ class TerminalCodeEater:
|
||||
if is_final_char(c):
|
||||
self._transition(State.ExpectESC)
|
||||
continue
|
||||
elif is_intermediate_char(c):
|
||||
if is_intermediate_char(c):
|
||||
self._transition(State.InCSIIntermediates)
|
||||
continue
|
||||
elif is_param_char(c):
|
||||
if is_param_char(c):
|
||||
continue
|
||||
else:
|
||||
raise ValueError(f'Corrupt escape sequence, at {c:x}')
|
||||
msg = f"Corrupt escape sequence, at {c:x}"
|
||||
raise ValueError(msg)
|
||||
case State.InCSIIntermediates:
|
||||
if is_final_char(c):
|
||||
self._transition(State.ExpectESC)
|
||||
continue
|
||||
elif is_intermediate_char(c):
|
||||
if is_intermediate_char(c):
|
||||
continue
|
||||
else:
|
||||
raise ValueError(
|
||||
f'Corrupt escape sequence in intermediates, at {c:x}'
|
||||
)
|
||||
msg = f"Corrupt escape sequence in intermediates, at {c:x}"
|
||||
raise ValueError(msg)
|
||||
# An OSC is OSC [\x20-\x7e]* ST per ECMA-48
|
||||
# where OSC is \x1b ] and ST is \x1b \.
|
||||
case State.InOSCParams:
|
||||
# first part of ST
|
||||
if c == 0x1b:
|
||||
if c == 0x1B:
|
||||
self._transition(State.InOSCST)
|
||||
continue
|
||||
# OSC sequences can be ended by BEL on old xterms
|
||||
elif c == 0x07:
|
||||
if c == 0x07:
|
||||
self._transition(State.ExpectESC)
|
||||
continue
|
||||
elif c < 0x20 or c == 0x7f:
|
||||
raise ValueError(f'Corrupt OSC sequence, at {c:x}')
|
||||
if c < 0x20 or c == 0x7F:
|
||||
msg = f"Corrupt OSC sequence, at {c:x}"
|
||||
raise ValueError(msg)
|
||||
# either way, eat it
|
||||
continue
|
||||
case State.InOSCST:
|
||||
# ST ends by \
|
||||
if c == 0x5c: # \
|
||||
if c == 0x5C: # \
|
||||
self._transition(State.ExpectESC)
|
||||
elif c < 0x20 or c > 0x7e:
|
||||
raise ValueError(
|
||||
f'Corrupt OSC sequence in ST, at {c:x}')
|
||||
elif c < 0x20 or c > 0x7E:
|
||||
msg = f"Corrupt OSC sequence in ST, at {c:x}"
|
||||
raise ValueError(msg)
|
||||
else:
|
||||
self._transition(State.InOSCParams)
|
||||
continue
|
||||
|
||||
@@ -2,7 +2,11 @@ from functional2.testlib.terminal_code_eater import eat_terminal_codes
|
||||
|
||||
|
||||
def test_eats_color():
|
||||
assert eat_terminal_codes(b'\x1b[7mfoo blah bar\x1b[0m') == b'foo blah bar'
|
||||
assert eat_terminal_codes(b"\x1b[7mfoo blah bar\x1b[0m") == b"foo blah bar"
|
||||
|
||||
|
||||
def test_eats_osc():
|
||||
assert eat_terminal_codes(b'\x1b]8;;http://example.com\x1b\\This is a link\x1b]8;;\x1b\\') == b'This is a link'
|
||||
assert (
|
||||
eat_terminal_codes(b"\x1b]8;;http://example.com\x1b\\This is a link\x1b]8;;\x1b\\")
|
||||
== b"This is a link"
|
||||
)
|
||||
|
||||
@@ -2,3 +2,15 @@
|
||||
command = "nixfmt"
|
||||
includes = ["*.nix"]
|
||||
excludes = ["tests/**"]
|
||||
|
||||
[formatter.ruff-format]
|
||||
command = "ruff"
|
||||
options = ["format"]
|
||||
includes = ["tests/functional2/**/*.py"]
|
||||
priority = 0
|
||||
|
||||
[formatter.ruff]
|
||||
command = "ruff"
|
||||
options = ["check"]
|
||||
includes = ["tests/functional2/**/*.py"]
|
||||
priority = 1
|
||||
|
||||
Reference in New Issue
Block a user