test_actor: add unittests for SimActor

* also implicitly tests for the access of signals during simulation that are
not referenced in any statements

* before, if the busy signal is never used, it is stripped
  and could not be accessed in simulation

diff --git a/migen/test/test_actor.py b/migen/test/test_actor.py
new file mode 100644 (file)
index 0000000..d3856a0
--- /dev/null
+++ b/migen/test/test_actor.py
@@ -0,0 +1,65 @@
+import unittest
+
+from migen.fhdl.std import *
+from migen.flow.actor import *
+from migen.flow.transactions import *
+from migen.flow.network import *
+from migen.actorlib.sim import *
+
+from migen.test.support import SimCase, SimBench
+
+def source_gen(sent):
+       for i in range(10):
+               yield Token("source", {"value": i})
+               sent.append(i)
+
+class SimSource(SimActor):
+       def __init__(self):
+               self.source = Source([("value", 32)])
+               self.sent = []
+               SimActor.__init__(self, source_gen(self.sent))
+
+def sink_gen(received):
+       while True:
+               t = Token("sink")
+               yield t
+               received.append(t.value["value"])
+
+class SimSink(SimActor):
+       def __init__(self):
+               self.sink = Sink([("value", 32)])
+               self.received = []
+               SimActor.__init__(self, sink_gen(self.received))
+
+class SourceSinkCase(SimCase, unittest.TestCase):
+       class TestBench(SimBench):
+               def __init__(self):
+                       self.source = SimSource()
+                       self.sink = SimSink()
+                       g = DataFlowGraph()
+                       g.add_connection(self.source, self.sink)
+                       self.submodules.comp = CompositeActor(g)
+
+               def do_simulation(self, selfp):
+                       if self.source.token_exchanger.done:
+                               raise StopSimulation
+
+       def test_equal(self):
+               self.run_with(lambda tb, tbp: None)
+               self.assertEqual(self.tb.source.sent, self.tb.sink.received)
+
+class SourceSinkDirectCase(SimCase, unittest.TestCase):
+       class TestBench(SimBench):
+               def __init__(self):
+                       self.source = SimSource()
+                       self.sink = SimSink()
+                       self.submodules += self.source, self.sink
+                       self.comb += self.sink.sink.connect(self.source.source)
+
+               def do_simulation(self, selfp):
+                       if self.source.token_exchanger.done:
+                               raise StopSimulation
+
+       def test_equal(self):
+               self.run_with(lambda tb, tbp: None)
+               self.assertEqual(self.tb.source.sent, self.tb.sink.received)