run_safe

gfe_integration_tests/test_profile/test_profile.py

@@ -13,41 +13,44 @@
 # limitations under the License.
 
 import os
-import spinnaker_graph_front_end as s
-from gfe_integration_tests.test_profile.profiled_vertex import ProfiledVertex
+
 from spinn_front_end_common.data.fec_data_view import FecDataView
+from spinnaker_testbase import BaseTestCase
 
+import spinnaker_graph_front_end as s
+from gfe_integration_tests.test_profile.profiled_vertex import ProfiledVertex
 
-def test_profile():
-    s.setup(model_binary_folder=os.path.dirname(__file__))
-    vertex = ProfiledVertex()
-    s.add_machine_vertex_instance(vertex)
-    s.run(50)
+class  TestProfile(BaseTestCase):
 
-    place = FecDataView.get_placement_of_vertex(vertex)
-    folder = FecDataView.get_app_provenance_dir_path()
+    def check_profile_data(self):
+        s.setup(model_binary_folder=os.path.dirname(__file__))
+        vertex = ProfiledVertex()
+        s.add_machine_vertex_instance(vertex)
+        s.run(50)
 
-    s.stop()
+        place = FecDataView.get_placement_of_vertex(vertex)
+        folder = FecDataView.get_app_provenance_dir_path()
 
-    profile_file = os.path.join(
-        folder, f"{place.x}_{place.y}_{place.p}_profile.txt")
-    assert os.path.exists(profile_file)
+        s.stop()
 
-    with open(profile_file, "r") as f:
-        # Headings and lines
-        f.readline()
-        f.readline()
+        profile_file = os.path.join(
+            folder, f"{place.x}_{place.y}_{place.p}_profile.txt")
+        assert os.path.exists(profile_file)
 
-        # Next 2 lines should have profile data
-        for i in range(2):
-            line = f.readline()
-            parts = line.split()
-            assert len(parts) == 5
-            assert parts[0] == "SDRAMWrite" or parts[0] == "DTCMWrite"
-            assert int(parts[1]) == 25
-            assert float(parts[3]) == 1.0
-            assert float(parts[2]) == float(parts[4])
+        with open(profile_file, "r") as f:
+            # Headings and lines
+            f.readline()
+            f.readline()
 
+            # Next 2 lines should have profile data
+            for i in range(2):
+                line = f.readline()
+                parts = line.split()
+                assert len(parts) == 5
+                assert parts[0] == "SDRAMWrite" or parts[0] == "DTCMWrite"
+                assert int(parts[1]) == 25
+                assert float(parts[3]) == 1.0
+                assert float(parts[2]) == float(parts[4])
 
-if __name__ == "__main__":
-    test_profile()
+    def test_profile_data(self):
+        self.runsafe(self.check_profile_data)

gfe_integration_tests/test_rte/test_rte_during_run.py

@@ -15,20 +15,28 @@
 import os
 import traceback
 import pytest
+
 from spinnman.exceptions import SpinnmanException
 from spinnman.model.enums import ExecutableType
+from spinnaker_testbase import BaseTestCase
+
 import spinnaker_graph_front_end as s
 from gfe_integration_tests.test_rte.run_vertex import RunVertex
 
 
-def test_rte_during_run():
-    s.setup(model_binary_folder=os.path.dirname(__file__))
-    s.add_machine_vertex_instance(RunVertex(
-        "test_rte_during_run.aplx",
-        ExecutableType.USES_SIMULATION_INTERFACE))
-    with pytest.raises(SpinnmanException):
-        try:
-            s.run(1000)
-        except Exception:
-            traceback.print_exc()
-            raise
+class TestRteDuringRun(BaseTestCase):
+
+    def check_rte_during_run(self):
+        s.setup(model_binary_folder=os.path.dirname(__file__))
+        s.add_machine_vertex_instance(RunVertex(
+            "test_rte_during_run.aplx",
+            ExecutableType.USES_SIMULATION_INTERFACE))
+        with pytest.raises(SpinnmanException):
+            try:
+                s.run(1000)
+            except Exception:
+                traceback.print_exc()
+                raise
+
+    def test_rte_during_run(self):
+        self.runsafe(self.check_rte_during_run)

gfe_integration_tests/test_rte/test_rte_during_run_forever.py

@@ -13,32 +13,36 @@
 # limitations under the License.
 
 import os
+import pytest
 from time import sleep
+
 from spinnman.model.enums import ExecutableType
 from spinn_front_end_common.utilities.exceptions import (
     ExecutableFailedToStopException)
 from spinn_front_end_common.utilities.database import DatabaseConnection
+from spinnaker_testbase import BaseTestCase
+
 import spinnaker_graph_front_end as s
 from gfe_integration_tests.test_rte.run_vertex import RunVertex
-import pytest
 
 
-def test_rte_during_run_forever():
+class TestRteDuringRunForever(BaseTestCase):
 
-    def start():
-        sleep(3.0)
-        s.stop_run()
+    def check_rte_during_run_forever(self):
 
-    with DatabaseConnection(start, local_port=None) as conn:
-        s.setup(model_binary_folder=os.path.dirname(__file__))
-        s.add_machine_vertex_instance(RunVertex(
-            "test_rte_during_run.aplx",
-            ExecutableType.USES_SIMULATION_INTERFACE))
-        s.add_socket_address(None, "localhost", conn.local_port)
-        s.run(None)
-        with pytest.raises(ExecutableFailedToStopException):
-            s.stop()
+        def start():
+            sleep(3.0)
+            s.stop_run()
 
+        with DatabaseConnection(start, local_port=None) as conn:
+            s.setup(model_binary_folder=os.path.dirname(__file__))
+            s.add_machine_vertex_instance(RunVertex(
+                "test_rte_during_run.aplx",
+                ExecutableType.USES_SIMULATION_INTERFACE))
+            s.add_socket_address(None, "localhost", conn.local_port)
+            s.run(None)
+            with pytest.raises(ExecutableFailedToStopException):
+                s.stop()
 
-if __name__ == "__main__":
-    test_rte_during_run_forever()
+    def test_rte_during_run_forever(self):
+        self.runsafe(self.check_rte_during_run_forever)

gfe_integration_tests/test_rte/test_rte_start.py

@@ -14,17 +14,25 @@
 
 import os
 import pytest
+
 from spinnman.exceptions import SpinnmanException
 from spinnman.model.enums import ExecutableType
+from spinnaker_testbase import BaseTestCase
+
 import spinnaker_graph_front_end as s
 from gfe_integration_tests.test_rte.run_vertex import RunVertex
 
 
-def test_rte_at_start():
-    s.setup(model_binary_folder=os.path.dirname(__file__))
-    s.add_machine_vertex_instance(
-        RunVertex(
-            "test_rte_start.aplx",
-            ExecutableType.USES_SIMULATION_INTERFACE))
-    with pytest.raises(SpinnmanException):
-        s.run(1000)
+class TestRteStart(BaseTestCase):
+
+    def check_rte_at_start(self):
+        s.setup(model_binary_folder=os.path.dirname(__file__))
+        s.add_machine_vertex_instance(
+            RunVertex(
+                "test_rte_start.aplx",
+                ExecutableType.USES_SIMULATION_INTERFACE))
+        with pytest.raises(SpinnmanException):
+            s.run(1000)
+
+    def test_rte_at_start(self):
+        self.runsafe(self.check_rte_at_start)

gfe_integration_tests/test_rte/test_run_too_long.py

@@ -14,16 +14,24 @@
 
 import os
 import pytest
+
 from spinnman.exceptions import SpinnmanTimeoutException
 from spinnman.model.enums import ExecutableType
+from spinnaker_testbase import BaseTestCase
+
 import spinnaker_graph_front_end as s
 from gfe_integration_tests.test_rte.run_vertex import RunVertex
 
 
-def test_run_too_long():
-    s.setup(model_binary_folder=os.path.dirname(__file__))
-    s.add_machine_vertex_instance(RunVertex(
-        "test_run_too_long.aplx",
-        ExecutableType.USES_SIMULATION_INTERFACE))
-    with pytest.raises(SpinnmanTimeoutException):
-        s.run(1000)
+class TestRunTooLong(BaseTestCase):
+
+    def check_run_too_long(slef):
+        s.setup(model_binary_folder=os.path.dirname(__file__))
+        s.add_machine_vertex_instance(RunVertex(
+            "test_run_too_long.aplx",
+            ExecutableType.USES_SIMULATION_INTERFACE))
+        with pytest.raises(SpinnmanTimeoutException):
+            s.run(1000)
+
+    def test_run_too_long(self):
+        self.runsafe(self.check_run_too_long)