bslathi19/PeakRDL-BusDecoder
1
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

add colorized build/sim log propgate on error to all runners (#26)

Browse Source 
* add colorized build/sim log propgate on error to all runners

* add doctoring

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Arnav Sacheti
2025-11-26 08:01:41 -08:00
committed by GitHub
parent f0f25a6d92
commit 88827c65b5
10 changed files with 311 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
tests/cocotb/apb3/smoke/test_register_access.py
View File

@@ -11,6 +11,7 @@ from cocotb.triggers import Timer
from tests.cocotb_lib.handle_utils import SignalHandle, resolve_handle
class _Apb3SlaveShim:
"""Accessor for the APB3 slave signals on the DUT."""
@@ -128,16 +129,20 @@ async def test_apb3_address_decoding(dut) -> None:
assert _get_int(entry["outputs"]["PSEL"], index) == 1, f"{master_name} should assert PSEL for write"
assert _get_int(entry["outputs"]["PWRITE"], index) == 1, f"{master_name} should see write direction"
assert _get_int(entry["outputs"]["PADDR"], index) == address, f"{master_name} must receive write address"
assert _get_int(entry["outputs"]["PWDATA"], index) == write_data, f"{master_name} must receive write data"
assert _get_int(entry["outputs"]["PADDR"], index) == address, (
f"{master_name} must receive write address"
)
assert _get_int(entry["outputs"]["PWDATA"], index) == write_data, (
f"{master_name} must receive write data"
)
for other_name, other_idx in _all_index_pairs(masters):
if other_name == master_name and other_idx == index:
continue
other_entry = masters[other_name]
assert (
_get_int(other_entry["outputs"]["PSEL"], other_idx) == 0
), f"{other_name}{other_idx} should remain idle during {txn['label']}"
assert _get_int(other_entry["outputs"]["PSEL"], other_idx) == 0, (
f"{other_name}{other_idx} should remain idle during {txn['label']}"
)
assert int(slave.PREADY.value) == 1, "Slave ready should mirror selected master"
assert int(slave.PSLVERR.value) == 0, "Write should complete without error"
@@ -164,16 +169,20 @@ async def test_apb3_address_decoding(dut) -> None:
await Timer(1, units="ns")
assert _get_int(entry["outputs"]["PSEL"], index) == 1, f"{master_name} must assert PSEL for read"
assert _get_int(entry["outputs"]["PWRITE"], index) == 0, f"{master_name} should clear write during read"
assert _get_int(entry["outputs"]["PADDR"], index) == address, f"{master_name} must receive read address"
assert _get_int(entry["outputs"]["PWRITE"], index) == 0, (
f"{master_name} should clear write during read"
)
assert _get_int(entry["outputs"]["PADDR"], index) == address, (
f"{master_name} must receive read address"
)
for other_name, other_idx in _all_index_pairs(masters):
if other_name == master_name and other_idx == index:
continue
other_entry = masters[other_name]
assert (
_get_int(other_entry["outputs"]["PSEL"], other_idx) == 0
), f"{other_name}{other_idx} must stay idle during read of {txn['label']}"
assert _get_int(other_entry["outputs"]["PSEL"], other_idx) == 0, (
f"{other_name}{other_idx} must stay idle during read of {txn['label']}"
)
assert int(slave.PRDATA.value) == read_data, "Read data should propagate back to the slave"
assert int(slave.PREADY.value) == 1, "Slave ready should acknowledge the read"

53
tests/cocotb/apb3/smoke/test_runner.py
View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import json
from pathlib import Path
import logging
import pytest
@@ -15,7 +16,7 @@ except ImportError: # pragma: no cover
from cocotb_tools.runner import get_runner
from tests.cocotb_lib import RDL_CASES
from tests.cocotb_lib.utils import get_verilog_sources, prepare_cpuif_case
from tests.cocotb_lib.utils import get_verilog_sources, prepare_cpuif_case, colorize_cocotb_log
@pytest.mark.simulation
@@ -44,16 +45,42 @@ def test_apb3_smoke(tmp_path: Path, rdl_file: str, top_name: str) -> None:
runner = get_runner("verilator")
sim_build = build_root / "sim_build"
runner.build(
sources=sources,
hdl_toplevel=module_path.stem,
build_dir=sim_build,
)
build_log_file = build_root / "build.log"
sim_log_file = build_root / "simulation.log"
runner.test(
hdl_toplevel=module_path.stem,
test_module="tests.cocotb.apb3.smoke.test_register_access",
build_dir=sim_build,
log_file=str(build_root / "simulation.log"),
extra_env={"RDL_TEST_CONFIG": json.dumps(config)},
)
try:
runner.build(
sources=sources,
hdl_toplevel=module_path.stem,
build_dir=sim_build,
log_file=str(build_log_file),
)
except SystemExit as e:
# Print build log on failure for easier debugging
if build_log_file.exists():
logging.error(f"""
=== Build Log ===
{colorize_cocotb_log(build_log_file.read_text())}
=== End Build Log ===
""")
if e.code != 0:
raise
try:
runner.test(
hdl_toplevel=module_path.stem,
test_module="tests.cocotb.apb3.smoke.test_register_access",
build_dir=sim_build,
log_file=str(sim_log_file),
extra_env={"RDL_TEST_CONFIG": json.dumps(config)},
)
except SystemExit as e:
# Print simulation log on failure for easier debugging
if sim_log_file.exists():
logging.error(f"""
=== Simulation Log ===
{colorize_cocotb_log(sim_log_file.read_text())}
=== End Simulation Log ===
""")
if e.code != 0:
raise

33
tests/cocotb/apb4/smoke/test_register_access.py
View File

@@ -11,6 +11,7 @@ from cocotb.triggers import Timer
from tests.cocotb_lib.handle_utils import SignalHandle, resolve_handle
class _Apb4SlaveShim:
"""Lightweight accessor for the APB4 slave side of the DUT."""
@@ -141,17 +142,23 @@ async def test_apb4_address_decoding(dut) -> None:
assert _get_int(entry["outputs"]["PSEL"], index) == 1, f"{master_name} should assert PSEL for write"
assert _get_int(entry["outputs"]["PWRITE"], index) == 1, f"{master_name} should see write intent"
assert _get_int(entry["outputs"]["PADDR"], index) == address, f"{master_name} must receive write address"
assert _get_int(entry["outputs"]["PWDATA"], index) == write_data, f"{master_name} must receive write data"
assert _get_int(entry["outputs"]["PSTRB"], index) == strobe_mask, f"{master_name} must receive full strobes"
assert _get_int(entry["outputs"]["PADDR"], index) == address, (
f"{master_name} must receive write address"
)
assert _get_int(entry["outputs"]["PWDATA"], index) == write_data, (
f"{master_name} must receive write data"
)
assert _get_int(entry["outputs"]["PSTRB"], index) == strobe_mask, (
f"{master_name} must receive full strobes"
)
for other_name, other_idx in _all_index_pairs(masters):
if other_name == master_name and other_idx == index:
continue
other_entry = masters[other_name]
assert (
_get_int(other_entry["outputs"]["PSEL"], other_idx) == 0
), f"{other_name}{other_idx} should remain idle during {txn['label']}"
assert _get_int(other_entry["outputs"]["PSEL"], other_idx) == 0, (
f"{other_name}{other_idx} should remain idle during {txn['label']}"
)
assert int(slave.PREADY.value) == 1, "Slave ready should reflect selected master"
assert int(slave.PSLVERR.value) == 0, "No error expected during write"
@@ -179,16 +186,20 @@ async def test_apb4_address_decoding(dut) -> None:
await Timer(1, units="ns")
assert _get_int(entry["outputs"]["PSEL"], index) == 1, f"{master_name} must assert PSEL for read"
assert _get_int(entry["outputs"]["PWRITE"], index) == 0, f"{master_name} should deassert write for reads"
assert _get_int(entry["outputs"]["PADDR"], index) == address, f"{master_name} must receive read address"
assert _get_int(entry["outputs"]["PWRITE"], index) == 0, (
f"{master_name} should deassert write for reads"
)
assert _get_int(entry["outputs"]["PADDR"], index) == address, (
f"{master_name} must receive read address"
)
for other_name, other_idx in _all_index_pairs(masters):
if other_name == master_name and other_idx == index:
continue
other_entry = masters[other_name]
assert (
_get_int(other_entry["outputs"]["PSEL"], other_idx) == 0
), f"{other_name}{other_idx} must stay idle during read of {txn['label']}"
assert _get_int(other_entry["outputs"]["PSEL"], other_idx) == 0, (
f"{other_name}{other_idx} must stay idle during read of {txn['label']}"
)
assert int(slave.PRDATA.value) == read_data, "Slave should observe readback data from master"
assert int(slave.PREADY.value) == 1, "Slave ready should follow responding master"

32
tests/cocotb/apb4/smoke/test_runner.py
View File

@@ -5,6 +5,7 @@ from __future__ import annotations
import json
from pathlib import Path
import logging
import pytest
from peakrdl_busdecoder.cpuif.apb4.apb4_cpuif_flat import APB4CpuifFlat
@@ -15,7 +16,7 @@ except ImportError: # pragma: no cover
from cocotb_tools.runner import get_runner
from tests.cocotb_lib import RDL_CASES
from tests.cocotb_lib.utils import get_verilog_sources, prepare_cpuif_case
from tests.cocotb_lib.utils import get_verilog_sources, prepare_cpuif_case, colorize_cocotb_log
@pytest.mark.simulation
@@ -44,20 +45,24 @@ def test_apb4_smoke(tmp_path: Path, rdl_file: str, top_name: str) -> None:
runner = get_runner("verilator")
sim_build = build_root / "sim_build"
build_log_file = build_root / "build.log"
sim_log_file = build_root / "simulation.log"
try:
runner.build(
sources=sources,
hdl_toplevel=module_path.stem,
build_dir=sim_build,
log_file=str(build_root / "build.log"),
log_file=str(build_log_file),
)
except SystemExit as e:
# Print build log on failure for easier debugging
log_path = build_root / "build.log"
if log_path.exists():
logging.error("\n\n=== Build Log ===\n")
logging.error(log_path.read_text())
logging.error("\n=== End Build Log ===\n")
if build_log_file.exists():
logging.error(f"""
=== Build Log ===
{colorize_cocotb_log(build_log_file.read_text())}
=== End Build Log ===
""")
if e.code != 0:
raise
@@ -66,15 +71,16 @@ def test_apb4_smoke(tmp_path: Path, rdl_file: str, top_name: str) -> None:
hdl_toplevel=module_path.stem,
test_module="tests.cocotb.apb4.smoke.test_register_access",
build_dir=sim_build,
log_file=str(build_root / "simulation.log"),
log_file=str(sim_log_file),
extra_env={"RDL_TEST_CONFIG": json.dumps(config)},
)
except SystemExit as e:
# Print simulation log on failure for easier debugging
log_path = build_root / "simulation.log"
if log_path.exists():
logging.error("\n\n=== Simulation Log ===\n")
logging.error(log_path.read_text())
logging.error("\n=== End Simulation Log ===\n")
if sim_log_file.exists():
logging.error(f"""
=== Simulation Log ===
{colorize_cocotb_log(sim_log_file.read_text())}
=== End Simulation Log ===
""")
if e.code != 0:
raise

19
tests/cocotb/axi4lite/smoke/test_register_access.py
View File

@@ -11,6 +11,7 @@ from cocotb.triggers import Timer
from tests.cocotb_lib.handle_utils import SignalHandle, resolve_handle
class _AxilSlaveShim:
"""Accessor for AXI4-Lite slave ports on the DUT."""
@@ -167,12 +168,12 @@ async def test_axi4lite_address_decoding(dut) -> None:
if other_name == master_name and other_idx == index:
continue
other_entry = masters[other_name]
assert (
_get_int(other_entry["outputs"]["AWVALID"], other_idx) == 0
), f"{other_name}{other_idx} AWVALID should remain low during {txn['label']}"
assert (
_get_int(other_entry["outputs"]["WVALID"], other_idx) == 0
), f"{other_name}{other_idx} WVALID should remain low during {txn['label']}"
assert _get_int(other_entry["outputs"]["AWVALID"], other_idx) == 0, (
f"{other_name}{other_idx} AWVALID should remain low during {txn['label']}"
)
assert _get_int(other_entry["outputs"]["WVALID"], other_idx) == 0, (
f"{other_name}{other_idx} WVALID should remain low during {txn['label']}"
)
slave.AWVALID.value = 0
slave.WVALID.value = 0
@@ -198,9 +199,9 @@ async def test_axi4lite_address_decoding(dut) -> None:
if other_name == master_name and other_idx == index:
continue
other_entry = masters[other_name]
assert (
_get_int(other_entry["outputs"]["ARVALID"], other_idx) == 0
), f"{other_name}{other_idx} ARVALID should remain low during read of {txn['label']}"
assert _get_int(other_entry["outputs"]["ARVALID"], other_idx) == 0, (
f"{other_name}{other_idx} ARVALID should remain low during read of {txn['label']}"
)
assert int(slave.RVALID.value) == 1, "Slave should observe RVALID when master responds"
assert int(slave.RDATA.value) == read_data, "Read data must fold back to slave"

53
tests/cocotb/axi4lite/smoke/test_runner.py
View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import json
from pathlib import Path
import logging
import pytest
@@ -15,7 +16,7 @@ except ImportError: # pragma: no cover
from cocotb_tools.runner import get_runner
from tests.cocotb_lib import RDL_CASES
from tests.cocotb_lib.utils import get_verilog_sources, prepare_cpuif_case
from tests.cocotb_lib.utils import get_verilog_sources, prepare_cpuif_case, colorize_cocotb_log
@pytest.mark.simulation
@@ -44,16 +45,42 @@ def test_axi4lite_smoke(tmp_path: Path, rdl_file: str, top_name: str) -> None:
runner = get_runner("verilator")
sim_build = build_root / "sim_build"
runner.build(
sources=sources,
hdl_toplevel=module_path.stem,
build_dir=sim_build,
)
build_log_file = build_root / "build.log"
sim_log_file = build_root / "simulation.log"
runner.test(
hdl_toplevel=module_path.stem,
test_module="tests.cocotb.axi4lite.smoke.test_register_access",
build_dir=sim_build,
log_file=str(build_root / "simulation.log"),
extra_env={"RDL_TEST_CONFIG": json.dumps(config)},
)
try:
runner.build(
sources=sources,
hdl_toplevel=module_path.stem,
build_dir=sim_build,
log_file=str(build_log_file),
)
except SystemExit as e:
# Print build log on failure for easier debugging
if build_log_file.exists():
logging.error(f"""
=== Build Log ===
{colorize_cocotb_log(build_log_file.read_text())}
=== End Build Log ===
""")
if e.code != 0:
raise
try:
runner.test(
hdl_toplevel=module_path.stem,
test_module="tests.cocotb.axi4lite.smoke.test_register_access",
build_dir=sim_build,
log_file=str(sim_log_file),
extra_env={"RDL_TEST_CONFIG": json.dumps(config)},
)
except SystemExit as e:
# Print simulation log on failure for easier debugging
if sim_log_file.exists():
logging.error(f"""
=== Simulation Log ===
{colorize_cocotb_log(sim_log_file.read_text())}
=== End Simulation Log ===
""")
if e.code != 0:
raise

59
tests/cocotb_lib/utils.py
View File

@@ -4,6 +4,7 @@ from __future__ import annotations
from collections import defaultdict
from pathlib import Path
import re
from typing import Any
from systemrdl import RDLCompiler
@@ -12,6 +13,64 @@ from systemrdl.node import AddressableNode, AddrmapNode, RegNode
from peakrdl_busdecoder.cpuif.base_cpuif import BaseCpuif
from peakrdl_busdecoder.exporter import BusDecoderExporter
RESET = "\x1b[0m"
DIM = "\x1b[2m"
LEVEL_COLORS = {
"DEBUG": "\x1b[35m", # magenta
"INFO": "\x1b[36m", # cyan
"WARNING": "\x1b[33m", # yellow
"ERROR": "\x1b[31m", # red
"CRITICAL": "\x1b[1;31m", # bold red
}
# Matches lines like:
# " 0.00ns INFO cocotb ..." or "-.--ns INFO gpi ..."
LINE_RE = re.compile(
r"^(?P<prefix>\s*)" # leading spaces
r"(?P<time>[-0-9\.]+[a-zA-Z]+)" # timestamp (e.g. 0.00ns, -.--ns)
r"\s+"
r"(?P<level>[A-Z]+)" # log level
r"(?P<rest>.*)$" # the rest of the line
)
def colorize_cocotb_log(text: str) -> str:
"""
Colorizes cocotb log lines for improved readability in terminal output.
Each log line is parsed to identify the timestamp and log level, which are then
colorized using ANSI escape codes. The timestamp is dimmed, and the log level
is colored according to its severity (e.g., INFO, WARNING, ERROR).
Args:
text: The input string containing cocotb log lines.
Returns:
A string with colorized log lines.
"""
def _color_line(match: re.Match) -> str:
prefix = match.group("prefix")
time = match.group("time")
level = match.group("level")
rest = match.group("rest")
level_color = LEVEL_COLORS.get(level, "")
# dim timestamp, color level
time_colored = f"{DIM}{time}{RESET}"
level_colored = f"{level_color}{level}{RESET}" if level_color else level
return f"{prefix}{time_colored} {level_colored}{rest}"
lines = []
for line in text.splitlines():
m = LINE_RE.match(line)
if m:
lines.append(_color_line(m))
else:
lines.append(line)
return "\n".join(lines)
def compile_rdl_and_export(
rdl_source: str, top_name: str, output_dir: Path, cpuif_cls: type[BaseCpuif], **kwargs: Any