458 lines
25 KiB
Python
458 lines
25 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome
|
|
import subprocess
|
|
import sys
|
|
import re
|
|
import json
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from dataclasses import dataclass, replace
|
|
from pathlib import Path
|
|
from typing import Sequence, cast
|
|
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.test_environment import TEST_ENV_VARIABLES_TO_CLEAR, sanitized_test_environment
|
|
from band_aider.judgement.test_runner_strategy import (
|
|
TestRunnerStrategy,
|
|
TestRunResult,
|
|
bounded_failure_lines,
|
|
bounded_failure_evidence,
|
|
)
|
|
from band_aider.judgement.repo_exec import RepoExec
|
|
from band_aider.judgement.source_discovery_strategy import SearchableBeat, SourceDiscoveryStrategy
|
|
from band_aider.strategies.language_strategy_set import LanguageStrategySet
|
|
from band_aider.judgement.stack_trace_order_strategy import StackTraceOrderStrategy
|
|
|
|
|
|
PYTEST_NO_TESTS_COLLECTED_EXIT_CODE = 5
|
|
|
|
# Only pytest's diagnostic/summary forms establish a plugin timeout; source
|
|
# literals, fixture reprs and ordinary log messages do not establish execution.
|
|
_PYTEST_TIMEOUT = re.compile(
|
|
r"(?m)^(?:E\s+Failed: |(?:FAILED|ERROR)\s+\S.*? - Failed: )"
|
|
r"Timeout \(>(?P<seconds>[\d.]+)s\) from pytest-timeout\.?\s*$"
|
|
r"|^\+{3,}\s+Timeout\s+\+{3,}\s*$"
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class PytestRunnerStrategy(TestRunnerStrategy):
|
|
"""PytestRunnerStrategy runs Python tests through pytest."""
|
|
|
|
TARGETED_TIMEOUT_SECONDS = 60
|
|
ALL_TESTS_TIMEOUT_SECONDS = 7200
|
|
|
|
file_manager: FileManager
|
|
repo_exec: RepoExec
|
|
|
|
@staticmethod
|
|
def classify(returncode: int, output: str) -> TestRunOutcome:
|
|
"""Distinguish timeout and error diagnostics from ordinary call failures."""
|
|
if returncode == 0:
|
|
return TestRunOutcome.Success
|
|
# A test deadline is repairable execution evidence, even without frames.
|
|
if _PYTEST_TIMEOUT.search(output) or (returncode == 124 and output.startswith("TIMEOUT: pytest exceeded ")):
|
|
return TestRunOutcome.Timeout
|
|
if returncode != 1 or re.search(r"(?m)^(?:_+\s+ERROR (?:at |collecting )|ERROR\s+\S+)", output):
|
|
return TestRunOutcome.Error
|
|
return TestRunOutcome.Failure
|
|
|
|
def failure_reports(self, test_files: list[ClassMeta], result: TestRunResult) -> list[FailingTest]:
|
|
"""Select the first outermost *intersected* method independently per report.
|
|
|
|
Pytest locals can contain other files, node IDs and entire trace strings.
|
|
Only report headings and actual frame lines identify execution. Production
|
|
blame may use broader evidence; it cannot replace this originating test.
|
|
Error reports retain known file scope without requiring a parseable test AST.
|
|
"""
|
|
if result.passed:
|
|
return []
|
|
output = result.output
|
|
headings = tuple(re.finditer(
|
|
r"(?m)^_+\s+((?:ERROR collecting \S+)|(?:(?:ERROR at (?:setup|teardown) of )?[\w.]+(?:\[[^\n]*\])?))\s+_+[ \t]*$",
|
|
output,
|
|
))
|
|
blocks = [(match[1], output[match.start():headings[index + 1].start() if index + 1 < len(headings) else len(output)])
|
|
for index, match in enumerate(headings)] or [("", output)]
|
|
if len(headings) == 1:
|
|
blocks = [(headings[0][1], output)]
|
|
# With traceback disabled, each summary node is an independent report.
|
|
if not headings and not re.search(r'(?m)^(?:(?: )?File "|[^\s\'"{}]+:\d+:)', output):
|
|
summaries = re.findall(r"(?m)^(?:FAILED|ERROR)\s+.+$", output)
|
|
blocks = [("ERROR summary" if line.startswith("ERROR ") else "FAILED", line) for line in summaries] or blocks
|
|
discovery = cast(SourceDiscoveryStrategy, LanguageStrategySet.get(self.file_manager)[SourceDiscoveryStrategy])
|
|
error_nodes: dict[tuple[TestRunOutcome, str], list[str]] = {}
|
|
for match in re.finditer(r"(?m)^(?:ERROR|FAILED)\s+(\S+?)::((?:\w+::)*\w+(?:\[[^\n]*\])?)(?:\s+-[^\n]*|\s*$)", output):
|
|
name, marker, parameters = match[2].partition("[")
|
|
identity = name.replace("::", ".") + marker + parameters
|
|
error_nodes.setdefault((self.classify(1, match[0]), identity), []).append(match[1])
|
|
definitions: dict[str, tuple[SearchableBeat, ...]] = {}
|
|
reports: list[FailingTest] = []
|
|
for heading, block in blocks:
|
|
body = re.split(r"(?m)^={3,}\s+short test summary info\s+=+", block)[0] if heading else block
|
|
outcome = TestRunOutcome.Error if heading.startswith("ERROR ") else (
|
|
TestRunOutcome.Failure if heading else result.outcome)
|
|
# Classify each body independently; the final summary can name other timeouts.
|
|
if _PYTEST_TIMEOUT.search(body):
|
|
outcome = TestRunOutcome.Timeout
|
|
captured = replace(result, outcome=outcome, output=block,
|
|
stack_trace=list(result.stack_trace) if block == output and result.stack_trace else bounded_failure_lines(block),
|
|
test_name="")
|
|
evidence = replace(captured, output=body, stack_trace=bounded_failure_lines(body))
|
|
if not test_files:
|
|
raise ValueError(f"Cannot identify a managed test file:\n{result.run_info}\n{block}")
|
|
frames = cast(StackTraceOrderStrategy, LanguageStrategySet.get(self.file_manager)[StackTraceOrderStrategy]).frames_for_failure(
|
|
file_manager=self.file_manager, failure=FailingTest.from_result(test_files[0], evidence),
|
|
)
|
|
files = {test.location: test for test in test_files}
|
|
frame_files = [frame.location for frame in frames if frame.location in files]
|
|
# Collection names its file directly, without an executing test frame.
|
|
explicit = re.findall(r"(?m)^(?:ERROR|FAILED)\s+(\S+?)(?:::|\s+-|\s*$)", body)
|
|
if heading.startswith("ERROR collecting "):
|
|
explicit.insert(0, heading.removeprefix("ERROR collecting "))
|
|
# A setup failure need not execute any line in the test's file. Pytest's
|
|
# error summary binds the heading to its collected file without AST
|
|
# method discovery (which collection errors may make impossible).
|
|
if heading.startswith(("ERROR at setup of ", "ERROR at teardown of ")) or (heading and outcome is TestRunOutcome.Timeout):
|
|
identity = re.sub(r"^ERROR at (?:setup|teardown) of ", "", heading)
|
|
paths = error_nodes.get((outcome, identity), [])
|
|
explicit = [paths.pop(0)] if paths else explicit
|
|
locations = [test.location for path in explicit for test in test_files
|
|
if path.replace("\\", "/") in {test.location, str(test.resolve_path(self.file_manager))}]
|
|
if outcome.is_error:
|
|
owners = tuple(dict.fromkeys(locations or frame_files or (list(files) if len(files) == 1 else [])))
|
|
if not owners:
|
|
raise ValueError(f"Cannot identify the known test file for pytest error:\n{result.run_info}\n{block}")
|
|
# One setup/teardown/collection report belongs to one test file.
|
|
reports.append(FailingTest.from_result(files[owners[0]], captured))
|
|
continue
|
|
candidates: list[tuple[str, str]] = []
|
|
for frame in frames:
|
|
if frame.location not in files:
|
|
continue
|
|
test = files[frame.location]
|
|
if test.location not in definitions:
|
|
path = test.resolve_path(self.file_manager)
|
|
if not self.file_manager.repo_exec.exists(path):
|
|
continue
|
|
source = self.file_manager.repo_exec.read_text(path)
|
|
definitions[test.location] = tuple(beat for beat in discovery.searchable_beats(str(path), source)
|
|
if isinstance(beat, SearchableBeat) and beat.kind in {"function", "method"}
|
|
and (re.fullmatch(r"test_\w+", beat.text) or re.fullmatch(r"Test\w*\.test_\w+", beat.text)))
|
|
names = [beat.text for beat in definitions[test.location]
|
|
if (frame.name and frame.name in {beat.text, beat.text.rsplit('.', 1)[-1]})
|
|
or (not frame.name and not heading and frame.line is not None
|
|
and (beat.start_line or 1) <= frame.line <= (beat.end_line or 0))]
|
|
# A long traceback footer can omit the callable; its heading supplies
|
|
# identity only after intersecting the actual frame's current file.
|
|
if not frame.name and not names:
|
|
identity = heading.partition("[")[0].replace("::", ".")
|
|
names = [beat.text for beat in definitions[test.location] if beat.text == identity]
|
|
if len(names) > 1:
|
|
raise ValueError(f"Ambiguous test in {test.location}: {names}")
|
|
for name in names:
|
|
suffix = "[" + heading.partition("[")[2] if "[" in heading and heading.partition("[")[0] in {name, name.rsplit('.', 1)[-1]} else ""
|
|
candidates.append((test.location, name + suffix))
|
|
if candidates:
|
|
break
|
|
# Explicit summary nodes cover pytest's short/no-traceback presentation.
|
|
if not candidates:
|
|
matches = self.matching_tests(test_files, evidence)
|
|
candidates = [(location, name) for location, names in matches.items() for name in names]
|
|
# A named invocation can identify a failure without a printed traceback.
|
|
# It must still intersect the current AST; errors bypass this branch.
|
|
if not candidates and result.test_name and len(test_files) == 1 and not headings:
|
|
reported = f"FAILED {test_files[0].location}::{result.test_name.partition('[')[0].replace('.', '::')}{'[' + result.test_name.partition('[')[2] if '[' in result.test_name else ''}"
|
|
matches = self.matching_tests(test_files, replace(captured, output=reported))
|
|
candidates = [(location, name) for location, names in matches.items() for name in names]
|
|
if not candidates:
|
|
raise ValueError(f"Cannot identify a current test method in pytest failure:\n{block}")
|
|
location, name = candidates[0]
|
|
reports.append(FailingTest.from_result(files[location], replace(captured, test_name=name)))
|
|
# A process deadline after earlier reports is still an unresolved invocation;
|
|
# it cannot be replaced by an already printed assertion failure.
|
|
if result.outcome is TestRunOutcome.Timeout and result.returncode == 124 and any(heading for heading, _ in blocks):
|
|
if len(test_files) != 1:
|
|
raise ValueError(f"Cannot identify the known test file for pytest timeout:\n{result.run_info}\n{output}")
|
|
reports.append(FailingTest.from_result(test_files[0], replace(result, test_name="")))
|
|
return reports
|
|
|
|
def matching_tests(
|
|
self, test_files: list[ClassMeta], result: TestRunResult,
|
|
) -> dict[str, tuple[str, ...]]:
|
|
"""Match pytest evidence against current AST callables, preserving case IDs.
|
|
|
|
``matching_tests(files, suite)`` returns candidates only. A subsequent
|
|
singleton execution establishes whether each candidate still fails.
|
|
"""
|
|
discovery = cast(SourceDiscoveryStrategy, LanguageStrategySet.get(self.file_manager)[SourceDiscoveryStrategy])
|
|
locations = dict(map(lambda test: (
|
|
test.location, self.file_manager.test_path(test.location).relative_to(self.file_manager.repo_root).as_posix(),
|
|
), test_files))
|
|
output = result.output
|
|
reported: list[tuple[set[str], str]] = []
|
|
# Pytest summary nodes bind the path explicitly, including parameter IDs with spaces.
|
|
for match in re.finditer(r"(?m)^(?:FAILED|ERROR)\s+(\S+?)::((?:\w+::)*\w+(?:\[[^\n]*\])?)(?:\s+-|\s*$)", output):
|
|
paths = set(filter(lambda location: match[1].replace("\\", "/") in {
|
|
locations[location], str(self.file_manager.repo_root / locations[location]),
|
|
}, locations))
|
|
reported.append((paths, match[2]))
|
|
headers = tuple(re.finditer(
|
|
r"(?m)^\s*_+\s+(?:ERROR at (?:setup|teardown) of )?([\w.]+(?:\[[^\n]*\])?)\s+_+\s*$",
|
|
output,
|
|
))
|
|
# Long tracebacks carry their test path inside the block rather than its heading.
|
|
for index, header in enumerate(headers):
|
|
end = headers[index + 1].start() if index + 1 < len(headers) else len(output)
|
|
block = output[header.end():end].replace("\\", "/")
|
|
paths = set(filter(lambda location: re.search(
|
|
rf'(?m)^(?: File ")?(?:{re.escape(str(self.file_manager.repo_root))}/)?{re.escape(locations[location])}(?::\d+|", line \d+)', block,
|
|
), locations))
|
|
reported.append((paths or (set(locations) if len(locations) == 1 else set()), header[1]))
|
|
matches: dict[str, tuple[str, ...]] = {}
|
|
reported_paths = set().union(*(paths for paths, _ in reported))
|
|
# AST discovery is fresh for this suite observation; no compiler inventory is shared.
|
|
for test in test_files:
|
|
if test.location not in reported_paths:
|
|
continue
|
|
path = self.file_manager.test_path(test.location)
|
|
if not self.file_manager.repo_exec.exists(path):
|
|
continue
|
|
source = self.file_manager.repo_exec.read_text(path)
|
|
names: set[str] = set()
|
|
# Read callable identities from the language-owned AST discovery boundary.
|
|
for beat in discovery.searchable_beats(str(path), source):
|
|
# Match only supported pytest functions and class methods, not nested helpers.
|
|
if isinstance(beat, SearchableBeat) and beat.kind in {"function", "method"}:
|
|
parts = beat.text.split(".")
|
|
# Pytest methods require a test class; nested helper functions are not nodes.
|
|
if (len(parts) == 1 and parts[0].startswith("test_")) or (
|
|
len(parts) == 2 and parts[0].startswith("Test") and parts[1].startswith("test_")
|
|
):
|
|
names.add(beat.text)
|
|
selected: dict[str, None] = {}
|
|
# Intersect each reported identity with its possible current file owners.
|
|
for paths, identity in reported:
|
|
name, marker, parameters = identity.partition("[")
|
|
name = name.replace("::", ".")
|
|
# A qualified report restricts class ownership; bare names admit every real owner.
|
|
owners = names & {name} if "." in name else set(filter(lambda current: current.rsplit(".", 1)[-1] == name, names))
|
|
# An explicit pytest file path narrows otherwise ambiguous method names.
|
|
if test.location in paths:
|
|
# Distinct parameter cases remain individually executable selectors.
|
|
for owner in sorted(owners):
|
|
selected[owner + marker + parameters] = None
|
|
# Prefer reported cases over a redundant whole-method selection from the same output.
|
|
parameterized = set(map(lambda name: name.partition("[")[0], filter(lambda name: "[" in name, selected)))
|
|
identities = tuple(filter(lambda name: name not in parameterized, selected))
|
|
# Files without current matched methods create no execution candidates.
|
|
if identities:
|
|
matches[test.location] = identities
|
|
return matches
|
|
|
|
@staticmethod
|
|
def test_identity(output: str) -> str:
|
|
"""Recover one reported failing node without inventing identities for passes."""
|
|
reported = re.findall(r"^FAILED\s+\S+?::([^\s]+)(?:\s+-|$)", output, re.MULTILINE)
|
|
headers = re.findall(r"^_+\s+([\w.]+(?:\[[^\n]*\])?)\s+_+$", output, re.MULTILINE)
|
|
names = tuple(dict.fromkeys(name.replace("::", ".") for name in (*reported, *headers)
|
|
if any(part.startswith("test_") for part in name.replace("::", ".").split("."))))
|
|
return names[0] if len(names) == 1 else ""
|
|
|
|
@staticmethod
|
|
def trace_identity(stack_trace: Sequence[str]) -> tuple[str, ...]:
|
|
"""Number object addresses by first occurrence across the supplied text fragments."""
|
|
|
|
addresses: dict[str, str] = {}
|
|
|
|
def label(match: re.Match[str]) -> str:
|
|
"""Replace the captured address while retaining repeated object references."""
|
|
address = addresses.setdefault(match[1], f"address{len(addresses) + 1}")
|
|
return f"object at {address}"
|
|
|
|
return tuple(map(
|
|
lambda text: re.sub(r"\bobject at (0x[0-9a-fA-F]+)\b", label, text),
|
|
stack_trace,
|
|
))
|
|
|
|
def _run(
|
|
self,
|
|
paths: list[Path],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Run pytest for concrete paths and return the full result."""
|
|
exec_adapter = self.repo_exec
|
|
mapped_paths = [exec_adapter.path_in_exec(p) for p in paths]
|
|
base_cmd = [
|
|
exec_adapter.python_bin(),
|
|
"-B",
|
|
"-m",
|
|
"pytest",
|
|
"--no-header",
|
|
"-qq",
|
|
"-rfE",
|
|
"--tb=long",
|
|
"--showlocals",
|
|
"--show-capture=no",
|
|
"--disable-warnings",
|
|
"--color=no",
|
|
"--continue-on-collection-errors",
|
|
]
|
|
# Limit only callers that explicitly need one repairable failure at a time.
|
|
if max_failures is not None:
|
|
base_cmd.append(f"--maxfail={max(1, max_failures)}")
|
|
# Restrict collection only when the caller supplied generic callable identities.
|
|
if test_names:
|
|
base_cmd.extend(
|
|
f"{path}::{test_name.partition('[')[0].replace('.', '::')}{'[' + test_name.partition('[')[2] if '[' in test_name else ''}"
|
|
for path in mapped_paths
|
|
for test_name in test_names
|
|
)
|
|
else:
|
|
base_cmd.extend(mapped_paths)
|
|
timeout_seconds = (
|
|
self.ALL_TESTS_TIMEOUT_SECONDS
|
|
if paths == [self.file_manager.test_root]
|
|
else self.TARGETED_TIMEOUT_SECONDS
|
|
)
|
|
started_at = datetime.now(timezone.utc).isoformat()
|
|
started = time.monotonic()
|
|
# Keep invocation facts separate from test frames and stable failure identity.
|
|
run_info = "\n".join((
|
|
f"EXECUTABLE: {base_cmd[0]}",
|
|
f"ARGV: {json.dumps(base_cmd)}",
|
|
f"EXECUTOR ARGV: {json.dumps(exec_adapter.decorate_cli(base_cmd, cwd=self.file_manager.repo_root))}",
|
|
f"CWD: {self.file_manager.repo_root}",
|
|
f"REQUESTED PATHS: {json.dumps(list(map(str, paths)))}",
|
|
f"REQUESTED TEST NAMES: {json.dumps(tuple(test_names))}",
|
|
f"STARTED UTC: {started_at}",
|
|
f"PROCESS DEADLINE SECONDS: {timeout_seconds}",
|
|
))
|
|
try:
|
|
completed = exec_adapter.run_cli(
|
|
base_cmd,
|
|
cwd=self.file_manager.repo_root,
|
|
env=(
|
|
sanitized_test_environment(repo_exec=exec_adapter)
|
|
if exec_adapter.preserves_host_environment
|
|
else {}
|
|
),
|
|
**(
|
|
{"unset_env": TEST_ENV_VARIABLES_TO_CLEAR}
|
|
if not exec_adapter.preserves_host_environment
|
|
else {}
|
|
),
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=timeout_seconds,
|
|
)
|
|
except subprocess.TimeoutExpired as exc:
|
|
stdout = exc.stdout or ""
|
|
stderr = exc.stderr or ""
|
|
stdout = stdout.decode(errors="replace") if isinstance(stdout, bytes) else stdout
|
|
stderr = stderr.decode(errors="replace") if isinstance(stderr, bytes) else stderr
|
|
observed_name = self.test_identity("\n".join((stdout, stderr)))
|
|
test_name = test_names[0] if len(paths) == len(test_names) == 1 else self.test_identity("\n".join((stdout, stderr)))
|
|
stdout, stderr = self.trace_identity((stdout, stderr))
|
|
combined_output = "\n".join(
|
|
part
|
|
for part in [
|
|
f"TIMEOUT: pytest exceeded {timeout_seconds} seconds while running \"{' '.join(exec_adapter.decorate_cli(base_cmd, cwd=self.file_manager.repo_root))}\"",
|
|
f"command: {' '.join(exec_adapter.decorate_cli(base_cmd, cwd=self.file_manager.repo_root))}",
|
|
f"cwd: {self.file_manager.repo_root}",
|
|
f"stdout:\n{stdout}" if stdout else "stdout:",
|
|
f"stderr:\n{stderr}" if stderr else "stderr:",
|
|
]
|
|
if part != ""
|
|
)
|
|
return TestRunResult(
|
|
command=" ".join(exec_adapter.decorate_cli(base_cmd, cwd=self.file_manager.repo_root)),
|
|
outcome=TestRunOutcome.Timeout,
|
|
returncode=124,
|
|
output=combined_output,
|
|
stack_trace=bounded_failure_lines(combined_output),
|
|
trace_identity=self.trace_identity,
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
test_name=test_name,
|
|
run_info="\n".join((run_info,
|
|
f"ELAPSED SECONDS: {time.monotonic() - started:.6f}",
|
|
"TIMEOUT MECHANISM: runner process deadline",
|
|
f"TIMEOUT SECONDS: {timeout_seconds}",
|
|
"RETURN CODE: 124 (synthetic; no completed process exit status)",
|
|
f"REPORTED TEST NAME: {observed_name or 'unknown'}",
|
|
)),
|
|
)
|
|
stdout, stderr = completed.stdout or "", completed.stderr or ""
|
|
test_name = test_names[0] if len(paths) == len(test_names) == 1 else self.test_identity("\n".join((stdout, stderr)))
|
|
stdout, stderr = self.trace_identity((stdout, stderr))
|
|
combined_output = "\n".join(
|
|
part
|
|
for part in [stdout, stderr]
|
|
if part is not None and part != ""
|
|
)
|
|
timeout_reports = tuple(_PYTEST_TIMEOUT.finditer(combined_output))
|
|
reported_limits = tuple(dict.fromkeys(filter(None, map(lambda match: match['seconds'], timeout_reports))))
|
|
observed_name = self.test_identity(combined_output)
|
|
return TestRunResult(
|
|
command=" ".join(exec_adapter.decorate_cli(base_cmd, cwd=self.file_manager.repo_root)),
|
|
outcome=TestRunOutcome.Success if completed.returncode == 0 or (
|
|
not test_names and completed.returncode == PYTEST_NO_TESTS_COLLECTED_EXIT_CODE
|
|
) else self.classify(completed.returncode, combined_output),
|
|
returncode=completed.returncode,
|
|
output=combined_output,
|
|
stack_trace=bounded_failure_lines(combined_output),
|
|
trace_identity=self.trace_identity,
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
test_name=test_name,
|
|
run_info="\n".join((run_info,
|
|
f"ELAPSED SECONDS: {time.monotonic() - started:.6f}",
|
|
f"RETURN CODE: {completed.returncode} (process exit status)",
|
|
f"REPORTED TEST NAME: {observed_name or 'unknown'}",
|
|
f"TIMEOUT MECHANISM: {'pytest-reported timeout' if timeout_reports else 'none reported'}",
|
|
f"PYTEST TIMEOUT SECONDS: {', '.join(reported_limits) or 'unknown'}",
|
|
)),
|
|
)
|
|
|
|
def run_tests_for(
|
|
self,
|
|
test_files: list[ClassMeta],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Run pytest for the supplied test files."""
|
|
|
|
paths = [
|
|
self.file_manager.test_path(test_file.location)
|
|
for test_file in test_files
|
|
]
|
|
|
|
# An empty requested test scope has no validation failure.
|
|
if not paths:
|
|
return TestRunResult(
|
|
command=f"{sys.executable} -B -m pytest",
|
|
outcome=TestRunOutcome.Success,
|
|
returncode=PYTEST_NO_TESTS_COLLECTED_EXIT_CODE,
|
|
output="No test files were supplied for pytest.",
|
|
)
|
|
|
|
return self._run(
|
|
paths,
|
|
max_failures=max_failures,
|
|
test_names=test_names,
|
|
)
|
|
|
|
def run_all_tests_under(self, test_root: Path) -> TestRunResult:
|
|
"""Run pytest for every test under the supplied root."""
|
|
|
|
return self._run([test_root])
|