1  from .memory import *
  2  from .common import SimulationBackend
  3  from .conftest import riscv_tests_dir, profile_dir
  4  from test.regression.pysim import PySimulation
  5  import xml.etree.ElementTree as eT
  6  import asyncio
  7  from typing import Literal
  8  import os
  9  import pytest
 10  import subprocess
 11  import json
 12  import tempfile
 13  from filelock import FileLock
 14  
 15  REGRESSION_TESTS_PREFIX = "test.regression."
 16  
 17  
 18  # disable write protection for specific tests with writes to .text section
 19  exclude_write_protection = ["rv32uc-rvc"]
 20  
 21  # force executable bit for memory segments in specific tests
 22  force_executable_memory = ["rv32ui-fence_i"]
 23  
 24  
 25  class MMIO(MemorySegment):
 26      def __init__(self, on_finish: Callable[[], None]):
 27          super().__init__(range(0x80000000, 0x80000000 + 4), SegmentFlags.READ | SegmentFlags.WRITE)
 28          self.on_finish = on_finish
 29          self.failed_test = 0
 30  
 31      def read(self, req: ReadRequest) -> ReadReply:
 32          return ReadReply()
 33  
 34      def write(self, req: WriteRequest) -> WriteReply:
 35          self.failed_test = req.data
 36          self.on_finish()
 37          return WriteReply()
 38  
 39  
 40  async def run_test(sim_backend: SimulationBackend, test_name: str):
 41      mmio = MMIO(lambda: sim_backend.stop())
 42  
 43      mem_segments: list[MemorySegment] = []
 44      mem_segments += load_segments_from_elf(
 45          str(riscv_tests_dir.joinpath("test-" + test_name)),
 46          disable_write_protection=test_name in exclude_write_protection,
 47          force_executable=test_name in force_executable_memory,
 48      )
 49      mem_segments.append(mmio)
 50  
 51      mem_model = CoreMemoryModel(mem_segments)
 52  
 53      result = await sim_backend.run(mem_model, timeout_cycles=5000)
 54  
 55      if result.profile is not None:
 56          os.makedirs(profile_dir, exist_ok=True)
 57          result.profile.encode(f"{profile_dir}/test.regression.{test_name}.json")
 58  
 59      if not result.success:
 60          raise RuntimeError("Simulation timed out")
 61  
 62      if mmio.failed_test:
 63          raise RuntimeError("Failing test: %d" % mmio.failed_test)
 64  
 65  
 66  def regression_body_with_cocotb(test_name: str, traces: bool):
 67      arglist = ["make", "-C", "test/regression/cocotb", "-f", "test.Makefile"]
 68      arglist += [f"TESTCASE={test_name}"]
 69  
 70      verilog_code = os.path.join(os.getcwd(), "core.v")
 71      gen_info_path = f"{verilog_code}.json"
 72      arglist += [f"_COREBLOCKS_GEN_INFO={gen_info_path}"]
 73      arglist += [f"VERILOG_SOURCES={verilog_code}"]
 74      tmp_result_file = tempfile.NamedTemporaryFile("r")
 75      arglist += [f"COCOTB_RESULTS_FILE={tmp_result_file.name}"]
 76  
 77      if traces:
 78          arglist += ["TRACES=1"]
 79  
 80      my_env = dict(os.environ)
 81      my_env["PATH"] = os.path.join(os.getcwd(), "test/regression/cocotb") + ":" + my_env["PATH"]
 82  
 83      res = subprocess.run(arglist, env=my_env)
 84  
 85      assert res.returncode == 0
 86  
 87      tree = eT.parse(tmp_result_file.name)
 88      assert len(list(tree.iter("failure"))) == 0
 89  
 90  
 91  def regression_body_with_pysim(test_name: str, traces: bool):
 92      traces_file = None
 93      if traces:
 94          traces_file = REGRESSION_TESTS_PREFIX + test_name
 95      asyncio.run(run_test(PySimulation(traces_file=traces_file), test_name))
 96  
 97  
 98  @pytest.fixture(scope="session")
 99  def verilate_model(worker_id, request: pytest.FixtureRequest):
100      """
101      Fixture to prevent races on verilating the coreblocks model. It is run only in
102      distributed, cocotb, mode. It executes a 'SKIP' regression test which verilates the model.
103      """
104      if request.session.config.getoption("coreblocks_backend") != "cocotb" or worker_id == "master":
105          # pytest expect yield on every path in fixture
106          yield None
107          return
108  
109      lock_path = "_coreblocks_regression.lock"
110      counter_path = "_coreblocks_regression.counter"
111      with FileLock(lock_path):
112          regression_body_with_cocotb("SKIP", False)
113          if os.path.exists(counter_path):
114              with open(counter_path, "r") as counter_file:
115                  c = json.load(counter_file)
116          else:
117              c = 0
118          with open(counter_path, "w") as counter_file:
119              json.dump(c + 1, counter_file)
120      yield
121      # Session teardown
122      deferred_remove = False
123      with FileLock(lock_path):
124          with open(counter_path, "r") as counter_file:
125              c = json.load(counter_file)
126          if c == 1:
127              deferred_remove = True
128          else:
129              with open(counter_path, "w") as counter_file:
130                  json.dump(c - 1, counter_file)
131      if deferred_remove:
132          os.remove(lock_path)
133          os.remove(counter_path)
134  
135  
136  @pytest.fixture
137  def sim_backend(request: pytest.FixtureRequest):
138      return request.config.getoption("coreblocks_backend")
139  
140  
141  @pytest.fixture
142  def traces_enabled(request: pytest.FixtureRequest):
143      return request.config.getoption("coreblocks_traces")
144  
145  
146  def test_entrypoint(test_name: str, sim_backend: Literal["pysim", "cocotb"], traces_enabled: bool, verilate_model):
147      if sim_backend == "cocotb":
148          regression_body_with_cocotb(test_name, traces_enabled)
149      elif sim_backend == "pysim":
150          regression_body_with_pysim(test_name, traces_enabled)