From f825409b22b67a45034967960ade68e177fb487c Mon Sep 17 00:00:00 2001 From: Ostrzyciel Date: Mon, 26 Oct 2026 22:00:38 +0100 Subject: [PATCH] Implement Jelly-SPARQL conformance reports --- .../generate_conformance_report.py | 190 ++++++++++-------- 1 file changed, 109 insertions(+), 81 deletions(-) diff --git a/tests/conformance_tests/generate_conformance_report.py b/tests/conformance_tests/generate_conformance_report.py index 4fc9c1b9..206ef80f 100644 --- a/tests/conformance_tests/generate_conformance_report.py +++ b/tests/conformance_tests/generate_conformance_report.py @@ -5,10 +5,10 @@ import re import shutil import subprocess +import sys from collections import defaultdict -from dataclasses import dataclass +from collections.abc import Collection from pathlib import Path -from typing import cast import pytest from _pytest.nodes import Item @@ -23,19 +23,58 @@ ASSERTOR = URIRef("#assertor") IMPL = URIRef("#impl") DEVELOPER = URIRef("#developer") + +GENERIC = "generic" +RDFLIB = "rdflib" +IMPLEMENTATIONS = (GENERIC, RDFLIB) + _outcome_map = { "passed": EARL.passed, "failed": EARL.failed, "inapplicable": EARL.inapplicable, } +_outcome_priority = ("failed", "passed", "inapplicable") + + +def implementation_of(node_id: str) -> str | None: + """ + Indicate which integration is tested by the conformance test. + + Every conformance test says so in its name: `rdflib` for the RDFLib + integration, `generic` for the Generic API. + + Args: + node_id (str): pytest node ID of the test + + Returns: + str | None: the integration, or nothing if the name tells us neither + + """ + name = node_id.rsplit("::", 1)[-1] + if RDFLIB in name: + return RDFLIB + if GENERIC in name: + return GENERIC + return None + + +def combine_outcomes(outcomes: Collection[str]) -> str: + """ + Sum up what one integration did with one case. + + Args: + outcomes (Collection[str]): outcomes of the tests that integration ran + for the case; empty if it ran none -@dataclass(frozen=True) -class Result: - """Single test result.""" + Returns: + str: the outcome to report for the case - name: str - outcome: str + """ + for outcome in _outcome_priority: + if outcome in outcomes: + return outcome + return "inapplicable" def get_release_version_date() -> tuple[str, str]: @@ -70,7 +109,8 @@ class ConformanceReportPlugin: Plugin class used during pytest execution for conformance test reporting. Pytest Hooks: - pytest_runtest_makereport: store results data from each test + pytest_itemcollected: remember which case each test runs + pytest_runtest_logreport: store results data from each test pytest_sessionfinish: build RDF graph with all test results """ @@ -97,8 +137,12 @@ def __init__( self.path_to_generic_report = path_to_generic_report self.path_to_rdflib_sparql_report = path_to_rdflib_sparql_report self.path_to_generic_sparql_report = path_to_generic_sparql_report - self._results: defaultdict[str, list[Result]] = defaultdict(list) - self._case_by_id: dict[str, object | None] = defaultdict(list) + # Case IRI -> integration -> outcomes of that integration's tests. + self._results: defaultdict[str, defaultdict[str, list[str]]] = defaultdict( + lambda: defaultdict(list) + ) + self._case_by_id: dict[str, object | None] = {} + self.unclassified: set[str] = set() @staticmethod def _outcome_from_report(report: pytest.TestReport) -> str: @@ -237,7 +281,8 @@ def _add_test_record_to_graph( g.add((result, EARL.outcome, _outcome_map[outcome])) return g - def pytest_runtest_call(self, item: Item) -> None: + def pytest_itemcollected(self, item: Item) -> None: + # Every collected test, so that the ones a mark skips are in here too. case = None if hasattr(item, "callspec"): case = item.callspec.params.get("case") @@ -251,19 +296,22 @@ def pytest_runtest_logreport(self, report: TestReport) -> None: report (TestReport): single test report """ - if report.when != "call": + if report.when == "teardown": + return + if report.when == "setup" and report.outcome == "passed": + # The test itself is yet to run. A test that never gets that far, + # because a mark skips it, is only ever reported here. return - node_id = report.nodeid - case = self._case_by_id.get(node_id) - outcome = self._outcome_from_report(report) - test_name = report.nodeid.rsplit("::", 1)[-1] + case = self._case_by_id.get(report.nodeid) uri = getattr(case, "uri", None) - self._results[cast(str, uri)].append( - Result( - name=test_name, - outcome=outcome, - ) - ) + if uri is None: + # Not a conformance case -- a doctest, say. + return + implementation = implementation_of(report.nodeid) + if implementation is None: + self.unclassified.add(report.nodeid) + return + self._results[uri][implementation].append(self._outcome_from_report(report)) def pytest_sessionfinish(self) -> None: """ @@ -271,67 +319,44 @@ def pytest_sessionfinish(self) -> None: Builds RDF Graph using RDFLib and serializes to Turtle. """ - rdf_graphs = ( - self._initialize_graph(implementation="generic"), - self._initialize_graph(implementation="RDFLib"), + rdf_graphs = { + implementation: self._initialize_graph(implementation) + for implementation in IMPLEMENTATIONS + } + sparql_graphs = { + implementation: self._initialize_graph(implementation) + for implementation in IMPLEMENTATIONS + } + + for uri, outcomes in self._results.items(): + # The tests of Jelly-SPARQL have reports of their own + graphs = sparql_graphs if "/tests/sparql/" in uri else rdf_graphs + for implementation, g in graphs.items(): + # An integration that ran no test for the case gets an + # inapplicable record, so that every report covers every case. + self._add_test_record_to_graph( + g, + combine_outcomes(outcomes.get(implementation, ())), + uri, + ASSERTOR, + IMPL, + ) + + rdf_graphs[RDFLIB].serialize(self.path_to_rdflib_report, format="turtle") + rdf_graphs[GENERIC].serialize(self.path_to_generic_report, format="turtle") + sparql_graphs[RDFLIB].serialize( + self.path_to_rdflib_sparql_report, format="turtle" ) - sparql_graphs = ( - self._initialize_graph(implementation="generic"), - self._initialize_graph(implementation="RDFLib"), + sparql_graphs[GENERIC].serialize( + self.path_to_generic_sparql_report, format="turtle" ) - for k, v in self._results.items(): - # The tests of Jelly-SPARQL have reports of their own - is_sparql = "/tests/sparql/" in str(k) - g_generic, g_rdflib = sparql_graphs if is_sparql else rdf_graphs - if len(v) == 1: - r = v[0] - if ( - "test_generic" in r.name - or "test_serializes_generic" in r.name - or "test_serializing_fails_generic" in r.name - ): - g_generic = self._add_test_record_to_graph( - g_generic, r.outcome, k, ASSERTOR, IMPL - ) - g_rdflib = self._add_test_record_to_graph( - g_rdflib, "inapplicable", k, ASSERTOR, IMPL - ) - elif ( - "test_rdflib" in r.name - or "test_serializes_rdf" in r.name - or "test_serializing_fails_rdf" in r.name - ): - g_rdflib = self._add_test_record_to_graph( - g_rdflib, r.outcome, k, ASSERTOR, IMPL - ) - g_generic = self._add_test_record_to_graph( - g_generic, "inapplicable", k, ASSERTOR, IMPL - ) - - if len(v) > 1: - for r in v: - if ( - "test_generic" in r.name - or "test_serializes_generic" in r.name - or "test_serializing_fails_generic" in r.name - ): - g_generic = self._add_test_record_to_graph( - g_generic, r.outcome, k, ASSERTOR, IMPL - ) - elif ( - "test_rdflib" in r.name - or "test_serializes_rdf" in r.name - or "test_serializing_fails_rdf" in r.name - ): - g_rdflib = self._add_test_record_to_graph( - g_rdflib, r.outcome, k, ASSERTOR, IMPL - ) - - rdf_graphs[1].serialize(self.path_to_rdflib_report, format="turtle") - rdf_graphs[0].serialize(self.path_to_generic_report, format="turtle") - sparql_graphs[1].serialize(self.path_to_rdflib_sparql_report, format="turtle") - sparql_graphs[0].serialize(self.path_to_generic_sparql_report, format="turtle") + if self.unclassified: + sys.stderr.write( + "Left out of the reports, because their names say nothing about " + "which integration they run:\n" + + "".join(f" {node_id}\n" for node_id in sorted(self.unclassified)) + ) def main() -> int: @@ -344,7 +369,10 @@ def main() -> int: path_to_generic_sparql_report="pyjelly Generic API SPARQL.ttl", ) - return pytest.main(args=[tests_dir], plugins=[conformance_report_plugin]) + code = int(pytest.main(args=[tests_dir], plugins=[conformance_report_plugin])) + if conformance_report_plugin.unclassified: + return max(code, 1) + return code if __name__ == "__main__":