diff --git a/cmake/sphinx.cmake b/cmake/sphinx.cmake index dd80002..fbcd716 100644 --- a/cmake/sphinx.cmake +++ b/cmake/sphinx.cmake @@ -49,6 +49,12 @@ set( CACHE STRING "Directory holding twister's own output (twister.json, twister_report.xml, per-scenario handler.log) for the testreport/twisterinfo directives" ) +set( + ZDOCS_COVERAGE_OUT + "" + CACHE STRING + "Directory of a per-test coverage run (twister.json, coverage/test_matrix.json, zephyr.sha) for the testcoverage directive" +) separate_arguments(ZDOCS_SPHINXOPTS) separate_arguments(ZDOCS_SPHINXOPTS_EXTRA) @@ -225,6 +231,9 @@ function(add_sphinx_target doc_name) if(NOT ZDOCS_TWISTER_OUT STREQUAL "") list(APPEND SPHINX_ENV ZDOCS_TWISTER_OUT=${ZDOCS_TWISTER_OUT}) endif() + if(NOT ZDOCS_COVERAGE_OUT STREQUAL "") + list(APPEND SPHINX_ENV ZDOCS_COVERAGE_OUT=${ZDOCS_COVERAGE_OUT}) + endif() # The Zephyr that find_package(Zephyr) found. zdocs_conf loads Zephyr's doc # extensions from it, including external_content, which stages the sources # into ${DOCS_SRC_DIR}. find_package sets only the CMake variable, never the diff --git a/doc/api/python/extensions.rst b/doc/api/python/extensions.rst index 37020b6..c1f6c92 100644 --- a/doc/api/python/extensions.rst +++ b/doc/api/python/extensions.rst @@ -49,6 +49,14 @@ directives. .. automodule:: test_module :members: +``test_coverage`` +------------------- + +The ``.. testcoverage::`` directive. ``test_module`` loads it. + +.. automodule:: test_coverage + :members: + ``symbol_needs`` ------------------ @@ -114,6 +122,14 @@ Twister output parsing, with no Sphinx dependency of its own. :members: :exclude-members: parse_twister_results +``adequacy`` +-------------- + +Coverage adequacy, with no Sphinx dependency of its own. + +.. automodule:: adequacy + :members: + .. parse_twister_results is excluded: its docstring's own example text, "leading 'test_' prefix stripped", ends in a bare trailing underscore diff --git a/doc/manual/explanation/decisions/0009-need-type-role-mapping.rst b/doc/manual/explanation/decisions/0009-need-type-role-mapping.rst index b88725b..b18e4b2 100644 --- a/doc/manual/explanation/decisions/0009-need-type-role-mapping.rst +++ b/doc/manual/explanation/decisions/0009-need-type-role-mapping.rst @@ -41,6 +41,11 @@ Two fields were later made roles too: a test result's ``depends_met`` and review filters and reports by, not a record of what twister wrote, so a consumer names them like the types and links. +The fields of an adequacy need (``testcoverage_need_fields``: ``verdict``, +``evidence``, ``coverage_run``, ``judged_symbols``, ``symbol_hits``) are roles +for the same reason. The ``adequacy`` type and the ``assesses`` link are roles +like the other types and links. + Consequences ------------ diff --git a/doc/manual/reference/consumer-contract.rst b/doc/manual/reference/consumer-contract.rst index e009e7a..4b37130 100644 --- a/doc/manual/reference/consumer-contract.rst +++ b/doc/manual/reference/consumer-contract.rst @@ -167,6 +167,17 @@ from outside your ``CMakeLists.txt``. the build still succeeds — a documentation build legitimately outrunning its test run is normal. +``ZDOCS_COVERAGE_OUT`` + Default empty. The directory of a per-test coverage run + (``west twister --coverage-per-test``), read by the ``testcoverage`` + directive (:doc:`directives-and-roles`). The directive reads three files + from it: ``twister.json``, ``coverage/test_matrix.json`` and ``zephyr.sha`` + (the run commit). It is not the run of ``ZDOCS_TWISTER_OUT``: a coverage + run builds with instrumentation, usually on one board. The wiring is the + same as for ``ZDOCS_TWISTER_OUT``: CMake passes it only when it is not + empty, and ``zdocs_conf.py`` reads it as ``coverage_output_dir``. If it is + unset, the directive renders "no coverage run configured" and does not warn. + ``ZDOCS_LATEXOPTS`` Default ``"-interaction=nonstopmode -halt-on-error"``. Passed to ``xelatex`` through the ``latexmk``-generated ``latexmkrc``. Changing this is rarely diff --git a/doc/manual/reference/directives-and-roles.rst b/doc/manual/reference/directives-and-roles.rst index 839e584..0c3666b 100644 --- a/doc/manual/reference/directives-and-roles.rst +++ b/doc/manual/reference/directives-and-roles.rst @@ -283,6 +283,125 @@ field and a warning instead. ... } +``.. testcoverage::`` +---------------------- + +One ``adequacy`` need per requirement that a per-test coverage run can assess. +The verdict says if the requirement's own verifying tests run the code that +satisfies it. ``test_module`` loads the directive. + +.. code-block:: rst + + .. testcoverage:: + :run: nightly-cov + :layout: adequacy + +The optional argument is the run directory. Without it, the directive reads +``ZDOCS_COVERAGE_OUT`` (:doc:`consumer-contract`). The run directory holds +``twister.json``, ``coverage/test_matrix.json`` and ``zephyr.sha``. The +``:run:`` option names the run in the need ids. Without it, the name is the +first tag (sorted) on the run commit. If the commit has no tag, the name is +the name of the run directory. The ``:layout:`` option sets the sphinx-needs +layout of each need. + +The directive joins these inputs: + +* The requirement's verifying test cases, through the ``verifies`` link of + the case needs. +* The requirement's satisfying symbols, through the ``satisfies`` link of the + implementation needs (``IMPL-``, see ``symbolneeds``). +* The test cases that the run ran. Each twister case goes to its spec case + by (suite, function), as a test result does. Its matrix key is built from + the scenario and the C function name (``kernel.lifo.usage`` + + ``test_x`` gives ``kernel_lifo_usage_test_x``). The directive does not parse + keys, because scenario names are prefixes of other scenario names. +* The needs come from ``needs_external_needs`` and ``testspec_needs_json``. + +The directive finds the bodies of each symbol in the sources of the run +commit (``git show :`` in ``testmodule_root``). The commit comes +from ``zephyr.sha``, else from the ``-g`` of +``environment.zephyr_version`` in ``twister.json``. If the commit is not in +the tree, the directive reads the working tree and warns. A body is +``z_impl_``, ``z_vrfy_`` (the verifier that a user-mode test +reaches), a plain definition, or a header ``static inline``. A macro has no +body. + +``testcoverage_impl_files`` sets the files that hold the bodies. Each entry is +a glob pattern relative to ``testmodule_root``: ``**/`` is zero or more +directories, and ``*`` stays in one directory. In a ``.h`` file, only a +``static inline`` definition is a body. The default is the set of the +original resolver: + +.. code-block:: python + + testcoverage_impl_files = [ # default + "kernel/*.c", + "kernel/**/*.c", + "include/zephyr/kernel.h", + "include/zephyr/kernel/**/*.h", + "include/zephyr/sys/**/*.h", + ] + +An empty list gives the default. A symbol with its body outside these files +gets the verdict ``unresolved``. The summary of the run lists the files that +the directive searched. + +The verdicts: + +``true`` + The own tests run every symbol that coverage can judge. +``partial`` + The own tests run some of these symbols, not all. +``broken`` + Other tests of the run reach the code. The own tests never do. +``unattributed`` + No test of the run covers any body. Coverage cannot judge the link. +``unresolved`` + No satisfying symbol maps to a body (a macro). +``no-cov`` + The verifying tests have no coverage data in the run. +``no-impl`` + No symbol satisfies the requirement. + +The directive assesses a requirement if the run ran at least one of its +verifying test cases. It renders a summary of the run, a table of the +verdicts, and one section per verdict. Each need lists its symbols and +bodies. For each body, it gives the lines that each own test ran, and the +other tests that ran the body. + +The id of a need is ``ADQ-/``. ``testcoverage_id_prefix`` +sets the prefix. The type, the link and the fields are roles, as for the other +directives: + +.. code-block:: python + + testcoverage_need_types = {"adequacy": "adequacy"} # defaults + testcoverage_need_links = {"assesses": "assesses"} + testcoverage_need_fields = { + "verdict": "verdict", "evidence": "evidence", "coverage_run": "coverage_run", + "judged_symbols": "judged_symbols", "symbol_hits": "symbol_hits", + } + +Declare the type and the link in your ``needs_config.toml``. The directive +sets a field only if your ``needs_config.toml`` declares it (string fields). +The body of the need always shows the same information. The fields: + +``verdict`` + One of the verdicts above. +``evidence`` + The state of the verifying tests in the coverage run: ``passing``, + ``failing``, ``skipped``, ``no-run`` or ``untested``. +``coverage_run`` + The name of the run. +``judged_symbols`` + The satisfying symbols, joined with ``"; "``. +``symbol_hits`` + Per symbol, the body lines that the own tests ran and that any test ran + (``k_sem_init: own 11, any 11``). + +A parameterized test (``ZTEST_P``) has one matrix key for all its values: the +per-test dump of Zephyr has no value in its tag. + ``.. symbolneeds::`` -------------------- diff --git a/sphinx/_extensions/_tests/test_adequacy.py b/sphinx/_extensions/_tests/test_adequacy.py new file mode 100644 index 0000000..eb43f00 --- /dev/null +++ b/sphinx/_extensions/_tests/test_adequacy.py @@ -0,0 +1,569 @@ +# Copyright (c) 2026 inovex GmbH +# +# SPDX-License-Identifier: Apache-2.0 + +"""Coverage adequacy: the oracle fixtures (Phase 4 order 002, part 1). + +Each case has a verdict that is known in advance. The fixture is a git +repository with a small kernel tree, a twister.json, a test_matrix.json and the +needs of the spec (test cases, implementation needs). The test builds it in +``tmp_path`` because a nested repository cannot live in the engine's own tree. + +Requirements, one per verdict and one per body form: + +========== ================== ===================================== ============ +Requirement Satisfied by Verified by (what its tests cover) Verdict +========== ================== ===================================== ============ +R-VRFY k_obj_init TSPEC-1: the z_vrfy body only true +R-IMPL k_impl_only TSPEC-2: the z_impl body true +R-DEF k_plain TSPEC-3: the plain definition true +R-INLINE k_inline TSPEC-4: the header static inline true +R-MACRO k_macro TSPEC-3 unresolved +R-PARTIAL k_plain, k_inline TSPEC-3: k_plain only partial +R-BROKEN k_plain TSPEC-5: none of it (TSPEC-3 does) broken +R-UNATTR k_boot TSPEC-5 (no test covers k_boot) unattributed +R-NOIMPL — TSPEC-5 no-impl +R-NOCOV k_plain TSPEC-6: ran, no matrix entry no-cov +R-PREFIX k_plain TSPEC-9 (scenario demo, test_hit) broken +R-PARAM k_plain TSPEC-10, two parameter values true +R-NOTRUN k_plain TSPEC-7: not in the run not assessed +========== ================== ===================================== ============ + +R-PREFIX is the prefix trap: TSPEC-8 is ``test_hit`` in scenario ``demo.usage`` +and covers k_plain; its key ``demo_usage_test_hit`` starts with the slug of +scenario ``demo``. A reader that parses keys by prefix can give that coverage +to TSPEC-9. Keys built from (scenario, function) do not. +""" + +import json +import subprocess +from pathlib import Path + +import adequacy as A +import pytest +import test_coverage as tc +from rst_builders import _DEFAULT_NEED_NAMES + +OBJ_C = """\ +#include + +int z_impl_k_obj_init(struct k_obj *o) +{ + o->x = 0; + return 0; +} + +static inline int z_vrfy_k_obj_init(struct k_obj *o) +{ + K_OOPS(o == NULL); + return z_impl_k_obj_init(o); +} + +int z_impl_k_impl_only(int v) +{ + return v + 1; +} +""" + +PLAIN_C = """\ +void k_plain(void) +{ + do_something(); +} + +void k_boot(void) +{ + early_init(); +} +""" + +# A direct child of include/zephyr/sys/: `sys/**/*.h` must find it. +INLINE_H = """\ +static inline int k_inline(int x) +{ + return x * 2; +} +""" + +KERNEL_H = """\ +int k_obj_init(struct k_obj *o); +#define k_macro(x) k_plain() +""" + +# A kernel-internal header: outside the default file set. No requirement uses +# k_priv, so the verdicts above do not change. A non-static definition in a +# header is not a body. +PRIV_H = """\ +static inline bool k_priv(int x) +{ + return x > 0; +} + +int k_priv_extern(int x) +{ + return x; +} +""" + + +def _lines(text, first, last): + """The 1-based line numbers of ``first`` .. ``last`` (by content) in ``text``.""" + rows = text.split("\n") + a = next(i for i, r in enumerate(rows) if first in r) + 1 + b = next(i for i, r in enumerate(rows) if i >= a and last in r) + 1 + return a, b + + +# Line numbers at the commit. +VRFY = _lines(OBJ_C, "z_vrfy_k_obj_init", "}") +IMPL = _lines(OBJ_C, "int z_impl_k_obj_init", "}") +IMPL_ONLY = _lines(OBJ_C, "z_impl_k_impl_only", "}") +PLAIN = _lines(PLAIN_C, "void k_plain", "}") +BOOT = _lines(PLAIN_C, "void k_boot", "}") +INLINE = _lines(INLINE_H, "k_inline", "}") + +CASES = { # id: (suite, C function, scenario or None, verifies) + "TSPEC-1": ("s1", "test_obj_init", "demo", ["R-VRFY"]), + "TSPEC-2": ("s1", "test_impl_only", "demo", ["R-IMPL"]), + "TSPEC-3": ("s1", "test_plain", "demo", ["R-DEF", "R-MACRO", "R-PARTIAL"]), + "TSPEC-4": ("s1", "test_inline", "demo", ["R-INLINE"]), + "TSPEC-5": ("s1", "test_other", "demo", ["R-BROKEN", "R-UNATTR", "R-NOIMPL"]), + "TSPEC-6": ("s1", "test_nocov", "demo", ["R-NOCOV"]), + "TSPEC-7": ("s1", "test_notrun", None, ["R-NOTRUN"]), + "TSPEC-8": ("s2", "test_hit", "demo.usage", []), + "TSPEC-9": ("s1", "test_hit", "demo", ["R-PREFIX"]), + "TSPEC-10": ("s1", "test_param_case", "demo", ["R-PARAM"]), +} +SATISFIES = { + "k_obj_init": ["R-VRFY"], + "k_impl_only": ["R-IMPL"], + "k_plain": ["R-DEF", "R-PARTIAL", "R-BROKEN", "R-NOCOV", "R-PREFIX", "R-PARAM", "R-NOTRUN"], + "k_inline": ["R-INLINE", "R-PARTIAL"], + "k_macro": ["R-MACRO"], + "k_boot": ["R-UNATTR"], +} +REQS = sorted({r for c in CASES.values() for r in c[3]}) + +EXPECTED = { + "R-VRFY": "true", + "R-IMPL": "true", + "R-DEF": "true", + "R-INLINE": "true", + "R-MACRO": "unresolved", + "R-PARTIAL": "partial", + "R-BROKEN": "broken", + "R-UNATTR": "unattributed", + "R-NOIMPL": "no-impl", + "R-NOCOV": "no-cov", + "R-PREFIX": "broken", + "R-PARAM": "true", +} + + +def _span(ab, pick=None): + a, b = ab + return list(range(a, b + 1)) if pick is None else [a + i for i in pick] + + +def _matrix(): + """by_test for the fixture: per key, the lines it covers.""" + return { + "demo_test_obj_init": {"kernel/obj.c": _span(VRFY, [0, 1, 2])}, + "demo_test_impl_only": {"kernel/obj.c": _span(IMPL_ONLY)}, + "demo_test_plain": {"kernel/plain.c": _span(PLAIN)}, + "demo_test_inline": {"include/zephyr/sys/k_inline.h": _span(INLINE)}, + # Covers something, but none of the bodies. + "demo_test_other": {"kernel/obj.c": [1]}, + "demo_usage_test_hit": {"kernel/plain.c": _span(PLAIN, [0, 2])}, + "demo_test_hit": {"kernel/obj.c": [1]}, + "demo_test_param_case": {"kernel/plain.c": _span(PLAIN, [2])}, + # A test outside the tree's filter is dropped. + "demo_test_plain_lib": {"../modules/x.c": [1]}, + } + + +def _write_matrix(path, by_test): + by_line = {} + for key, files in by_test.items(): + for f, lines in files.items(): + for ln in lines: + by_line.setdefault(f, {}).setdefault(str(ln), []).append(key) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"by_test": by_test, "by_line": by_line})) + + +def _twister(): + suites = {} + for cid, (suite, fn, scenario, _) in CASES.items(): + if scenario is None: + continue + ts = suites.setdefault(scenario, {"name": scenario, "platform": "demo_board", + "path": "tests/demo", "testcases": []}) + bare = fn[len("test_"):] + if cid == "TSPEC-10": + for n in (0, 1): + ts["testcases"].append({"identifier": f"{scenario}.{bare}[cases/{n}]", + "status": "passed"}) + continue + status = "failed" if cid == "TSPEC-5" else "passed" + ts["testcases"].append({"identifier": f"{scenario}.{suite}.{bare}", "status": status}) + return {"environment": {"zephyr_version": "v1.0.0-1-gdeadbee"}, + "testsuites": list(suites.values())} + + +def _needs(): + needs = {} + for cid, (suite, fn, _, verifies) in CASES.items(): + needs[cid] = {"id": cid, "type": "test_case", "title": fn, "test_function": fn, + "suite": suite, "test_module": "tests/demo", "verifies": verifies} + for sym, reqs in SATISFIES.items(): + needs[f"IMPL-{sym}"] = {"id": f"IMPL-{sym}", "type": "impl", "title": sym, + "satisfies": reqs} + for r in REQS: + needs[r] = {"id": r, "type": "req", "title": r} + return {"current_version": "1.0", "versions": {"1.0": {"needs": needs}}} + + +def _git(root, *args): + subprocess.run(["git", *args], cwd=root, check=True, capture_output=True) + + +@pytest.fixture +def tree(tmp_path): + """``(root, sha, run_dir, needs_json)``: a committed tree, then a working tree 20 lines off.""" + root = tmp_path / "zephyr" + files = { + "kernel/obj.c": OBJ_C, + "kernel/plain.c": PLAIN_C, + "include/zephyr/sys/k_inline.h": INLINE_H, + "include/zephyr/kernel.h": KERNEL_H, + "kernel/include/k_priv.h": PRIV_H, + "lib/unrelated.c": "void k_plain(void)\n{\n}\n", # not a searched path + } + for rel, text in files.items(): + (root / rel).parent.mkdir(parents=True, exist_ok=True) + (root / rel).write_text(text) + _git(root, "init", "-q") + _git(root, "add", ".") + _git(root, "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "-m", "run") + sha = subprocess.run(["git", "rev-parse", "HEAD"], cwd=root, check=True, + capture_output=True, text=True).stdout.strip() + # The working tree moves on after the run: every body 20 lines further down. + for rel in ("kernel/obj.c", "kernel/plain.c", "include/zephyr/sys/k_inline.h"): + (root / rel).write_text("\n" * 20 + files[rel]) + + run_dir = tmp_path / "cov-run" + run_dir.mkdir() + (run_dir / "twister.json").write_text(json.dumps(_twister())) + (run_dir / "zephyr.sha").write_text(sha + "\n") + _write_matrix(run_dir / "coverage" / "test_matrix.json", _matrix()) + needs_json = tmp_path / "needs.json" + needs_json.write_text(json.dumps(_needs())) + return root, sha, run_dir, needs_json + + +def _assess(tree, ref="commit", name=None): + root, sha, run_dir, needs_json = tree + spec, verified_by, satisfied_by, ids = A.collect_links([needs_json]) + run, inputs = A.load_coverage_run(run_dir, spec, root, name=name) + source = A.Source(root, run.sha if ref == "commit" else None) + results, impl_loc = A.assess(run, verified_by, satisfied_by, source, ids) + return run, results, impl_loc, inputs + + +# --------------------------------------------------------------------------- +# One fixture per verdict and per body form +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("req, verdict", sorted(EXPECTED.items())) +def test_verdict(tree, req, verdict): + _, results, _, _ = _assess(tree) + assert results[req]["verdict"] == verdict + + +def test_a_requirement_whose_tests_did_not_run_is_not_assessed(tree): + _, results, _, _ = _assess(tree) + assert set(results) == set(EXPECTED) + assert "R-NOTRUN" not in results + + +def test_body_forms(tree): + _, _, loc, _ = _assess(tree) + assert [(b["variant"], b["a"], b["b"]) for b in loc["k_obj_init"]] == [ + ("impl", *IMPL), ("vrfy", *VRFY), + ] + assert [(b["variant"], b["file"]) for b in loc["k_impl_only"]] == [("impl", "kernel/obj.c")] + # lib/ is not searched: one plain definition only. + assert [(b["variant"], b["file"], b["a"], b["b"]) for b in loc["k_plain"]] == [ + ("def", "kernel/plain.c", *PLAIN), + ] + assert [(b["variant"], b["file"]) for b in loc["k_inline"]] == [ + ("inline", "include/zephyr/sys/k_inline.h"), + ] + # A macro and a prototype have no body. + assert "k_macro" not in loc + + +def test_a_user_mode_test_reaches_only_the_verifier(tree): + _, results, _, _ = _assess(tree) + by_variant = {d["variant"]: d for d in results["R-VRFY"]["impls"]} + assert by_variant["impl"]["own"] == 0 + assert by_variant["vrfy"]["own_tests"] == {"TSPEC-1": _span(VRFY, [0, 1, 2])} + + +def test_partial_and_broken_name_the_tests_that_do_reach_the_code(tree): + _, results, _, _ = _assess(tree) + inline = next(d for d in results["R-PARTIAL"]["impls"] if d["sym"] == "k_inline") + assert (inline["own"], inline["any"]) == (0, len(_span(INLINE))) + assert set(inline["other_tests"]) == {"demo_test_inline"} + broken = results["R-BROKEN"]["impls"][0] + assert set(broken["other_tests"]) == { + "demo_test_plain", "demo_usage_test_hit", "demo_test_param_case", + } + + +def test_keys_are_built_from_scenario_and_function(tree): + run, results, _, _ = _assess(tree) + assert run.cases["TSPEC-8"]["keys"] == ["demo_usage_test_hit"] + assert run.cases["TSPEC-9"]["keys"] == ["demo_test_hit"] + assert results["R-PREFIX"]["impls"][0]["own"] == 0 + + +def test_parameter_values_share_one_key(tree): + run, results, _, _ = _assess(tree) + assert run.cases["TSPEC-10"] == {"statuses": ["passed", "passed"], + "keys": ["demo_test_param_case"]} + assert results["R-PARAM"]["evidence"] == "passing" + + +def test_evidence(tree): + _, results, _, _ = _assess(tree) + assert results["R-BROKEN"]["evidence"] == "failing" + assert results["R-DEF"]["evidence"] == "passing" + run = A.CoverageRun("r", None, {}, {}, {}, {"C": {"statuses": ["skipped"], "keys": []}}, []) + assert A.evidence([], run) == "untested" + assert A.evidence(["X"], run) == "no-run" + assert A.evidence(["C"], run) == "skipped" + + +def test_matrix_keeps_the_tree_files_only(tree): + by_test, by_line = A.load_matrix(tree[2] / "coverage" / "test_matrix.json") + assert by_test["demo_test_plain_lib"] == {} + assert "../modules/x.c" not in by_line + + +# --------------------------------------------------------------------------- +# Sources at the run commit +# --------------------------------------------------------------------------- + + +def test_sources_come_from_the_run_commit(tree): + root, sha, _, _ = tree + run, results, loc, inputs = _assess(tree) + assert run.sha == sha + assert loc["k_plain"][0]["a"] == PLAIN[0] + assert {r: v["verdict"] for r, v in results.items()} == EXPECTED + assert [p.name for p in inputs] == ["twister.json", "test_matrix.json", "zephyr.sha"] + + +def test_the_working_tree_gives_other_lines_and_other_verdicts(tree): + # The control: the same run read against the moved working tree. + _, results, loc, _ = _assess(tree, ref="working tree") + assert loc["k_plain"][0]["a"] == PLAIN[0] + 20 + assert results["R-DEF"]["verdict"] == "unattributed" + + +def test_run_commit_falls_back_to_zephyr_version(tree, tmp_path): + root, sha, run_dir, _ = tree + (run_dir / "zephyr.sha").unlink() + assert A.run_commit(run_dir, {"zephyr_version": f"v1.0.0-3-g{sha[:12]}"}, root) == sha + assert A.run_commit(run_dir, {"zephyr_version": "v1.0.0-3-gdeadbee"}, root) is None + (run_dir / "zephyr.sha").write_text("0" * 40) + assert A.run_commit(run_dir, {}, root) is None + + +def test_run_name_is_the_tag_on_the_commit_else_the_directory(tree): + root, sha, run_dir, _ = tree + assert A.run_name(run_dir, sha, root) == "cov-run" + _git(root, "tag", "z-run") + _git(root, "tag", "a.run/1") + assert A.run_name(run_dir, sha, root) == "a-run-1" + run, _, _, _ = _assess(tree, name="given") + assert run.name == "given" + + +def test_glob_patterns_take_zero_or_more_directories(): + rx = A._glob_regex("include/zephyr/sys/**/*.h") + assert rx.match("include/zephyr/sys/slist.h") + assert rx.match("include/zephyr/sys/a/b/x.h") + assert not rx.match("include/zephyr/sysx.h") + assert not A._glob_regex("kernel/*.c").match("kernel/sub/x.c") + + +def test_the_default_file_set_is_the_set_of_the_original(tree): + root, sha, _, _ = tree + loc = A.resolve_impl_symbols(A.Source(root, sha), ["k_plain", "k_priv"]) + assert "k_priv" not in loc + assert [b["file"] for b in loc["k_plain"]] == ["kernel/plain.c"] + + +def test_impl_files_set_the_files_that_hold_the_bodies(tree): + root, sha, _, _ = tree + files = (*A.IMPL_PATTERNS, "kernel/include/*.h", "lib/*.c") + loc = A.resolve_impl_symbols(A.Source(root, sha), ["k_plain", "k_priv", "k_priv_extern"], + files) + # A .h file outside include/ is a header too: static inline only. + assert [(b["variant"], b["file"], b["a"], b["b"]) for b in loc["k_priv"]] == [ + ("inline", "kernel/include/k_priv.h", 1, 4), + ] + assert "k_priv_extern" not in loc + assert [b["file"] for b in loc["k_plain"]] == ["kernel/plain.c", "lib/unrelated.c"] + # Only the given files: without kernel/*.c, k_plain is in lib/ only. + loc = A.resolve_impl_symbols(A.Source(root, sha), ["k_plain"], ["lib/*.c"]) + assert [b["file"] for b in loc["k_plain"]] == ["lib/unrelated.c"] + + +def test_the_matrix_keeps_the_files_of_every_pattern(): + assert A.keep_prefixes() == A._KEEP + assert A.keep_prefixes(["arch/**/*.c", "soc/x.c", "kernel/include/*.h"]) == ( + *A._KEEP, "arch/", "soc/x.c", + ) + # Nothing outside the tree, nothing for a pattern that starts with a wildcard. + assert A.keep_prefixes(["../modules/*.c", "/abs/*.c", "*.c", "ar*/x.c"]) == A._KEEP + + +def test_the_matrix_keeps_the_lines_of_a_wider_file_set(tree, tmp_path): + _, _, run_dir, _ = tree + m = tmp_path / "m.json" + _write_matrix(m, {"k": {"arch/a.c": [3], "kernel/obj.c": [1], "../x/y.c": [1]}}) + assert A.load_matrix(m)[0]["k"] == {"kernel/obj.c": {1}} + by_test, _ = A.load_matrix(m, A.keep_prefixes(["arch/**/*.c"])) + assert by_test["k"] == {"arch/a.c": {3}, "kernel/obj.c": {1}} + + +def test_matrix_key(): + assert A.matrix_key("kernel.semaphore", "test_sem_init_validity") == ( + "kernel_semaphore_test_sem_init_validity" + ) + assert A.matrix_key("kernel.lifo.usage", "test_x", suite="s") == "kernel_lifo_usage_s_test_x" + + +def test_line_ranges(): + assert A.line_ranges([45, 51, 57, 58, 62]) == "45, 51, 57-58, 62" + assert A.line_ranges([]) == "" + + +# --------------------------------------------------------------------------- +# The directive +# --------------------------------------------------------------------------- + + +def test_adequacy_rst_sets_the_declared_fields_under_the_consumers_names(tree): + run, results, _, _ = _assess(tree) + names = {**_DEFAULT_NEED_NAMES, "adequacy": "judgement", "assesses": "judges", + "verdict": "outcome"} + rst = "\n".join(tc.build_adequacy_rst("R-VRFY", results["R-VRFY"], run, names, {"verdict"})) + assert ".. judgement:: Adequacy of R-VRFY" in rst + assert " :id: ADQ-cov-run/R-VRFY" in rst + assert " :judges: R-VRFY" in rst + assert " :outcome: true" in rst + assert ":evidence:" not in rst # not declared + assert ":layout:" not in rst + with_layout = tc.build_adequacy_rst("R-VRFY", results["R-VRFY"], run, names, (), "Q", "adq") + assert " :layout: adq" in with_layout and " :id: Q-cov-run/R-VRFY" in with_layout + assert f"own :need:`TSPEC-1`: lines {VRFY[0]}-{VRFY[0] + 2}" in rst + + +_CONF = """\ +import sys +sys.path.insert(0, {ext!r}) +extensions = ["sphinx_needs", "test_module"] +master_doc = "index" +exclude_patterns = ["_build"] +needs_types = [ + dict(directive=d, title=d, prefix=d.upper() + "_", color="#FFF", style="node") + for d in ("adequacy", "test_case", "impl", "req") +] +_s = {{"schema": {{"type": "string"}}, "nullable": True}} +needs_fields = {{"verdict": _s, "evidence": _s, "coverage_run": _s, + "judged_symbols": _s, "symbol_hits": _s, + "test_function": _s, "suite": _s, "test_module": _s}} +needs_id_regex = r"^[A-Za-z][A-Za-z0-9_-]+" +needs_links = {{ + "assesses": {{"incoming": "assessed by", "outgoing": "assesses"}}, + "verifies": {{"incoming": "verified by", "outgoing": "verifies"}}, + "satisfies": {{"incoming": "satisfied by", "outgoing": "satisfies"}}, +}} +needs_external_needs = [{{"json_path": {needs!r}, "base_url": "http://localhost/", + "version": "1.0"}}] +coverage_output_dir = {run!r} +testmodule_root = {root!r} +needs_build_json = True +suppress_warnings = ["config.cache"] +""" + + +def test_directive_emits_one_adequacy_need_per_requirement(tree, make_app, tmp_path): + root, _, run_dir, needs_json = tree + src = tmp_path / "doc" + src.mkdir() + (src / "conf.py").write_text(_CONF.format( + ext=str(Path(A.__file__).parent), needs=str(needs_json), run=str(run_dir), + root=str(root), + )) + (src / "index.rst").write_text("Adequacy\n########\n\n.. testcoverage::\n") + app = make_app("html", srcdir=src) + app.build() + data = json.loads((Path(app.outdir) / "needs.json").read_text()) + needs = next(iter(data["versions"].values()))["needs"] + adq = {n["id"]: n for n in needs.values() if n["type"] == "adequacy"} + assert {i.split("/", 1)[1]: n["verdict"] for i, n in adq.items()} == EXPECTED + five = adq["ADQ-cov-run/R-PARTIAL"] + assert five["assesses"] == ["R-PARTIAL"] + assert five["judged_symbols"] == "k_inline; k_plain" + assert five["coverage_run"] == "cov-run" + log = app._warning.getvalue() + assert "WARNING" not in log, log + html = (Path(app.outdir) / "index.html").read_text() + assert "Verdict broken" in html and 'id="ADQ-cov-run/R-VRFY"' in html + + +def test_directive_reads_the_consumers_file_set(tree, make_app, tmp_path): + root, _, run_dir, needs_json = tree + src = tmp_path / "doc" + src.mkdir() + (src / "conf.py").write_text(_CONF.format( + ext=str(Path(A.__file__).parent), needs=str(needs_json), run=str(run_dir), + root=str(root), + ) + 'testcoverage_impl_files = ["include/zephyr/sys/**/*.h"]\n') + (src / "index.rst").write_text("Adequacy\n########\n\n.. testcoverage::\n") + app = make_app("html", srcdir=src) + app.build() + data = json.loads((Path(app.outdir) / "needs.json").read_text()) + needs = next(iter(data["versions"].values()))["needs"] + verdicts = {n["id"].split("/", 1)[1]: n["verdict"] for n in needs.values() + if n["type"] == "adequacy"} + # Only k_inline has a body in these files. The own test of R-PARTIAL runs + # k_plain only, so k_inline alone judges it: another test runs it. + assert verdicts["R-INLINE"] == "true" + assert verdicts["R-DEF"] == "unresolved" + assert verdicts["R-PARTIAL"] == "broken" + html = (Path(app.outdir) / "index.html").read_text() + assert "Files searched for bodies" in html + assert "include/zephyr/sys/**/*.h" in html and "kernel/*.c" not in html + + +def test_directive_without_a_run_says_so(make_app, tmp_path): + src = tmp_path / "doc" + src.mkdir() + (src / "conf.py").write_text( + "import sys\n" + f"sys.path.insert(0, {str(Path(A.__file__).parent)!r})\n" + "extensions = ['sphinx_needs', 'test_module']\n" + ) + (src / "index.rst").write_text("X\n#\n\n.. testcoverage::\n") + app = make_app("html", srcdir=src) + app.build() + assert "no coverage run configured" in (Path(app.outdir) / "index.html").read_text() + assert "WARNING" not in app._warning.getvalue() diff --git a/sphinx/_extensions/adequacy.py b/sphinx/_extensions/adequacy.py new file mode 100644 index 0000000..5ca26a1 --- /dev/null +++ b/sphinx/_extensions/adequacy.py @@ -0,0 +1,640 @@ +# Copyright (c) 2026 inovex GmbH +# Copyright The Zephyr Project Contributors +# +# SPDX-License-Identifier: Apache-2.0 + +"""Coverage adequacy: do the own tests of a requirement run the code that satisfies it. + +This module has no Sphinx dependency. A ``verifies`` link and a ``satisfies`` +link are claims. A per-test coverage run (``west twister --coverage-per-test``) +checks them against execution. For each requirement, the module finds the +bodies of the satisfying symbols in the sources of the run commit. Then it +compares their lines with the lines that the own verifying tests covered. + +The port starts from ``doc/_scripts/traceability_app.py`` by Anas (zephyr +collab-safety 0cc56a35003). `Source`, `resolve_impl_symbols` and the verdicts +(`evidence`, `adequacy`) are close to the original. The keys are the keys of +zdocs: + +* A requirement gets its verifying test cases through the ``verifies`` link + of the case needs. +* A requirement gets its satisfying symbols through the ``satisfies`` link of + the implementation needs (``-``, `rst_builders.symbol_need_id`). +* A test case gets its matrix keys through the ``twister.json`` of the run. + Each twister case goes to its spec case by (suite, function), as a test + result does. The key is built from (scenario, C function name), as twister + writes it. The module does not parse keys, because scenario names are + prefixes of other scenario names (``kernel.lifo``, ``kernel.lifo.usage``). + +Two changes from the original correct errors: + +* File patterns use glob rules in both modes: ``**/`` is zero or more + directories, and ``*`` stays in one directory. The original used + ``fnmatch`` on the ``git ls-tree`` output. There, ``**/`` needs at least one + directory, so ``include/zephyr/sys/**/*.h`` did not find ``sys/slist.h``. +* The run commit comes from ``zephyr.sha`` beside the twister output first. + Then it comes from ``environment.zephyr_version``. + +One change from the original is a setting: the files that hold the bodies. +`IMPL_PATTERNS` is the set of the original and the default. A consumer gives +its own set to `assess` (``testcoverage_impl_files`` in Sphinx). + +The verdicts are the verdicts of the original: + +``true`` + Every symbol that coverage can judge has a body that the own tests ran. +``partial`` + Some of these symbols have such a body, others do not. +``broken`` + Other tests of the run reach the code. The own tests never do. +``unattributed`` + No test of the run covers any body, so coverage cannot judge the link. +``unresolved`` + Satisfying symbols exist, but none maps to a body (a macro). +``no-impl`` + No symbol satisfies the requirement. +``no-cov`` + The verifying tests have no coverage data in this run. +""" + +import json +import re +import subprocess +from collections import defaultdict +from pathlib import Path + +from rst_builders import _need_name, symbol_need_id +from twister_reader import SpecLookup, split_case_name + +__all__ = [ + "VERDICTS", + "matrix_key", + "IMPL_PATTERNS", + "keep_prefixes", + "load_matrix", + "run_commit", + "run_name", + "Source", + "resolve_impl_symbols", + "CoverageRun", + "load_coverage_run", + "collect_links", + "evidence", + "adequacy", + "assess", + "line_ranges", +] + +#: In the order a report lists them: the findings first. +VERDICTS = ("broken", "partial", "unattributed", "unresolved", "no-cov", "no-impl", "true") + +# Test-result rollup precedence (worst wins for reporting a test's state). +_FAIL = {"failed", "error"} +_SKIP = {"skipped", "blocked", "not run", "filtered"} + +#: Where the original looks for bodies. The default of ``impl_files``. +IMPL_PATTERNS = ( + "kernel/*.c", + "kernel/**/*.c", + "include/zephyr/kernel.h", + "include/zephyr/kernel/**/*.h", + "include/zephyr/sys/**/*.h", +) + +#: Files the matrix is read for (the original's ``keep``). `keep_prefixes` +#: adds the fixed start of each pattern of ``impl_files``. +_KEEP = ("kernel/", "include/", "lib/", "tests/") + +#: A body ends at the first column-0 ``}`` within this many lines. +_MAX_BODY = 500 + + +def _slug(name): + """A coverage test name as twister writes it (coverage.py, ``TN:``).""" + return re.sub(r"[^A-Za-z0-9_]", "_", name) + + +def matrix_key(scenario, function, suite=None): + """The ``test_matrix.json`` key of C function ``function`` in ``scenario``. + + Twister names a per-test tracefile ``.``. If two suites of + one scenario have the same test name, it uses ``..``. + The key is that name, with ``_`` for each character outside + ``[A-Za-z0-9_]``. Give ``suite`` for the second form. ``function`` keeps + its ``test_`` prefix. + """ + name = f"{scenario}.{suite}.{function}" if suite else f"{scenario}.{function}" + return _slug(name) + + +# --- test_matrix.json --------------------------------------------------------- + + +def keep_prefixes(impl_files=IMPL_PATTERNS): + """`_KEEP` and the fixed start of each pattern, up to its last ``/`` before a wildcard. + + A body file must be in the matrix that `load_matrix` keeps, so that its + lines can count as covered. ``arch/**/*.c`` adds ``arch/``. A pattern with + a wildcard in its first part adds nothing, so that the matrix does not keep + files outside the tree (``../modules/...``). + """ + out = list(_KEEP) + for pat in impl_files: + fixed = re.split(r"[*?]", pat, maxsplit=1)[0] + if fixed != pat: + fixed = fixed[: fixed.rfind("/") + 1] + if fixed and not fixed.startswith(("/", "../")) and not fixed.startswith(tuple(out)): + out.append(fixed) + return tuple(out) + + +def load_matrix(matrix_path, keep=_KEEP): + """``(by_test, by_line)`` of a ``test_matrix.json``, for the files under ``keep``. + + * ``by_test[key] = {file: set of covered lines}`` + * ``by_line[file] = {line (int): [keys]}`` + + ``keep``: path prefixes (`_KEEP`, or `keep_prefixes` of the body files). + """ + data = json.loads(Path(matrix_path).read_text()) + by_test = {} + for key, files in data.get("by_test", {}).items(): + by_test[key] = { + f: {int(x) for x in lines} for f, lines in files.items() if f.startswith(keep) + } + by_line = { + f: {int(ln): list(keys) for ln, keys in lines.items()} + for f, lines in data.get("by_line", {}).items() + if f.startswith(keep) + } + return by_test, by_line + + +# --- the run's commit and name ------------------------------------------------ + + +def _git(root, *args): + return subprocess.run( + ["git", *args], cwd=root, capture_output=True, text=True, errors="replace" + ) + + +def run_commit(run_dir, environment, root): + """The commit that the run built, as a full sha in ``root``, or ``None``. + + The first source is ``zephyr.sha`` beside the twister output. The second + source is the ``-g`` of ``environment.zephyr_version`` in twister.json. + """ + candidates = [] + sha_file = Path(run_dir) / "zephyr.sha" + if sha_file.is_file(): + candidates.append(sha_file.read_text().strip()) + m = re.search(r"-g([0-9a-f]{7,})$", (environment or {}).get("zephyr_version", "") or "") + if m: + candidates.append(m.group(1)) + for ref in candidates: + if not ref: + continue + r = _git(root, "rev-parse", "--verify", "--quiet", ref + "^{commit}") + if r.returncode == 0: + return r.stdout.strip() + return None + + +def run_name(run_dir, sha, root): + """The name of the run: the first tag (sorted) on its commit. + + If the commit has no tag, the name is the name of the run directory. Each + run of characters outside ``[A-Za-z0-9_-]`` becomes one ``-``. + """ + name = "" + if sha: + r = _git(root, "tag", "--points-at", sha) + tags = sorted(t for t in r.stdout.split() if t) if r.returncode == 0 else [] + name = tags[0] if tags else "" + name = name or Path(run_dir).resolve().name + return re.sub(r"[^A-Za-z0-9_-]+", "-", name).strip("-") + + +# --- source access at the coverage build's commit ------------------------------- + + +def _glob_regex(pattern): + """A regex for a glob ``pattern``: ``**/`` is zero or more directories, ``*`` is in one.""" + out, i = [], 0 + while i < len(pattern): + if pattern.startswith("**/", i): + out.append("(?:[^/]+/)*") + i += 3 + elif pattern[i] == "*": + out.append("[^/]*") + i += 1 + elif pattern[i] == "?": + out.append("[^/]") + i += 1 + else: + out.append(re.escape(pattern[i])) + i += 1 + return re.compile("".join(out) + r"\Z") + + +class Source: + """Reads the files of the tree at the commit of the coverage run. + + Coverage line numbers are correct only for the sources of the build that + made them. With ``ref`` (the run commit, `run_commit`), `read` uses + ``git show :``. The line ranges are then correct also after + the working tree changes. Without ``ref``, `read` uses the working tree + under ``root``, and the caller warns. + """ + + def __init__(self, root, ref=None): + self.root = Path(root) + self.ref = ref + self._ls = None + + def describe(self): + return self.ref or "working tree" + + def list(self, patterns): + """The relative paths that match one of the glob patterns.""" + regexes = [_glob_regex(p) for p in patterns] + if not self.ref: + files = ( + p.relative_to(self.root).as_posix() + for pat in patterns + for p in self.root.glob(pat) + if p.is_file() + ) + else: + if self._ls is None: + r = _git(self.root, "ls-tree", "-r", "--name-only", self.ref) + if r.returncode != 0: + raise RuntimeError(f"git ls-tree {self.ref} failed in {self.root}: {r.stderr}") + self._ls = r.stdout.split("\n") + files = self._ls + return sorted({f for f in files if any(rx.match(f) for rx in regexes)}) + + def read(self, rel): + """The text of the file at the ref, or None.""" + if not self.ref: + p = (self.root / rel).resolve() + if not str(p).startswith(str(self.root.resolve())) or not p.is_file(): + return None + return p.read_text(errors="replace") + r = _git(self.root, "show", f"{self.ref}:{rel}") + return r.stdout if r.returncode == 0 else None + + +# --- implementation symbol resolution ----------------------------------------- + +#: The identifier before the first ``(`` of a line. The pattern of the +#: original can match only this identifier: its ``[\w \t\*]*`` stops at ``(``. +_CALLEE = re.compile(r"[^(]*?\b(\w+)\s*\(") + + +def resolve_impl_symbols(source, symbols, impl_files=IMPL_PATTERNS): + """The function bodies of the satisfying symbols (best effort). + + A system call has more than one body. ``z_impl_`` is the + implementation in supervisor mode. ``z_vrfy_`` is the verifier that a + ZTEST_USER test reaches instead. The verifier can do the operation itself + and never call the z_impl body (z_vrfy_k_thread_create does this). The + function collects all bodies. A test that runs one of them runs the + implementation. + + A definition starts at column 0 (``static [ALWAYS_INLINE] inline`` is + permitted), and its line does not end in ``;``. Its body ends at the first + ``}`` in column 0, in 500 lines or less. In a header, only a ``static`` + line counts: other lines are prototypes or macros. So a macro has no body. + A header is a ``.h`` file anywhere in the tree (``kernel/include/`` too). + + ``impl_files``: the glob patterns of the files to search (`IMPL_PATTERNS`). + + Returns {sym: [{"file", "a", "b", "variant"}, ...]}. + """ + sources = {} + for rel in source.list(impl_files): + text = source.read(rel) + if text is not None: + sources[rel] = text.split("\n") + + # Candidate lines by the identifier before their first "(". The patterns + # of a symbol then run on a small number of lines, not on each line of the + # tree. The full pattern decides. + candidates = defaultdict(list) + for f, lines in sources.items(): + in_header = f.endswith(".h") + for i, ln in enumerate(lines): + if ln.rstrip().endswith(";"): + continue + # in headers only accept static-inline bodies (anything else + # is a prototype or macro), in .c files anything definition-like + if in_header and not ln.lstrip().startswith("static"): + continue + m = _CALLEE.match(ln) + if m: + candidates[m.group(1)].append((f, i)) + + resolved = {} + for sym in symbols: + variants = [(f"z_impl_{sym}", "impl"), (f"z_vrfy_{sym}", "vrfy"), (sym, "def")] + bodies = [] + for n, variant in variants: + p = re.compile( + r"^(static +(ALWAYS_INLINE +|__?always_inline +)?inline +)?" + r"[A-Za-z_][\w \t\*]*\b" + re.escape(n) + r"\s*\(" + ) + for f, i in candidates.get(n, ()): + lines = sources[f] + if not p.match(lines[i]): + continue + stop = min(i + _MAX_BODY, len(lines)) + end = next((j + 1 for j in range(i + 1, stop) if lines[j] == "}"), None) + if end: + bodies.append({ + "file": f, "a": i + 1, "b": end, + "variant": "inline" if f.endswith(".h") else variant, + }) + if bodies: + # stable order: impl first, then vrfy, inline, plain definitions + order = {"impl": 0, "vrfy": 1, "inline": 2, "def": 3} + bodies.sort(key=lambda x: (order[x["variant"]], x["file"], x["a"])) + resolved[sym] = bodies + return resolved + + +# --- the coverage run ------------------------------------------------------------ + + +class CoverageRun: + """One per-test coverage run, joined to the test cases of the spec. + + * ``cases[case id] = {"statuses": [...], "keys": [matrix keys]}``, for each + spec case that twister ran in this run. + * ``case_of_key[key] = {case ids}``: the reverse map, for the question + "which tests ran this line". + * ``unmatched``: the twister cases with no unique spec case (sorted names). + """ + + def __init__(self, name, sha, environment, by_test, by_line, cases, unmatched): + self.name = name + self.sha = sha + self.environment = environment + self.by_test = by_test + self.by_line = by_line + self.cases = cases + self.unmatched = unmatched + self.case_of_key = defaultdict(set) + for cid, c in cases.items(): + for k in c["keys"]: + self.case_of_key[k].add(cid) + + +def join_cases(twister, spec_lookup, by_test): + """``(cases, unmatched)`` for `CoverageRun`, from twister.json and the spec. + + The matrix key of a case comes from the scenario and the C function name + of the spec (`matrix_key`). The form with the suite is used only if the + plain key is not in the matrix and is not the plain key of another case. + The values of a parameterized test share one key, because the per-test + dump has no value in its tag. + """ + ran = [] # (scenario, suite, fn, status, name) + for ts in twister.get("testsuites", []): + scenario = ts.get("name", "") + for tc in ts.get("testcases", []): + name = tc.get("identifier", "") + suite, fn, _ = split_case_name(name, scenario) + ran.append((scenario, suite, fn, (tc.get("status") or "").lower(), name)) + + cases, unmatched, plain_owned = {}, set(), set() + joined = [] + for scenario, suite, fn, status, name in ran: + info = spec_lookup.find(suite, fn) + if info is None: + unmatched.add(name) + continue + joined.append((scenario, suite, info, status)) + plain_owned.add(matrix_key(scenario, info["test_function"])) + for scenario, suite, info, status in joined: + c = cases.setdefault(info["id"], {"statuses": [], "keys": []}) + c["statuses"].append(status) + key = matrix_key(scenario, info["test_function"]) + if key not in by_test and suite: + qualified = matrix_key(scenario, info["test_function"], suite) + if qualified in by_test and qualified not in plain_owned: + key = qualified + if key in by_test and key not in c["keys"]: + c["keys"].append(key) + return cases, sorted(unmatched) + + +def load_coverage_run(run_dir, spec_lookup, root, name=None, impl_files=IMPL_PATTERNS): + """Read a coverage run directory: twister.json, coverage/test_matrix.json, zephyr.sha. + + ``impl_files``: the body files of `resolve_impl_symbols`. The matrix keeps + their lines (`keep_prefixes`). + + Returns ``(run, inputs)``: the `CoverageRun` and the files that it reads. + """ + run_dir = Path(run_dir) + twister_json = run_dir / "twister.json" + matrix_json = run_dir / "coverage" / "test_matrix.json" + inputs = [twister_json, matrix_json, run_dir / "zephyr.sha"] + twister = json.loads(twister_json.read_text()) + by_test, by_line = load_matrix(matrix_json, keep_prefixes(impl_files)) + env = twister.get("environment", {}) + sha = run_commit(run_dir, env, root) + cases, unmatched = join_cases(twister, spec_lookup, by_test) + return ( + CoverageRun(name or run_name(run_dir, sha, root), sha, env, by_test, by_line, cases, + unmatched), + inputs, + ) + + +# --- links from the needs ----------------------------------------------------------- + + +def _needs_of(json_path): + data = json.loads(Path(json_path).read_text()) + versions = data.get("versions", {}) + current = data.get("current_version") or next(iter(versions), None) + return versions.get(current, {}).get("needs", {}) if current is not None else {} + + +def collect_links(json_paths, need_names=None): + """``(spec_lookup, verified_by, satisfied_by, ids)`` from needs.json files. + + * ``spec_lookup``: a `SpecLookup` of the test cases (type role ``case``). + * ``verified_by[req] = [case ids]``, through the ``verifies`` role. + * ``satisfied_by[req] = [symbols]``: the implementation needs (type role + ``implementation``, ids ``-``), through ``satisfies``. + * ``ids``: each need id in the files. A link target that is not a need + gets no assessment. + + If several files export one need, it counts once. + """ + case_type = _need_name(need_names, "case") + impl_type = _need_name(need_names, "implementation") + verifies = _need_name(need_names, "verifies") + satisfies = _need_name(need_names, "satisfies") + impl_prefix = symbol_need_id("", need_names) + seen = {} + for path in json_paths: + for nid, need in _needs_of(path).items(): + seen.setdefault(nid, need) + entries, verified_by, satisfied_by = [], defaultdict(list), defaultdict(list) + for nid, need in seen.items(): + if need.get("type") == case_type and need.get("test_function"): + entries.append({ + "id": nid, + "test_function": need["test_function"], + "test_module": need.get("test_module", ""), + "suite": need.get("suite", ""), + }) + for req in need.get(verifies, []) or []: + verified_by[req].append(nid) + elif need.get("type") == impl_type and nid.startswith(impl_prefix): + for req in need.get(satisfies, []) or []: + satisfied_by[req].append(nid[len(impl_prefix):]) + return SpecLookup(entries), dict(verified_by), dict(satisfied_by), set(seen) + + +# --- verdicts -------------------------------------------------------------------------- + + +def _rollup(statuses): + npass = sum(s == "passed" for s in statuses) + nfail = sum(s in _FAIL for s in statuses) + nskip = sum(s in _SKIP for s in statuses) + return ("failing" if nfail else + "passing" if npass else + "skipped" if nskip else "no-run") + + +def evidence(case_ids, run): + """The state of the verifying cases: untested, no-run, failing, passing or skipped.""" + if not case_ids: + return "untested" + roll = [_rollup(run.cases[c]["statuses"]) for c in case_ids if c in run.cases] + if not roll: + return "no-run" + if "failing" in roll: + return "failing" + if "passing" in roll: + return "passing" + return "skipped" + + +def adequacy(symbols, case_ids, run, impl_loc): + """``{"verdict", "impls"}`` of a requirement from its ``symbols`` and ``case_ids``. + + ``impls`` has one entry for each symbol and body, with these keys: + + * ``sym`` and ``variant``. + * ``file``, ``a`` and ``b``: ``None`` for a symbol with no body. + * ``own``: the number of body lines that the own tests ran. + * ``any``: the number of body lines that any test of the run ran. + * ``own_tests``: ``{case id: [lines]}``. + * ``other_tests``: ``{matrix key: [lines]}``, for the keys of other tests. + """ + if not symbols: + return {"verdict": "no-impl", "impls": []} + own_keys = [k for c in case_ids if c in run.cases for k in run.cases[c]["keys"]] + tests_with_cov = [c for c in case_ids if c in run.cases and run.cases[c]["keys"]] + detail, resolved, adjudicable, own_hits = [], 0, 0, 0 + for sym in sorted(set(symbols)): + bodies = impl_loc.get(sym) + if not bodies: + detail.append({"sym": sym, "variant": None, "file": None, "a": None, "b": None, + "own": 0, "any": 0, "own_tests": {}, "other_tests": {}}) + continue + resolved += 1 + sym_hit = sym_seen = False + for body in bodies: + f, a, b = body["file"], body["a"], body["b"] + fl = run.by_line.get(f, {}) + exec_lines = sorted(ln for ln in fl if a <= ln <= b) + own, own_tests = set(), {} + for c in tests_with_cov: + hit = set() + for k in run.cases[c]["keys"]: + hit.update(ln for ln in run.by_test[k].get(f, ()) if a <= ln <= b) + if hit: + own_tests[c] = sorted(hit) + own |= hit + other_tests = defaultdict(set) + for ln in exec_lines: + for k in fl[ln]: + if k not in own_keys: + other_tests[k].add(ln) + if own: + sym_hit = True + if exec_lines: + sym_seen = True + detail.append({"sym": sym, "variant": body["variant"], "file": f, "a": a, "b": b, + "own": len(own), "any": len(exec_lines), "own_tests": own_tests, + "other_tests": {k: sorted(v) for k, v in sorted(other_tests.items())}}) + # If no test of the run covers the bodies of a symbol (boot-time + # code, inlined code, code that the configuration removes), coverage + # cannot judge it. It must not count as "broken". + if sym_seen: + adjudicable += 1 + if sym_hit: + own_hits += 1 + if not resolved: + verdict = "unresolved" + elif not tests_with_cov: + verdict = "no-cov" + elif not adjudicable: + verdict = "unattributed" + elif own_hits == adjudicable: + verdict = "true" + elif own_hits: + verdict = "partial" + else: + verdict = "broken" + return {"verdict": verdict, "impls": detail} + + +def assess(run, verified_by, satisfied_by, source, ids=None, impl_files=IMPL_PATTERNS): + """The assessment of each requirement in the scope of the run: ``({req: result}, impl_loc)``. + + A requirement is in the scope if twister ran at least one of its verifying + cases in this run. If ``ids`` is given, the requirement must also be a need. + Each result has ``verdict`` and ``impls`` (`adequacy`), ``evidence``, + ``symbols`` and ``cases``. ``impl_files``: the body files + (`resolve_impl_symbols`). + """ + reqs = sorted( + r for r, cs in verified_by.items() + if any(c in run.cases for c in cs) and (ids is None or r in ids) + ) + symbols = sorted({s for r in reqs for s in satisfied_by.get(r, [])}) + impl_loc = resolve_impl_symbols(source, symbols, impl_files) + out = {} + for r in reqs: + cases = sorted(set(verified_by[r])) + syms = sorted(set(satisfied_by.get(r, []))) + res = adequacy(syms, cases, run, impl_loc) + res.update(evidence=evidence(cases, run), symbols=syms, cases=cases) + out[r] = res + return out, impl_loc + + +def line_ranges(lines): + """``"79-81, 84"`` for sorted line numbers.""" + parts, start, prev = [], None, None + for ln in lines: + if start is None: + start = prev = ln + elif ln == prev + 1: + prev = ln + else: + parts.append(f"{start}-{prev}" if prev != start else f"{start}") + start = prev = ln + if start is not None: + parts.append(f"{start}-{prev}" if prev != start else f"{start}") + return ", ".join(parts) diff --git a/sphinx/_extensions/rst_builders.py b/sphinx/_extensions/rst_builders.py index ff553f0..581e15a 100644 --- a/sphinx/_extensions/rst_builders.py +++ b/sphinx/_extensions/rst_builders.py @@ -52,6 +52,15 @@ def slugify(s): # depends_on, and why a skipped result was skipped. "depends_met": "depends_met", "skip_class": "skip_class", + # testcoverage: one adequacy need per requirement and coverage run, its + # link to the requirement, and its fields (`testcoverage_need_*`). + "adequacy": "adequacy", + "assesses": "assesses", + "verdict": "verdict", + "evidence": "evidence", + "coverage_run": "coverage_run", + "judged_symbols": "judged_symbols", + "symbol_hits": "symbol_hits", } #: The result-field roles `build_result_rst` can set (see its ``fields``). diff --git a/sphinx/_extensions/test_coverage.py b/sphinx/_extensions/test_coverage.py new file mode 100644 index 0000000..f1f373a --- /dev/null +++ b/sphinx/_extensions/test_coverage.py @@ -0,0 +1,334 @@ +# Copyright (c) 2026 inovex GmbH +# +# SPDX-License-Identifier: Apache-2.0 + +"""Sphinx extension: the testcoverage directive (coverage adequacy as needs). + +``.. testcoverage::`` reads a per-test coverage run (`adequacy`) and emits one +adequacy need per requirement that the run can assess. Each need links to its +requirement (link role ``assesses``) and carries the verdict. The directive +also writes a summary of the run and the distribution of the verdicts. +""" + +import re +from collections import Counter +from pathlib import Path + +from adequacy import ( + IMPL_PATTERNS, + VERDICTS, + Source, + assess, + collect_links, + line_ranges, + load_coverage_run, +) +from docutils import nodes +from docutils.parsers.rst import Directive, directives +from docutils.statemachine import ViewList +from input_tracking import _note_input +from needs_fields import field_type +from rst_builders import _need_name + +from sphinx.util import logging + +logger = logging.getLogger(__name__) + +#: The adequacy field roles (`testcoverage_need_fields`). +ADEQUACY_FIELD_ROLES = ("verdict", "evidence", "coverage_run", "judged_symbols", "symbol_hits") + +#: What each verdict says, for the distribution table. +VERDICT_MEANING = { + "broken": "Tests of the run reach the code. The requirement's own tests never do.", + "partial": "The own tests run some of the satisfying symbols, not all.", + "unattributed": "No test of the run covers any body. Coverage cannot judge the link.", + "unresolved": "No satisfying symbol maps to a body (a macro).", + "no-cov": "The verifying tests have no coverage data in this run.", + "no-impl": "No symbol satisfies the requirement.", + "true": "The own tests run every symbol that coverage can judge.", +} + +#: At most this many other tests are named for one body. +_OTHER_TESTS_SHOWN = 10 + + +def _need_names(config): + """The role->name mapping of every vocabulary the directive reads or writes.""" + names = {} + for key in ( + "testmodule_need_types", "testmodule_need_links", + "symbolneeds_need_types", "symbolneeds_need_links", + "testcoverage_need_types", "testcoverage_need_links", "testcoverage_need_fields", + ): + names.update(getattr(config, key, {}) or {}) + return names + + +def adequacy_need_id(run, req, prefix="ADQ"): + """``-/``, the id of ``req``'s adequacy need in ``run``.""" + return f"{prefix}-{run}/{req}" + + +def _natural(uid): + return [int(t) if t.isdigit() else t for t in re.split(r"(\d+)", uid)] + + +def _need_ref(key, run): + """A matrix key as the case needs it stands for, else as a literal.""" + cases = sorted(run.case_of_key.get(key, ())) + return ", ".join(f":need:`{c}`" for c in cases) if cases else f"``{key}``" + + +def _symbol_hits(impls): + """``"k_sem_init: own 11, any 11"`` per symbol, joined with ``"; "``.""" + own, any_, order = Counter(), Counter(), [] + for d in impls: + if d["sym"] not in order: + order.append(d["sym"]) + own[d["sym"]] += d["own"] + any_[d["sym"]] += d["any"] + return "; ".join(f"{s}: own {own[s]}, any {any_[s]}" for s in order) + + +def build_adequacy_rst(req, res, run, need_names=None, fields=(), prefix="ADQ", layout=""): + """RST lines for the adequacy need of ``req`` (``res`` from `adequacy.assess`). + + ``layout``: the sphinx-needs layout of the need, if not empty. + """ + values = { + "verdict": res["verdict"], + "evidence": res["evidence"], + "coverage_run": run.name, + "judged_symbols": "; ".join(res["symbols"]), + "symbol_hits": _symbol_hits(res["impls"]), + } + lines = [ + f".. {_need_name(need_names, 'adequacy')}:: Adequacy of {req}", + f" :id: {adequacy_need_id(run.name, req, prefix)}", + f" :{_need_name(need_names, 'assesses')}: {req}", + ] + for role in ADEQUACY_FIELD_ROLES: + if role in fields and values[role]: + lines.append(f" :{_need_name(need_names, role)}: {values[role]}") + if layout: + lines.append(f" :layout: {layout}") + lines += [ + "", + f" Verdict ``{res['verdict']}``, evidence ``{res['evidence']}``, " + f"coverage run ``{run.name}``.", + "", + ] + cases = ", ".join(f":need:`{c}`" for c in res["cases"]) + lines += [f" Verifying test cases: {cases}.", ""] + if not res["impls"]: + lines += [" No symbol satisfies this requirement.", ""] + return lines + lines += [ + " .. list-table:: Satisfying symbols", + " :header-rows: 1", + " :widths: 20 10 30 10 10", + "", + " * - Symbol", + " - Body", + " - Location", + " - Own lines", + " - Any lines", + ] + for d in res["impls"]: + loc = f"``{d['file']}:{d['a']}-{d['b']}``" if d["file"] else "no body found" + lines += [ + f" * - ``{d['sym']}``", + f" - {d['variant'] or '—'}", + f" - {loc}", + f" - {d['own']}", + f" - {d['any']}", + ] + lines.append("") + for d in res["impls"]: + if not d["file"] or not (d["own_tests"] or d["other_tests"]): + continue + lines += [f" ``{d['sym']}`` ({d['variant']}, ``{d['file']}:{d['a']}-{d['b']}``):", ""] + for case, hit in d["own_tests"].items(): + lines.append(f" * own :need:`{case}`: lines {line_ranges(hit)}") + others = list(d["other_tests"].items()) + for key, hit in others[:_OTHER_TESTS_SHOWN]: + lines.append(f" * other {_need_ref(key, run)}: lines {line_ranges(hit)}") + if len(others) > _OTHER_TESTS_SHOWN: + lines.append(f" * and {len(others) - _OTHER_TESTS_SHOWN} other tests") + lines.append("") + return lines + + +def build_coverage_rst( + results, run, source, impl_loc, need_names=None, fields=(), prefix="ADQ", layout="", + impl_files=IMPL_PATTERNS, +): + """RST lines for the whole directive: run summary, distribution, needs by verdict. + + ``impl_files``: the body files that the run searched, for the summary. + """ + counts = Counter(r["verdict"] for r in results.values()) + symbols = sorted({s for r in results.values() for s in r["symbols"]}) + lines = [ + ".. list-table:: Coverage run", + " :header-rows: 0", + " :widths: 30 70", + "", + " * - Run", + f" - ``{run.name}``", + " * - Sources read at", + f" - ``{source.describe()}``", + " * - Files searched for bodies", + " - " + ", ".join(f"``{p}``" for p in impl_files), + " * - Tests in the coverage matrix", + f" - {len(run.by_test)}", + " * - Spec test cases the run ran", + f" - {len(run.cases)}", + " * - Requirements assessed", + f" - {len(results)}", + " * - Satisfying symbols (with a body)", + f" - {len(symbols)} ({sum(1 for s in symbols if s in impl_loc)})", + "", + ".. list-table:: Verdicts", + " :header-rows: 1", + " :widths: 15 10 75", + "", + " * - Verdict", + " - Requirements", + " - Meaning", + ] + for v in VERDICTS: + lines += [f" * - ``{v}``", f" - {counts.get(v, 0)}", f" - {VERDICT_MEANING[v]}"] + lines.append("") + for v in VERDICTS: + reqs = sorted((r for r, res in results.items() if res["verdict"] == v), key=_natural) + if not reqs: + continue + heading = f"Verdict {v}" + lines += [heading, "-" * len(heading), ""] + for req in reqs: + lines += build_adequacy_rst( + req, results[req], run, need_names, fields, prefix, layout + ) + return lines + + +class TestCoverageDirective(Directive): + """Emit one adequacy need per requirement that a per-test coverage run can assess. + + Usage:: + + .. testcoverage:: + :run: my-run + :layout: adequacy + + The optional argument is the run directory. Without it, the directive reads + ``coverage_output_dir`` (``ZDOCS_COVERAGE_OUT``). ``:run:`` names the run in + the need ids. Without it, the name is the first tag on the run commit, or + else the name of the run directory. ``:layout:`` sets the sphinx-needs + layout of each need. ``testcoverage_impl_files`` sets the files that hold + the bodies of the satisfying symbols. + """ + + required_arguments = 0 + optional_arguments = 1 + has_content = False + option_spec = {"run": directives.unchanged, "layout": directives.unchanged} + + def _paragraph(self, text): + return [nodes.paragraph(text=text)] + + def run(self): + env = self.state.document.settings.env + app = env.app + config = app.config + run_dir = (self.arguments[0].strip() if self.arguments else "") or getattr( + config, "coverage_output_dir", "" + ) + if not run_dir: + return self._paragraph("[testcoverage: no coverage run configured]") + run_dir = Path(run_dir) + for name in ("twister.json", "coverage/test_matrix.json", "zephyr.sha"): + _note_input(env, run_dir / name) + if not (run_dir / "coverage" / "test_matrix.json").is_file(): + logger.warning(f"testcoverage: no coverage/test_matrix.json in {run_dir}") + return self._paragraph( + f"[testcoverage: coverage matrix not found in {run_dir.name}]" + ) + + need_names = _need_names(config) + json_paths = [ + e["json_path"] for e in getattr(config, "needs_external_needs", []) or [] + if e.get("json_path") + ] + spec_json = getattr(config, "testspec_needs_json", "") + if spec_json: + json_paths.append(spec_json) + json_paths = [p for p in dict.fromkeys(json_paths) if Path(p).is_file()] + for p in json_paths: + _note_input(env, p) + + root = getattr(config, "testmodule_root", "") or env.srcdir + impl_files = tuple(getattr(config, "testcoverage_impl_files", None) or IMPL_PATTERNS) + try: + spec_lookup, verified_by, satisfied_by, ids = collect_links(json_paths, need_names) + run, _ = load_coverage_run( + run_dir, spec_lookup, root, name=self.options.get("run", "").strip() or None, + impl_files=impl_files, + ) + except Exception as exc: + logger.warning(f"testcoverage: cannot read the coverage run {run_dir}: {exc}") + return self._paragraph(f"[testcoverage: cannot read {run_dir.name}]") + if not run.sha: + logger.warning( + f"testcoverage: the commit of {run_dir} is not in {root}; the sources " + f"come from the working tree, and line ranges can be wrong for files " + f"changed since the run" + ) + source = Source(root, run.sha) + results, impl_loc = assess(run, verified_by, satisfied_by, source, ids, impl_files) + if not results: + return self._paragraph("[testcoverage: the run ran no verifying test case]") + + fields = { + role for role in ADEQUACY_FIELD_ROLES + if field_type(env, _need_name(need_names, role)) is not None + } + lines = build_coverage_rst( + results, run, source, impl_loc, need_names, fields, + getattr(config, "testcoverage_id_prefix", "ADQ"), + self.options.get("layout", "").strip(), + impl_files, + ) + dump_dir = getattr(config, "dump_generated_rst", "") + if dump_dir: + out = Path(dump_dir) + out.mkdir(parents=True, exist_ok=True) + doc_slug = env.docname.replace("/", "__") + (out / f"{doc_slug}__testcoverage__{run.name}.rst").write_text( + "\n".join(lines), encoding="utf-8" + ) + container = nodes.container() + self.state.nested_parse( + ViewList(lines, source=""), self.content_offset, container, + match_titles=True, + ) + return container.children + + +def setup(app): + # The per-test coverage run directory (ZDOCS_COVERAGE_OUT): twister.json, + # coverage/test_matrix.json and zephyr.sha. + app.add_config_value("coverage_output_dir", "", "env") + app.add_config_value("testcoverage_id_prefix", "ADQ", "env") + # Glob patterns of the files that hold the bodies, relative to testmodule_root. + app.add_config_value("testcoverage_impl_files", list(IMPL_PATTERNS), "env") + app.add_config_value("testcoverage_need_types", {"adequacy": "adequacy"}, "env") + app.add_config_value("testcoverage_need_links", {"assesses": "assesses"}, "env") + # Field roles -> names. A field is set only where the consumer declares it. + app.add_config_value( + "testcoverage_need_fields", {role: role for role in ADEQUACY_FIELD_ROLES}, "env" + ) + app.add_directive("testcoverage", TestCoverageDirective) + app.setup_extension("input_tracking") + return {"version": "0.1", "parallel_read_safe": True} diff --git a/sphinx/_extensions/test_module.py b/sphinx/_extensions/test_module.py index 6cf020d..06b8b66 100644 --- a/sphinx/_extensions/test_module.py +++ b/sphinx/_extensions/test_module.py @@ -908,4 +908,6 @@ def setup(app): app.add_directive("testreport", TestReportDirective) app.add_directive("twisterinfo", TwisterInfoDirective) app.setup_extension("input_tracking") + # The testcoverage directive: coverage adequacy beside the test results. + app.setup_extension("test_coverage") return {"version": "0.2", "parallel_read_safe": True} diff --git a/sphinx/_extensions/twister_reader.py b/sphinx/_extensions/twister_reader.py index d3c74d7..2d5edc2 100644 --- a/sphinx/_extensions/twister_reader.py +++ b/sphinx/_extensions/twister_reader.py @@ -18,6 +18,7 @@ from rst_builders import _need_name, _values_summary __all__ = [ + "split_case_name", "parse_twister_results", "normalise_test_path", "scenario_selected", @@ -100,6 +101,29 @@ def scenario_selected( return True +def split_case_name(name, scenario): + """``(suite, function, instance)`` of twister test case ``name`` in ``scenario``. + + Twister names a case ``..``, with ztest's ``test_`` + stripped from ``fn`` (and so from ``function`` here). A parameterized test + (ZTEST_P) reports one case per value as ``.[/ + ]``: no suite segment (``suite`` is ``""``), and the value may + contain anything, dots included, so it is split off before the name is. + ``instance`` is the part in brackets, or ``None``. + """ + base, instance = name, None + if name.endswith("]") and "[" in name: + cut = name.index("[") + base, instance = name[:cut], name[cut + 1 : -1] + suffix = base[len(scenario) + 1 :] if base.startswith(scenario + ".") else base + parts = suffix.rsplit(".", 1) + suite = parts[0] if len(parts) == 2 else "" + function = parts[-1] + if function.startswith("test_"): + function = function[5:] + return suite, function, instance + + def parse_twister_results( xml_path, module_filter=None, exact=False, path_filter=None, suite_paths=None ): @@ -121,20 +145,7 @@ def parse_twister_results( continue name = tc.get("name", "") scenario = classname - # A parameterized test (ZTEST_P) reports one result per value as - # `.[/]`: no suite segment, and - # the value may contain anything, dots included, so it is split - # off before the name is. - base, instance = name, None - if name.endswith("]") and "[" in name: - cut = name.index("[") - base, instance = name[:cut], name[cut + 1 : -1] - suffix = base[len(scenario) + 1 :] if base.startswith(scenario + ".") else base - parts = suffix.rsplit(".", 1) - suite = parts[0] if len(parts) == 2 else "" - function = parts[-1] - if function.startswith("test_"): - function = function[5:] + suite, function, instance = split_case_name(name, scenario) failure = tc.find("failure") error = tc.find("error") skipped = tc.find("skipped") diff --git a/sphinx/zdocs_conf.py b/sphinx/zdocs_conf.py index 929f975..ca760ff 100644 --- a/sphinx/zdocs_conf.py +++ b/sphinx/zdocs_conf.py @@ -427,6 +427,7 @@ def configure( namespace["testspec_needs_json"] = testmodule["needs_json"] namespace["testmodule_root"] = str(project_base) if project_base else "" namespace["twister_output_dir"] = os.environ.get("ZDOCS_TWISTER_OUT", "") + namespace["coverage_output_dir"] = os.environ.get("ZDOCS_COVERAGE_OUT", "") namespace["twisterinfo_project_name"] = project namespace["twisterinfo_project_version"] = version if symbol_needs is not None: