1"""
2Test collection.
3
4Conventions: files named `test_*.py` (searched recursively from the target),
5and in them functions named `test_*`. A test is a function, and a file is
6the group. Test modules get assertion rewriting when imported; helper
7modules do not.
8
9Nothing that looks like a test is left out without a word. What can't be
10run as it is written (a test that takes parameters nothing passes in, one
11that yields, a class with tests in it, a test imported from another file)
12is a collection error for its file, which says what to write instead.
13
14Helper modules are imported by their path from one directory, the helper
15directory. The caller puts it on `sys.path` before collecting. A test module
16can't reach them any other way: see `loading.import_problems`.
17
18Tests aren't kept in the application. A test file that is in it is not run,
19and the run says so: `problems.tests_are_not_kept_in_the_application`.
20"""
21
22import ast
23import functools
24import inspect
25import os
26import types
27from collections.abc import Callable
28from dataclasses import dataclass
29from pathlib import Path
30
31from ..decorators import (
32 TEST_CASES_ATTRIBUTE,
33 TEST_SKIP_ATTRIBUTE,
34 TEST_TAGS_ATTRIBUTE,
35)
36from ..definition import TestDefinitionError
37from ..lifecycle import CollectedTest
38from .layout import Layout, find_tests_directory
39from .loading import ImportsAnotherWay, load_test_module
40from .output_capture import NO_OUTPUT, OutputCapture
41from .problems import (
42 A_TEST_TAKES_ONLY_ITS_CASES,
43 CantBeRunAsWritten,
44 ProblemsInAFile,
45 tests_are_not_kept_in_the_application,
46)
47from .targets import TargetError, name_of_the_test_at, read_target
48
49__all__ = []
50
51_SKIP_DIR_NAMES = {"__pycache__", "node_modules"}
52
53
54class CollectionError(Exception):
55 """
56 One file the runner couldn't collect from, and why. The cause is a
57 TestDefinitionError when the file is written in a way the runner can't
58 run, and whatever was raised (an ImportError, a SyntaxError) when the
59 file couldn't be loaded at all.
60 """
61
62 def __init__(self, path: Path, error: BaseException) -> None:
63 self.path = path
64 self.error = error
65 # What loading the file wrote, when the run is holding output.
66 self.output = NO_OUTPUT
67 # The parameters of the file's tests that nothing passes in, and how
68 # many of its tests take each. The run adds them up.
69 self.parameters: dict[str, int] = {}
70 super().__init__(f"Failed to collect {path}: {error!r}")
71
72
73@dataclass(frozen=True, kw_only=True)
74class RunnableTest(CollectedTest):
75 """A collected test, with what the runner needs to run it."""
76
77 func: Callable # zero-argument callable that runs the test body
78 skip_reason: str | None = None # from `@skip`
79 # The test as the test file defined it. `func` calls it, with a case's
80 # values. A failure is traced back to the frame that ran this function's
81 # code.
82 function: types.FunctionType | None = None
83
84
85def collect_tests(
86 targets: list[str],
87 *,
88 root: Path | None = None,
89 application_directory: Path | None = None,
90 helper_directory: Path | None = None,
91 capture: OutputCapture | None = None,
92) -> tuple[list[RunnableTest], list[CollectionError]]:
93 """
94 Collect tests from the given targets (directories, files, a test by
95 `path::name`, or the test at `path:line`), relative to `root` (default:
96 cwd). A target that can't be used raises `TargetError`.
97
98 `application_directory` is where the application is, when there is one.
99 Tests aren't kept in it. A test file the search comes to there, or a
100 target that is in it, is not run, and the run has a collection error that
101 says which files and where they belong.
102
103 `helper_directory` is the tests directory, where helper modules live. A
104 test module imports `<helper_directory>/helpers.py` as `helpers`, and is
105 refused an import that goes through the directory's own name. Without
106 one, helper modules are found from `root` and no import is refused for
107 its name, since `root` can be called anything. Either way the caller has
108 already put that directory on `sys.path`.
109
110 `capture` is the run's hold on what is written to stdout and stderr. What
111 loading a file writes is kept with that file's error, and thrown away
112 when the file loads.
113
114 Returns the collected tests plus any per-file collection errors — one
115 unimportable file shouldn't stop every other file's tests from running.
116 """
117 root = (root or Path.cwd()).resolve()
118 layout = Layout(
119 root=root,
120 helper_directory=(helper_directory or root).resolve(),
121 refused_import_name=helper_directory.name if helper_directory else None,
122 )
123 if application_directory is not None:
124 application_directory = application_directory.resolve()
125 kept_in_the_application: list[Path] = []
126
127 # Every target is found before anything is loaded, so a target that
128 # isn't there stops the run before a test file has been run.
129 files_of_targets: list[tuple[list[Path], str]] = []
130 for written in targets or ["."]:
131 target = read_target(written)
132 base = (root / target.path).resolve() if target.path not in ("", ".") else root
133 name_part = target.name
134
135 if application_directory is not None and base.is_relative_to(
136 application_directory
137 ):
138 # Named, and not run for being named. A test that runs only when
139 # it is asked for by name is one a plain `plain test` leaves out.
140 if base.is_file():
141 kept_in_the_application.append(base)
142 elif base.is_dir():
143 kept_in_the_application.extend(_test_files_kept_in(base))
144 else:
145 raise TargetError(f"No such test target: {written}")
146 continue
147
148 if base.is_file():
149 files = [base]
150 if target.line is not None:
151 # None for a file that can't be read, which has an error of
152 # its own to report.
153 name_part = (
154 name_of_the_test_at(base, line=target.line, written=written) or ""
155 )
156 elif base.is_dir() and target.line is None:
157 files = _find_test_files(base, leaving_out=application_directory)
158 if (
159 application_directory is not None
160 and application_directory.is_relative_to(base)
161 ):
162 kept_in_the_application.extend(
163 _test_files_kept_in(application_directory)
164 )
165 elif base.is_dir():
166 raise TargetError(
167 f"No test at {written}: {target.path} is a directory, and a"
168 " line is a line of a file."
169 )
170 else:
171 raise TargetError(f"No such test target: {written}")
172 files_of_targets.append((files, name_part))
173
174 errors: list[CollectionError] = []
175 if application_directory is not None and kept_in_the_application:
176 tests_directory = find_tests_directory(root)
177 # First: it is about where the run's tests are, not about one file.
178 errors.append(
179 CollectionError(
180 application_directory,
181 TestDefinitionError(
182 tests_are_not_kept_in_the_application(
183 _once_each_as_shown(kept_in_the_application, layout=layout),
184 application=layout.shown(application_directory),
185 tests=(
186 None
187 if tests_directory == root
188 else layout.shown(tests_directory)
189 ),
190 )
191 ),
192 )
193 )
194
195 collected: list[RunnableTest] = []
196 for files, name_part in files_of_targets:
197 for file in files:
198 if capture is not None:
199 capture.discard()
200 try:
201 tests = _collect_file(file, layout=layout)
202 except CollectionError as error:
203 if capture is not None:
204 error.output = capture.take()
205 errors.append(error)
206 continue
207 if name_part:
208 tests = [t for t in tests if _matches_target(t.name, name_part)]
209 collected.extend(tests)
210
211 _say_each_thing_once(errors, layout=layout)
212
213 # De-duplicate (overlapping targets) while preserving order.
214 seen: set[str] = set()
215 unique = []
216 for test in collected:
217 if test.id not in seen:
218 seen.add(test.id)
219 unique.append(test)
220 return unique, errors
221
222
223def _say_each_thing_once(errors: list[CollectionError], *, layout: Layout) -> None:
224 """
225 Say what is wrong with each file that is written in a way that can't be
226 run, and say why once.
227
228 A file's error has what is wrong with that file: the imports it can't
229 use, the tests it defines wrongly. Eighty files whose tests take
230 parameters have one thing to be told between them. The first says it,
231 and the rest say where the first is.
232 """
233 written_wrongly = [
234 error
235 for error in errors
236 if isinstance(error.error, ImportsAnotherWay | CantBeRunAsWritten)
237 ]
238 importing_another_way = [
239 e for e in written_wrongly if isinstance(e.error, ImportsAnotherWay)
240 ]
241 taking_parameters = [
242 e
243 for e in written_wrongly
244 if e.error.tests.tests_taking_parameters # ty: ignore[unresolved-attribute]
245 ]
246
247 for error in written_wrongly:
248 cause = error.error
249 assert isinstance(cause, ImportsAnotherWay | CantBeRunAsWritten)
250 sections = []
251
252 if isinstance(cause, ImportsAnotherWay):
253 sections.append(cause.what_is_wrong())
254
255 if cause.tests:
256 sections.append(cause.tests.what_is_wrong())
257 if error in taking_parameters:
258 first = taking_parameters[0]
259 if error is first:
260 sections.append(A_TEST_TAKES_ONLY_ITS_CASES)
261 else:
262 sections.append(
263 "What to write instead is in the error for "
264 f"{layout.shown(first.path)}."
265 )
266
267 if isinstance(cause, ImportsAnotherWay) and error is importing_another_way[0]:
268 sections.append(cause.why)
269
270 error.parameters = dict(cause.tests.parameters)
271
272 # Handed on as the one kind of error there is, not as the kind made
273 # here to keep what was known.
274 error.error = TestDefinitionError("\n\n".join(sections), line=cause.line)
275
276
277def _matches_target(name: str, target: str) -> bool:
278 """Whether a test name matches a `::`-target: exact, or a case of it."""
279 return name == target or name.startswith(f"{target}[")
280
281
282def _find_test_files(directory: Path, *, leaving_out: Path | None = None) -> list[Path]:
283 """
284 The files named `test_*.py` under a directory.
285
286 Not looked in: directories whose name starts with a dot, `node_modules`
287 and `__pycache__`, where nobody keeps tests, and `leaving_out`, which is
288 the application.
289 """
290 files = []
291 for dirpath, dirnames, filenames in os.walk(directory):
292 # Prune skipped directories in place so os.walk never descends into
293 # them (rglob can't prune — a .venv or node_modules would get a full
294 # tree walk).
295 dirnames[:] = sorted(
296 d
297 for d in dirnames
298 if d not in _SKIP_DIR_NAMES
299 and not d.startswith(".")
300 and Path(dirpath, d) != leaving_out
301 )
302 files.extend(
303 Path(dirpath) / f
304 for f in sorted(filenames)
305 if f.startswith("test_") and f.endswith(".py")
306 )
307 return files
308
309
310def _test_files_kept_in(directory: Path) -> list[Path]:
311 """
312 The test files in a directory of the application: the ones named
313 `test_*.py` that have tests in them. An application can have a module
314 called `test_connection.py` that checks a connection and tests nothing.
315 """
316 return [file for file in _find_test_files(directory) if _has_tests_in_it(file)]
317
318
319def _has_tests_in_it(file: Path) -> bool:
320 """
321 Whether a file defines a test, as far as reading it can tell: a
322 function named `test_*`, or a class with one in it. A file that can't be
323 read as Python is taken to, since its name says it is a test file and
324 nothing says it isn't.
325 """
326 try:
327 tree = ast.parse(file.read_text(), filename=str(file))
328 except SyntaxError, UnicodeDecodeError, OSError:
329 return True
330
331 for node in tree.body:
332 if isinstance(node, ast.ClassDef):
333 defined = node.body
334 else:
335 defined = [node]
336 for member in defined:
337 is_a_function = isinstance(member, ast.FunctionDef | ast.AsyncFunctionDef)
338 if is_a_function and member.name.startswith("test_"):
339 return True
340 return False
341
342
343def _once_each_as_shown(files: list[Path], *, layout: Layout) -> list[str]:
344 shown = []
345 for file in files:
346 name = layout.shown(file)
347 if name not in shown:
348 shown.append(name)
349 return shown
350
351
352def _collect_file(path: Path, *, layout: Layout) -> list[RunnableTest]:
353 module = _import_test_module(path, layout=layout)
354 relative = layout.shown(path)
355
356 tests: list[RunnableTest] = []
357 # Every test the file defines wrongly is reported together, so a file
358 # that needs the same fix twenty times says so once.
359 problems = ProblemsInAFile()
360
361 # Only functions and classes are ever looked at. Anything else in the
362 # module's namespace is left alone entirely — a test module can hold
363 # objects that object to being probed for attributes.
364 named = [
365 (name, obj)
366 for name, obj in vars(module).items()
367 if inspect.isfunction(obj) or inspect.isclass(obj)
368 ]
369
370 for name, obj in named:
371 defined_here = obj.__module__ == module.__name__
372
373 if inspect.isfunction(obj):
374 if not name.startswith("test_"):
375 continue
376 if not defined_here:
377 problems.defined_elsewhere.append((name, obj.__module__))
378 continue
379 _check_a_test(obj, name=name, problems=problems)
380 tests.extend(_expand(obj, base_id=f"{relative}::{name}"))
381 continue
382
383 # A class the file defines, with tests in it. One that was imported
384 # is someone else's class, and what it holds was never going to be
385 # run by this file.
386 tests_in_it = _tests_defined_in(obj)
387 if defined_here and tests_in_it:
388 line = getattr(obj, "__firstlineno__", None)
389 problems.classes_with_tests.append((line, name, len(tests_in_it)))
390
391 if problems:
392 raise CollectionError(path, CantBeRunAsWritten(problems))
393
394 return tests
395
396
397def _tests_defined_in(cls: type) -> list[str]:
398 """The names of the tests a class's own body defines."""
399 return [
400 name
401 for name, obj in vars(cls).items()
402 if name.startswith("test_")
403 and (inspect.isfunction(obj) or isinstance(obj, staticmethod | classmethod))
404 ]
405
406
407def _yields(func: types.FunctionType) -> bool:
408 """Whether calling a function makes a generator and runs none of it."""
409 functions = [func]
410 try:
411 # A decorator that uses `functools.wraps` hands back a function that
412 # isn't a generator function, around one that is.
413 functions.append(inspect.unwrap(func))
414 except ValueError:
415 pass # wrapped in itself
416 return any(
417 inspect.isgeneratorfunction(function) or inspect.isasyncgenfunction(function)
418 for function in functions
419 )
420
421
422def _check_a_test(
423 func: types.FunctionType, *, name: str, problems: ProblemsInAFile
424) -> None:
425 """
426 Add what is wrong with how one test is written: it yields, it takes
427 parameters nothing passes in, or its @cases don't fit them.
428 """
429 if _yields(func):
430 problems.tests_that_yield.append(name)
431 return
432
433 # follow_wrapped=False: a decorator that passes arguments in itself
434 # (`@mock.patch(...)`) wraps the test in a function that takes anything,
435 # and that wrapper is what the runner calls.
436 signature = inspect.signature(func, follow_wrapped=False)
437 parameters = list(signature.parameters.values())
438 written = f"{name}({', '.join(str(parameter) for parameter in parameters)})"
439
440 case_list = getattr(func, TEST_CASES_ATTRIBUTE, None)
441 if case_list is None:
442 try:
443 signature.bind()
444 except TypeError:
445 problems.tests_taking_parameters.append(written)
446 for parameter in parameters:
447 if parameter.default is not inspect.Parameter.empty:
448 continue
449 if parameter.kind in (
450 inspect.Parameter.VAR_POSITIONAL,
451 inspect.Parameter.VAR_KEYWORD,
452 ):
453 continue
454 problems.parameters[parameter.name] = (
455 problems.parameters.get(parameter.name, 0) + 1
456 )
457 return
458
459 for values, case_id in case_list:
460 try:
461 signature.bind(*values)
462 except TypeError:
463 problems.of_one_test.append(
464 f"{written} doesn't fit its @cases: case [{case_id}] "
465 f"{_how_a_case_misses(parameters, values)}"
466 )
467
468
469def _how_a_case_misses(
470 parameters: list[inspect.Parameter], values: tuple[object, ...]
471) -> str:
472 """
473 What a case passes and what that leaves: "passes 2 values, for plan
474 and amount. Nothing fills currency."
475 """
476 count = "1 value" if len(values) == 1 else f"{len(values)} values"
477 names = [
478 parameter.name
479 for parameter in parameters
480 if parameter.kind
481 in (
482 inspect.Parameter.POSITIONAL_ONLY,
483 inspect.Parameter.POSITIONAL_OR_KEYWORD,
484 )
485 ]
486
487 if len(values) > len(names):
488 if not names:
489 return f"passes {count}, and the test takes none."
490 takes = "1" if len(names) == 1 else str(len(names))
491 return f"passes {count}, and the test takes {takes}: {_listed(names)}."
492
493 filled = names[: len(values)]
494 unfilled = [
495 parameter.name
496 for parameter in parameters
497 if parameter.name not in filled
498 and parameter.default is inspect.Parameter.empty
499 and parameter.kind
500 not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
501 ]
502 passes = f"passes {count}, for {_listed(filled)}." if filled else f"passes {count}."
503 return f"{passes} Nothing fills {_listed(unfilled)}."
504
505
506def _listed(names: list[str]) -> str:
507 """`a`, `a and b`, `a, b and c`."""
508 if len(names) <= 1:
509 return "".join(names)
510 return f"{', '.join(names[:-1])} and {names[-1]}"
511
512
513def _expand(func: types.FunctionType, *, base_id: str) -> list[RunnableTest]:
514 """Expand @cases into one test per case."""
515 tags = tuple(getattr(func, TEST_TAGS_ATTRIBUTE, ()))
516 skip_reason = getattr(func, TEST_SKIP_ATTRIBUTE, None)
517 case_list = getattr(func, TEST_CASES_ATTRIBUTE, None)
518
519 if case_list is None:
520 return [
521 RunnableTest(
522 id=base_id,
523 func=func,
524 tags=tags,
525 skip_reason=skip_reason,
526 function=func,
527 )
528 ]
529
530 return [
531 RunnableTest(
532 id=f"{base_id}[{case_id}]",
533 func=functools.partial(func, *values),
534 tags=tags,
535 skip_reason=skip_reason,
536 function=func,
537 )
538 for values, case_id in case_list
539 ]
540
541
542def _import_test_module(path: Path, *, layout: Layout) -> types.ModuleType:
543 try:
544 return load_test_module(path, layout=layout)
545 except (ImportsAnotherWay, CantBeRunAsWritten) as e:
546 raise CollectionError(path, e) from e
547 except TestDefinitionError as e:
548 raise CollectionError(path, _with_its_line(e, path=path)) from e
549 except Exception as e:
550 raise CollectionError(path, e) from e
551
552
553def _with_its_line(error: TestDefinitionError, *, path: Path) -> TestDefinitionError:
554 """
555 A definition error raised while the test file ran, such as a `@skip` with
556 no reason, with the line of the file that raised it in front. The
557 message is all that gets printed, and "@skip requires a reason" doesn't
558 say which of a file's tests it means.
559 """
560 line = None
561 frame = error.__traceback__
562 while frame is not None:
563 if frame.tb_frame.f_code.co_filename == str(path):
564 line = frame.tb_lineno
565 frame = frame.tb_next
566 if line is None:
567 return error
568 return TestDefinitionError(f"line {line}: {error}", line=line)