Updating bench flipper search algorithm (#3742)

* initial commit, changed flipper search function and unit test runner script
* increased timeouts due to new unpacking delays
* python formatting fixes
* fixed flipper name

Co-authored-by: doomwastaken <k.volkov@flipperdevices.com>
Co-authored-by: あく <alleteam@gmail.com>

.github/workflows/unit_tests.yml

@@ -1,5 +1,4 @@
 name: 'Unit tests'
-
 on:
   pull_request:
 
@@ -25,7 +24,7 @@ jobs:
       - name: 'Get flipper from device manager (mock)'
         id: device
         run: |
-          echo "flipper=/dev/ttyACM0" >> $GITHUB_OUTPUT
+          echo "flipper=auto" >> $GITHUB_OUTPUT
 
       - name: 'Flash unit tests firmware'
         id: flashing
@@ -40,7 +39,7 @@ jobs:
         timeout-minutes: 5
         run: |
           source scripts/toolchain/fbtenv.sh
-          python3 scripts/testing/await_flipper.py ${{steps.device.outputs.flipper}}
+          python3 scripts/testops.py -p=${{steps.device.outputs.flipper}} -t=120 await_flipper 
           python3 scripts/storage.py -p ${{steps.device.outputs.flipper}} format_ext
 
       - name: 'Copy assets and unit data, reboot and wait for flipper'
@@ -49,11 +48,11 @@ jobs:
         timeout-minutes: 7
         run: |
           source scripts/toolchain/fbtenv.sh
-          python3 scripts/testing/await_flipper.py ${{steps.device.outputs.flipper}}
+          python3 scripts/testops.py -p=${{steps.device.outputs.flipper}} -t=15 await_flipper 
           rm -rf build/latest/resources/dolphin
           python3 scripts/storage.py -p ${{steps.device.outputs.flipper}} -f send build/latest/resources /ext
           python3 scripts/power.py -p ${{steps.device.outputs.flipper}} reboot
-          python3 scripts/testing/await_flipper.py ${{steps.device.outputs.flipper}}
+          python3 scripts/testops.py -p=${{steps.device.outputs.flipper}} -t=15 await_flipper 
 
       - name: 'Run units and validate results'
         id: run_units
@@ -61,7 +60,7 @@ jobs:
         timeout-minutes: 7
         run: |
           source scripts/toolchain/fbtenv.sh
-          python3 scripts/testing/units.py ${{steps.device.outputs.flipper}}
+          python3 scripts/testops.py run_units -p ${{steps.device.outputs.flipper}}
 
       - name: 'Check GDB output'
         if: failure() && steps.flashing.outcome == 'success'

.github/workflows/updater_test.yml

@@ -1,8 +1,6 @@
 name: 'Updater test'
-
 on:
   pull_request:
-
 env:
   TARGETS: f7
   DEFAULT_TARGET: f7
@@ -26,7 +24,7 @@ jobs:
       - name: 'Get flipper from device manager (mock)'
         id: device
         run: |
-          echo "flipper=Rekigyn" >> $GITHUB_OUTPUT
+          echo "flipper=auto" >> $GITHUB_OUTPUT
           echo "stlink=0F020D026415303030303032" >> $GITHUB_OUTPUT
 
       - name: 'Flashing target firmware'
@@ -35,7 +33,7 @@ jobs:
         run: |
           source scripts/toolchain/fbtenv.sh
           ./fbt flash_usb_full PORT=${{steps.device.outputs.flipper}} FORCE=1
-          python3 scripts/testing/await_flipper.py ${{steps.device.outputs.flipper}}
+          python3 scripts/testops.py -p=${{steps.device.outputs.flipper}} -t=120 await_flipper 
 
       - name: 'Validating updater'
         id: second_full_flash
@@ -44,7 +42,7 @@ jobs:
         run: |
           source scripts/toolchain/fbtenv.sh
           ./fbt flash_usb PORT=${{steps.device.outputs.flipper}} FORCE=1
-          python3 scripts/testing/await_flipper.py ${{steps.device.outputs.flipper}}
+          python3 scripts/testops.py -p=${{steps.device.outputs.flipper}} -t=120 await_flipper 
 
       - name: 'Get last release tag'
         id: release_tag
@@ -71,5 +69,5 @@ jobs:
         if: failure()
         run: |
           source scripts/toolchain/fbtenv.sh
-          python3 scripts/testing/await_flipper.py ${{steps.device.outputs.flipper}}
+          python3 scripts/testops.py -p=${{steps.device.outputs.flipper}} -t=120 await_flipper 
           python3 scripts/storage.py -p ${{steps.device.outputs.flipper}} format_ext

scripts/flipper/utils/cdc.py

@@ -15,4 +15,3 @@ def resolve_port(logger, portname: str = "auto"):
         logger.error("Failed to find connected Flipper")
     elif len(flippers) > 1:
         logger.error("More than one Flipper is attached")
-    logger.error("Failed to guess which port to use")

scripts/testing/await_flipper.py

@@ -1,61 +0,0 @@
-#!/usr/bin/env python3
-import logging
-import os
-import sys
-import time
-
-
-def flp_serial_by_name(flp_name):
-    if sys.platform == "darwin":  # MacOS
-        flp_serial = "/dev/cu.usbmodemflip_" + flp_name + "1"
-        logging.info(f"Darwin, looking for {flp_serial}")
-    elif sys.platform == "linux":  # Linux
-        flp_serial = (
-            "/dev/serial/by-id/usb-Flipper_Devices_Inc._Flipper_"
-            + flp_name
-            + "_flip_"
-            + flp_name
-            + "-if00"
-        )
-        logging.info(f"linux, looking for {flp_serial}")
-
-    if os.path.exists(flp_serial):
-        return flp_serial
-    else:
-        logging.info(f"Couldn't find {flp_name} on this attempt.")
-        if os.path.exists(flp_name):
-            return flp_name
-        else:
-            return ""
-
-
-UPDATE_TIMEOUT = 30 * 4  # 4 minutes
-
-
-def main():
-    flipper_name = sys.argv[1]
-    elapsed = 0
-    flipper = flp_serial_by_name(flipper_name)
-    logging.basicConfig(
-        format="%(asctime)s %(levelname)-8s %(message)s",
-        level=logging.INFO,
-        datefmt="%Y-%m-%d %H:%M:%S",
-    )
-    logging.info(f"Waiting for Flipper {flipper_name} to be ready...")
-
-    while flipper == "" and elapsed < UPDATE_TIMEOUT:
-        elapsed += 1
-        time.sleep(1)
-        flipper = flp_serial_by_name(flipper_name)
-
-    if flipper == "":
-        logging.error("Flipper not found!")
-        exit(1)
-
-    logging.info(f"Found Flipper at {flipper}")
-
-    sys.exit(0)
-
-
-if __name__ == "__main__":
-    main()

scripts/testing/units.py

@@ -1,87 +0,0 @@
-#!/usr/bin/env python3
-import logging
-import re
-import sys
-
-import serial
-from await_flipper import flp_serial_by_name
-
-
-def main():
-    logging.basicConfig(
-        format="%(asctime)s %(levelname)-8s %(message)s",
-        level=logging.INFO,
-        datefmt="%Y-%m-%d %H:%M:%S",
-    )
-    logging.info("Trying to run units on flipper")
-    flp_serial = flp_serial_by_name(sys.argv[1])
-
-    if flp_serial == "":
-        logging.error("Flipper not found!")
-        sys.exit(1)
-
-    with serial.Serial(flp_serial, timeout=150) as flipper:
-        logging.info(f"Found Flipper at {flp_serial}")
-        flipper.baudrate = 230400
-        flipper.flushOutput()
-        flipper.flushInput()
-
-        flipper.read_until(b">: ").decode("utf-8")
-        flipper.write(b"unit_tests\r")
-        data = flipper.read_until(b">: ").decode("utf-8")
-
-        lines = data.split("\r\n")
-
-        tests_re = r"Failed tests: \d{0,}"
-        time_re = r"Consumed: \d{0,}"
-        leak_re = r"Leaked: \d{0,}"
-        status_re = r"Status: \w{3,}"
-
-        tests_pattern = re.compile(tests_re)
-        time_pattern = re.compile(time_re)
-        leak_pattern = re.compile(leak_re)
-        status_pattern = re.compile(status_re)
-
-        tests, time, leak, status = None, None, None, None
-        total = 0
-
-        for line in lines:
-            logging.info(line)
-            if "()" in line:
-                total += 1
-
-            if not tests:
-                tests = re.match(tests_pattern, line)
-            if not time:
-                time = re.match(time_pattern, line)
-            if not leak:
-                leak = re.match(leak_pattern, line)
-            if not status:
-                status = re.match(status_pattern, line)
-
-        if None in (tests, time, leak, status):
-            logging.error(f"Failed to parse output: {leak} {time} {leak} {status}")
-            sys.exit(1)
-
-        leak = int(re.findall(r"[- ]\d+", leak.group(0))[0])
-        status = re.findall(r"\w+", status.group(0))[1]
-        tests = int(re.findall(r"\d+", tests.group(0))[0])
-        time = int(re.findall(r"\d+", time.group(0))[0])
-
-        if tests > 0 or status != "PASSED":
-            logging.error(f"Got {tests} failed tests.")
-            logging.error(f"Leaked (not failing on this stat): {leak}")
-            logging.error(f"Status: {status}")
-            logging.error(f"Time: {time/1000} seconds")
-            sys.exit(1)
-
-        logging.info(f"Leaked (not failing on this stat): {leak}")
-        logging.info(
-            f"Tests ran successfully! Time elapsed {time/1000} seconds. Passed {total} tests."
-        )
-
-        sys.exit(0)
-
-
-if __name__ == "__main__":
-    main()

scripts/testops.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+
+import re
+import sys
+import time
+from typing import Optional
+
+from flipper.app import App
+from flipper.storage import FlipperStorage
+from flipper.utils.cdc import resolve_port
+
+
+class Main(App):
+    # this is basic use without sub-commands, simply to reboot flipper / power it off, not meant as a full CLI wrapper
+    def init(self):
+        self.parser.add_argument("-p", "--port", help="CDC Port", default="auto")
+        self.parser.add_argument(
+            "-t", "--timeout", help="Timeout in seconds", type=int, default=10
+        )
+
+        self.subparsers = self.parser.add_subparsers(help="sub-command help")
+
+        self.parser_await_flipper = self.subparsers.add_parser(
+            "await_flipper", help="Wait for Flipper to connect or reconnect"
+        )
+        self.parser_await_flipper.set_defaults(func=self.await_flipper)
+
+        self.parser_run_units = self.subparsers.add_parser(
+            "run_units", help="Run unit tests and post result"
+        )
+        self.parser_run_units.set_defaults(func=self.run_units)
+
+    def _get_flipper(self, retry_count: Optional[int] = 1):
+        port = None
+        self.logger.info(f"Attempting to find flipper with {retry_count} attempts.")
+
+        for i in range(retry_count):
+            self.logger.info(f"Attempt to find flipper #{i}.")
+
+            if port := resolve_port(self.logger, self.args.port):
+                self.logger.info(f"Found flipper at {port}")
+                break
+            time.sleep(1)
+
+        if not port:
+            self.logger.info(f"Failed to find flipper {port}")
+            return None
+
+        flipper = FlipperStorage(port)
+        flipper.start()
+        return flipper
+
+    def await_flipper(self):
+        if not (flipper := self._get_flipper(retry_count=self.args.timeout)):
+            return 1
+
+        self.logger.info("Flipper started")
+        flipper.stop()
+        return 0
+
+    def run_units(self):
+        if not (flipper := self._get_flipper(retry_count=10)):
+            return 1
+
+        self.logger.info("Running unit tests")
+        flipper.send("unit_tests" + "\r")
+        self.logger.info("Waiting for unit tests to complete")
+        data = flipper.read.until(">: ")
+        self.logger.info("Parsing result")
+
+        lines = data.decode().split("\r\n")
+
+        tests_re = r"Failed tests: \d{0,}"
+        time_re = r"Consumed: \d{0,}"
+        leak_re = r"Leaked: \d{0,}"
+        status_re = r"Status: \w{3,}"
+
+        tests_pattern = re.compile(tests_re)
+        time_pattern = re.compile(time_re)
+        leak_pattern = re.compile(leak_re)
+        status_pattern = re.compile(status_re)
+
+        tests, elapsed_time, leak, status = None, None, None, None
+        total = 0
+
+        for line in lines:
+            self.logger.info(line)
+            if "()" in line:
+                total += 1
+
+            if not tests:
+                tests = re.match(tests_pattern, line)
+            if not elapsed_time:
+                elapsed_time = re.match(time_pattern, line)
+            if not leak:
+                leak = re.match(leak_pattern, line)
+            if not status:
+                status = re.match(status_pattern, line)
+
+        if None in (tests, elapsed_time, leak, status):
+            self.logger.error(
+                f"Failed to parse output: {tests} {elapsed_time} {leak} {status}"
+            )
+            sys.exit(1)
+
+        leak = int(re.findall(r"[- ]\d+", leak.group(0))[0])
+        status = re.findall(r"\w+", status.group(0))[1]
+        tests = int(re.findall(r"\d+", tests.group(0))[0])
+        elapsed_time = int(re.findall(r"\d+", elapsed_time.group(0))[0])
+
+        if tests > 0 or status != "PASSED":
+            self.logger.error(f"Got {tests} failed tests.")
+            self.logger.error(f"Leaked (not failing on this stat): {leak}")
+            self.logger.error(f"Status: {status}")
+            self.logger.error(f"Time: {elapsed_time/1000} seconds")
+            flipper.stop()
+            return 1
+
+        self.logger.info(f"Leaked (not failing on this stat): {leak}")
+        self.logger.info(
+            f"Tests ran successfully! Time elapsed {elapsed_time/1000} seconds. Passed {total} tests."
+        )
+
+        flipper.stop()
+        return 0
+
+
+if __name__ == "__main__":
+    Main()()