Check for unconnected ports at definition time

magma/circuit.py

@@ -18,7 +18,7 @@
 from .config import get_debug_mode
 from .debug import get_callee_frame_info, debug_info
 from .logging import warning
-from .port import report_wiring_warning
+from .port import report_wiring_warning, report_wiring_error
 from .is_definition import isdefinition
 
 
@@ -476,11 +476,22 @@ def __new__(metacls, name, bases, dct):
             if hasattr(self, 'definition'):
                  pushDefinition(self)
                  self.definition()
+                 self.check_unconnected()
                  self._is_definition = True
                  EndCircuit()
 
         return self
 
+    def check_unconnected(self):
+        for port in self.interface.ports.values():
+            if port.isinput() and not port.driven():
+                report_wiring_error(f"Output port {self.name}.{port.name} not driven", self.debug_info)
+
+        for inst in self.instances:
+            for port in inst.interface.ports.values():
+                if port.isinput() and not port.driven():
+                    report_wiring_error(f"Input port {inst.name}.{port.name} not driven", inst.debug_info)
+
     @property
     def is_definition(self):
         return self._is_definition or self.verilog or self.verilogFile

magma/testing/__init__.py

@@ -1,2 +1,2 @@
-from .utils import check_files_equal
+from .utils import check_files_equal, magma_debug_section
 from .compile import compile_and_regress

magma/testing/utils.py

@@ -5,6 +5,7 @@
 import os
 import sys
 import difflib
+from ..config import get_debug_mode, set_debug_mode
 
 
 def check_files_equal(callee_file, file1_name, file2_name):
@@ -29,3 +30,18 @@ def check_files_equal(callee_file, file1_name, file2_name):
                 for line in diff:
                     sys.stderr.write(line)
     return result
+
+
+class _MagmaDebugSection:
+    def __init__(self):
+        self.__restore = get_debug_mode()
+
+    def __enter__(self):
+        set_debug_mode(True)
+
+    def __exit__(self, typ, value, traceback):
+        set_debug_mode(self.__restore)
+
+
+def magma_debug_section():
+    return _MagmaDebugSection()

tests/test_circuit/test_unconnected.py

@@ -0,0 +1,55 @@
+import pytest
+import magma as m
+from magma.testing import magma_debug_section
+
+
+def _make_unconnected_io():
+    class _Circuit(m.Circuit):
+        IO = ["I", m.In(m.Bits[1]), "O", m.Out(m.Bits[2])]
+
+        @classmethod
+        def definition(io):
+            # Leave io.O[1] unwired.
+            io.O[0] <= io.I[0]
+
+    return _Circuit
+
+
+def _make_unconnected_instance():
+    class _Buffer(m.Circuit):
+        IO = ["I", m.In(m.Bit), "O", m.Out(m.Bit)]
+
+        @classmethod
+        def definition(io):
+            io.O <= io.I
+
+    class _Circuit(m.Circuit):
+        IO = ["I", m.In(m.Bit), "O", m.Out(m.Bit)]
+
+        @classmethod
+        def definition(io):
+            buf = _Buffer()
+            # Leave buf.I unwired.
+            io.O <= buf.O
+
+
+def test_unconnected_io(caplog):
+    with magma_debug_section():
+        Circuit = _make_unconnected_io()
+        logs = caplog.records
+        assert len(logs) == 2
+        for log in logs:
+            assert log.levelname == "ERROR"
+        assert logs[0].msg == "\x1b[1mtests/test_circuit/test_unconnected.py:7: Output port _Circuit.O not driven"  # nopep8
+        assert logs[1].msg == "    class _Circuit(m.Circuit):\n"
+
+
+def test_unconnected_instance(caplog):
+    with magma_debug_section():
+        Circuit = _make_unconnected_instance()
+        logs = caplog.records
+        assert len(logs) == 2
+        for log in logs:
+            assert log.levelname == "ERROR"
+        assert logs[0].msg == "\x1b[1mtests/test_circuit/test_unconnected.py:31: Input port buf.I not driven"  # nopep8
+        assert logs[1].msg == "            buf = _Buffer()\n"