Merge branch '1-port-interface' into 'develop'

Resolve #1 "Port Interface", #8 "Port Coupling"

See merge request PUM_TDDD96/B-ASIC!9

b_asic/abstract_graph_component.py

@@ -5,6 +5,7 @@ TODO: More info.
 
 from b_asic.graph_component import GraphComponent, Name
 
+
 class AbstractGraphComponent(GraphComponent):
     """Abstract Graph Component class which is a component of a signal flow graph.
 

b_asic/abstract_operation.py

@@ -4,7 +4,7 @@ TODO: More info.
 """
 
 from abc import abstractmethod
-from typing import List, Set, Dict, Optional, Any
+from typing import List, Dict, Optional, Any
 from numbers import Number
 
 from b_asic.port import InputPort, OutputPort
@@ -13,6 +13,8 @@ from b_asic.operation import Operation
 from b_asic.simulation import SimulationState, OperationState
 from b_asic.utilities import breadth_first_search
 from b_asic.abstract_graph_component import AbstractGraphComponent
+from b_asic.graph_component import Name
+
 
 class AbstractOperation(Operation, AbstractGraphComponent):
     """Generic abstract operation class which most implementations will derive from.
@@ -23,15 +25,16 @@ class AbstractOperation(Operation, AbstractGraphComponent):
     _output_ports: List[OutputPort]
     _parameters: Dict[str, Optional[Any]]
 
-    def __init__(self, **kwds):
-        super().__init__(**kwds)
+    def __init__(self, name: Name = ""):
+        super().__init__(name)
         self._input_ports = []
         self._output_ports = []
         self._parameters = {}
 
     @abstractmethod
     def evaluate(self, inputs: list) -> list:
-        """Evaluate the operation and generate a list of output values given a list of input values."""
+        """Evaluate the operation and generate a list of output values given a 
+        list of input values."""
         raise NotImplementedError
 
     def inputs(self) -> List[InputPort]:

b_asic/core_operations.py

@@ -29,10 +29,10 @@ class Constant(AbstractOperation):
     TODO: More info.
     """
 
-    def __init__(self, value: Number, **kwds):
-        """Construct a Constant."""
-        super().__init__(**kwds)
-        self._output_ports = [OutputPort(1, self)] # TODO: Generate appropriate ID for ports.
+    def __init__(self, value: Number = 0, name: Name = ""):
+        super().__init__(name)
+
+        self._output_ports = [OutputPort(0, self)] # TODO: Generate appropriate ID for ports.
         self._parameters["value"] = value
 
     def evaluate(self, inputs: list) -> list:
@@ -48,11 +48,16 @@ class Addition(AbstractOperation):
     TODO: More info.
     """
 
-    def __init__(self, **kwds):
-        """Construct an Addition."""
-        super().__init__(**kwds)
-        self._input_ports = [InputPort(1, self), InputPort(1, self)] # TODO: Generate appropriate ID for ports.
-        self._output_ports = [OutputPort(1, self)] # TODO: Generate appropriate ID for ports.
+    def __init__(self, source1: OutputPort = None, source2: OutputPort = None, name: Name = ""):
+        super().__init__(name)
+
+        self._input_ports = [InputPort(0, self), InputPort(1, self)] # TODO: Generate appropriate ID for ports.
+        self._output_ports = [OutputPort(0, self)] # TODO: Generate appropriate ID for ports.
+
+        if source1 is not None:
+            self._input_ports[0].connect_to_port(source1)
+        if source2 is not None:
+            self._input_ports[1].connect_to_port(source2)
 
     def evaluate(self, inputs: list) -> list:
         return [inputs[0] + inputs[1]]
@@ -67,13 +72,15 @@ class ConstantMultiplication(AbstractOperation):
     TODO: More info.
     """
 
-    def __init__(self, coefficient: Number, **kwds):
-        """Construct a ConstantMultiplication."""
-        super().__init__(**kwds)
-        self._input_ports = [InputPort(1), self] # TODO: Generate appropriate ID for ports.
-        self._output_ports = [OutputPort(1, self)] # TODO: Generate appropriate ID for ports.
+    def __init__(self, coefficient: Number, source1: OutputPort = None, name: Name = ""):
+        super().__init__(name)
+        self._input_ports = [InputPort(0, self)] # TODO: Generate appropriate ID for ports.
+        self._output_ports = [OutputPort(0, self)] # TODO: Generate appropriate ID for ports.
         self._parameters["coefficient"] = coefficient
 
+        if source1 is not None:
+            self._input_ports[0].connect_to_port(source1)
+
     def evaluate(self, inputs: list) -> list:
         return [inputs[0] * self.param("coefficient")]
 

b_asic/graph_component.py

@@ -18,17 +18,17 @@ class GraphComponent(ABC):
     @property
     @abstractmethod
     def type_name(self) -> TypeName:
-        """Returns the type name of the graph component"""
+        """Return the type name of the graph component"""
         raise NotImplementedError
 
     @property
     @abstractmethod
     def name(self) -> Name:
-        """Returns the name of the graph component."""
+        """Return the name of the graph component."""
         raise NotImplementedError
 
     @name.setter
     @abstractmethod
     def name(self, name: Name) -> None:
-        """Sets the name of the graph component to the entered name."""
+        """Set the name of the graph component to the entered name."""
         raise NotImplementedError

b_asic/graph_id.py

@@ -20,7 +20,7 @@ class GraphIDGenerator:
         self._next_id_number = defaultdict(lambda: 1)       # Initalises every key element to 1
 
     def get_next_id(self, graph_id_type: GraphIDType) -> GraphID:
-        """Returns the next graph id for a certain graph id type."""
+        """Return the next graph id for a certain graph id type."""
         graph_id = graph_id_type + str(self._next_id_number[graph_id_type])
         self._next_id_number[graph_id_type] += 1            # Increase the current id number
         return graph_id

b_asic/operation.py

@@ -43,13 +43,11 @@ class Operation(GraphComponent):
         """Get the input port at index i."""
         raise NotImplementedError
 
-
     @abstractmethod
     def output(self, i: int) -> "OutputPort":
         """Get the output port at index i."""
         raise NotImplementedError
 
-
     @abstractmethod
     def params(self) -> Dict[str, Optional[Any]]:
         """Get a dictionary of all parameter values."""

b_asic/port.py

@@ -6,15 +6,16 @@ TODO: More info.
 from abc import ABC, abstractmethod
 from typing import NewType, Optional, List
 
-from b_asic.signal import Signal
 from b_asic.operation import Operation
+from b_asic.signal import Signal
 
 PortId = NewType("PortId", int)
 
 
 class Port(ABC):
     """Abstract port class.
-    TODO: More info.
+
+    Handles functionality for port id and saves the connection to the parent operation.
     """
 
     _port_id: PortId
@@ -25,39 +26,62 @@ class Port(ABC):
         self._operation = operation
 
     @property
-    def identifier(self) -> PortId:
-        """Get the unique identifier."""
+    def id(self) -> PortId:
+        """Return the unique portid."""
         return self._port_id
 
     @property
     def operation(self) -> Operation:
-        """Get the connected operation."""
+        """Return the connected operation."""
         return self._operation
 
     @property
     @abstractmethod
     def signals(self) -> List[Signal]:
-        """Get a list of all connected signals."""
+        """Return a list of all connected signals."""
         raise NotImplementedError
 
     @abstractmethod
     def signal(self, i: int = 0) -> Signal:
-        """Get the connected signal at index i."""
+        """Return the connected signal at index i.
+
+        Keyword argumens:
+        i: integer index of the signal requsted.
+        """
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def connected_ports(self) -> List["Port"]:
+        """Return a list of all connected Ports."""
         raise NotImplementedError
 
     @abstractmethod
     def signal_count(self) -> int:
-        """Get the number of connected signals."""
+        """Return the number of connected signals."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def connect_port(self, port: "Port") -> Signal:
+        """Create and return a signal that is connected to this port and the entered
+        port and connect this port to the signal and the entered port to the signal."""
         raise NotImplementedError
 
     @abstractmethod
-    def connect(self, signal: Signal) -> None:
-        """Connect a signal."""
+    def connect_signal(self, signal: Signal) -> None:
+        """Connect this port to the entered signal. If the entered signal isn't connected to
+        this port then connect the entered signal to the port aswell."""
         raise NotImplementedError
 
     @abstractmethod
-    def disconnect(self, i: int = 0) -> None:
-        """Disconnect a signal."""
+    def disconnect_signal(self, i: int = 0) -> None:
+        """Disconnect a signal from the port. If the port is still connected to the entered signal
+        then the port is disconnected from the the entered signal aswell."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def is_connected_to_signal(self, signal: Signal) -> bool:
+        """Return true if the port is connected to the entered signal else false."""
         raise NotImplementedError
 
 
@@ -65,34 +89,51 @@ class InputPort(Port):
     """Input port.
     TODO: More info.
     """
-    _source_signal: Optional[Signal]
+
+    _signal: Optional[Signal]
 
     def __init__(self, port_id: PortId, operation: Operation):
         super().__init__(port_id, operation)
-        self._source_signal = None
+        self._signal = None
 
     @property
     def signals(self) -> List[Signal]:
-        return [] if self._source_signal is None else [self._source_signal]
+        return [] if self._signal is None else [self._signal]
 
     def signal(self, i: int = 0) -> Signal:
-        assert 0 <= i < self.signal_count() # TODO: Error message.
-        assert self._source_signal is not None # TODO: Error message.
-        return self._source_signal
+        assert 0 <= i < self.signal_count(), "Signal index out of bound."
+        assert self._signal is not None, "No Signal connect to InputPort."
+        return self._signal
+
+    @property
+    def connected_ports(self) -> List[Port]:
+        return [] if self._signal is None else [self._signal.source]
 
     def signal_count(self) -> int:
-        return 0 if self._source_signal is None else 1
+        return 0 if self._signal is None else 1
 
-    def connect(self, signal: Signal) -> None:
-        self._source_signal = signal
-        signal.destination = self
+    def connect_port(self, port: "OutputPort") -> Signal:
+        assert self._signal is None, "Connecting new port to already connected input port."
+        return Signal(port, self)        # self._signal is set by the signal constructor
 
-    def disconnect(self, i: int = 0) -> None:
-        assert 0 <= i < self.signal_count() # TODO: Error message.
-        self._source_signal.disconnect_source()
-        self._source_signal = None
+    def connect_signal(self, signal: Signal) -> None:
+        assert self._signal is None, "Connecting new port to already connected input port."
+        self._signal = signal
+        if self is not signal.destination:
+            # Connect this inputport as destination for this signal if it isn't already.
+            signal.connect_destination(self)
 
-    # TODO: More stuff.
+    def disconnect_signal(self, i: int = 0) -> None:
+        assert 0 <= i < self.signal_count(), "Signal Index out of range."
+        old_signal: Signal = self._signal
+        self._signal = None
+        if self is old_signal.destination:
+            # Disconnect the dest of the signal if this inputport currently is the dest
+            old_signal.disconnect_destination()
+        old_signal.disconnect_destination()
+
+    def is_connected_to_signal(self, signal: Signal) -> bool:
+        return self._signal is signal
 
 
 class OutputPort(Port):
@@ -100,30 +141,57 @@ class OutputPort(Port):
     TODO: More info.
     """
 
-    _destination_signals: List[Signal]
+    _signals: List[Signal]
 
     def __init__(self, port_id: PortId, operation: Operation):
         super().__init__(port_id, operation)
-        self._destination_signals = []
+        self._signals = []
 
     @property
     def signals(self) -> List[Signal]:
-        return self._destination_signals.copy()
+        return self._signals.copy()
 
     def signal(self, i: int = 0) -> Signal:
-        assert 0 <= i < self.signal_count() # TODO: Error message.
-        return self._destination_signals[i]
-
-    def signal_count(self) -> int:
-        return len(self._destination_signals)
+        assert 0 <= i < self.signal_count(), "Signal index out of bounds."
+        return self._signals[i]
 
-    def connect(self, signal: Signal) -> None:
-        assert signal not in self._destination_signals # TODO: Error message.
-        self._destination_signals.append(signal)
-        signal.source = self
-
-    def disconnect(self, i: int = 0) -> None:
-        assert 0 <= i < self.signal_count() # TODO: Error message.
-        del self._destination_signals[i]
+    @property
+    def connected_ports(self) -> List[Port]:
+        return [signal.destination for signal in self._signals \
+            if signal.destination is not None]
 
-    # TODO: More stuff.
+    def signal_count(self) -> int:
+        return len(self._signals)
+
+    def connect_port(self, port: InputPort) -> Signal:
+        return Signal(self, port)      # Signal is added to self._signals in signal constructor
+
+    def connect_signal(self, signal: Signal) -> None:
+        assert not self.is_connected_to_signal(signal), \
+                "Attempting to connect to Signal already connected."
+        self._signals.append(signal)
+        if self is not signal.source:
+            # Connect this outputport to the signal if it isn't already
+            signal.connect_source(self)
+
+    def disconnect_signal(self, i: int = 0) -> None:
+        assert 0 <= i < self.signal_count(), "Signal index out of bounds."
+        old_signal: Signal = self._signals[i]
+        del self._signals[i]
+        if self is old_signal.source:
+            # Disconnect the source of the signal if this outputport currently is the source
+            old_signal.disconnect_source()
+
+    def disconnect_signal_by_ref(self, signal: Signal) -> None:
+        """Remove the signal that was entered from the OutputPorts signals.
+        If the entered signal still is connected to this port then disconnect the
+        entered signal from the port aswell.
+
+        Keyword arguments:
+        - signal: Signal to remove.
+        """
+        i: int = self._signals.index(signal)
+        self.disconnect_signal(i)
+
+    def is_connected_to_signal(self, signal: Signal) -> bool:
+        return signal in self._signals                  # O(n) complexity

b_asic/signal.py

 """@package docstring
 B-ASIC Signal Module.
 """
-from typing import TYPE_CHECKING, Optional
+from typing import Optional, TYPE_CHECKING
 
 from b_asic.graph_component import TypeName
 from b_asic.abstract_graph_component import AbstractGraphComponent
+from b_asic.graph_component import Name
 
 if TYPE_CHECKING:
-    from b_asic import OutputPort, InputPort
+    from b_asic.port import InputPort, OutputPort
 
 class Signal(AbstractGraphComponent):
     """A connection between two ports."""
+
     _source: "OutputPort"
     _destination: "InputPort"
 
-    def __init__(self, src: Optional["OutputPort"] = None, dest: Optional["InputPort"] = None, **kwds):
-        super().__init__(**kwds)
-        self._source = src
-        self._destination = dest
+    def __init__(self, source: Optional["OutputPort"] = None, \
+                destination: Optional["InputPort"] = None, name: Name = ""):
+
+        super().__init__(name)
+
+        self._source = source
+        self._destination = destination
+
+        if source is not None:
+            self.connect_source(source)
+
+        if destination is not None:
+            self.connect_destination(destination)
 
     @property
     def source(self) -> "OutputPort":
-        """Returns the source OutputPort of the signal."""
+        """Return the source OutputPort of the signal."""
         return self._source
 
     @property
     def destination(self) -> "InputPort":
-        """Returns the destination InputPort of the signal."""
+        """Return the destination "InputPort" of the signal."""
         return self._destination
 
-    @source.setter
-    def source(self, src: "OutputPort") -> None:
-        """Sets the value of the source OutputPort of the signal."""
+    def connect_source(self, src: "OutputPort") -> None:
+        """Disconnect the previous source OutputPort of the signal and
+        connect to the entered source OutputPort. Also connect the entered
+        source port to the signal if it hasn't already been connected.
+
+        Keyword arguments:
+        - src: OutputPort to connect as source to the signal.
+        """
+        self.disconnect_source()
         self._source = src
+        if not src.is_connected_to_signal(self):
+            # If the new source isn't connected to this signal then connect it.
+            src.connect_signal(self)
 
-    @destination.setter
-    def destination(self, dest: "InputPort") -> None:
-        """Sets the value of the destination InputPort of the signal."""
+    def connect_destination(self, dest: "InputPort") -> None:
+        """Disconnect the previous destination InputPort of the signal and
+        connect to the entered destination InputPort. Also connect the entered
+        destination port to the signal if it hasn't already been connected.
+
+        Keywords argments:
+        - dest: InputPort to connect as destination to the signal.
+        """
+        self.disconnect_destination()
         self._destination = dest
+        if not dest.is_connected_to_signal(self):
+            # If the new destination isn't connected to tis signal then connect it.
+            dest.connect_signal(self)
 
     @property
     def type_name(self) -> TypeName:
         return "s"
 
     def disconnect_source(self) -> None:
-        """Disconnects the source OutputPort of the signal."""
-        self._source = None
+        """Disconnect the source OutputPort of the signal. If the source port
+        still is connected to this signal then also disconnect the source port."""
+        if self._source is not None:
+            old_source: "OutputPort" = self._source
+            self._source = None
+            if old_source.is_connected_to_signal(self):
+                # If the old destination port still is connected to this signal, then disconnect it.
+                old_source.disconnect_signal_by_ref(self)
 
     def disconnect_destination(self) -> None:
-        """Disconnects the destination InputPort of the signal."""
-        self._destination = None
+        """Disconnect the destination InputPort of the signal."""
+        if self._destination is not None:
+            old_destination: "InputPort" = self._destination
+            self._destination = None
+            if old_destination.is_connected_to_signal(self):
+                # If the old destination port still is connected to this signal, then disconnect it.
+                old_destination.disconnect_signal()
+
+    def is_connected(self) -> bool:
+        """Returns true if the signal is connected to both a source and a destination,
+        else false."""
+        return self._source is not None and self._destination is not None

b_asic/signal_flow_graph.py

@@ -50,8 +50,8 @@ class SFG(AbstractOperation):
         return [] # TODO: Implement
 
     def _add_graph_component(self, graph_component: GraphComponent) -> GraphID:
-        """Adds the entered graph component to the SFG's dictionary of graph objects and
-         returns a generated GraphID for it.
+        """Add the entered graph component to the SFG's dictionary of graph objects and
+         return a generated GraphID for it.
 
         Keyword arguments:
         graph_component: Graph component to add to the graph.
@@ -65,8 +65,8 @@ class SFG(AbstractOperation):
         return graph_id
 
     def find_by_id(self, graph_id: GraphID) -> Optional[GraphComponent]:
-        """Finds a graph object based on the entered Graph ID and returns it. If no graph
-        object with the entered ID was found then returns None.
+        """Find a graph object based on the entered Graph ID and return it. If no graph
+        object with the entered ID was found then return None.
 
         Keyword arguments:
         graph_id: Graph ID of the wanted object.
@@ -77,8 +77,8 @@ class SFG(AbstractOperation):
         return None
 
     def find_by_name(self, name: Name) -> List[GraphComponent]:
-        """Finds all graph objects that have the entered name and returns them
-        in a list. If no graph object with the entered name was found then returns an
+        """Find all graph objects that have the entered name and return them
+        in a list. If no graph object with the entered name was found then return an
         empty list.
 
         Keyword arguments:

test/fixtures/operation_tree.py

@@ -10,9 +10,9 @@ def operation():
 def create_operation(_type, dest_oper, index, **kwargs):
     oper = _type(**kwargs)
     oper_signal = Signal()
-    oper._output_ports[0].connect(oper_signal)
+    oper._output_ports[0].connect_signal(oper_signal)
 
-    dest_oper._input_ports[index].connect(oper_signal)
+    dest_oper._input_ports[index].connect_signal(oper_signal)
     return oper
 
 @pytest.fixture
@@ -50,11 +50,11 @@ def large_operation_tree():
     create_operation(Constant, add_oper_2, 1, value=5)
 
     add_oper_3 = Addition()
-    add_oper_signal = Signal(add_oper, add_oper_3)
-    add_oper._output_ports[0].connect(add_oper_signal)
-    add_oper_3._input_ports[0].connect(add_oper_signal)
+    add_oper_signal = Signal(add_oper.output(0), add_oper_3.output(0))
+    add_oper._output_ports[0].connect_signal(add_oper_signal)
+    add_oper_3._input_ports[0].connect_signal(add_oper_signal)
 
-    add_oper_2_signal = Signal(add_oper_2, add_oper_3)
-    add_oper_2._output_ports[0].connect(add_oper_2_signal)
-    add_oper_3._input_ports[1].connect(add_oper_2_signal)
+    add_oper_2_signal = Signal(add_oper_2.output(0), add_oper_3.output(0))
+    add_oper_2._output_ports[0].connect_signal(add_oper_2_signal)
+    add_oper_3._input_ports[1].connect_signal(add_oper_2_signal)
     return const_oper

test/graph_id/conftest.py

-import pytest

test/port/test_inputport.py

@@ -2,19 +2,83 @@
 B-ASIC test suite for Inputport
 """
 
-from b_asic import InputPort
-
 import pytest
 
-def test_connect_multiple_signals(signals):
-    """
-    test if only one signal can connect to an input port
-    """
+from b_asic import InputPort, OutputPort
+from b_asic import Signal
+
+@pytest.fixture
+def inp_port():
+    return InputPort(0, None)
+
+@pytest.fixture
+def out_port():
+    return OutputPort(0, None)
+
+@pytest.fixture
+def out_port2():
+    return OutputPort(1, None)
+
+@pytest.fixture
+def dangling_sig():
+    return Signal()
+
+@pytest.fixture
+def s_w_source():
+    out_port = OutputPort(0, None)
+    return Signal(source=out_port)
+
+@pytest.fixture
+def sig_with_dest():
+    inp_port = InputPort(0, None)
+    return Signal(destination=out_port)
+
+@pytest.fixture
+def connected_sig():
+    out_port = OutputPort(0, None)
     inp_port = InputPort(0, None)
+    return Signal(source=out_port, destination=inp_port)
+
+def test_connect_port_then_disconnect(inp_port, out_port):
+    """Test connect unused port to port."""
+    s1 = inp_port.connect_port(out_port)
+
+    assert inp_port.connected_ports == [out_port]
+    assert out_port.connected_ports == [inp_port]
+    assert inp_port.signals == [s1]
+    assert out_port.signals == [s1]
+    assert s1.source is out_port
+    assert s1.destination is inp_port
+
+    inp_port.disconnect_signal()
+
+    assert inp_port.connected_ports == []
+    assert out_port.connected_ports == []
+    assert inp_port.signals == []
+    assert out_port.signals == [s1]
+    assert s1.source is out_port
+    assert s1.destination is None
+
+def test_connect_used_port_to_new_port(inp_port, out_port, out_port2):
+    """Does connecting multiple ports to an inputport throw error?"""
+    inp_port.connect_port(out_port)
+    with pytest.raises(AssertionError):
+        inp_port.connect_port(out_port2)
+
+def test_connect_signal_then_disconnect(inp_port, s_w_source):
+    inp_port.connect_signal(s_w_source)
+
+    assert inp_port.connected_ports == [s_w_source.source]
+    assert s_w_source.source.connected_ports == [inp_port]
+    assert inp_port.signals == [s_w_source]
+    assert s_w_source.source.signals == [s_w_source]
+    assert s_w_source.destination is inp_port
 
-    for s in signals:
-        inp_port.connect(s)
+    inp_port.disconnect_signal()
 
-    assert inp_port.signal_count() == 1
-    assert inp_port.signals[0] == signals[-1]
+    assert inp_port.connected_ports == []
+    assert s_w_source.source.connected_ports == []
+    assert inp_port.signals == []
+    assert s_w_source.source.signals == [s_w_source]
+    assert s_w_source.destination is None
 

test/port/test_outputport.py

 """
-B-ASIC test suite for InputPort
+B-ASIC test suite for OutputPort
 TODO: More info
 """
-from b_asic import OutputPort
+from b_asic import InputPort, OutputPort
 import pytest
 
-def test_connect_multiple_signals(signals):
-    """
-    test if multiple signals can connect to an output port
-    """
-    outp_port = OutputPort(0, None)
+@pytest.fixture
+def inp_ports():
+    return [InputPort(_, None) for _ in range(0,3)]
 
-    for s in signals:
-        outp_port.connect(s)
+def test_connect_multiple_signals(inp_ports):
+    """Can multiple ports connect to an output port?"""
+    out_port = OutputPort(0, None)
 
-    assert outp_port.signal_count() == 3
-    assert outp_port.signals == signals
+    for port in inp_ports:
+        out_port.connect_port(port)
+    
+    assert out_port.signal_count() == len(inp_ports)
+
+def test_disconnect_multiple_signals(inp_ports):
+    """Can multiple ports disconnect from an output port?"""
+    out_port = OutputPort(0, None)
+
+    for port in inp_ports:
+        out_port.connect_port(port)
+    
+    for _ in inp_ports:
+        out_port.disconnect_signal(0)
+
+    assert out_port.signal_count() == 0
\ No newline at end of file

test/signal/test_signal.py

+"""
+B-ASIC test suit for the signal module which consists of the Signal class.
+"""
+
+from b_asic.port import InputPort, OutputPort
+from b_asic.signal import Signal
+
+import pytest
+
+def test_signal_creation_and_disconnction_and_connection_changing():
+    in_port = InputPort(0, None)
+    out_port = OutputPort(1, None)
+    s = Signal(out_port, in_port)
+
+    assert in_port.signals == [s]
+    assert out_port.signals == [s]
+    assert s.source is out_port
+    assert s.destination is in_port
+
+    in_port1 = InputPort(0, None)
+    s.connect_destination(in_port1)
+
+    assert in_port.signals == []
+    assert in_port1.signals == [s]
+    assert out_port.signals == [s]
+    assert s.source is out_port
+    assert s.destination is in_port1
+
+    s.disconnect_source()
+
+    assert out_port.signals == []
+    assert in_port1.signals == [s]
+    assert s.source is None
+    assert s.destination is in_port1
+
+    s.disconnect_destination()
+
+    assert out_port.signals == []
+    assert in_port1.signals == []
+    assert s.source is None
+    assert s.destination is None
+
+    out_port1 = OutputPort(0, None)
+    s.connect_source(out_port1)
+
+    assert out_port1.signals == [s]
+    assert s.source is out_port1
+    assert s.destination is None
+
+    s.connect_source(out_port)
+
+    assert out_port.signals == [s]
+    assert out_port1.signals == []
+    assert s.source is out_port
+    assert s.destination is None
+
+    s.connect_destination(in_port)
+
+    assert out_port.signals == [s]
+    assert in_port.signals == [s]
+    assert s.source is out_port
+    assert s.destination is in_port

test/signal_flow_graph/conftest.py

-import pytest

test/traverse/test_traverse_tree.py

@@ -24,6 +24,7 @@ def test_traverse_type(large_operation_tree):
 
 def test_traverse_loop(operation_tree):
     add_oper_signal = Signal()
-    operation_tree._output_ports[0].connect(add_oper_signal)
-    operation_tree._input_ports[0].connect(add_oper_signal)
+    operation_tree._output_ports[0].connect_signal(add_oper_signal)
+    operation_tree._input_ports[0].disconnect_signal()
+    operation_tree._input_ports[0].connect_signal(add_oper_signal)
     assert len(list(operation_tree.traverse())) == 2