Merge branch 'develop' into feature/error_prints

Jenkinsfile

@@ -1,5 +1,27 @@
 @Library('[email protected]') _
 
+def runningOn(machine) {
+  println "Stage running on:"
+  println machine
+}
+
+def checkSkipLink() {
+    def skip_linkcheck = ""
+    if (env.GH_LABEL_ALL.contains("skip_linkcheck")) {
+        println "skip_linkcheck set, skipping link check..."
+        skip_linkcheck = "clean html pdf"
+    }
+    return skip_linkcheck
+}
+
+def buildDocs(String zipFileName) {
+  withVenv {
+    sh 'pip install git+ssh://[email protected]/xmos/[email protected]'
+    sh "xmosdoc ${checkSkipLink()}"
+    zip zipFile: zipFileName, archive: true, dir: "doc/_build"
+  }
+}
+
 getApproval()
 
 pipeline {
@@ -33,24 +55,16 @@ pipeline {
         stage('Build and Docs') {
             parallel {
                 stage('Build Docs') {
-                    agent { label "docker" }
-                    environment { XMOSDOC_VERSION = "v5.0" }
+                    agent { label "documentation" }
                     steps {
-                        script {
-                            def skip_linkcheck = ""
-                            if (env.GH_LABEL_ALL.contains("skip_linkcheck")) {
-                                println "skip_linkcheck set, skipping link check..."
-                                skip_linkcheck = "clean html pdf"
-                            }
+                        runningOn(env.NODE_NAME)
+                        dir('fwk_rtos'){
                             checkout scm
-                            sh 'git submodule update --init --recursive --depth 1'
-                            sh "docker pull ghcr.io/xmos/xmosdoc:$XMOSDOC_VERSION"
-                            sh """docker run -u "\$(id -u):\$(id -g)" \
-                                --rm \
-                                -v ${WORKSPACE}:/build \
-                                ghcr.io/xmos/xmosdoc:$XMOSDOC_VERSION -v $skip_linkcheck"""
-                            archiveArtifacts artifacts: "doc/_build/**", allowEmptyArchive: true
-                        } // script
+                            createVenv()
+                            withTools(params.TOOLS_VERSION) {
+                                buildDocs("fwk_rtos_docs.zip")
+                            } // withTools
+                        } // dir
                     } // steps
                     post {
                         cleanup {
@@ -69,6 +83,7 @@ pipeline {
                     stages {
                         stage('Checkout') {
                             steps {
+                                runningOn(env.NODE_NAME)
                                 checkout scm
                                 sh 'git submodule update --init --recursive --depth 1 --jobs \$(nproc)'
                             }
@@ -161,10 +176,8 @@ pipeline {
                                 withTools(params.TOOLS_VERSION) {
                                     withVenv {
                                         script {
-                                            uid = sh(returnStdout: true, script: 'id -u').trim()
-                                            gid = sh(returnStdout: true, script: 'id -g').trim()
                                             withXTAG(["$RTOS_TEST_RIG_TARGET"]) { adapterIDs ->
-                                                sh "docker run --rm -u $uid:$gid --privileged -v /dev:/dev -w /fwk_rtos -v $WORKSPACE:/fwk_rtos ghcr.io/xmos/xcore_voice_tester:develop bash -l test/rtos_drivers/usb/check_usb.sh " + adapterIDs[0]
+                                                sh "bash -l test/rtos_drivers/usb/check_usb.sh " + adapterIDs[0]
                                             }
                                             sh "pytest test/rtos_drivers/usb"
                                         }

test/rtos_drivers/usb/check_usb.sh

@@ -51,6 +51,7 @@ DFU_RT_VID_PID=""
 DFU_MODE_VID_PID="20b1:4000"
 DFU_DOWNLOAD_BYTES=1024
 DFU_ALT=1
+DFU_SERIAL="123456"
 DFU_VERBOSITY="-e" # Set to "-v -v -v" libusb debug prints.
 APP_STARTUP_TIME_S=10
 APP_SHUTDOWN_TIME_S=5
@@ -150,6 +151,7 @@ function wait_for_lsusb_entry {
 }
 
 function run_cdc_tests {
+    dfu-util --list
     print_and_log_test_step "Writing CDC test data on both interfaces."
     ($TIMEOUT ${PY_TIMEOUT_S}s python3 "$REPO_ROOT/test/rtos_drivers/usb/serial_send_receive.py" -if0 "$SERIAL_TX0_FILE" -if1 "$SERIAL_TX1_FILE" -of0 "$SERIAL_RX0_FILE" -of1 "$SERIAL_RX1_FILE" 2>&1) >> "$HOST_REPORT"
     exit_code=$?
@@ -184,33 +186,33 @@ function run_dfu_tests {
     # First determine the state of the image currently on the target, then download
     # the test image, and then read it back a final time to verify that it changed.
     # Finally issue a DFU detach, to allow the device to terminate its application.
-
+    dfu-util --list # debug info
     print_and_log_test_step "Reading DFU initial state of device's test partition."
-    dfu-util $DFU_VERBOSITY -d "$DFU_RT_VID_PID,$DFU_MODE_VID_PID" -a $DFU_ALT -U "$FILE_PRE_DOWNLOAD" >> "$HOST_REPORT" 2>&1
+    dfu-util $DFU_VERBOSITY -d "$DFU_RT_VID_PID,$DFU_MODE_VID_PID" -a $DFU_ALT --serial="$DFU_SERIAL" -U "$FILE_PRE_DOWNLOAD" >> "$HOST_REPORT" 2>&1
     exit_code=$?
     if [ $exit_code -ne 0 ]; then
         print_and_log_failure "An error occurred during upload of pre-test image (exit code = $exit_code)."
         return 1
     fi
 
     print_and_log_test_step "Writing DFU test image to target's test partition."
-    dfu-util $DFU_VERBOSITY -d "$DFU_RT_VID_PID,$DFU_MODE_VID_PID" -a $DFU_ALT -D "$FILE_TO_DOWNLOAD" >> "$HOST_REPORT" 2>&1
+    dfu-util $DFU_VERBOSITY -d "$DFU_RT_VID_PID,$DFU_MODE_VID_PID" -a $DFU_ALT --serial="$DFU_SERIAL" -D "$FILE_TO_DOWNLOAD" >> "$HOST_REPORT" 2>&1
     exit_code=$?
     if [ $exit_code -ne 0 ]; then
         print_and_log_failure "An error occurred during download of test image (exit code = $exit_code)."
         return 1
     fi
 
     print_and_log_test_step "Reading back DFU test image from the device's test partition."
-    dfu-util $DFU_VERBOSITY -d "$DFU_RT_VID_PID,$DFU_MODE_VID_PID" -a $DFU_ALT -U "$FILE_POST_DOWNLOAD" >> "$HOST_REPORT" 2>&1
+    dfu-util $DFU_VERBOSITY -d "$DFU_RT_VID_PID,$DFU_MODE_VID_PID" -a $DFU_ALT --serial="$DFU_SERIAL" -U "$FILE_POST_DOWNLOAD" >> "$HOST_REPORT" 2>&1
     exit_code=$?
     if [ $exit_code -ne 0 ]; then
         print_and_log_failure "An error occurred during upload of post-test image (exit code = $exit_code)."
         return 1
     fi
 
     print_and_log_test_step "Issuing DFU detach request."
-    dfu-util $DFU_VERBOSITY -e -d "$DFU_RT_VID_PID,$DFU_MODE_VID_PID" -a $DFU_ALT >> "$HOST_REPORT" 2>&1
+    dfu-util $DFU_VERBOSITY -e -d "$DFU_RT_VID_PID,$DFU_MODE_VID_PID" -a $DFU_ALT --serial="$DFU_SERIAL" >> "$HOST_REPORT" 2>&1
     exit_code=$?
     if [ $exit_code -ne 0 ]; then
         print_and_log_failure "An error occurred during DFU detach request (exit code = $exit_code)."

test/rtos_drivers/usb/serial_send_receive.py

@@ -12,6 +12,9 @@
 pid=0x4000
 
 max_read_size=4096
+required_ports = 2
+
+N_PORTS_ERROR_MSG = "We need at least 2 COM ports to perform the tests"
 
 # example run
 # python serial_send_receive.py -if0 tx_data0 -if1 tx_data1 -of0 rx_data0 -of1 rx_data1
@@ -28,51 +31,47 @@ def parse_arguments():
 
     return args
 
-def main(if0, if1, of0, of1):
-
+def find_ports_by_vid_pid(rq=2, msg="COM ports missing"):
     all_ports = serial.tools.list_ports.comports()
-    test_ports = []
-    required_ports = 2
-
-    for port in all_ports:
-        if port.vid == vid and port.pid == pid:
-            test_ports.append(port)
-
-    if len(test_ports) != required_ports:
-        print(f'Error: Expected {required_ports} serial ports, found { len(test_ports) }.')
-        sys.exit(1)
+    test_ports = [port for port in all_ports if port.vid == vid and port.pid == pid]
+    assert (len(test_ports) >= rq), msg
+    return test_ports
+
+def transfer_data(input_file, output_file, write_port, read_port, max_read_size):
+    with open(input_file, 'rb') as in_file, open(output_file, 'wb') as out_file:
+        while True:
+            tx_data = in_file.read(max_read_size)
+            if not tx_data:
+                break
+            write_port.write(tx_data)
+            out_file.write(read_port.read(len(tx_data)))
 
+def main(if0, if1, of0, of1):
+
+    test_ports = find_ports_by_vid_pid(required_ports, N_PORTS_ERROR_MSG)
     port0 = None
     port1 = None
-
+        
     try:
-        print(f'Opening port 0 ({test_ports[0].device}).')
-        port0 = serial.Serial(test_ports[0].device)
-        print(f'Opening port 1 ({test_ports[1].device}).')
-        port1 = serial.Serial(test_ports[1].device)
-
-        print('Writing CDC test data to port 0, reading from port 1.')
-        with open(if0, 'rb') as in_file, open(of1, 'wb') as out_file:
-            while 1:
-                tx_data = in_file.read(max_read_size)
+        d0 = test_ports[0].device
+        d1 = test_ports[1].device
 
-                if not tx_data:
-                    break
+        print(f'Opening port 0 ({d0}).')
+        port0 = serial.Serial(d0)
+
+        print(f'Opening port 1 ({d1}).')
+        port1 = serial.Serial(d1)
 
-                port0.write(tx_data)
-                out_file.write(port1.read(len(tx_data)))
+        print('Writing CDC test data to port 0, reading from port 1.')
+        transfer_data(if0, of1, port0, port1, max_read_size)
 
         print('Writing CDC test data to port 1, reading from port 0.')
-        with open(if1, 'rb') as in_file, open(of0, 'wb') as out_file:
-            while 1:
-                tx_data = in_file.read(max_read_size)
-
-                if not tx_data:
-                    break
-
-                port1.write(tx_data)
-                out_file.write(port0.read(len(tx_data)))
+        transfer_data(if1, of0, port1, port0, max_read_size)
 
+    except Exception as e:
+        print(e)
+        sys.exit(1)
+
     finally:
         if port0 is not None:
             print('Closing port 0.')

test/rtos_drivers/usb/test_verify_usb_results.py

@@ -3,39 +3,43 @@
 # This Software is subject to the terms of the XMOS Public Licence: Version 1.
 
 import re
+from pathlib import Path
 
-target_test_results_filename = "testing/target.rpt"
-target_test_regex = r"^Tile\[(\d{1})\]\|FCore\[(\d{1})\]\|(\d+)\|TEST\|(\w{4}) USB$"
+top_level = Path(__file__).parents[3].absolute() # if changes, make sure this relative path is correct
 
-host_test_results_filename = "testing/host.rpt"
+target_test_results_filename = top_level / "testing/target.rpt"
+target_test_regex = r"^Tile\[\d\]\|FCore\[\d\]\|\d+\|TEST\|PASS USB$"
+target_test_required_count = 2
+
+host_test_results_filename = top_level / "testing/host.rpt"
 host_test_regex = r"^\[TEST PASS\]:"
+host_test_required_count = 1
 
-def test_results():
-    with open(target_test_results_filename, "r") as f:
-        cnt = 0
-        while 1:
-            line = f.readline()
 
-            if not line:
-                assert cnt == 2 # each tile should report PASS
-                break
+def validate_results(filename, regex, required_count):
+    """Counts the number of matching elements in a file given a regex
+    and ensures mathes the required count, will raise an exception if not. 
 
-            p = re.match(target_test_regex, line)
+    Args:
+        filename (str): path to the file
+        regex (regex): regex expression
+        required_count (int): number of elements expected
+    """
+    with open(filename, "r") as file:
+        content = file.read()
+
+    matches = re.findall(regex, content, re.MULTILINE)
+    pass_count = len(matches)
 
-            if p:
-                cnt += 1
-                assert p.group(4).find("PASS") != -1
+    assert pass_count == required_count, f"Expected {required_count} passes, but found {pass_count}"
+    return pass_count
 
-    with open(host_test_results_filename, "r") as f:
-        cnt = 0
-        while 1:
-            line = f.readline()
 
-            if not line:
-                assert cnt == 1
-                break
+def test_results():
+    validate_results(target_test_results_filename, target_test_regex, target_test_required_count)
+    validate_results(host_test_results_filename, host_test_regex, host_test_required_count)
 
-            p = re.match(host_test_regex, line)
 
-            if p:
-                cnt += 1
+if __name__ == "__main__":
+    test_results() # manual debug
+