PK! hpytest_fastest/__init__.py# -*- coding: utf-8 -*- """Use coverage data and Git to determine which tests may be skipped.""" import contextlib import enum import json import pathlib import sys from typing import Dict, List, Set, Tuple # noqa: F401, pylint: disable=unused-import import pytest try: from _pytest.config import ArgumentError except ImportError: from _pytest.config.argparsing import ArgumentError from _pytest.runner import runtestprotocol from . import git STOREFILE = ".fastest.coverage" COVERAGE = {} # type: Dict[str, List[str]] STOREVERSION = 1 # Configuration class Mode(enum.Enum): """Enumerated running modes.""" # Run all tests without collecting coverage data: normal pytest behavior ALL = "all" # Skip tests that don't need to be run, but update coverage data on the ones that do SKIP = "skip" # Don't skip tests, but gather coverage data GATHER = "gather" # Skip tests, but don't gather coverage data CACHE = "cache" def pytest_addoption(parser): """Add command line options.""" group = parser.getgroup("fastest") group.addoption( "--fastest-mode", default="all", choices=[mode.value for mode in Mode], action="store", dest="fastest_mode", help=( "Set the running mode." " `all` runs all tests but does not collect coverage data." " `skip` skips tests that can be skipped, and updates coverage data on the rest." " `gather` runs all tests and gathers coverage data on them." " `cache` skips tests and does not collect coverage data." ), ) group.addoption( "--fastest-commit", action="store", dest="fastest_commit", help="Git commit to compare current work against.", ) parser.addini("fastest_commit", "Git commit to compare current work against") # Helpers @contextlib.contextmanager def tracer(rootdir: str, own_file: str): """Collect call graphs for modules within the rootdir.""" result = set() base_path = str(pathlib.Path(rootdir)) def trace_calls(frame, event, arg): # pylint: disable=unused-argument """settrace calls this every time something interesting happens.""" if event != "call": return func_filename = frame.f_code.co_filename if func_filename == own_file: return if not func_filename.endswith(".py"): return if not func_filename.startswith(base_path): return result.add(func_filename) sys.settrace(trace_calls) try: yield result finally: sys.settrace(None) # type: ignore def load_coverage(): """Load the coverage data from disk.""" try: with open(STOREFILE, "r") as infile: data = json.load(infile) except FileNotFoundError: return {} try: if data["version"] != STOREVERSION: return {} except KeyError: return {} return data["coverage"] def save_coverage(coverage): """Save the coverage data to disk.""" with open(STOREFILE, "w") as outfile: json.dump({"coverage": coverage, "version": STOREVERSION}, outfile, indent=2) # Hooks def pytest_configure(config): """Process the configuration.""" config.cache.fastest_commit = config.getoption("fastest_commit") or config.getini( "fastest_commit" ) config.cache.fastest_mode = config.getoption("fastest_mode") config.cache.fastest_skip, config.cache.fastest_gather = { Mode.ALL.value: (False, False), Mode.SKIP.value: (True, True), Mode.GATHER.value: (False, True), Mode.CACHE.value: (True, False), }[config.cache.fastest_mode] if config.cache.fastest_skip and not config.cache.fastest_commit: raise ArgumentError( "fastest_mode", "Mode {} requires fastest_commit to be set.".format(config.cache.fastest_mode), ) COVERAGE.clear() if config.cache.fastest_gather or config.cache.fastest_skip: COVERAGE.update(load_coverage()) def pytest_collection_modifyitems(config, items): """Mark unaffected tests as skippable.""" if not config.cache.fastest_skip: return None covered_test_files = {covdata["fspath"] for covdata in COVERAGE.values()} changed_files, changed_tests = git.changes_since(config.cache.fastest_commit) affected_nodes = set() for nodeid, covdata in COVERAGE.items(): if any(fname in changed_files for fname in covdata["files"]): affected_nodes.add(nodeid) skip = pytest.mark.skip(reason="skipper") for item in items: if not any( ( # Tests refer to modules that have changed item.nodeid in affected_nodes, # Tests that we don't already have coverage data for str(item.fspath) not in covered_test_files, # Tests that have themselves been changed (str(item.fspath), item.name) in changed_tests, ) ): item.add_marker(skip) return True def pytest_runtest_protocol(item, nextitem): """Gather coverage data for the item.""" if not item.config.cache.fastest_gather: return None with tracer(item.config.rootdir, str(item.fspath)) as coverage: reports = runtestprotocol(item, nextitem=nextitem) item.ihook.pytest_runtest_logfinish(nodeid=item.nodeid, location=item.location) outcomes = {report.when: report.outcome for report in reports} if outcomes["setup"] in {"failed", "skipped"}: return True if outcomes["call"] == "passed": COVERAGE[item.nodeid] = {"files": sorted(coverage), "fspath": str(item.fspath)} else: try: del COVERAGE[item.nodeid] except KeyError: pass return True def pytest_terminal_summary(terminalreporter, exitstatus): # pylint: disable=unused-argument """Save the coverage data we've collected.""" if COVERAGE: save_coverage(COVERAGE) PK!1tkpytest_fastest/git.py"""Git backend for pytest-fastest.""" import pathlib import subprocess from typing import List, Set, Tuple def cmd_output(args: List[str]) -> str: """Run a git command and return its output as a string.""" return subprocess.check_output(["git", *args]).decode("UTF-8") def find_toplevel() -> pathlib.Path: """Get the toplevel git directory.""" return pathlib.Path(cmd_output(["rev-parse", "--show-toplevel"]).strip()) def changes_since(commit: str) -> Tuple[Set[str], Set[Tuple[str, str]]]: """Get the set of changes between the given commit.""" toplevel = find_toplevel() diff = cmd_output(["diff", commit, "--"]) changed_files = set() changed_tests = set() current_file = "" for line in diff.splitlines(): try: test_index = line.index("def test_") except ValueError: pass else: test_name = line[test_index + 4 :].partition("(")[0] changed_tests.add((current_file, test_name)) if not (line.startswith("--- ") or line.startswith("+++ ")): continue if not line.endswith(".py"): continue fname = line[4:] if fname == "/dev/null": continue if not (fname.startswith("a/") or fname.startswith("b/")): raise ValueError("Pretty sure {!r} should start with a/ or b/".format(fname)) current_file = str(toplevel / fname[2:]) changed_files.add(current_file) return changed_files, changed_tests PK!HF#/pytest_fastest-0.0.9.dist-info/entry_points.txt.,I-.14JK,1m!"P.PK!so[99&pytest_fastest-0.0.9.dist-info/LICENSE The MIT License (MIT) Copyright (c) 2018 Kirk Strauser Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HڽTU$pytest_fastest-0.0.9.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!HغɿN'pytest_fastest-0.0.9.dist-info/METADATAXmo70K\ᄦns9;Er%ֻ_$3]ɫI̙3<$QY ߔښkIj&mP>xU*]Ƕn;_g4(Z9T"X4[aT%(e[w0UMKVqQdM6ԥ2N/nw1kw'{jnfk?VKQ;-PďoO7^Fzk\UcV2D/f3Z="X\q J+S*ʑm_Yz}!u}FWA6bJugs p)2RR~;}}{G7%-\:lS7}},uzÛՋ ʐ|2R j̈́ fIX8%R-}Ti#ia:Yu`v$+_:Q%fہA݇}`d1Viۏn_iT\Et}X_/3Ynޛ .|X;&P.:ƟNjhRZ_eυ\xb.ԃnAz gGEluk7aO=,pNS.ࣵA~Y3$e7+3OoEĥ6b#X*C 3k*cR6~ZN-EgW>0Rd8Bf;"˲_ʀp2)⃄>8VQ;ۦԩ&VމJlQGSd[[Ft؇(WLKke3j 06« l=$Q`Fmr7 V3Ȓna w i;^iXn\c#1B'q2׉Xk 4,cE`m}0lD5% @=$#6n56;p\ɵA3iB{75Gȶwĺ M3XʊoݰKA" JՃn%uc74R8$w 77iLA`[נMF;Nb&hCTJO!m6mwOLbltHn*~(C S!C=7+(o!:A EݙQP##tK7rQkI噾'm\z EUBț3/h(5f:7ul|'wx>Dt&'Yv#NѹB.%@!%o8As)c%nQ<|HE!Xy%k.}Cq^ bKIf]AxZ$EĎG\;(+̡J@+kr"*ݟ;!e F䈀yjhR7yz aGaMި65PA <Ш_IZ&s< >#l>I.MXҍA!Og R9PN)utd q2C<ґjJ:  /^xD-M~.8jMNevrN5..1۝PCϔZAKp3IiBB&b9s婪G%Os}!g(ǖȭc0vpjt&W~BzڨZSUI-'sg6<}T\]ijlce*}⻎ؿߛ|J8ϟa})Ok8NV:BSjEzCg?nePK!HME_cI%pytest_fastest-0.0.9.dist-info/RECORD͒0vfYP[W@]E$۳g\b$@GVG'g끛 `!l` ^Mv @\~O[ sR3X%6||å 95$|ˢk)?hāu 2lPK! hpytest_fastest/__init__.pyPK!1tkpytest_fastest/git.pyPK!HF#/8pytest_fastest-0.0.9.dist-info/entry_points.txtPK!so[99&pytest_fastest-0.0.9.dist-info/LICENSEPK!HڽTU$#pytest_fastest-0.0.9.dist-info/WHEELPK!HغɿN'#pytest_fastest-0.0.9.dist-info/METADATAPK!HME_cI% ,pytest_fastest-0.0.9.dist-info/RECORDPK6-