update existing tests to updated test framework

docs/testing.md

@@ -63,15 +63,17 @@ If your test rig differs, you may be able to run some but not all of the tests.
  4. Remove the following arguments from `/boot/cmdline.txt`:
     - `console=serial0,115200`
 
- 5. Reboot.
+ 5. Append `ODRIVE_TEST_RIG_NAME=[test-rig-name]` to `/etc/environment`. The HWIL tests use this to look up the the file `[test-rig-name].yaml` which is supposed to describe your test rig.
 
- 6. Install the prerequisites:
+ 6. Reboot.
+
+ 7. Install the prerequisites:
 
         sudo apt-get install ipython3 python3-appdirs python3-yaml python3-usb python3-serial python3-can python3-scipy python3-matplotlib python3-ipdb git openocd
         # Optionally, to be able to compile the firmware:
         sudo apt-get install gcc-arm-none-eabi
 
- 7. Install Teensyduino and teensy-loader-cli:
+ 8. Install Teensyduino and teensy-loader-cli:
 
         sudo apt-get install libfontconfig libxft2 libusb-dev
 
@@ -91,20 +93,20 @@ If your test rig differs, you may be able to run some but not all of the tests.
         popd
         curl https://www.pjrc.com/teensy/49-teensy.rules | sudo tee /etc/udev/rules.d/49-teensy.rules
 
- 8. Add the following lines to `/etc/udev/rules.d/49-stlinkv2.rules`:
+ 9. Add the following lines to `/etc/udev/rules.d/49-stlinkv2.rules`:
 
         SUBSYSTEMS=="usb", ATTRS{idVendor}=="0483", ATTRS{idProduct}=="374b", MODE:="0666"
         SUBSYSTEMS=="usb", ATTRS{idVendor}=="0483", ATTRS{idProduct}=="3748", MODE:="0666"
 
- 9.  `sudo mkdir /opt/odrivetest && sudo chown $USER /opt/odrivetest`
+ 10. `sudo mkdir /opt/odrivetest && sudo chown $USER /opt/odrivetest`
 
- 10. At this point you need the ODrive repository. See next section to sync it from your main PC. We assume now that you navigated to `tools/odrive/tests/`.
+ 11. At this point you need the ODrive repository. See next section to sync it from your main PC. We assume now that you navigated to `tools/odrive/tests/`.
 
- 11. `sudo ../../odrivetool udev-setup`
+ 12. `sudo ../../odrivetool udev-setup`
 
- 12. `sudo udevadm trigger`
+ 13. `sudo udevadm trigger`
 
- 13. Run once after every reboot: `sudo ipython3 --pdb test_runner.py -- --setup-host --test-rig-yaml ../../test-rig-rpi.yaml`
+ 14. Run once after every reboot: `sudo -E ipython3 --pdb test_runner.py -- --setup-host`
 
 ## SSH testing flow
 
@@ -125,5 +127,5 @@ To run a test:
 
     ssh odrv
     > cd /opt/odrivetest/tools/odrive/tests/
-    > ipython3 --pdb uart_ascii_test.py -- --test-rig-yaml ../../test-rig-rpi.yaml
+    > ipython3 --pdb uart_ascii_test.py
 

tools/odrive/tests/analog_input_test.py

@@ -56,17 +56,16 @@ class TestAnalogInput():
     def get_test_cases(self, testrig: TestRig):
         for odrive in testrig.get_components(ODriveComponent):
             for odrive_gpio_num, odrive_gpio in [(2, odrive.gpio3), (3, odrive.gpio4)]:
-                analog_out_options = []
-                lpf_gpio = [gpio for lpf in testrig.get_connected_components(odrive_gpio, LowPassFilterComponent)
-                                 for gpio in testrig.get_connected_components(lpf.en, LinuxGpioComponent)]
-                for teensy_gpio in testrig.get_connected_components(odrive_gpio, TeensyGpio):
-                    teensy = teensy_gpio.parent
-                    analog_reset_options = []
-                    for gpio in teensy.gpios:
-                        for local_gpio in testrig.get_connected_components(gpio, LinuxGpioComponent):
-                            analog_reset_options.append((gpio, local_gpio))
-                    analog_out_options.append((teensy, teensy_gpio, analog_reset_options))
-                yield (odrive, lpf_gpio, odrive_gpio_num, analog_out_options)
+                alternatives = []
+                lpfs = [(gpio, TestFixture.all_of(tf1, tf2)) for lpf, tf1 in testrig.get_connected_components(odrive_gpio, LowPassFilterComponent)
+                                                             for gpio, tf2 in testrig.get_connected_components(lpf.en, LinuxGpioComponent)]
+                for lpf, tf1 in lpfs:
+                    for teensy_gpio, tf2 in testrig.get_connected_components(odrive_gpio, TeensyGpio):
+                        teensy = teensy_gpio.parent
+                        for gpio in teensy.gpios:
+                            for local_gpio, tf3 in testrig.get_connected_components(gpio, LinuxGpioComponent):
+                                alternatives.append([odrive, lpf, odrive_gpio_num, teensy, teensy_gpio, gpio, local_gpio, TestFixture.all_of(tf1, tf2, tf3)])
+                yield AnyTestCase(*alternatives)
 
 
     def run_test(self, odrive: ODriveComponent, lpf_enable: LinuxGpioComponent, analog_in_num: int, teensy: TeensyComponent, teensy_analog_out: Component, teensy_analog_reset: Component, analog_reset_gpio: LinuxGpioComponent, logger: Logger):

tools/odrive/tests/calibration_test.py

@@ -18,10 +18,9 @@ class TestMotorCalibration():
 
     def get_test_cases(self, testrig: TestRig):
         """Returns all axes that are connected to a motor, along with the corresponding motor(s)"""
-        for odrive in testrig.get_components(ODriveComponent):
-            for axis in odrive.axes:
-                for motor in testrig.get_connected_components(axis, MotorComponent):
-                    yield (axis, motor)
+        for axis in testrig.get_components(ODriveAxisComponent):
+            for motor, tf in testrig.get_connected_components({'phases': axis}, MotorComponent):
+                yield (axis, motor, tf)
 
     def run_test(self, axis_ctx: ODriveAxisComponent, motor_ctx: MotorComponent, logger: Logger):
         # reset old calibration values
@@ -57,10 +56,9 @@ class TestDisconnectedMotorCalibration():
 
     def get_test_cases(self, testrig: TestRig):
         """Returns all axes that are disconnected"""
-        for odrive in testrig.get_components(ODriveComponent):
-            for axis in odrive.axes:
-                if axis.yaml == 'floating':
-                    yield (axis,)
+        for axis in testrig.get_components(ODriveAxisComponent):
+            if axis.yaml == 'floating':
+                yield (axis, None)
 
     def run_test(self, axis_ctx: ODriveAxisComponent, logger: Logger):
         axis = axis_ctx.handle
@@ -85,17 +83,7 @@ class TestEncoderDirFind():
     """
 
     def get_test_cases(self, testrig: TestRig):
-        for odrive in testrig.get_components(ODriveComponent):
-            for num in range(len(odrive.axes)):
-                encoders = testrig.get_connected_components({
-                    'a': (odrive.encoders[num].a, False),
-                    'b': (odrive.encoders[num].b, False)
-                }, EncoderComponent)
-                motors = testrig.get_connected_components(odrive.axes[num], MotorComponent)
-
-                for motor, encoder in itertools.product(motors, encoders):
-                    if encoder.impl in testrig.get_connected_components(motor):
-                        yield (odrive.axes[num], motor, encoder)
+        return testrig.get_closed_loop_combos(init=False)
 
     def run_test(self, axis_ctx: ODriveAxisComponent, motor_ctx: MotorComponent, enc_ctx: EncoderComponent, logger: Logger):
         axis = axis_ctx.handle
@@ -129,17 +117,7 @@ class TestEncoderOffsetCalibration():
     """
 
     def get_test_cases(self, testrig: TestRig):
-        for odrive in testrig.get_components(ODriveComponent):
-            for num in range(len(odrive.axes)):
-                encoders = testrig.get_connected_components({
-                    'a': (odrive.encoders[num].a, False),
-                    'b': (odrive.encoders[num].b, False)
-                }, EncoderComponent)
-                motors = testrig.get_connected_components(odrive.axes[num], MotorComponent)
-
-                for motor, encoder in itertools.product(motors, encoders):
-                    if encoder.impl in testrig.get_connected_components(motor):
-                        yield (odrive.axes[num], motor, encoder)
+        return testrig.get_closed_loop_combos(init=False)
 
     def run_test(self, axis_ctx: ODriveAxisComponent, motor_ctx: MotorComponent, enc_ctx: EncoderComponent, logger: Logger):
         axis = axis_ctx.handle
@@ -178,18 +156,11 @@ class TestEncoderIndexSearch():
     """
 
     def get_test_cases(self, testrig: TestRig):
-        for odrive in testrig.get_components(ODriveComponent):
-            for num in range(len(odrive.axes)):
-                encoders = testrig.get_connected_components({
-                    'a': (odrive.encoders[num].a, False),
-                    'b': (odrive.encoders[num].b, False)
-                }, EncoderComponent)
-                motors = testrig.get_connected_components(odrive.axes[num], MotorComponent)
-                z_gpio = list(testrig.get_connected_components((odrive.encoders[num].z, False), LinuxGpioComponent))
-
-                for motor, encoder in itertools.product(motors, encoders):
-                    if encoder.impl in testrig.get_connected_components(motor):
-                        yield (odrive.axes[num], motor, encoder, z_gpio)
+        for axis, motor, encoder, tf1 in testrig.get_closed_loop_combos(init=False):
+            alternatives = []
+            for z_gpio, tf2 in testrig.get_connected_components((axis.parent.encoders[axis.num].z, False), LinuxGpioComponent):
+                alternatives.append((axis, motor, encoder, z_gpio, TestFixture.all_of(tf1, tf2)))
+            yield AnyTestCase(*alternatives)
 
     def run_test(self, axis_ctx: ODriveAxisComponent, motor_ctx: MotorComponent, enc_ctx: EncoderComponent, z_gpio: LinuxGpioComponent, logger: Logger):
         axis = axis_ctx.handle

tools/odrive/tests/can_test.py

@@ -104,8 +104,8 @@ class TestSimpleCAN():
     def get_test_cases(self, testrig: TestRig):
         for odrive in testrig.get_components(ODriveComponent):
             can_interfaces = list(testrig.get_connected_components(odrive.can, CanInterfaceComponent))
-            yield (odrive, can_interfaces, 0, False) # standard ID
-            yield (odrive, can_interfaces, 0xfedcba, True) # extended ID
+            yield AnyTestCase(*[(odrive, intf, 0, False, tf) for intf, tf in can_interfaces]) # standard ID
+            yield AnyTestCase(*[(odrive, intf, 0xfedcba, True, tf) for intf, tf in can_interfaces]) # extended ID
 
     def run_test(self, odrive: ODriveComponent, canbus: CanInterfaceComponent, node_id: int, extended_id: bool, logger: Logger):
         odrive.disable_mappings()

tools/odrive/tests/encoder_test.py

@@ -107,12 +107,12 @@ class TestIncrementalEncoder(TestEncoderBase):
                 ]
 
                 valid_combinations = [
-                    (combination[0].parent,) + tuple(combination)
+                    (encoder, combination[0].parent,) + tuple(combination) + (None,)
                     for combination in itertools.product(*gpio_conns)
                     if ((len(set(c.parent for c in combination)) == 1) and isinstance(combination[0].parent, TeensyComponent))
                 ]
 
-                yield (encoder, valid_combinations)
+                yield AnyTestCase(*valid_combinations)
 
 
     def run_test(self, enc: ODriveEncoderComponent, teensy: TeensyComponent, teensy_gpio_a: TeensyGpio, teensy_gpio_b: TeensyGpio, logger: Logger):
@@ -167,12 +167,12 @@ class TestSinCosEncoder(TestEncoderBase):
             ]
 
             valid_combinations = [
-                (combination[0].parent,) + tuple(combination)
+                (odrive.encoders[0], combination[0].parent,) + tuple(combination) + (None,)
                 for combination in itertools.product(*gpio_conns)
                 if ((len(set(c.parent for c in combination)) == 1) and isinstance(combination[0].parent, TeensyComponent))
             ]
 
-            yield (odrive.encoders[0], valid_combinations)
+            yield AnyTestCase(*valid_combinations)
 
 
     def run_test(self, enc: ODriveEncoderComponent, teensy: TeensyComponent, teensy_gpio_sin: TeensyGpio, teensy_gpio_cos: TeensyGpio, logger: Logger):
@@ -236,12 +236,12 @@ class TestHallEffectEncoder(TestEncoderBase):
                 ]
 
                 valid_combinations = [
-                    (combination[0].parent,) + tuple(combination)
+                    (encoder, combination[0].parent,) + tuple(combination) + (None,)
                     for combination in itertools.product(*gpio_conns)
                     if ((len(set(c.parent for c in combination)) == 1) and isinstance(combination[0].parent, TeensyComponent))
                 ]
 
-                yield (encoder, valid_combinations)
+                yield AnyTestCase(*valid_combinations)
 
 
     def run_test(self, enc: ODriveEncoderComponent, teensy: TeensyComponent, teensy_gpio_a: TeensyGpio, teensy_gpio_b: TeensyGpio, teensy_gpio_c: TeensyGpio, logger: Logger):
@@ -413,28 +413,24 @@ class TestSpiEncoder(TestEncoderBase):
         self.mode = mode
 
     def get_test_cases(self, testrig: TestRig):
-        for odrive in testrig.get_components(ODriveComponent):
-            for encoder in odrive.encoders:
-                odrive_ncs_gpio = odrive.gpio7 # this GPIO choice is completely arbitrary
+        for encoder in testrig.get_components(ODriveEncoderComponent):
+            odrive = encoder.parent
+            odrive_ncs_gpio = odrive.gpio7 # this GPIO choice is completely arbitrary
+
+            for teensy in testrig.get_components(TeensyComponent):
                 gpio_conns = [
-                    testrig.get_connected_components(odrive.sck, TeensyGpio),
-                    testrig.get_connected_components(odrive.miso, TeensyGpio),
-                    testrig.get_connected_components(odrive.mosi, TeensyGpio),
-                    testrig.get_connected_components(odrive_ncs_gpio, TeensyGpio),
+                    testrig.net_by_component.get(odrive.sck, set()).intersection(set(teensy.gpios)),
+                    testrig.net_by_component.get(odrive.miso, set()).intersection(set(teensy.gpios)),
+                    testrig.net_by_component.get(odrive.mosi, set()).intersection(set(teensy.gpios)),
+                    testrig.net_by_component.get(odrive_ncs_gpio, set()).intersection(set(teensy.gpios)),
+                    teensy.gpios
                 ]
 
-                valid_combinations = []
-                for combination in itertools.product(*gpio_conns):
-                    if (len(set(c.parent for c in combination)) != 1):
-                        continue
-                    teensy = combination[0].parent
-                    reset_pin_options = []
-                    for gpio in teensy.gpios:
-                        for local_gpio in testrig.get_connected_components(gpio, LinuxGpioComponent):
-                            reset_pin_options.append((gpio, local_gpio))
-                    valid_combinations.append((teensy, *combination, reset_pin_options))
-
-                yield (encoder, 7, valid_combinations)
+                alternatives = []
+                for gpio1, gpio2, gpio3, gpio4, gpio5 in itertools.product(*gpio_conns):
+                    for local_reset_gpio, tf in testrig.get_connected_components(gpio5, LinuxGpioComponent):
+                        alternatives.append((encoder, 7, teensy, gpio1, gpio2, gpio3, gpio4, gpio5, local_reset_gpio, tf))
+                yield AnyTestCase(*alternatives)
 
 
     def run_test(self, enc: ODriveEncoderComponent, odrive_ncs_gpio: int, teensy: TeensyComponent, teensy_gpio_sck: TeensyGpio, teensy_gpio_miso: TeensyGpio, teensy_gpio_mosi: TeensyGpio, teensy_gpio_ncs: TeensyGpio, teensy_gpio_reset: TeensyGpio, reset_gpio: LinuxGpioComponent, logger: Logger):

tools/odrive/tests/fibre_test.py

@@ -13,7 +13,7 @@ class FibreFunctionalTest():
     """
 
     def get_test_cases(self, testrig: TestRig):
-        return testrig.get_components(ODriveComponent)
+        return [(odrv, None) for odrv in testrig.get_components(ODriveComponent)]
 
     def run_test(self, odrive: ODriveComponent, logger: Logger):
         # Test property read/write
@@ -41,7 +41,7 @@ class FibreBurnInTest():
     """
 
     def get_test_cases(self, testrig: TestRig):
-        return testrig.get_components(ODriveComponent)
+        return [(odrv, None) for odrv in testrig.get_components(ODriveComponent)]
 
     def run_test(self, odrive: ODriveComponent, logger: Logger):
         data = record_log(lambda: [odrive.handle.vbus_voltage], duration=10.0)

tools/odrive/tests/integration_test.py

@@ -129,17 +129,11 @@ class TestSimpleCANClosedLoop():
 
 
     def get_test_cases(self, testrig: TestRig):
-        for odrive in testrig.get_components(ODriveComponent):
-            can_interfaces = list(testrig.get_connected_components(odrive.can, CanInterfaceComponent))
-            for num in range(2):
-                encoders = testrig.get_connected_components({
-                    'a': (odrive.encoders[num].a, False),
-                    'b': (odrive.encoders[num].b, False)
-                }, EncoderComponent)
-                motors = testrig.get_connected_components(odrive.axes[num], MotorComponent)
-                for motor, encoder in itertools.product(motors, encoders):
-                    if encoder.impl in testrig.get_connected_components(motor):
-                        yield (odrive, can_interfaces, odrive.axes[num], motor, encoder, 0, False)
+        for axis, motor, encoder, tf1 in testrig.get_closed_loop_combos(init=False):
+            yield AnyTestCase(*[
+                (axis.parent, canbus, axis, motor, encoder, 0, False, TestFixture.all_of(tf1, tf2))
+                for canbus, tf2 in testrig.get_connected_components(axis.parent.can, CanInterfaceComponent)
+            ])
 
     def run_test(self, odrive: ODriveComponent, canbus: CanInterfaceComponent, axis_ctx: ODriveAxisComponent, motor_ctx: MotorComponent, enc_ctx: EncoderComponent, node_id: int, extended_id: bool, logger: Logger):
         # this test is a sanity check to make sure that closed loop operation works

tools/odrive/tests/nvm_test.py

@@ -16,7 +16,7 @@ class TestStoreAndReboot():
 
     def get_test_cases(self, testrig: TestRig):
         for odrive in testrig.get_components(ODriveComponent):
-            yield (odrive,)
+            yield (odrive, None)
 
     def run_with_values(self, odrive: ODriveComponent, values: list, logger: Logger):
         logger.debug("storing configuration and rebooting...")

tools/odrive/tests/pwm_input_test.py

@@ -51,19 +51,22 @@ class TestPwmInput():
         for odrive in testrig.get_components(ODriveComponent):
             if odrive.yaml['board-version'].startswith('v3.'):
                 # Run a separate test for each PWM-capable GPIO. Use different min/max settings for each test.
-                yield (odrive, 1, -50, 200, list(testrig.get_connected_components(odrive.gpio1, TeensyGpio)))
-                yield (odrive, 2, 20, 400, list(testrig.get_connected_components(odrive.gpio2, TeensyGpio)))
-                yield (odrive, 3, -1000, 0, list(testrig.get_connected_components(odrive.gpio3, TeensyGpio)))
-                yield (odrive, 4, -20000, 20000, list(testrig.get_connected_components(odrive.gpio4, TeensyGpio)))
+                test_cases = [(1, -50, 200, odrive.gpio1),
+                              (2, 20, 400, odrive.gpio2),
+                              (3, -1000, 0, odrive.gpio3),
+                              (4, -20000, 20000, odrive.gpio4)]
             elif odrive.yaml['board-version'].startswith('v4.'):
                 # Run a separate test for each PWM-capable GPIO. Use different min/max settings for each test.
-                yield (odrive, 14, -50, 200, list(testrig.get_connected_components(odrive.gpio14, TeensyGpio)))
-                yield (odrive, 19, 20, 400, list(testrig.get_connected_components(odrive.gpio19, TeensyGpio)))
-                yield (odrive, 20, -20000, 20000, list(testrig.get_connected_components(odrive.gpio20, TeensyGpio)))
-                yield (odrive, 21, -1000, 0, list(testrig.get_connected_components(odrive.gpio21, TeensyGpio)))
+                test_cases = [(14, -50, 200, odrive.gpio14),
+                              (19, 20, 400, odrive.gpio19),
+                              (20, -20000, 20000, odrive.gpio20),
+                              (21, -1000, 0, odrive.gpio21)]
             else:
                 raise Exception(f"unknown board version {odrive.yaml['board-version']}")
 
+            for test_case in test_cases:
+                yield AnyTestCase(*[(odrive,) + tuple(test_case[:-1]) + (teensy_gpio,tf,) for teensy_gpio, tf in testrig.get_connected_components(test_case[-1], TeensyGpio)])
+
     def run_test(self, odrive: ODriveComponent, odrive_gpio_num: int, min_val: float, max_val: float, teensy_gpio: Component, logger: Logger):
         teensy = teensy_gpio.parent
         code = teensy_code_template.replace("{pwm_gpio}", str(teensy_gpio.num))

tools/odrive/tests/run_all_tests.sh

@@ -16,7 +16,7 @@ declare -a tests=('analog_input_test.py'
 summary=""
 
 for test in "${tests[@]}"; do
-    (ipython3 "$test" -- --test-rig-yaml ../../test-rig-rpi.yaml || true) | tee /tmp/odrivetest.log
+    (ipython3 "$test" || true) | tee /tmp/odrivetest.log
     if grep "All tests passed!" /tmp/odrivetest.log; then
         summary="$summary - $test: passed"$'\n'
     else

tools/odrive/tests/step_dir_test.py

@@ -19,25 +19,20 @@ class TestStepDir():
     """
     def get_test_cases(self, testrig: TestRig):
         for odrive in testrig.get_components(ODriveComponent):
-            gpio_conns = [
-                list(testrig.get_connected_components((odrive.gpio1, False), LinuxGpioComponent)),
-                list(testrig.get_connected_components((odrive.gpio2, False), LinuxGpioComponent)),
-                #list(testrig.get_connected_components((odrive.gpio3, False), LinuxGpioComponent)), # connected to LPF on test rig
-                #list(testrig.get_connected_components((odrive.gpio4, False), LinuxGpioComponent)), # connected to LPF on test rig
-                list(testrig.get_connected_components((odrive.gpio5, False), LinuxGpioComponent)),
-                list(testrig.get_connected_components((odrive.gpio6, False), LinuxGpioComponent)),
-                list(testrig.get_connected_components((odrive.gpio7, False), LinuxGpioComponent)),
-                list(testrig.get_connected_components((odrive.gpio8, False), LinuxGpioComponent)),
-            ]
+            def stepdir_test_case(axis, step_gpio_num, dir_gpio_num):
+                alternatives = []
+                for step_ctrl_gpio, tf1 in testrig.get_connected_components((getattr(odrive, f'gpio{step_gpio_num}'), False), LinuxGpioComponent):
+                    for dir_ctrl_gpio, tf2 in testrig.get_connected_components((getattr(odrive, f'gpio{dir_gpio_num}'), False), LinuxGpioComponent):
+                        alternatives.append((axis, step_gpio_num, step_ctrl_gpio, dir_gpio_num, dir_ctrl_gpio, TestFixture.all_of(tf1, tf2)))
+                return AnyTestCase(*alternatives)
 
-            yield (odrive.axes[0], 1, gpio_conns[0], 2, gpio_conns[1])
-            yield (odrive.axes[0], 5, gpio_conns[2], 6, gpio_conns[3])
-            yield (odrive.axes[0], 7, gpio_conns[4], 8, gpio_conns[5])
-           # yield (odrive.axes[0], 7, gpio_conns[6], 8, gpio_conns[7]) # broken
+            yield stepdir_test_case(odrive.axes[0], 1, 2)
+            yield stepdir_test_case(odrive.axes[0], 5, 6)
+            yield stepdir_test_case(odrive.axes[0], 7, 8)
             
             # test other axes
             for i in range(1, len(odrive.axes)):
-                yield (odrive.axes[i], 7, gpio_conns[4], 8, gpio_conns[5])
+                yield stepdir_test_case(odrive.axes[i], 7, 8)
 
     def run_test(self, axis: ODriveAxisComponent, step_gpio_num: int, step_gpio: LinuxGpioComponent, dir_gpio_num: int, dir_gpio: LinuxGpioComponent, logger: Logger):
         step_gpio.config(output=True)

tools/odrive/tests/test_runner.py

@@ -393,10 +393,12 @@ class SerialPortComponent(Component):
     def __init__(self, parent: Component, yaml: dict):
         Component.__init__(self, parent)
         self.yaml = yaml
+        self.tx = Component(self)
+        self.rx = Component(self)
 
     def get_subcomponents(self):
-        yield 'tx', Component(self)
-        yield 'rx', Component(self)
+        yield 'tx', self.tx
+        yield 'rx', self.rx
 
     def open(self, baudrate: int):
         import serial
@@ -571,10 +573,16 @@ class TeensyForwardingFixture(TestFixture):
     def prepare(self, logger: Logger):
         self.teensy.add_route(self.high_z, self.low_z, self.noise_enable)
 
+    def get_resources(self):
+        return [(self.teensy, False), (self.high_z, True), (self.low_z, True), (self.noise_enable, True)]
+
 class ClosedLoopControlFixture(TestFixture):
     def __init__(self, axis_ctx: ODriveAxisComponent, motor_ctx: MotorComponent, enc_ctx: EncoderComponent):
         self.axis_ctx, self.motor_ctx, self.enc_ctx = (axis_ctx, motor_ctx, enc_ctx)
 
+    def get_resources(self):
+        return []
+
     def prepare(self, logger: Logger):
         # Make sure no funny configuration is active
         logger.debug('Setting up clean configuration...')
@@ -667,7 +675,7 @@ class TestRig():
             for port in s:
                 self.net_by_component[port] = s
 
-    def get_closed_loop_combos(self):
+    def get_closed_loop_combos(self, init: bool = True):
         """
         Fetches all connected (odrive axis, motor, encoder) combos in the test rig.
 
@@ -678,23 +686,19 @@ class TestRig():
         all_motors = self.get_components(MotorComponent)
         all_encoders = self.get_components(EncoderComponent)
         for axis, motor, encoder in itertools.product(all_axes, all_motors, all_encoders):
-            is_connected, test_fixture1 = self.check_connection(axis, motor.phases)
+            is_connected, tf1 = self.check_connections([
+                (axis, motor.phases),
+                (motor.shaft, encoder.shaft),
+                (encoder.a, axis.parent.encoders[axis.num].a),
+                (encoder.b, axis.parent.encoders[axis.num].b)
+            ])
             if not is_connected:
                 continue
-            is_connected, test_fixture2 = self.check_connection(motor.shaft, encoder.shaft)
-            if not is_connected:
-                continue
-            is_connected, test_fixture3 = self.check_connection(encoder.a, axis.parent.encoders[axis.num].a)
-            if not is_connected:
-                continue
-            is_connected, test_fixture4 = self.check_connection(encoder.b, axis.parent.encoders[axis.num].b)
-            if not is_connected:
-                continue
-            test_fixture = TestFixture.all_of(test_fixture1, test_fixture2,
-                test_fixture3, test_fixture4, ClosedLoopControlFixture(axis, motor, encoder))
+            tf2 = ClosedLoopControlFixture(axis, motor, encoder) if init else None
+            test_fixture = TestFixture.all_of(tf1, tf2)
             yield axis, motor, encoder, test_fixture
     
-    def check_connection(self, component1: Component, component2: Component):
+    def check_connection(self, component1: Component, component2: Component, mode: str = 'indirect'):
         """
         Checks if the specified components are connected or connectable through
         a test fixture.
@@ -706,9 +710,14 @@ class TestRig():
                  connectable is True then the test_fixture, if not None, must
                  be prepared before the components are actually connected.
         """
+        assert(mode in ['direct', 'indirect'])
+
         net1 = self.net_by_component.get(component1, set([component1]))
         if component2 in net1:
-            return True, None
+            return True, None # The components are directly connected
+        if mode == 'direct':
+            return False, None
+        
         net2 = self.net_by_component.get(component2, set([component2]))
 
         possible_test_fixtures = []
@@ -722,6 +731,15 @@ class TestRig():
         else:
             return True, TestFixture.any_of(*possible_test_fixtures)
 
+    def check_connections(self, components: list, mode: str = 'indirect'):
+        tfs = []
+        for component1, component2 in components:
+            is_connected, tf = self.check_connection(component1, component2, mode)
+            if not is_connected:
+                return False, None
+            tfs.append(tf)
+        return True, TestFixture.all_of(*tfs)
+
 
     def get_components(self, t: type):
         """Returns a tuple (name, component) for all components that are of the specified type"""
@@ -761,29 +779,38 @@ class TestRig():
         A type can be specified to filter the connected components.
         """
 
-        if isinstance(src, dict):
-            candidates = self.get_components(comp_type)
+        candidates = self.get_components(comp_type)
 
+        if isinstance(src, dict):
             for candidate in candidates:
                 subcomponents = dict(candidate.get_subcomponents())
+
                 if len(set(src.keys()) - set(subcomponents.keys())):
                     continue # candidate doesn't have all of the requested ports
 
                 test_fixtures = []
-                for name, (srcport, direction) in src.items():
+                for name, subsrc in src.items():
+                    srcport, direction = subsrc if isinstance(subsrc, tuple) else (subsrc, False)
                     is_connected, test_fixture = (self.check_connection(srcport, subcomponents[name]) if direction else
                                                   self.check_connection(subcomponents[name], srcport))
                     if not is_connected:
                         break
                     test_fixtures.append(test_fixture)
 
+
                 if len(test_fixtures) < len(src.items()):
                     continue # not all of the ports are connected to the candidate's ports
                 
                 yield candidate, TestFixture.all_of(*test_fixtures)
 
         else:
-            raise Exception("not supported")
+            src, direction = src if isinstance(src, tuple) else (src, False)
+            assert(isinstance(src, Component))
+            for candidate in candidates:
+                is_connected, test_fixture = (self.check_connection(src, candidate) if direction else
+                                              self.check_connection(candidate, src))
+                if is_connected:
+                    yield candidate, test_fixture
 
 # Helper functions ------------------------------------------------------------#
 
@@ -853,8 +880,27 @@ def select_params(param_options):
 
     return None
 
-def is_feasible(params):
-    return True # TODO
+def is_feasible(params, test_fixtures):
+    """
+    Checks if the specified parameter and test fixture combination is feasible.
+    A combination is feasible if none of the test fixture resources appear in 
+    the parameters and if all of the exclusive-use test fixture resources are
+    only used by one test fixture.
+    """
+    exclusive_tf_resources = []
+    shared_tf_resources = set()
+    for r, ex in [(r, ex) for tf in test_fixtures if not tf is None for r, ex in tf.get_resources() if not r is None]:
+        if ex:
+            exclusive_tf_resources.append(r)
+        else:
+            shared_tf_resources.add(r)
+    if len(exclusive_tf_resources + list(shared_tf_resources)) > len(set(exclusive_tf_resources).union(shared_tf_resources)):
+        return False # At least one exclusive-use resource is used twice in the test fixtures
+    if len(set(exclusive_tf_resources).union(shared_tf_resources).intersection(params)):
+        return False # At least one test fixture resource appears in the params too
+    if len(shared_tf_resources.intersection(params)):
+        return False # At least one test fixture resource appears in the params too
+    return True
 
 def run(tests):
     if not isinstance(tests, list):
@@ -874,6 +920,7 @@ def run(tests):
         # test case no alternative is feasible (e.g. because there are component
         # conflicts).
 
+        logger.debug("loading...")
         test_cases = list(test.get_test_cases(testrig))
 
         if len(test_cases) == 0:
@@ -889,17 +936,17 @@ def run(tests):
                 if isinstance(test_fixture, CompositeTestFixture):
                     candidates += [tuple(candidate[:-1]) + (tf,) for tf in test_fixture._subfixtures]
                 else:
-                    candidates.append(candidate)
+                    candidates.append(tuple(candidate[:-1]) + (([] if test_fixture is None else [test_fixture]),))
 
             # Select the first candidate that is feasible
             params, test_fixture = (None, None)
             for candidate in candidates:
-                if is_feasible(candidate):
+                if is_feasible(candidate[:-1], candidate[-1]):
                     params, test_fixture = (candidate[:-1], candidate[-1])
                     break
 
             if params is None:
-                logger.warn(f'I found a {type(test).__name__} test case with {len(candidate)} possible parameter combination candidates but none of them is feasible.')
+                logger.warn(f'I found a {type(test).__name__} test case with {len(candidates)} possible parameter combination candidates but none of them is feasible.')
                 continue
 
             logger.notify('* preparing {} with {}...'.format(type(test).__name__,
@@ -948,14 +995,24 @@ parser = argparse.ArgumentParser(description='ODrive automated test tool\n')
 parser.add_argument("--ignore", metavar='DEVICE', action='store', nargs='+',
                     help="Ignore (disable) one or more components of the test rig")
                     # TODO: implement
-parser.add_argument("--test-rig-yaml", type=argparse.FileType('r'), required=True,
-                    help="test rig YAML file")
+parser.add_argument("--test-rig-yaml", type=argparse.FileType('r'),
+                    help="Test rig YAML file. Can be omitted if the environment variable ODRIVE_TEST_RIG_NAME is set.")
 parser.add_argument("--setup-host", action='store_true', default=False,
                     help="configure operating system functions such as GPIOs (requires root)")
 parser.set_defaults(ignore=[])
 
 args = parser.parse_args()
 
+if args.test_rig_yaml is None:
+    test_rig_name = os.environ.get('ODRIVE_TEST_RIG_NAME', '')
+    if test_rig_name == '':
+        print("You must either provide a --test-rig-yaml argument or set the environment variable ODRIVE_TEST_RIG_NAME.")
+        sys.exit(1)
+    path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))), test_rig_name + '.yaml')
+    args.test_rig_yaml = open(path, 'r')
+    
+
+
 # Load objects
 test_rig_yaml = yaml.load(args.test_rig_yaml, Loader=yaml.BaseLoader)
 logger = Logger()

tools/odrive/tests/uart_ascii_test.py

@@ -30,19 +30,19 @@ def reset_state(ser):
     time.sleep(0.1) # wait for any response that this may generate
     ser.flushInput() # discard response
 
-class TestUartAscii():
+class UartTest():
     """
-    Tests the most important functions of the ASCII protocol.
+    Base class for UART tests
     """
 
     def get_test_cases(self, testrig: TestRig):
         for odrive in testrig.get_components(ODriveComponent):
             if odrive.yaml['board-version'].startswith('v3.'):
-                ports = list(testrig.get_connected_components({
+                ports = testrig.get_connected_components({
                     'rx': (odrive.gpio1, True),
                     'tx': (odrive.gpio2, False)
-                }, SerialPortComponent))
-                yield (odrive, 0, 1, 2, ports)
+                }, SerialPortComponent)
+                yield AnyTestCase(*[(odrive, 0, 1, 2, port, tf) for port, tf in ports])
 
                 # Enable the line below to manually test UART_B. For this you need
                 # to manually move to the wires go to GPIO1/2 to GPIO3/4. The ones
@@ -53,11 +53,11 @@ class TestUartAscii():
                     'rx': (odrive.gpio15, True),
                     'tx': (odrive.gpio14, False)
                 }, SerialPortComponent))
-                yield (odrive, 0, 15, 14, ports)
+                yield AnyTestCase(*[(odrive, 0, 15, 14, port, tf) for port, tf in ports])
             else:
                 raise TestFailed("unknown board version")
 
-    def run_test(self, odrive: ODriveComponent, uart_num: int, tx_gpio: list, rx_gpio: list, port: SerialPortComponent, logger: Logger):
+    def prepare(self, odrive: ODriveComponent, uart_num: int, tx_gpio: list, rx_gpio: list, logger: Logger):
         logger.debug('Enabling UART {}...'.format(chr(ord('A') + uart_num)))
         
         # GPIOs might be in use by something other than UART and some components
@@ -86,6 +86,14 @@ class TestUartAscii():
 
         odrive.save_config_and_reboot()
 
+class TestUartAscii(UartTest):
+    """
+    Tests the most important functions of the ASCII protocol.
+    """
+
+    def run_test(self, odrive: ODriveComponent, uart_num: int, tx_gpio: list, rx_gpio: list, port: SerialPortComponent, logger: Logger):
+        self.prepare(odrive, uart_num, tx_gpio, rx_gpio, logger)
+
         with port.open(115200) as ser:
             # reset port to known state
             reset_state(ser)
@@ -183,23 +191,13 @@ class TestUartAscii():
             # TODO: test cases for 't', 'ss', 'se', 'sr' commands
 
 
-class TestUartBaudrate():
+class TestUartBaudrate(UartTest):
     """
     Tests if the UART baudrate setting works as intended.
     """
 
-    def get_test_cases(self, testrig: TestRig):
-        for odrive in testrig.get_components(ODriveComponent):
-            ports = list(testrig.get_connected_components({
-                'rx': (odrive.gpio1, True),
-                'tx': (odrive.gpio2, False)
-            }, SerialPortComponent))
-            yield (odrive, ports)
-
-    def run_test(self, odrive: ODriveComponent, port: SerialPortComponent, logger: Logger):
-        odrive.handle.config.enable_uart_a = True
-        odrive.handle.config.gpio1_mode = GPIO_MODE_UART_A
-        odrive.handle.config.gpio2_mode = GPIO_MODE_UART_A
+    def run_test(self, odrive: ODriveComponent, uart_num: int, tx_gpio: list, rx_gpio: list, port: SerialPortComponent, logger: Logger):
+        self.prepare(odrive, uart_num, tx_gpio, rx_gpio, logger)
 
         odrive.handle.config.uart_a_baudrate = 9600
         odrive.save_config_and_reboot()
@@ -225,23 +223,13 @@ class TestUartBaudrate():
         odrive.save_config_and_reboot()
 
 
-class TestUartBurnIn():
+class TestUartBurnIn(UartTest):
     """
     Tests if the ASCII protocol can handle 64kB of random data being thrown at it.
     """
 
-    def get_test_cases(self, testrig: TestRig):
-        for odrive in testrig.get_components(ODriveComponent):
-            ports = list(testrig.get_connected_components({
-                'rx': (odrive.gpio1, True),
-                'tx': (odrive.gpio2, False)
-            }, SerialPortComponent))
-            yield (odrive, ports)
-
-    def run_test(self, odrive: ODriveComponent, port: SerialPortComponent, logger: Logger):
-        odrive.handle.config.enable_uart_a = True
-        odrive.handle.config.gpio1_mode = GPIO_MODE_UART_A
-        odrive.handle.config.gpio2_mode = GPIO_MODE_UART_A
+    def run_test(self, odrive: ODriveComponent, uart_num: int, tx_gpio: list, rx_gpio: list, port: SerialPortComponent, logger: Logger):
+        self.prepare(odrive, uart_num, tx_gpio, rx_gpio, logger)
 
         with port.open(115200) as ser:
             with open('/dev/random', 'rb') as rand:
@@ -264,32 +252,30 @@ class TestUartNoise():
     
     def get_test_cases(self, testrig: TestRig):
         for odrive in testrig.get_components(ODriveComponent):
-            # For every ODrive, find a connected serial port which has a teensy
-            # in between, so that we can inject noise,
+            alternatives = []
 
-            ports = list(testrig.get_connected_components({
-                'rx': (odrive.gpio1, True),
-                'tx': (odrive.gpio2, False)
-            }, SerialPortComponent))
+            # Find a Teensy that sits between a linux serial port and the ODrive
+            # with an additional wire that runs to the linux PC and will be used
+            # as noise-enable line
 
-            # Hack the bus objects to enable noise_enable functionality on the TX line.
+            for teensy in testrig.get_components(TeensyComponent):
+                for port in testrig.get_components(SerialPortComponent):
+                    gpio_conns = [
+                        testrig.net_by_component.get(odrive.gpio1, set()).intersection(set(teensy.gpios)),
+                        testrig.net_by_component.get(odrive.gpio2, set()).intersection(set(teensy.gpios)),
+                        testrig.net_by_component.get(port.tx, set()).intersection(set(teensy.gpios)),
+                        testrig.net_by_component.get(port.rx, set()).intersection(set(teensy.gpios)),
+                        teensy.gpios
+                    ]
 
-            def get_noise_gpio(bus):
-                teensy = bus.gpio_tuples[1][0]
-                for teensy_gpio in teensy.gpios:
-                    for other_gpio in testrig.get_directly_connected_components(teensy_gpio):
-                        if isinstance(other_gpio, LinuxGpioComponent):
-                            return teensy_gpio, other_gpio
-                return None
+                    for gpio1, gpio2, gpio3, gpio4, gpio5 in itertools.product(*gpio_conns):
+                        for noise_ctrl_gpio, tf3 in testrig.get_connected_components(gpio5, LinuxGpioComponent):
+                            tf1 = TeensyForwardingFixture(teensy, gpio3, gpio2)
+                            tf2 = TeensyForwardingFixture(teensy, gpio1, gpio4)
+                            tf1.noise_enable = gpio5
+                            alternatives.append((odrive, port, noise_ctrl_gpio, TestFixture.all_of(tf1, tf2, tf3)))
 
-            for idx, bus in enumerate(ports):
-                noise_gpio_on_teensy, noise_gpio_on_rpi = get_noise_gpio(bus)
-                assert(noise_gpio_on_rpi)
-                t, i, o, _ = bus.gpio_tuples[1]
-                bus.gpio_tuples[1] = (t, i, o, noise_gpio_on_teensy)
-                ports[idx] = (bus, noise_gpio_on_rpi)
-                
-            yield (odrive, ports)
+            yield AnyTestCase(*alternatives)
 
     def run_test(self, odrive: ODriveComponent, port: SerialPortComponent, noise_enable: LinuxGpioComponent, logger: Logger):
         noise_enable.config(output=True)
@@ -304,6 +290,11 @@ class TestUartNoise():
             # reset port to known state
             reset_state(ser)
 
+            # First try
+            ser.write(b'r vbus_voltage\n')
+            response = float(ser.readline().strip())
+            test_assert_eq(response, odrive.handle.vbus_voltage, accuracy=0.1)
+
             # Enable square wave of ~1.6MHz on the ODrive's RX line
             noise_enable.write(True)