fxreader.online/freelance-project-34-marketing-blog
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

[+] refactor test runner into commands_typed/tests.py with pyproject.toml config

Browse Source 
1. add commands_typed/tests.py with run_tests(), collect_tests(), filter_tests();
  2. support test_names as dotted prefixes or * glob patterns via fnmatch;
  3. add dry-run mode via collect subcommand in __main__ with argparse;
  4. add [tool.online-fxreader-pr34.tests] section: search_paths, test_names, discovery_paths;
  5. extend PyProject dataclass with Tests nested class in cli_bootstrap.py;
  6. parse tests config in pyproject_load via check_dict/check_list;
  7. add tests() method to commands_typed/cli.py base CLI class;
  8. simplify python/cli.py to delegate to self.tests(project_name, argv);
  9. auto-patch PYTHONPATH from search_paths before running tests;
  10. narrow unittest discover start_dir from common prefix of test_names;
This commit is contained in:
LLM 2026-04-09 09:00:00 +00:00
parent bf3fd46953
commit 12c9ad8fe0
5 changed files with 327 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
python/cli.py
View File

@ -236,45 +236,12 @@ class CLI(_cli.CLI):
argv=args,
)
elif options.command is Command.tests:
from online.fxreader.pr34.commands_typed import argparse as pr34_argparse
assert not options.project is None
tests_parser = argparse.ArgumentParser()
tests_parser.add_argument(
'--timeout',
default=16,
type=int,
help='test timeout in seconds, default = 16',
self.tests(
project_name=options.project,
argv=args,
)
tests_options, tests_args = pr34_argparse.parse_args(tests_parser, args)
for k, v in self.projects.items():
if len(tests_args) > 0:
cmd = [
sys.executable,
'-m',
'unittest',
*tests_args,
]
else:
cmd = [
sys.executable,
'-m',
'unittest',
'discover',
'-s',
str(v.source_dir),
'-p',
'test_*.py',
'-t',
str(v.source_dir),
]
subprocess.check_call(
cmd,
cwd=str(v.source_dir),
timeout=tests_options.timeout,
)
elif options.command is Command.module_switch:
assert not options.project is None

23
python/m.py
View File

@ -595,7 +595,10 @@ def whl_cache_download(
if parsed is None:
continue
if py_tag_prefix is not None and parsed.python_tag is not None:
if not parsed.python_tag.startswith(py_tag_prefix) and parsed.python_tag not in ('py3', 'py2.py3'):
if not parsed.python_tag.startswith(py_tag_prefix) and parsed.python_tag not in (
'py3',
'py2.py3',
):
continue
cached_pkgs.add((parsed.name, parsed.version))
@ -796,9 +799,7 @@ def env_bootstrap(
with contextlib.ExitStack() as stack:
if overrides and len(constraint_args) > 0:
patched = stack.enter_context(
tempfile.NamedTemporaryFile(
mode='w', prefix='constraints_', suffix='.txt'
)
tempfile.NamedTemporaryFile(mode='w', prefix='constraints_', suffix='.txt')
)
packaging_t.apply_overrides_to_constraints(
requirements_path, overrides, patched
@ -925,8 +926,18 @@ class argv_extract_t:
if action is not None:
matched_argv.append(argv[i])
i += 1
if action.nargs in (None, 1) and action.const is None and not isinstance(
action, (_argparse._StoreTrueAction, _argparse._StoreFalseAction, _argparse._CountAction, _argparse._HelpAction)
if (
action.nargs in (None, 1)
and action.const is None
and not isinstance(
action,
(
_argparse._StoreTrueAction,
_argparse._StoreFalseAction,
_argparse._CountAction,
_argparse._HelpAction,
),
)
):
if i < len(argv):
matched_argv.append(argv[i])

38
python/online/fxreader/pr34/commands_typed/cli.py
View File

@ -213,6 +213,44 @@ class CLI(abc.ABC):
cwd=str(project.source_dir),
)
def tests(
self,
project_name: str,
argv: list[str],
) -> None:
from . import argparse as pr34_argparse
from . import cli_bootstrap
from . import tests as _tests
tests_parser = argparse.ArgumentParser()
tests_parser.add_argument(
'--timeout',
default=16,
type=int,
help='test timeout in seconds, default = 16',
)
tests_parser.add_argument(
'--dry-run',
default=False,
action=argparse.BooleanOptionalAction,
help='list tests to run without executing them',
)
tests_options, tests_args = pr34_argparse.parse_args(tests_parser, argv)
project = self.projects[project_name]
pyproject = cli_bootstrap.pyproject_load(project.source_dir / 'pyproject.toml')
_tests.run_tests(
source_dir=project.source_dir,
tests_args=tests_args,
timeout=tests_options.timeout,
search_paths=pyproject.tests.search_paths if pyproject.tests else None,
test_names=pyproject.tests.test_names if pyproject.tests else None,
discovery_paths=pyproject.tests.discovery_paths if pyproject.tests else None,
dry_run=tests_options.dry_run,
)
def pip_sync(
self,
project: str,

45
python/online/fxreader/pr34/commands_typed/cli_bootstrap.py
View File

@ -93,6 +93,14 @@ class PyProject:
default_factory=lambda: [],
)
@dataclasses.dataclass
class Tests:
search_paths: list[pathlib.Path] = dataclasses.field(default_factory=list)
test_names: list[str] = dataclasses.field(default_factory=list)
discovery_paths: list[pathlib.Path] = dataclasses.field(default_factory=list)
tests: Optional[Tests] = None
tool: dict[str, Any] = dataclasses.field(
default_factory=lambda: dict(),
)
@ -311,6 +319,20 @@ def pyproject_load(
for k, v in check_dict(pr34_tool['requirements'], str, str).items()
}
if 'tests' in pr34_tool:
tests_section = check_dict(pr34_tool['tests'], str)
res.tests = PyProject.Tests(
search_paths=[
d.parent / pathlib.Path(o)
for o in check_list(tests_section.get('search_paths', []), str)
],
test_names=check_list(tests_section.get('test_names', []), str),
discovery_paths=[
d.parent / pathlib.Path(o)
for o in check_list(tests_section.get('discovery_paths', []), str)
],
)
if 'modules' in pr34_tool:
modules = check_list(pr34_tool['modules'])
# res.modules = []
@ -595,7 +617,10 @@ def whl_cache_download(
if parsed is None:
continue
if py_tag_prefix is not None and parsed.python_tag is not None:
if not parsed.python_tag.startswith(py_tag_prefix) and parsed.python_tag not in ('py3', 'py2.py3'):
if not parsed.python_tag.startswith(py_tag_prefix) and parsed.python_tag not in (
'py3',
'py2.py3',
):
continue
cached_pkgs.add((parsed.name, parsed.version))
@ -796,9 +821,7 @@ def env_bootstrap(
with contextlib.ExitStack() as stack:
if overrides and len(constraint_args) > 0:
patched = stack.enter_context(
tempfile.NamedTemporaryFile(
mode='w', prefix='constraints_', suffix='.txt'
)
tempfile.NamedTemporaryFile(mode='w', prefix='constraints_', suffix='.txt')
)
packaging_t.apply_overrides_to_constraints(
requirements_path, overrides, patched
@ -925,8 +948,18 @@ class argv_extract_t:
if action is not None:
matched_argv.append(argv[i])
i += 1
if action.nargs in (None, 1) and action.const is None and not isinstance(
action, (_argparse._StoreTrueAction, _argparse._StoreFalseAction, _argparse._CountAction, _argparse._HelpAction)
if (
action.nargs in (None, 1)
and action.const is None
and not isinstance(
action,
(
_argparse._StoreTrueAction,
_argparse._StoreFalseAction,
_argparse._CountAction,
_argparse._HelpAction,
),
)
):
if i < len(argv):
matched_argv.append(argv[i])

229
python/online/fxreader/pr34/commands_typed/tests.py Normal file
View File

@ -0,0 +1,229 @@
import argparse
import fnmatch
import logging
import os
import pathlib
import subprocess
import sys
import unittest
from typing import Optional
logger = logging.getLogger(__name__)
def _build_discover_cmd(
source_dir: pathlib.Path,
test_names: Optional[list[str]] = None,
discovery_paths: Optional[list[pathlib.Path]] = None,
) -> list[str]:
top_level = str(source_dir)
start_dir = str(source_dir)
if discovery_paths is not None and len(discovery_paths) > 0:
top_level = str(discovery_paths[0].resolve())
start_dir = top_level
# narrow start_dir to common prefix of test_names
if test_names is not None and len(test_names) > 0:
# find the first non-glob prefix
prefixes = [n for n in test_names if '*' not in n]
if len(prefixes) > 0:
# find common prefix of all dotted prefixes
common = prefixes[0].split('.')
for p in prefixes[1:]:
parts = p.split('.')
common = common[: min(len(common), len(parts))]
for i in range(len(common)):
if common[i] != parts[i]:
common = common[:i]
break
if len(common) > 0:
start_dir = str(pathlib.Path(top_level) / '/'.join(common))
return [
sys.executable,
'-m',
'unittest',
'discover',
'-s',
start_dir,
'-p',
'test_*.py',
'-t',
top_level,
]
def _build_env(
search_paths: Optional[list[pathlib.Path]] = None,
) -> dict[str, str]:
env = dict(os.environ)
if search_paths is not None and len(search_paths) > 0:
extra = [str(p.resolve()) for p in search_paths]
existing = env.get('PYTHONPATH', '')
env['PYTHONPATH'] = os.pathsep.join(extra + ([existing] if existing else []))
return env
def _parse_discover_args(cmd: list[str]) -> tuple[str, str, str]:
start_dir = '.'
top_level = '.'
pattern = 'test_*.py'
for i, arg in enumerate(cmd):
if arg == '-s' and i + 1 < len(cmd):
start_dir = cmd[i + 1]
elif arg == '-t' and i + 1 < len(cmd):
top_level = cmd[i + 1]
elif arg == '-p' and i + 1 < len(cmd):
pattern = cmd[i + 1]
return start_dir, top_level, pattern
def collect_tests(
start_dir: str,
top_level: str,
pattern: str = 'test_*.py',
) -> list[str]:
loader = unittest.TestLoader()
suite = loader.discover(start_dir, pattern=pattern, top_level_dir=top_level)
result: list[str] = []
def _collect(s: unittest.TestSuite) -> None:
for t in s:
if isinstance(t, unittest.TestSuite):
_collect(t)
else:
result.append(t.id())
_collect(suite)
return result
def filter_tests(
all_tests: list[str],
test_names: list[str],
) -> list[str]:
"""Filter tests by prefix or glob pattern.
Each entry in test_names is either:
- a dotted prefix (e.g. 'pkg.tests.test_models') — matches tests starting with it
- a glob with * (e.g. '*TestResolver*') — matched via fnmatch against full test id
"""
result: list[str] = []
for t in all_tests:
for name in test_names:
if '*' in name:
if fnmatch.fnmatch(t, name):
result.append(t)
break
else:
if t.startswith(name):
result.append(t)
break
return result
def _collect_via_subprocess(
cmd: list[str],
env: dict[str, str],
source_dir: pathlib.Path,
) -> list[str]:
start_dir, top_level, pattern = _parse_discover_args(cmd)
collect_cmd = [
sys.executable,
'-m',
__name__,
'collect',
'-s',
start_dir,
'-t',
top_level,
'-p',
pattern,
]
logger.info(dict(collect_cmd=collect_cmd, cwd=str(source_dir)))
result = subprocess.run(
collect_cmd,
cwd=str(source_dir),
env=env,
capture_output=True,
text=True,
)
logger.info(dict(collect_rc=result.returncode, stdout_lines=len(result.stdout.splitlines())))
if result.returncode != 0:
logger.error(dict(stderr=result.stderr))
raise RuntimeError('test collection failed')
return [line for line in result.stdout.splitlines() if line.strip()]
def run_tests(
source_dir: pathlib.Path,
tests_args: list[str],
timeout: int = 16,
search_paths: Optional[list[pathlib.Path]] = None,
test_names: Optional[list[str]] = None,
discovery_paths: Optional[list[pathlib.Path]] = None,
dry_run: bool = False,
) -> None:
env = _build_env(search_paths)
if len(tests_args) > 0:
cmd = [sys.executable, '-m', 'unittest', *tests_args]
else:
discover_cmd = _build_discover_cmd(source_dir, test_names, discovery_paths)
all_tests = _collect_via_subprocess(discover_cmd, env, source_dir)
if test_names is not None and len(test_names) > 0:
all_tests = filter_tests(all_tests, test_names)
if len(all_tests) == 0:
logger.warning('no tests matched')
return
cmd = [sys.executable, '-m', 'unittest', *all_tests]
logger.info(dict(cmd=cmd, PYTHONPATH=env.get('PYTHONPATH', '')))
if dry_run:
for t in cmd[3:]:
print(t)
return
subprocess.check_call(
cmd,
cwd=str(source_dir),
timeout=timeout,
env=env,
)
def main() -> None:
parser = argparse.ArgumentParser(
prog='online.fxreader.pr34.commands_typed.tests',
)
subparsers = parser.add_subparsers(dest='command')
collect_parser = subparsers.add_parser('collect', help='discover and list test names')
collect_parser.add_argument('-s', dest='start_dir', default='.', help='start directory')
collect_parser.add_argument('-t', dest='top_level', default='.', help='top level directory')
collect_parser.add_argument('-p', dest='pattern', default='test_*.py', help='file pattern')
args = parser.parse_args()
if args.command == 'collect':
for t in collect_tests(args.start_dir, args.top_level, args.pattern):
print(t)
else:
parser.print_help()
sys.exit(1)
if __name__ == '__main__':
main()