Adapt to c/10315

Depends-On: 10315,10346

Change-Id: I6c6c4659d62b03a5e420c8e73d8596726432c0b7

test/test_hwsimtests_vx.py

@@ -3,7 +3,8 @@ import unittest
 from numbers import Integral
 from typing import Set, ClassVar
 
-from dlens_vx.sta import PlaybackProgramExecutor, generate, DigitalInit
+from dlens_vx.hxcomm import ManagedConnection
+from dlens_vx.sta import generate, DigitalInit, run
 from dlens_vx.halco import PPUOnDLS, iter_all, JTAGIdCodeOnDLS, TimerOnDLS
 from dlens_vx.tools.run_ppu_program import load_and_start_program, \
     stop_program, wait_until_ppu_finished, PPUTimeoutError
@@ -15,27 +16,28 @@ from helpers.hwtest_common import get_special_binaries, find_binaries, \
 
 class LibnuxHwSimTestsVx(unittest.TestCase):
     TESTS: Set[PpuHwTest] = set()
-    EXECUTOR = PlaybackProgramExecutor()
+    MANAGED_CONNECTION = ManagedConnection()
+    CONNECTION = None
     CHIP_REVISION: ClassVar[Integral]
 
     @classmethod
     def setUpClass(cls) -> None:
-        # Connect to some executor (sim or hardware)
-        cls.EXECUTOR.connect()
+        # Initialize remote connection (to sim or hardware)
+        cls.CONNECTION = cls.MANAGED_CONNECTION.__enter__()
 
         # Initialize the chip and find chip version
         init_builder, _ = generate(DigitalInit())
         jtag_id_ticket = init_builder.read(JTAGIdCodeOnDLS())
         init_builder.write(TimerOnDLS(), Timer(0))
         init_builder.wait_until(TimerOnDLS(), 1000)
-        cls.EXECUTOR.run(init_builder.done())
+        run(cls.CONNECTION, init_builder.done())
         jtag_id = jtag_id_ticket.get()
         cls.CHIP_REVISION = jtag_id.version
 
     @classmethod
     def tearDownClass(cls) -> None:
         # Disconnect the executor
-        cls.EXECUTOR.disconnect()
+        cls.MANAGED_CONNECTION.__exit__()
 
     @classmethod
     def generate_cases(cls):
@@ -53,9 +55,9 @@ class LibnuxHwSimTestsVx(unittest.TestCase):
                     ppu = random.choice(list(iter_all(PPUOnDLS)))
                     log.info("Running test on %s." % ppu)
 
-                    load_and_start_program(cls.EXECUTOR, ppu_test.path, ppu)
+                    load_and_start_program(cls.CONNECTION, ppu_test.path, ppu)
                     try:
-                        wait_until_ppu_finished(cls.EXECUTOR,
+                        wait_until_ppu_finished(cls.CONNECTION,
                                                 timeout=ppu_test.timeout,
                                                 ppu=ppu)
                         self.assertFalse(ppu_test.expect_timeout,
@@ -64,7 +66,7 @@ class LibnuxHwSimTestsVx(unittest.TestCase):
                         self.assertTrue(ppu_test.expect_timeout,
                                         "Unexpected timeout.")
                     finally:
-                        returncode = stop_program(cls.EXECUTOR, ppu=ppu)
+                        returncode = stop_program(cls.CONNECTION, ppu=ppu)
 
                     self.assertEqual(ppu_test.expected_exit_code, returncode,
                                      f"Exit code was {returncode}, expected "

test/with_hostcode/log_access_pattern_host.py

@@ -6,7 +6,8 @@ from typing import List, Iterator, ClassVar
 
 from dlens_vx.halco import iter_all, PPUOnDLS, SynapseRowOnSynram, \
     SynapseOnSynapseRow, TimerOnDLS, JTAGIdCodeOnDLS
-from dlens_vx.sta import PlaybackProgramExecutor, generate, DigitalInit
+from dlens_vx.hxcomm import ManagedConnection
+from dlens_vx.sta import generate, DigitalInit, run
 from dlens_vx import logger
 from dlens_vx.tools.run_ppu_program import load_and_start_program, \
     wait_until_ppu_finished, stop_program
@@ -38,7 +39,8 @@ FLANGE_LOG_PATH = os.environ.get("FLANGE_LOG_PATH",
 
 
 class LibnuxAccessPatternTestVx(unittest.TestCase):
-    EXECUTOR = PlaybackProgramExecutor()
+    MANAGED_CONNECTION = ManagedConnection()
+    CONNECTION = None
     CHIP_REVISION: ClassVar[int]
 
     class LogParser:
@@ -68,26 +70,26 @@ class LibnuxAccessPatternTestVx(unittest.TestCase):
     @classmethod
     def setUpClass(cls) -> None:
         # Connect to some executor (sim or hardware)
-        cls.EXECUTOR.connect()
+        cls.CONNECTION = cls.MANAGED_CONNECTION.__enter__()
 
         # Initialize the chip and find chip version
         init_builder, _ = generate(DigitalInit())
         jtag_id_ticket = init_builder.read(JTAGIdCodeOnDLS())
         init_builder.write(TimerOnDLS(), Timer(0))
         init_builder.wait_until(TimerOnDLS(), 1000)
-        cls.EXECUTOR.run(init_builder.done())
+        run(cls.CONNECTION, init_builder.done())
         jtag_id = jtag_id_ticket.get()
         cls.CHIP_REVISION = jtag_id.version
 
     @classmethod
     def tearDownClass(cls) -> None:
         # Disconnect the executor
-        cls.EXECUTOR.disconnect()
+        cls.MANAGED_CONNECTION.__exit__()
 
     def run_ppu_program(self, ppu: PPUOnDLS, program_path: str, timeout: int):
-        load_and_start_program(self.EXECUTOR, program_path, ppu)
-        wait_until_ppu_finished(self.EXECUTOR, timeout=timeout, ppu=ppu)
-        ret_code = stop_program(self.EXECUTOR, ppu=ppu,
+        load_and_start_program(self.CONNECTION, program_path, ppu)
+        wait_until_ppu_finished(self.CONNECTION, timeout=timeout, ppu=ppu)
+        ret_code = stop_program(self.CONNECTION, ppu=ppu,
                                 print_mailbox=False)
 
         self.assertEqual(0, ret_code,