mitmproxy/test/individual_coverage.py
dependabot[bot] 77d1f278e1
Update ruff requirement from <=0.8.6,>=0.5.0 to >=0.5.0,<=0.9.4 (#7528)
* Update ruff requirement from <=0.8.6,>=0.5.0 to >=0.5.0,<=0.9.4

Updates the requirements on [ruff](https://github.com/astral-sh/ruff) to permit the latest version.
- [Release notes](https://github.com/astral-sh/ruff/releases)
- [Changelog](https://github.com/astral-sh/ruff/blob/main/CHANGELOG.md)
- [Commits](https://github.com/astral-sh/ruff/compare/0.5.0...0.9.4)

---
updated-dependencies:
- dependency-name: ruff
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>

* [autofix.ci] apply automated fixes

* fixup

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Maximilian Hils <git@maximilianhils.com>
2025-02-01 23:21:33 +00:00

118 lines
3.6 KiB
Python
Executable file

#!/usr/bin/env python3
import ast
import asyncio
import fnmatch
import os
import re
import subprocess
import sys
import tomllib
from pathlib import Path
root = Path(__file__).parent.parent.absolute()
async def main():
with open("pyproject.toml", "rb") as f:
data = tomllib.load(f)
exclude = re.compile(
"|".join(
f"({fnmatch.translate(x)})"
for x in data["tool"]["pytest"]["individual_coverage"]["exclude"]
)
)
sem = asyncio.Semaphore(os.cpu_count() or 1)
async def run_tests(f: Path, should_fail: bool) -> None:
if f.name == "__init__.py":
mod = ast.parse(f.read_text())
full_cov_on_import = all(
isinstance(stmt, (ast.ImportFrom, ast.Import, ast.Assign))
for stmt in mod.body
)
if full_cov_on_import:
if should_fail:
raise RuntimeError(
f"Remove {f} from tool.pytest.individual_coverage in pyproject.toml."
)
else:
print(f"{f}: skip __init__.py file without logic")
return
test_file = Path("test") / f.parent.with_name(f"test_{f.parent.name}.py")
else:
test_file = Path("test") / f.with_name(f"test_{f.name}")
coverage_file = f".coverage-{str(f).replace('/', '-')}"
async with sem:
try:
proc = await asyncio.create_subprocess_exec(
"pytest",
"-qq",
"--disable-pytest-warnings",
"--cov",
str(f.with_suffix("")).replace("/", "."),
"--cov-fail-under",
"100",
"--cov-report",
"term-missing:skip-covered",
test_file,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env={
"COVERAGE_FILE": coverage_file,
**os.environ,
},
)
stdout, stderr = await asyncio.wait_for(proc.communicate(), 60)
except TimeoutError:
raise RuntimeError(f"{f}: timeout")
finally:
Path(coverage_file).unlink(missing_ok=True)
if should_fail:
if proc.returncode != 0:
print(f"{f}: excluded")
else:
raise RuntimeError(
f"{f} is now fully covered by {test_file}. Remove it from tool.pytest.individual_coverage in pyproject.toml."
)
else:
if proc.returncode == 0:
print(f"{f}: ok")
else:
raise RuntimeError(
f"{f} is not fully covered by {test_file}:\n{stdout.decode(errors='ignore')}\n{stderr.decode(errors='ignore')}"
)
tasks = []
for f in (root / "mitmproxy").glob("**/*.py"):
f = f.relative_to(root)
if len(sys.argv) > 1 and sys.argv[1] not in str(f):
continue
if f.name == "__init__.py" and f.stat().st_size == 0:
print(f"{f}: empty")
continue
tasks.append(
asyncio.create_task(run_tests(f, should_fail=exclude.match(str(f))))
)
exit_code = 0
for task in asyncio.as_completed(tasks):
try:
await task
except RuntimeError as e:
print(e)
exit_code = 1
sys.exit(exit_code)
if __name__ == "__main__":
asyncio.run(main())