Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -517,6 +517,15 @@ value.
'HELLO'
```

Both spellings of a reference work — the dot path above, and the `'pkg.mod:name'` colon
form the rest of cw uses (`python -m cw`, `mk_parser`'s `obj:`, `commands_from`):

```python
>>> import os.path
>>> resolve_to_function('os.path:join') is os.path.join
True
```

`parse_json_spec`, `parse_ast_spec` and `parse_spec_with_dot_path` are the three spec
grammars; `resource_inputs` wraps a function so that named parameters are resolved on the
way in. It is the one place cw touches a third-party package, and it is an optional extra:
Expand Down
53 changes: 48 additions & 5 deletions cw/resolution.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,9 +74,18 @@
from typing import Tuple, TypeVar, Union
from collections.abc import Callable, Mapping

# cw.commands owns the ``'pkg.mod:name'`` spelling and is stdlib-only, so importing it
# here costs nothing and keeps one implementation of the reference grammar. The graph
# stays acyclic: cw.commands -> cw.grammar -> cw.base, none of which import this module.
from cw.commands import REF_SEPARATOR, import_object

FuncSpec = TypeVar("FuncSpec")
FuncKey = TypeVar("FuncKey", bound=str)

#: What :func:`parse_spec_with_dot_path` accepts: a dot path, optionally split by one
#: :data:`~cw.commands.REF_SEPARATOR` into a module path and an attribute path.
_DOT_OR_COLON_REF = re.compile(rf"^[\w.]+(?:{re.escape(REF_SEPARATOR)}[\w.]+)?$")


def _get_builtin(name: str):
"""Get a built-in object, handling both dict and module __builtins__."""
Expand All @@ -88,8 +97,14 @@ def _get_builtin(name: str):
def resolve_func_from_dot_path(dot_path: str) -> Callable:
"""Resolve a function from a dot-separated import path.

Both spellings of a reference are accepted: the dot path this function is named
after, and the ``'pkg.mod:name'`` colon form the rest of cw documents (see
:func:`cw.commands.import_object`). The colon form is the unambiguous one -- it
says where the module ends and the attribute path begins.

Args:
dot_path: String like 'os.path.join', 'builtins.len', or 'str.upper'
dot_path: String like 'os.path.join', 'builtins.len', 'str.upper', or the
colon form 'os.path:join' / 'json:JSONDecoder.decode'

Returns:
The resolved callable function
Expand All @@ -110,7 +125,24 @@ def resolve_func_from_dot_path(dot_path: str) -> Callable:
>>> upper_func = resolve_func_from_dot_path('str.upper')
>>> upper_func('hello')
'HELLO'

The colon form resolves to the very same object:

>>> resolve_func_from_dot_path('os.path:join') is os.path.join
True
>>> resolve_func_from_dot_path('json:JSONDecoder.decode') # doctest: +ELLIPSIS
<function JSONDecoder.decode at ...>
"""
if REF_SEPARATOR in dot_path:
# The house spelling. One implementation of it, in cw.commands.
try:
func = import_object(dot_path)
except (ImportError, AttributeError) as e:
raise ValueError(f"Cannot resolve '{dot_path}': {e}")
if not callable(func):
raise ValueError(f"'{dot_path}' is not callable")
return func

if "." not in dot_path:
# Handle built-ins and single names
try:
Expand Down Expand Up @@ -158,8 +190,9 @@ def resolve_func_from_dot_path(dot_path: str) -> Callable:
def parse_spec_with_dot_path(func_spec: str) -> tuple[str, dict]:
"""Default parser for simple dot-path function specifications.

Validates that func_spec contains only word characters and dots,
then returns it as-is with empty kwargs.
Validates that func_spec is a dot path (``'pkg.mod.name'``) or a colon reference
(``'pkg.mod:name'``) -- word characters and dots, with at most one colon -- then
returns it as-is with empty kwargs.

Args:
func_spec: Function specification string
Expand All @@ -176,13 +209,18 @@ def parse_spec_with_dot_path(func_spec: str) -> tuple[str, dict]:

>>> parse_spec_with_dot_path('len')
('len', {})

>>> parse_spec_with_dot_path('os.path:join')
('os.path:join', {})
"""
if not isinstance(func_spec, str):
raise TypeError(f"func_spec must be a string, got {type(func_spec)}")

if not re.match(r"^[\w.]+$", func_spec):
if not _DOT_OR_COLON_REF.match(func_spec):
raise ValueError(
f"func_spec must contain only word characters and dots: '{func_spec}'"
f"func_spec must be a dot path ('pkg.mod.name') or a colon reference "
f"('pkg.mod:name') -- word characters and dots, with at most one colon: "
f"'{func_spec}'"
)

return func_spec, {}
Expand Down Expand Up @@ -367,6 +405,11 @@ def resolve_to_function(
>>> ast_func = resolve_to_function('str.upper()', parse_ast_spec)
>>> ast_func('hello')
'HELLO'

>>> # Colon reference -- the spelling the rest of cw documents
>>> import os.path
>>> resolve_to_function('os.path:join') is os.path.join
True
"""
# Resolve Mapping get_func into a function
if isinstance(get_func, Mapping):
Expand Down
99 changes: 98 additions & 1 deletion cw/testing.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,13 @@
a console script is installed as on Windows is scrubbed out of recorded text by
:func:`scrub_exe_suffix`, because ``argparse`` takes its ``prog`` from
``basename(sys.argv[0])`` and would otherwise report ``usage: opsward.EXE`` on the runner
and ``usage: opsward`` everywhere else. And :func:`parity` spawns no subprocess at all --
and ``usage: opsward`` everywhere else. What *cannot* be scrubbed automatically -- because
only the caller knows what belongs in its place -- is an absolute path under the recording
user's home directory, which ``argparse`` renders into ``--help`` whenever a default was
computed from ``$HOME``; :func:`characterize` warns about it at record time instead, via
:func:`local_path_hits`, since a golden is meant to be committed.

And :func:`parity` spawns no subprocess at all --
it runs in-process against shipped fixtures -- so the console-script shim and the cp1252
console never enter the picture there either.

Expand Down Expand Up @@ -90,6 +96,7 @@
"compare_case",
"diff_help",
"load_golden",
"local_path_hits",
"main",
"normalise_text",
"normalise_usage",
Expand Down Expand Up @@ -257,6 +264,12 @@ def normalise_usage(text: str) -> str:
#: A CPython object repr's memory address: ``<list_iterator object at 0x1017642e0>``.
_ADDRESS = re.compile(r"0x[0-9a-fA-F]{4,16}")

#: The home-directory roots :func:`local_path_hits` looks for, on top of the recording
#: user's own home. They catch the path of a *different* user -- a shared runner, a
#: container built as ``root``, a ``sudo``'d recording -- which is just as wrong in a
#: committed golden and which the recorder's own ``$HOME`` would miss.
HOME_ROOTS = ("/Users/", "/home/", "C:\\Users")


def scrub_addresses(text: str) -> str:
"""Replace object-repr memory addresses with a placeholder.
Expand Down Expand Up @@ -326,6 +339,46 @@ def scrub_exe_suffix(text: str, program) -> str:
return text


def local_path_hits(text: str, *, home="~") -> list:
"""Which markers of the recording machine's filesystem does ``text`` carry?

The third field with :func:`scrub_addresses`'s problem. ``argparse`` renders parameter
defaults into ``--help``, defaults are routinely computed from ``$HOME``, and a golden
is documented to be **committed** -- so a recorded body routinely carries an absolute
path under the recording user's home directory, into a public repo. Unlike an address
it cannot simply be scrubbed: what to put in its place is the caller's decision, not
this module's. So this reports, and :func:`characterize` warns.

Args:
text: A recorded ``stdout`` or ``stderr`` body.
home: The home directory to look for, ``~`` expanded. Defaults to the running
user's; pass ``None`` to look only for the generic :data:`HOME_ROOTS`.

Returns:
The offending substrings, most specific first, without duplicates. Empty when the
text is clean -- so it reads as a predicate too.

>>> local_path_hits('usage: x [-h]', home=None)
[]
>>> local_path_hits(' --rootdir ROOTDIR (default: /home/ada/.config/x)', home=None)
['/home/']
>>> local_path_hits(r'default: C:\\Users\\ada\\AppData', home=None)
['C:\\\\Users']
>>> local_path_hits('default: /opt/ada/.config/x', home='/opt/ada')
['/opt/ada']
"""
if not text:
return []
home = os.path.expanduser(home) if home else home
# A home of "/" or "" would match every path ever printed, which is noise, not a hit.
candidates = ([home] if home and home.strip("/\\") else []) + list(HOME_ROOTS)
hits = []
for candidate in candidates:
if candidate in text and candidate not in hits:
hits.append(candidate)
return hits


def _usage_of(stdout: str, stderr: str) -> str:
"""The normalised ``usage:`` line, from wherever the CLI put it.

Expand Down Expand Up @@ -664,6 +717,7 @@ def characterize(
cwd=None,
timeout=DFLT_TIMEOUT,
note=None,
warn_on_local_paths=True,
) -> dict:
"""Record a real console script's behaviour against ``cases``, as a golden.

Expand All @@ -677,6 +731,10 @@ def characterize(
cwd: Working directory for the subprocess.
timeout: Seconds one case may take.
note: Free text stored in the golden -- the commit you recorded at, say.
warn_on_local_paths: Warn when a recorded body carries a path from this machine
(see :func:`local_path_hits`). Nothing is rewritten either way -- the warning
exists because the golden is about to be committed and this is the last moment
anybody looks at it.

Returns:
The golden, as a plain dict. JSON-serialisable, with sorted keys when written, so
Expand All @@ -702,11 +760,50 @@ def run_one(argv):
"note": note,
"cases": [_record(run_one, _as_argv(case)) for case in cases],
}
if warn_on_local_paths:
_warn_about_local_paths(golden["cases"])
if out_path is not None:
write_golden(golden, out_path)
return golden


def _warn_about_local_paths(cases) -> None:
"""Warn about every recorded case whose body carries a path from this machine.

Record time, not compare time. A ``--help`` body is tier 3, so it is a snapshot and is
never asserted: :func:`replay` reports ``identical`` on every machine no matter whose
home directory the golden froze. The only reader who can still act on it is the one
running :func:`characterize`, before ``git add``.
"""
import warnings # local: the only caller, and D4 counts module-scope imports

def hits_in(case):
"""Every marker either stream of one case carries, without duplicates."""
found = local_path_hits(case.get("stdout") or "")
found += [
hit for hit in local_path_hits(case.get("stderr") or "") if hit not in found
]
return found

offenders = [
(case.get("argv", []), hits) for case in cases if (hits := hits_in(case))
]
if not offenders:
return
detail = "; ".join(
f"{argv!r} recorded {', '.join(hits)}" for argv, hits in offenders
)
warnings.warn(
f"cw.testing.characterize recorded a home directory path in {len(offenders)} "
f"case(s): {detail}. A golden is meant to be committed, so this freezes the "
f"recording machine's filesystem into the repo and is wrong everywhere else. "
f"Give the option a machine-independent default, drop the case from the corpus, "
f"or pass warn_on_local_paths=False if the path is genuinely intended.",
UserWarning,
stacklevel=3,
)


def _provenance() -> dict:
"""Who recorded this golden and with what -- so a stale one can be spotted."""
import platform # local: provenance is the only caller, and D4 counts module imports
Expand Down
47 changes: 47 additions & 0 deletions tests/test_resolution.py
Original file line number Diff line number Diff line change
Expand Up @@ -335,3 +335,50 @@ def test_the_custom_message_covers_both_type_errors(self):
resolve_object(
"b", object_map=self.MAP, expected_type=int, error_message="nope"
)


class TestTheColonRefSpelling:
"""``'pkg.mod:name'`` is the house spelling (``cw.commands.import_object``,
``cw.cli.mk_parser``'s ``obj:`` doc, ``python -m cw``), so the package's most
generically named resolver must accept it too -- i2mint/cw#40.
"""

def test_resolve_to_function_accepts_a_colon_ref(self):
import os.path

assert cw.resolve_to_function("os.path:join") is os.path.join

def test_a_colon_ref_may_walk_attributes_after_the_colon(self):
decode = cw.resolve_to_function("json:JSONDecoder.decode")
assert decode.__qualname__ == "JSONDecoder.decode"

def test_the_dot_path_spelling_still_resolves_identically(self):
assert cw.resolve_to_function("builtins.len") is builtins.len
assert cw.resolve_to_function("os.path.join") is __import__("os.path").path.join

def test_a_second_colon_is_still_rejected(self):
with pytest.raises(ValueError):
cw.resolve_to_function("a:b:c")

def test_an_unimportable_colon_ref_still_raises_value_error(self):
"""The failure *shape* is what dependents catch, so widening the grammar must not
change the exception type of a bad reference."""
with pytest.raises(ValueError):
cw.resolve_to_function("cw:no_such_attribute")

def test_a_non_callable_colon_ref_is_a_value_error(self):
with pytest.raises(ValueError, match="not callable"):
cw.resolve_to_function("cw.commands:REF_SEPARATOR")

def test_the_dot_path_resolver_takes_the_colon_form_too(self):
from cw.resolution import resolve_func_from_dot_path

assert resolve_func_from_dot_path("json:JSONDecoder.decode").__qualname__ == (
"JSONDecoder.decode"
)

def test_the_default_parser_passes_a_colon_ref_through_unchanged(self):
from cw.resolution import parse_spec_with_dot_path

assert parse_spec_with_dot_path("os.path:join") == ("os.path:join", {})
assert parse_ast_spec("os.path:join") == ("os.path:join", {})
Loading
Loading