Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions src/signaloid/benchmarking/automation/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -326,8 +326,13 @@ output):
`addDistValueTrace` `file:line` directives resolve against unoptimised debug
info. Optimisation could change the traced values. To guard against that,
each config is also built at `-O2` and its Ux strings are checked
(byte-for-byte) against the `-O0` ones. Any difference is reported as a
warning and does not stop the run. If a config's `-O2` build or run fails, that config is skipped and
(byte-for-byte) against the `-O0` ones. That build is compiled *without*
tracing, because the SDK forces `-O0` whenever tracing is on, so it has no
tracing database. Both sides are therefore read from the runs' stdout, where
printing an uncertain value emits its full Ux string. Any difference is
reported as a warning and does not stop the run. A config in which neither
build printed a Ux string is reported as a warning too, so a run that
verified nothing cannot be mistaken for a passing one. If a config's `-O2` build or run fails, that config is skipped and
reported as failing in an `-O2 ux-string verification FAILED` summary (the
run still continues). Set `TRACING_VERIFY_OPTFLAGS` to compare against a
different level.
Expand Down
186 changes: 186 additions & 0 deletions src/signaloid/benchmarking/automation/check_traced_values_printed.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,186 @@
# Copyright (c) 2026, Signaloid.
#
# 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.
"""
Check that every Ux string a tracing run recorded was also printed by it.

The ``-O2`` verification compares the two builds' **stdout**, because the
verification build is compiled without tracing and so has no tracing database
(see ``compare_tracing_ux_strings``). That substitution is only sound while the
SDK prints the same bytes it stores. This module checks exactly that, against
the one run that produces both: the ``-O0`` tracing build writes its traced
values to a database *and* prints them.

A mismatch here means the printed representation has drifted from the stored
one, which would quietly weaken the ``-O2`` check rather than break it. It is
therefore reported as a warning, and the caller continues.

Invoked from the bash tracing layer (``get-timings.sh``) as::

python3 -m signaloid.benchmarking.automation.check_traced_values_printed \
<tracing_db> <stdout> [--table TABLE] [--config SUFFIX]

Exit status is ``0`` when every recorded value was printed, ``1`` when any was
not or when the database recorded nothing, and ``2`` when the check could not
be performed.
"""

import argparse
import sqlite3
import sys
from contextlib import closing

from signaloid.benchmarking.automation.compare_tracing_ux_strings import (
load_ux_strings,
)

# Enough to name which traced expression is missing from stdout.
_IDENTITY_COLUMNS = (
"Expression_DeclarationFileName",
"Expression_Name",
"Expression_DeclarationLineNumber",
)

# A single Athens-16 value runs to several hundred characters, so reports
# truncate. See compare_tracing_ux_strings for the same reasoning.
_REPORTED_PREFIX_LENGTH = 80


def load_recorded_values(db_path: str, table: str) -> dict[str, tuple[object, ...]]:
"""
Load the distinct Ux strings a tracing run wrote, with one identity each.

Args:
db_path: Path to the tracing SQLite database.
table: Name of the traced-values table (e.g. ``TracingTable``).

Returns:
Mapping from Ux string to the identity tuple of an expression that
produced it. One identity is kept per distinct value, which is all the
report needs to point at the offending expression.
"""
columns = ", ".join(f'"{column}"' for column in _IDENTITY_COLUMNS)
query = f'SELECT {columns}, Dist_Value FROM "{table}"'
recorded: dict[str, tuple[object, ...]] = {}
with closing(sqlite3.connect(db_path)) as connection:
for row in connection.execute(query):
recorded.setdefault(row[-1], tuple(row[:-1]))
return recorded


def find_unprinted_values(
db_path: str, stdout_path: str, table: str
) -> tuple[list[tuple[str, tuple[object, ...]]], int]:
"""
Find recorded Ux strings that the same run did not print.

Args:
db_path: Path to the tracing SQLite database.
stdout_path: Captured stdout of the same run.
table: Name of the traced-values table.

Returns:
A tuple of (unprinted values with their identities, number of distinct
values recorded).
"""
recorded = load_recorded_values(db_path, table)
printed = set(load_ux_strings(stdout_path))
unprinted = [
(value, identity)
for value, identity in recorded.items()
if value not in printed
]
return unprinted, len(recorded)


def _format_identity(identity: tuple[object, ...]) -> str:
"""Render an identity tuple as ``expr @ file:line``."""
file_name, name, line = identity
return f"{name} @ {file_name}:{line}"


def main() -> int:
"""
Parse arguments, run the cross-check, and report the result.

Returns:
``0`` when every recorded value was printed, ``1`` when any was not or
when nothing was recorded, and ``2`` when the check could not run.
"""
parser = argparse.ArgumentParser(
description=(
"Check that every Ux string a tracing run recorded in its "
"database was also printed to stdout by that same run."
),
)
parser.add_argument("tracing_db", help="Tracing DB written by the -O0 run.")
parser.add_argument("stdout_path", help="Captured stdout of the same run.")
parser.add_argument(
"--table",
default="TracingTable",
help="Name of the traced-values table (default: TracingTable).",
)
parser.add_argument(
"--config",
default=None,
help="Optional config suffix, included in messages for context.",
)
args = parser.parse_args()

scope = f" for config '{args.config}'" if args.config else ""

try:
unprinted, recorded_count = find_unprinted_values(
args.tracing_db, args.stdout_path, args.table
)
except (OSError, sqlite3.Error) as exc:
print(
f"WARNING: db-vs-stdout check{scope} could not run on "
f"{args.tracing_db} and {args.stdout_path}: {exc}",
file=sys.stderr,
)
return 2

if recorded_count == 0:
print(
f"WARNING: db-vs-stdout check{scope}: the tracing run recorded no "
f"Ux strings, so nothing was cross-checked."
)
return 1

if not unprinted:
print(
f"db-vs-stdout check{scope}: OK — all {recorded_count} recorded "
f"Ux string(s) also appear in the run's stdout."
)
return 0

print(
f"WARNING: db-vs-stdout check{scope}: {len(unprinted)} of "
f"{recorded_count} recorded Ux string(s) were not printed by the same "
f"run. The -O2 check compares stdout, so it no longer covers these."
)
for value, identity in unprinted:
print(f" NOT PRINTED {_format_identity(identity)}")
print(f" recorded: {value[:_REPORTED_PREFIX_LENGTH]}")
return 1


if __name__ == "__main__":
sys.exit(main())
Original file line number Diff line number Diff line change
@@ -0,0 +1,140 @@
# Copyright (c) 2026, Signaloid.
#
# 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.
import sqlite3
import tempfile
import unittest
from collections.abc import Sequence
from pathlib import Path
from unittest.mock import patch

from signaloid.benchmarking.automation.check_traced_values_printed import (
find_unprinted_values,
main,
)

_TABLE = "TracingTable"

_UX_A = "Ux0400000000000000AAAA"
_UX_B = "Ux0400000000000000BBBB"


def _make_tracing_db(path: str, values: Sequence[str]) -> None:
"""Write a tracing DB holding one traced row per entry in *values*."""
with sqlite3.connect(path) as conn:
conn.execute(
f'CREATE TABLE "{_TABLE}" ('
"Expression_DeclarationFileName TEXT, "
"Expression_Name TEXT, "
"Expression_DeclarationLineNumber INTEGER, "
"Dist_Value TEXT)"
)
for index, value in enumerate(values):
conn.execute(
f'INSERT INTO "{_TABLE}" VALUES (?, ?, ?, ?)',
("main.c", f"outputVariables[{index}]", 48, value),
)
conn.commit()


class TestCheckTracedValuesPrinted(unittest.TestCase):
def setUp(self) -> None:
tmp_dir = tempfile.TemporaryDirectory()
self.addCleanup(tmp_dir.cleanup)
self.tmp = Path(tmp_dir.name)

def _db(self, name: str, values: Sequence[str]) -> str:
path = str(self.tmp / name)
_make_tracing_db(path, values)
return path

def _stdout(self, name: str, *ux_strings: str) -> str:
path = self.tmp / name
body = "".join(f" value: 1.5{ux}\n" for ux in ux_strings)
path.write_text(f"seed: 1024\n{body}", encoding="utf-8")
return str(path)

def test_every_recorded_value_printed(self) -> None:
unprinted, recorded = find_unprinted_values(
self._db("t.db", [_UX_A, _UX_B]),
self._stdout("run.out", _UX_A, _UX_B),
_TABLE,
)
self.assertEqual(unprinted, [])
self.assertEqual(recorded, 2)

def test_recorded_value_missing_from_stdout(self) -> None:
unprinted, recorded = find_unprinted_values(
self._db("t.db", [_UX_A, _UX_B]),
self._stdout("run.out", _UX_A),
_TABLE,
)
self.assertEqual(recorded, 2)
self.assertEqual(len(unprinted), 1)
value, identity = unprinted[0]
self.assertEqual(value, _UX_B)
self.assertEqual(identity, ("main.c", "outputVariables[1]", 48))

def test_print_order_and_extras_do_not_matter(self) -> None:
# stdout prints the recorded values in the other order, plus a value
# that was never traced. The check is containment, not equality.
unprinted, recorded = find_unprinted_values(
self._db("t.db", [_UX_A]),
self._stdout("run.out", _UX_B, _UX_A),
_TABLE,
)
self.assertEqual(unprinted, [])
self.assertEqual(recorded, 1)

def test_repeated_recordings_count_once(self) -> None:
# The same value written twice is one distinct value to cross-check.
unprinted, recorded = find_unprinted_values(
self._db("t.db", [_UX_A, _UX_A]),
self._stdout("run.out", _UX_A),
_TABLE,
)
self.assertEqual(unprinted, [])
self.assertEqual(recorded, 1)

def test_main_returns_zero_when_all_printed(self) -> None:
argv = [self._db("t.db", [_UX_A]), self._stdout("run.out", _UX_A)]
self.assertEqual(_run_main(argv), 0)

def test_main_returns_one_when_value_unprinted(self) -> None:
argv = [self._db("t.db", [_UX_A]), self._stdout("run.out", _UX_B)]
self.assertEqual(_run_main(argv), 1)

def test_main_returns_one_when_nothing_recorded(self) -> None:
argv = [self._db("t.db", []), self._stdout("run.out", _UX_A)]
self.assertEqual(_run_main(argv), 1)

def test_main_returns_two_on_missing_db(self) -> None:
missing = str(self.tmp / "does-not-exist.db")
argv = [missing, self._stdout("run.out", _UX_A)]
self.assertEqual(_run_main(argv), 2)


def _run_main(argv: list[str]) -> int:
"""Invoke ``main`` with a patched ``sys.argv`` and return its exit code."""
with patch("sys.argv", ["check_traced_values_printed", *argv]):
return main()


if __name__ == "__main__":
unittest.main()
Loading
Loading