6519f140a9
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 15s
CI / build (pull_request) Successful in 18s
CI / quality (pull_request) Successful in 18s
CI / typecheck (pull_request) Successful in 34s
CI / security (pull_request) Successful in 48s
CI / unit_tests (pull_request) Successful in 2m16s
CI / docker (pull_request) Successful in 39s
CI / integration_tests (pull_request) Successful in 3m0s
CI / coverage (pull_request) Successful in 3m58s
CI / lint (push) Successful in 13s
CI / build (push) Successful in 15s
CI / quality (push) Successful in 17s
CI / typecheck (push) Successful in 31s
CI / benchmark-regression (push) Has been skipped
CI / security (push) Successful in 35s
CI / unit_tests (push) Successful in 3m23s
CI / docker (push) Successful in 38s
CI / integration_tests (push) Successful in 4m7s
CI / coverage (push) Successful in 4m48s
CI / benchmark-publish (push) Successful in 14m14s
CI / benchmark-regression (pull_request) Successful in 24m27s
Implement hot/warm/cold context tiers with ContextTier, ActorRole, TieredFragment, TierBudget, ActorContextView, TierMetrics, and ScopedBackendView models. ContextTierService provides store/get, promotion/demotion with cold-tier summarisation hook, LRU eviction, per-actor filtered views (strategist/executor/reviewer), and project-scoped isolation via ScopedBackendView. Settings: context_max_tokens_hot, context_max_decisions_warm, context_max_decisions_cold. DI-wired as singleton context_tier_service. Includes Behave BDD scenarios (30), Robot Framework integration tests (7), ASV benchmarks (5 suites), and docs/reference/context_tiers.md. Closes #208
207 lines
7.4 KiB
Python
207 lines
7.4 KiB
Python
"""ASV benchmarks for coverage collection pipeline runtime.
|
|
|
|
Measures the performance of operations that contribute to the
|
|
``nox -s coverage_report`` wall-clock time:
|
|
- Slipcover configuration extraction from noxfile / pyproject
|
|
- Coverage threshold validation and percentage parsing
|
|
- Coverage JSON report parsing and summary extraction
|
|
- XML report generation overhead (file I/O proxy)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
_PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
_NOXFILE_PATH = _PROJECT_ROOT / "noxfile.py"
|
|
_PYPROJECT_PATH = _PROJECT_ROOT / "pyproject.toml"
|
|
_COVERAGE_JSON = _PROJECT_ROOT / "build" / "coverage.json"
|
|
_COVERAGE_XML = _PROJECT_ROOT / "build" / "coverage.xml"
|
|
|
|
# Threshold used by the project for coverage enforcement.
|
|
_COVERAGE_THRESHOLD = 97
|
|
|
|
|
|
class CoverageConfigExtractionSuite:
|
|
"""Benchmark extracting coverage configuration from project files.
|
|
|
|
The coverage report session reads noxfile.py and pyproject.toml to
|
|
determine source paths, omit patterns, and the fail-under threshold.
|
|
"""
|
|
|
|
timeout = 60
|
|
|
|
def setup(self) -> None:
|
|
"""Read configuration files once for reuse."""
|
|
if _NOXFILE_PATH.exists():
|
|
self.noxfile_text: str = _NOXFILE_PATH.read_text()
|
|
else:
|
|
self.noxfile_text = ""
|
|
if _PYPROJECT_PATH.exists():
|
|
self.pyproject_text: str = _PYPROJECT_PATH.read_text()
|
|
else:
|
|
self.pyproject_text = ""
|
|
|
|
def time_parse_noxfile_ast(self) -> None:
|
|
"""Benchmark full AST parse of noxfile.py."""
|
|
if self.noxfile_text:
|
|
ast.parse(self.noxfile_text)
|
|
|
|
def time_extract_coverage_threshold(self) -> None:
|
|
"""Benchmark extracting COVERAGE_THRESHOLD from noxfile AST."""
|
|
if self.noxfile_text:
|
|
tree = ast.parse(self.noxfile_text)
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Assign):
|
|
for target in node.targets:
|
|
if (
|
|
isinstance(target, ast.Name)
|
|
and target.id == "COVERAGE_THRESHOLD"
|
|
and isinstance(node.value, ast.Constant)
|
|
):
|
|
_ = node.value.value
|
|
|
|
def time_extract_omit_patterns(self) -> None:
|
|
"""Benchmark extracting omit patterns from noxfile text."""
|
|
if self.noxfile_text:
|
|
re.findall(r'"(\*[^"]+)"', self.noxfile_text)
|
|
|
|
def time_parse_pyproject_coverage_section(self) -> None:
|
|
"""Benchmark extracting [tool.coverage] from pyproject.toml."""
|
|
if self.pyproject_text:
|
|
import tomllib
|
|
|
|
data = tomllib.loads(self.pyproject_text)
|
|
_ = data.get("tool", {}).get("coverage", {})
|
|
|
|
def time_extract_source_paths(self) -> None:
|
|
"""Benchmark extracting coverage source paths from pyproject."""
|
|
if self.pyproject_text:
|
|
import tomllib
|
|
|
|
data = tomllib.loads(self.pyproject_text)
|
|
_ = (
|
|
data.get("tool", {})
|
|
.get("coverage", {})
|
|
.get("run", {})
|
|
.get("source", [])
|
|
)
|
|
|
|
def time_extract_fail_under_regex(self) -> None:
|
|
"""Benchmark regex extraction of --fail-under value from noxfile."""
|
|
if self.noxfile_text:
|
|
re.findall(r"--fail-under=(\d+)", self.noxfile_text)
|
|
|
|
|
|
class CoverageReportParsingSuite:
|
|
"""Benchmark parsing coverage report outputs.
|
|
|
|
After slipcover runs, the coverage pipeline parses JSON and XML
|
|
reports to extract totals and enforce thresholds.
|
|
"""
|
|
|
|
timeout = 60
|
|
|
|
def setup(self) -> None:
|
|
"""Load coverage report data if available."""
|
|
self.json_text: str = ""
|
|
self.xml_text: str = ""
|
|
if _COVERAGE_JSON.exists():
|
|
self.json_text = _COVERAGE_JSON.read_text()
|
|
if _COVERAGE_XML.exists():
|
|
self.xml_text = _COVERAGE_XML.read_text()
|
|
|
|
def time_parse_coverage_json(self) -> None:
|
|
"""Benchmark parsing the slipcover JSON report."""
|
|
if self.json_text:
|
|
json.loads(self.json_text)
|
|
|
|
def time_extract_coverage_percentage(self) -> None:
|
|
"""Benchmark extracting coverage percentage from JSON report."""
|
|
if self.json_text:
|
|
payload = json.loads(self.json_text)
|
|
summary = payload.get("summary") or payload.get("totals") or {}
|
|
for key in (
|
|
"percent_covered",
|
|
"percent_covered_display",
|
|
"line_coverage_percent",
|
|
):
|
|
value = summary.get(key)
|
|
if value is not None:
|
|
break
|
|
|
|
def time_threshold_comparison(self) -> None:
|
|
"""Benchmark the threshold enforcement logic."""
|
|
if self.json_text:
|
|
payload = json.loads(self.json_text)
|
|
summary = payload.get("summary") or payload.get("totals") or {}
|
|
percent_value = None
|
|
for key in (
|
|
"percent_covered",
|
|
"percent_covered_display",
|
|
"line_coverage_percent",
|
|
):
|
|
percent_value = summary.get(key)
|
|
if percent_value is not None:
|
|
break
|
|
if percent_value is None:
|
|
covered = summary.get("lines_covered") or summary.get("covered_lines")
|
|
total = summary.get("lines") or summary.get("total_lines")
|
|
if covered is not None and total:
|
|
percent_value = (float(covered) / float(total)) * 100.0
|
|
if isinstance(percent_value, str):
|
|
percent_value = percent_value.strip().rstrip("%")
|
|
if percent_value is not None:
|
|
pct = round(float(percent_value), 1)
|
|
_ = pct >= _COVERAGE_THRESHOLD
|
|
|
|
def time_parse_coverage_xml(self) -> None:
|
|
"""Benchmark parsing the coverage XML report."""
|
|
if self.xml_text:
|
|
import xml.etree.ElementTree as ET
|
|
|
|
ET.fromstring(self.xml_text)
|
|
|
|
|
|
class CoverageFileSuite:
|
|
"""Benchmark coverage-related file I/O operations."""
|
|
|
|
timeout = 60
|
|
|
|
def setup(self) -> None:
|
|
"""Prepare paths for file I/O benchmarks."""
|
|
self.build_dir: Path = _PROJECT_ROOT / "build"
|
|
|
|
def time_discover_slipcover_files(self) -> None:
|
|
"""Benchmark discovering .slipcover.*.json files in build/."""
|
|
if self.build_dir.exists():
|
|
list(self.build_dir.glob(".slipcover.*.json"))
|
|
|
|
def time_discover_coverage_artifacts(self) -> None:
|
|
"""Benchmark discovering all coverage-related artifacts."""
|
|
if self.build_dir.exists():
|
|
_json = list(self.build_dir.glob("coverage*.json"))
|
|
_xml = list(self.build_dir.glob("coverage*.xml"))
|
|
_txt = list(self.build_dir.glob("coverage*.txt"))
|
|
|
|
def track_source_file_count(self) -> int:
|
|
"""Track number of Python source files under coverage."""
|
|
src_dir = _PROJECT_ROOT / "src"
|
|
if src_dir.exists():
|
|
return len(list(src_dir.rglob("*.py")))
|
|
return 0
|
|
|
|
def track_source_total_bytes(self) -> int:
|
|
"""Track total byte size of Python source files."""
|
|
src_dir = _PROJECT_ROOT / "src"
|
|
if src_dir.exists():
|
|
return sum(f.stat().st_size for f in src_dir.rglob("*.py"))
|
|
return 0
|
|
|
|
|
|
CoverageFileSuite.track_source_file_count.unit = "files"
|
|
CoverageFileSuite.track_source_total_bytes.unit = "bytes"
|