Test objects

fec_integration_tests/interface/interface_functions/test_front_end_common_insert_edges_to_live_packet_gatherers.py

@@ -28,9 +28,7 @@
     LivePacketGatherParameters)
 from spinn_front_end_common.utility_models import (
     LivePacketGather, LivePacketGatherMachineVertex)
-from fec_integration_tests.interface.interface_functions.simple_test_vertex \
-    import (
-        SimpleTestVertex)
+from pacman_test_objects import SimpleTestVertex
 
 
 class TestInsertLPGEdges(unittest.TestCase):

fec_integration_tests/interface/interface_functions/test_front_end_common_load_executable_images.py

@@ -15,6 +15,7 @@
 
 import unittest
 from collections import defaultdict
+from spinn_utilities.overrides import overrides
 from spinnman.transceiver import Transceiver
 from spinnman.model import ExecutableTargets
 from spinn_front_end_common.utilities.utility_objs import (
@@ -32,6 +33,7 @@ def __init__(self, test_case):
         self._n_cores_in_app = defaultdict(lambda: 0)
         self._executable_on_core = dict()
 
+    @overrides(Transceiver.execute_flood)
     def execute_flood(
             self, core_subsets, executable, app_id,
             n_bytes=None, wait=False, is_filename=False):  # @UnusedVariable
@@ -44,9 +46,11 @@ def execute_flood(
                 self._executable_on_core[x, y, p] = executable
         self._n_cores_in_app[app_id] += len(core_subsets)
 
+    @overrides(Transceiver.get_core_state_count)
     def get_core_state_count(self, app_id, state):  # @UnusedVariable
         return self._n_cores_in_app[app_id]
 
+    @overrides(Transceiver.send_signal)
     def send_signal(self, app_id, signal):
         pass
 

fec_integration_tests/test_buffer_manager_listener_creation.py

@@ -16,16 +16,14 @@
 import tempfile
 import unittest
 
-from pacman.model.partitioner_interfaces import LegacyPartitionerAPI
-from spinn_utilities.overrides import overrides
 from pacman.model.placements import Placement, Placements
 from pacman.model.tags import Tags
-from pacman.model.graphs.application import ApplicationVertex
 from spinn_machine.tags import IPTag
 from spinnman.transceiver import Transceiver
 from spinnman.connections.udp_packet_connections import (
     SCAMPConnection, EIEIOConnection)
 from spinn_front_end_common.interface.buffer_management import BufferManager
+from pacman_test_objects import SimpleTestVertex
 
 
 class TestBufferManagerListenerCreation(unittest.TestCase):
@@ -37,8 +35,8 @@ def test_listener_creation(self):
         # a single listener
 
         # Create two vertices
-        v1 = _TestVertex(10, "v1", 256)
-        v2 = _TestVertex(10, "v2", 256)
+        v1 = SimpleTestVertex(10, "v1", 256)
+        v2 = SimpleTestVertex(10, "v2", 256)
 
         # Create two tags - important thing is port=None
         t1 = IPTag(board_address='127.0.0.1', destination_x=0,
@@ -97,32 +95,5 @@ def test_listener_creation(self):
         self.assertEqual(number_of_listeners, 1)
 
 
-class _TestVertex(ApplicationVertex, LegacyPartitionerAPI):
-    """
-    taken skeleton test vertex definition from PACMAN.uinit_test_objects
-    """
-    _model_based_max_atoms_per_core = None
-
-    def __init__(self, n_atoms, label=None, max_atoms_per_core=256):
-        super().__init__(label=label, max_atoms_per_core=max_atoms_per_core)
-        self._model_based_max_atoms_per_core = max_atoms_per_core
-        self._n_atoms = n_atoms
-
-    @overrides
-    def get_resources_used_by_atoms(self, vertex_slice):
-        pass
-
-    @overrides(LegacyPartitionerAPI.create_machine_vertex)
-    def create_machine_vertex(
-            self, vertex_slice, resources_required, label=None,
-            constraints=None):
-        pass
-
-    @property
-    @overrides(LegacyPartitionerAPI.n_atoms)
-    def n_atoms(self):
-        return self._n_atoms
-
-
 if __name__ == "__main__":
     unittest.main()

unittests/interface/interface_functions/test_front_end_common_app_finisher.py

@@ -13,12 +13,14 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+from spinn_utilities.overrides import overrides
 from spinn_machine import CoreSubsets
 from spinn_front_end_common.utilities.utility_objs import ExecutableType
 from spinnman.model.enums.cpu_state import CPUState
 from spinn_front_end_common.interface.interface_functions import (
     ApplicationFinisher)
 from spinnman.model.cpu_infos import CPUInfos
+from spinnman.transceiver import Transceiver
 
 
 class _MockTransceiver(object):
@@ -29,17 +31,19 @@ def __init__(self, core_states, time_between_states):
         self._current_state = 0
         self.sdp_send_count = 0
 
-    def get_core_state_count(self, _app_id, state):
+    @overrides(Transceiver.get_core_state_count)
+    def get_core_state_count(self, app_id, state):
         count = 0
         for core_state in self._core_states[self._current_state].values():
             if core_state == state:
                 count += 1
         return count
 
-    def get_cores_in_state(self, core_subsets, states):
+    @overrides(Transceiver.get_cores_in_state)
+    def get_cores_in_state(self, all_core_subsets, states):
         cores_in_state = CPUInfos()
         core_states = self._core_states[self._current_state]
-        for core_subset in core_subsets:
+        for core_subset in all_core_subsets:
             x = core_subset.x
             y = core_subset.y
 
@@ -54,9 +58,11 @@ def get_cores_in_state(self, core_subsets, states):
         self._current_state += 1
         return cores_in_state
 
-    def send_sdp_message(self, message):
+    @overrides(Transceiver.send_sdp_message)
+    def send_sdp_message(self, message, connection=None):
         self.sdp_send_count += 1
 
+    @overrides(Transceiver.send_signal)
     def send_signal(self, app_id, signal):
         pass
 

unittests/interface/interface_functions/test_front_end_common_chip_iobuf_extractor.py

@@ -17,18 +17,21 @@
 import tempfile
 import unittest
 from spinn_utilities.executable_finder import ExecutableFinder
+from spinn_utilities.overrides import overrides
 from spinn_machine import CoreSubsets, CoreSubset
 from spinnman.model import IOBuffer
 from spinn_front_end_common.interface.interface_functions import (
     ChipIOBufExtractor)
 from spinnman.model import ExecutableTargets
+from spinnman.transceiver import Transceiver
 
 
 class _PretendTransceiver(object):
     def __init__(self, iobuffers):
         self._iobuffers = iobuffers
 
-    def get_iobuf(self, core_subsets):
+    @overrides(Transceiver.get_iobuf)
+    def get_iobuf(self, core_subsets=None):
         for iobuf in self._iobuffers:
             if core_subsets.is_core(iobuf.x, iobuf.y, iobuf.p):
                 yield iobuf