7d581b1a-e6de-4b5b-ba5f-3d20fbd171e5

Committed 11 Jun 2025 10:43AM UTC coverage: 93.297% (-0.1%) from 93.401%

Build # 7d581b1a-e6de-4b5b-ba5f-3d20fbd171e5

Build Type

push

circleci

Committed by

Nic30

Commit Message

test: update test ref file

Run Details

5386 of 6460 branches covered (83.37%)

39902 of 42769 relevant lines covered (93.3%)

0.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.91

/hwtLib/examples/axi/debugbusmonitor_test.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from collections import deque
from io import StringIO
from math import ceil
import os
import threading
import unittest

from hwt.simulator.simTestCase import SimTestCase
from hwtLib.examples.axi.debugbusmonitor import DebugBusMonitorExampleAxi
from hwtLib.tools.debug_bus_monitor_ctl import DebugBusMonitorCtl, words_to_int
from hwtSimApi.constants import CLK_PERIOD
from hwtSimApi.triggers import Timer, StopSimumulation, WaitWriteOnly
from pyMathBitPrecise.bit_utils import ValidityError


class DebugBusMonitorCtlSim(DebugBusMonitorCtl):

    def __init__(self, tc):
        DebugBusMonitorCtl.__init__(self, 0)
        self.tc = tc

    def read(self, addr, size):
        axi = self.tc.dut.s
        word_size = axi.DATA_WIDTH // 8
        words = []
        for _ in range(ceil(size / word_size)):
            assert not self.tc.sim_done
            ar_req = axi.ar._ag.create_addr_req(addr)
            axi.ar._ag.data.append(ar_req)

            r_data = axi.r._ag.data
            while not r_data:
                assert not self.tc.sim_done
                self.tc.r_data_available.acquire()

            d = r_data.popleft()[0]
            try:
                d = int(d)
            except ValidityError:
                d = d.val & d.vld_mask

            words.append(d)
            addr += word_size

        return words_to_int(words, word_size, size).to_bytes(size, "little")


def run_DebugBusMonitorCtlSim(tc, out_txt, out_dot):
    db = DebugBusMonitorCtlSim(tc)
    db.dump_txt(out_txt)
    db.dump_dot(out_dot)
    tc.sim_done = True


class DebugBusMonitorExampleAxiTC(SimTestCase):

    @classmethod
    def setUpClass(cls):
        dut = cls.dut = DebugBusMonitorExampleAxi()
        dut.DATA_WIDTH = 32
        cls.compileSim(dut)

    def setUp(self):
        SimTestCase.setUp(self)
        self.sim_done = False
        self.r_data_available = threading.Lock()
        self.r_data_available.acquire()

    def test_dump(self):
        dut = self.dut
        tc = self

        class SpyDeque(deque):

            def __init__(self, tc):
                super(SpyDeque, self).__init__()
                self.tc = tc

            def append(self, x):
                if self.tc.r_data_available.locked():
                    self.tc.r_data_available.release()
                super(SpyDeque, self).append(x)

        dut.s.r._ag.data = SpyDeque(self)
        dut.din0._ag.data.extend([1, 2])
        dut.din1._ag.data.extend([3, 4])
        dut.din2._ag.data.extend([5, 6])

        def sim_init():
            yield WaitWriteOnly()
            dut.dout1._ag.setEnable(False)

        def time_sync():
            while True:
                if dut.s.r._ag.data and self.r_data_available.locked():
                    tc.r_data_available.release()
                yield Timer(CLK_PERIOD)
                if self.sim_done:
                    raise StopSimumulation()

        self.procs.extend([time_sync(), sim_init()])

        buff_txt = StringIO()
        buff_dot = StringIO()
        ctl_thread = threading.Thread(target=run_DebugBusMonitorCtlSim,
                                      args=(self, buff_txt, buff_dot))
        ctl_thread.start()
        # actually takes less time as the simulation is stopped after ctl_thread end
        self.runSim(12000 * CLK_PERIOD)
        # handle the case where something went wrong and ctl thread is still running
        self.sim_done = True
        if self.r_data_available.locked():
            self.r_data_available.release()
        ctl_thread.join()

        d = buff_txt.getvalue()
        self.assertEqual(d, """\
din0:
  data: 0x0
  vld: 0
  rd: 1
din0_snapshot:
  data: 0x2
  vld: 1
  rd: 1
dout0:
  data: 0x0
  vld: 0
  rd: 1
dout0_snapshot:
  data: 0x2
  vld: 1
  rd: 1
din1:
  data: 0x4
  vld: 1
  rd: 0
din1_snapshot:
  data: 0x3
  vld: 1
  rd: 1
reg:
  dataIn:
    data: 0x4
    vld: 1
    rd: 0
  dataIn_snapshot:
    data: 0x4
    vld: 1
    rd: 0
  dataOut:
    data: 0x3
    vld: 1
    rd: 0
  dataOut_snapshot:
    data: 0x0
    vld: 0
    rd: 0
dout1:
  data: 0x3
  vld: 1
  rd: 0
dout1_snapshot:
  data: 0x0
  vld: 0
  rd: 0
din2:
  data: 0x0
  vld: 0
  rd: 1
din2_snapshot:
  data: 0x0
  vld: 0
  rd: 1
dout2:
  data: 0x0
  vld: 0
  rd: 1
dout2_snapshot:
  data: 0x0
  vld: 0
  rd: 1
""")
        dot_file = os.path.join(os.path.dirname(__file__), "DebugBusMonitorExampleAxiTC.dot")
        #with open(dot_file, "w") as f:
        #    f.write(buff_dot.getvalue())
        with open(dot_file, "r") as f:
            self.assertEqual(buff_dot.getvalue(), f.read())

if __name__ == "__main__":
    testLoader = unittest.TestLoader()
    # suite = unittest.TestSuite([DebugBusMonitorExampleAxiTC("test_write")])
    suite = testLoader.loadTestsFromTestCase(DebugBusMonitorExampleAxiTC)
    runner = unittest.TextTestRunner(verbosity=3)
    runner.run(suite)

1	#!/usr/bin/env python3
2	# -- coding: utf-8 --
3
4	from collections import deque	1✔
5	from io import StringIO	1✔
6	from math import ceil	1✔
7	import os	1✔
8	import threading	1✔
9	import unittest	1✔
10
11	from hwt.simulator.simTestCase import SimTestCase	1✔
12	from hwtLib.examples.axi.debugbusmonitor import DebugBusMonitorExampleAxi	1✔
13	from hwtLib.tools.debug_bus_monitor_ctl import DebugBusMonitorCtl, words_to_int	1✔
14	from hwtSimApi.constants import CLK_PERIOD	1✔
15	from hwtSimApi.triggers import Timer, StopSimumulation, WaitWriteOnly	1✔
16	from pyMathBitPrecise.bit_utils import ValidityError	1✔
17
18
19	class DebugBusMonitorCtlSim(DebugBusMonitorCtl):	1✔
20
21	def __init__(self, tc):	1✔
22	DebugBusMonitorCtl.__init__(self, 0)	1✔
23	self.tc = tc	1✔
24
25	def read(self, addr, size):	1✔
26	axi = self.tc.dut.s	1✔
27	word_size = axi.DATA_WIDTH // 8	1✔
28	words = []	1✔
29	for _ in range(ceil(size / word_size)):	1✔
30	assert not self.tc.sim_done	1✔
31	ar_req = axi.ar._ag.create_addr_req(addr)	1✔
32	axi.ar._ag.data.append(ar_req)	1✔
33
34	r_data = axi.r._ag.data	1✔
35	while not r_data:	1✔
36	assert not self.tc.sim_done	1✔
37	self.tc.r_data_available.acquire()	1✔
38
39	d = r_data.popleft()[0]	1✔
40	try:	1✔
41	d = int(d)	1✔
42	except ValidityError:	1✔
43	d = d.val & d.vld_mask	1✔
44
45	words.append(d)	1✔
46	addr += word_size	1✔
47
48	return words_to_int(words, word_size, size).to_bytes(size, "little")	1✔
49
50
51	def run_DebugBusMonitorCtlSim(tc, out_txt, out_dot):	1✔
52	db = DebugBusMonitorCtlSim(tc)	1✔
53	db.dump_txt(out_txt)	1✔
54	db.dump_dot(out_dot)	1✔
55	tc.sim_done = True	1✔
56
57
58	class DebugBusMonitorExampleAxiTC(SimTestCase):	1✔
59
60	@classmethod	1✔
61	def setUpClass(cls):	1✔
62	dut = cls.dut = DebugBusMonitorExampleAxi()	1✔
63	dut.DATA_WIDTH = 32	1✔
64	cls.compileSim(dut)	1✔
65
66	def setUp(self):	1✔
67	SimTestCase.setUp(self)	1✔
68	self.sim_done = False	1✔
69	self.r_data_available = threading.Lock()	1✔
70	self.r_data_available.acquire()	1✔
71
72	def test_dump(self):	1✔
73	dut = self.dut	1✔
74	tc = self	1✔
75
76	class SpyDeque(deque):	1✔
77
78	def __init__(self, tc):	1✔
79	super(SpyDeque, self).__init__()	1✔
80	self.tc = tc	1✔
81
82	def append(self, x):	1✔
83	if self.tc.r_data_available.locked():	1!
84	self.tc.r_data_available.release()	1✔
85	super(SpyDeque, self).append(x)	1✔
86
87	dut.s.r._ag.data = SpyDeque(self)	1✔
88	dut.din0._ag.data.extend([1, 2])	1✔
89	dut.din1._ag.data.extend([3, 4])	1✔
90	dut.din2._ag.data.extend([5, 6])	1✔
91
92	def sim_init():	1✔
93	yield WaitWriteOnly()	1✔
94	dut.dout1._ag.setEnable(False)	1✔
95
96	def time_sync():	1✔
97	while True:	1✔
98	if dut.s.r._ag.data and self.r_data_available.locked():	1!
99	tc.r_data_available.release()	×
100	yield Timer(CLK_PERIOD)	1✔
101	if self.sim_done:	1✔
102	raise StopSimumulation()	1✔
103
104	self.procs.extend([time_sync(), sim_init()])	1✔
105
106	buff_txt = StringIO()	1✔
107	buff_dot = StringIO()	1✔
108	ctl_thread = threading.Thread(target=run_DebugBusMonitorCtlSim,	1✔
109	args=(self, buff_txt, buff_dot))
110	ctl_thread.start()	1✔
111	# actually takes less time as the simulation is stopped after ctl_thread end
112	self.runSim(12000 * CLK_PERIOD)	1✔
113	# handle the case where something went wrong and ctl thread is still running
114	self.sim_done = True	1✔
115	if self.r_data_available.locked():	1!
116	self.r_data_available.release()	1✔
117	ctl_thread.join()	1✔
118
119	d = buff_txt.getvalue()	1✔
120	self.assertEqual(d, """\	1✔
121	din0:
122	data: 0x0
123	vld: 0
124	rd: 1
125	din0_snapshot:
126	data: 0x2
127	vld: 1
128	rd: 1
129	dout0:
130	data: 0x0
131	vld: 0
132	rd: 1
133	dout0_snapshot:
134	data: 0x2
135	vld: 1
136	rd: 1
137	din1:
138	data: 0x4
139	vld: 1
140	rd: 0
141	din1_snapshot:
142	data: 0x3
143	vld: 1
144	rd: 1
145	reg:
146	dataIn:
147	data: 0x4
148	vld: 1
149	rd: 0
150	dataIn_snapshot:
151	data: 0x4
152	vld: 1
153	rd: 0
154	dataOut:
155	data: 0x3
156	vld: 1
157	rd: 0
158	dataOut_snapshot:
159	data: 0x0
160	vld: 0
161	rd: 0
162	dout1:
163	data: 0x3
164	vld: 1
165	rd: 0
166	dout1_snapshot:
167	data: 0x0
168	vld: 0
169	rd: 0
170	din2:
171	data: 0x0
172	vld: 0
173	rd: 1
174	din2_snapshot:
175	data: 0x0
176	vld: 0
177	rd: 1
178	dout2:
179	data: 0x0
180	vld: 0
181	rd: 1
182	dout2_snapshot:
183	data: 0x0
184	vld: 0
185	rd: 1
186	""")
187	dot_file = os.path.join(os.path.dirname(__file__), "DebugBusMonitorExampleAxiTC.dot")	1✔
188	#with open(dot_file, "w") as f:
189	# f.write(buff_dot.getvalue())
190	with open(dot_file, "r") as f:	1✔
191	self.assertEqual(buff_dot.getvalue(), f.read())	1✔
192
193	if __name__ == "__main__":
194	testLoader = unittest.TestLoader()
195	# suite = unittest.TestSuite([DebugBusMonitorExampleAxiTC("test_write")])
196	suite = testLoader.loadTestsFromTestCase(DebugBusMonitorExampleAxiTC)
197	runner = unittest.TextTestRunner(verbosity=3)
198	runner.run(suite)

Nic30 / hwtLib / 7d581b1a-e6de-4b5b-ba5f-3d20fbd171e5

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous