new f7 test

templates/f7/test

@@ -2,23 +2,17 @@
 
 
 import os
 import os
 import sys
 import sys
-import time
-import struct
-import serial
 import pylink
 import pylink
-sys.path.insert(0, '.')
 from test_common import (
 from test_common import (
-    SaleaeTimeMeasurements,
+    LogicMultiplexerTimeMeasurements,
     parse_nist_aead_test_vectors,
     parse_nist_aead_test_vectors,
     DeviceUnderTestAeadUARTP,
     DeviceUnderTestAeadUARTP,
-    run_nist_aead_test,
+    compare_dumps,
+    eprint,
+    run_nist_aead_test_line,
 )
 )
 
 
 
 
-def eprint(*args, **kargs):
-    print(*args, file=sys.stderr, **kargs)
-
-
 def get_serial():
 def get_serial():
     import serial.tools.list_ports
     import serial.tools.list_ports
     ports = serial.tools.list_ports.comports()
     ports = serial.tools.list_ports.comports()
@@ -38,17 +32,21 @@ def get_serial():
 
 
 class F7(DeviceUnderTestAeadUARTP):
 class F7(DeviceUnderTestAeadUARTP):
     RAM_SIZE = 0x50000
     RAM_SIZE = 0x50000
-    def __init__(self):
+
+    def __init__(self, firmware_path, ram_pattern_path):
         DeviceUnderTestAeadUARTP.__init__(self, get_serial())
         DeviceUnderTestAeadUARTP.__init__(self, get_serial())
         self.jlink = pylink.JLink()
         self.jlink = pylink.JLink()
         self.jlink.open(779340002)
         self.jlink.open(779340002)
-
+        self.firmware_path = firmware_path
+        self.ram_pattern_path = ram_pattern_path
 
 
     def flash(self):
     def flash(self):
         jlink = self.jlink
         jlink = self.jlink
         jlink.connect('STM32F746ZG')
         jlink.connect('STM32F746ZG')
-        jlink.flash_file('build/f7.bin', 0x8000000)
-        jlink.flash_file('ram_pattern.bin', 0x20000000)
+        jlink.flash_file(self.firmware_path, 0x8000000)
+        eprint("Firmware flashed.")
+        jlink.flash_file(self.ram_pattern_path, 0x20000000)
+        eprint("RAM flashed.")
         jlink.reset()
         jlink.reset()
         jlink.restart()
         jlink.restart()
 
 
@@ -65,27 +63,38 @@ def main(argv):
 
 
     eprint(argv[0])
     eprint(argv[0])
     script_dir = os.path.split(argv[0])[0]
     script_dir = os.path.split(argv[0])[0]
-    if len(script_dir) > 0:
-        os.chdir(script_dir)
 
 
-    dut = F7()
+    dut = F7(
+        os.path.join(script_dir, 'build', 'f7.bin'),
+        os.path.join(script_dir, 'ram_pattern.bin'))
+
+    try:
+        tool = LogicMultiplexerTimeMeasurements(0x000c)
+        tool.begin_measurement()
 
 
         dut.flash()
         dut.flash()
-    eprint("Flashed")
         dut.prepare()
         dut.prepare()
-    eprint("Board initialized properly")
-    sys.stdout.write("Hello, World!\n")
+        sys.stdout.write("Board prepared\n")
         sys.stdout.flush()
         sys.stdout.flush()
 
 
-    try:
-        run_nist_aead_test(dut, kat)
-        return 0
+        dump_a = dut.dump_ram()
+
+        for i, m, ad, k, npub, c in kat:
+            tool.arm()
+            run_nist_aead_test_line(dut, i, m, ad, k, npub, c)
+            tool.unarm()
+
+            if i == 1:
+                dump_b = dut.dump_ram()
+                longest = compare_dumps(dump_a, dump_b)
+                print("  longest chunk of untouched memory = %d" % longest)
 
 
     except Exception as ex:
     except Exception as ex:
         print("TEST FAILED")
         print("TEST FAILED")
         raise ex
         raise ex
 
 
     finally:
     finally:
+        tool.end_measurement()
         sys.stdout.flush()
         sys.stdout.flush()
         sys.stderr.flush()
         sys.stderr.flush()
 
 

test_common.py

@@ -141,23 +141,6 @@ class UARTP:
                         return buf
                         return buf
 
 
 
 
-def run_nist_aead_test(dut, kat):
-    dump_a = dut.dump_ram()
-
-    tool = SaleaeTimeMeasurements()
-    tool.begin_measurement()
-    try:
-        for i, m, ad, k, npub, c in kat:
-            run_nist_aead_test_line(dut, i, m, ad, k, npub, c)
-
-            if dump_a is not None and i == 1:
-                dump_b = dut.dump_ram()
-                longest = compare_dumps(dump_a, dump_b)
-                print("  longest chunk of untouched memory = %d" % longest)
-    finally:
-        tool.end_measurement()
-
-
 def run_nist_aead_test_line(dut, i, m, ad, k, npub, c):
 def run_nist_aead_test_line(dut, i, m, ad, k, npub, c):
     eprint()
     eprint()
     eprint("Count = %d" % i)
     eprint("Count = %d" % i)
@@ -268,12 +251,75 @@ def parse_nist_aead_test_vectors(test_file_path):
                     "ERROR: unparsed line in test vectors file: '%s'" % line)
                     "ERROR: unparsed line in test vectors file: '%s'" % line)
 
 
 
 
-class SaleaeTimeMeasurements:
+class TimeMeasurementTool:
+    def begin_measurement(self):
+        pass
+
+    def arm(self):
+        pass
+
+    def unarm(self):
+        pass
+
+    def end_measurement(self):
+        pass
+
+
+class LogicMultiplexerTimeMeasurements(TimeMeasurementTool):
+
+    def __init__(self, mask=0xffffffffffffffff):
+        self.mask = mask
+        self.sock = None
+        self.capture = []
+
+    def recv_samples(self):
+        import socket
+
+        capture = []
+        while 1:
+            try:
+                rcvd = self.sock.recv(16)
+            except socket.timeout:
+                break
+            except BlockingIOError:
+                break
+            if len(rcvd) != 16:
+                raise Exception("Could not receive 16 bytes of logic sample!")
+
+            time, value = struct.unpack("<dQ", rcvd)
+            eprint("%16.10f: %016x" % (time, value))
+            capture.append((time, value))
+        return capture
+
+    def begin_measurement(self):
+        import socket
+
+        server_addr = os.path.expandvars('$XDG_RUNTIME_DIR/lwc-logic-socket')
+        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self.sock.settimeout(0)
+        self.server_addr = server_addr
+        self.sock.connect(self.server_addr)
+        self.sock.send(struct.pack("<Q", self.mask))
+
+    def unarm(self):
+        self.capture.extend(self.recv_samples())
+
+    def end_measurement(self):
+        time.sleep(1)
+        self.capture.extend(self.recv_samples())
+        self.sock.close()
+
+
+class SaleaeTimeMeasurements(TimeMeasurementTool):
     __slots__ = ['sal']
     __slots__ = ['sal']
 
 
-    def __init__(self):
+    def __init__(self, channels=[0, 1]):
         import saleae
         import saleae
-        self.sal = saleae.Saleae()
+        sal = saleae.Saleae()
+        sal.set_active_channels(self.channels, [])
+        sal.set_sample_rate(sal.get_all_sample_rates()[0])
+        sal.set_capture_seconds(6000)
+        self.sal = sal
 
 
     def begin_measurement(self):
     def begin_measurement(self):
         # Channel 0 is reset
         # Channel 0 is reset
@@ -281,9 +327,6 @@ class SaleaeTimeMeasurements:
         import time
         import time
         sal = self.sal
         sal = self.sal
 
 
-        sal.set_active_channels([0, 1], [])
-        sal.set_sample_rate(sal.get_all_sample_rates()[0])
-        sal.set_capture_seconds(6000)
         sal.capture_start()
         sal.capture_start()
         time.sleep(1)
         time.sleep(1)
         if sal.is_processing_complete():
         if sal.is_processing_complete():
@@ -332,22 +375,34 @@ def main(argv):
     ser = serial.Serial(dev, baudrate=115200, timeout=5)
     ser = serial.Serial(dev, baudrate=115200, timeout=5)
     dut = DeviceUnderTestAeadUARTP(ser)
     dut = DeviceUnderTestAeadUARTP(ser)
 
 
+    try:
+        tool = SaleaeTimeMeasurements()
+        tool.begin_measurement()
         dut.flash()
         dut.flash()
         eprint("Flashed")
         eprint("Flashed")
         dut.prepare()
         dut.prepare()
-    eprint("Board initialized properly")
+        eprint("Prepared")
         sys.stdout.write("Hello, World!\n")
         sys.stdout.write("Hello, World!\n")
         sys.stdout.flush()
         sys.stdout.flush()
 
 
-    try:
-        run_nist_aead_test(dut, kat)
-        return 0
+        dump_a = dut.dump_ram()
+
+        for i, m, ad, k, npub, c in kat:
+            tool.arm()
+            run_nist_aead_test_line(dut, i, m, ad, k, npub, c)
+            tool.unarm()
+
+            if dump_a is not None and i == 1:
+                dump_b = dut.dump_ram()
+                longest = compare_dumps(dump_a, dump_b)
+                print("  longest chunk of untouched memory = %d" % longest)
 
 
     except Exception as ex:
     except Exception as ex:
         print("TEST FAILED")
         print("TEST FAILED")
         raise ex
         raise ex
 
 
     finally:
     finally:
+        tool.end_measurement()
         sys.stdout.flush()
         sys.stdout.flush()
         sys.stderr.flush()
         sys.stderr.flush()