Device setup: fixture-owned port lifecycle (realsenseai#15281)

Author: Nir-Az

unit-tests/conftest.py

@@ -15,7 +15,7 @@
 # Fake pyrealsense2 — track log_to_console calls so tests can verify --rslog
 import json as _json
 _tracking_log = os.path.join(os.path.dirname(os.path.abspath(__file__)), '_tracking.json')
-_tracking = {"rslog_calls": [], "query_kwargs": [], "enable_only_calls": []}
+_tracking = {"rslog_calls": [], "query_kwargs": [], "enable_only_calls": [], "disable_calls": []}
 def _save_tracking():
     with open(_tracking_log, 'w') as _f:
         _json.dump(_tracking, _f)
@@ -76,21 +76,37 @@ def _mock_get(sn):
 _dev.by_spec = _mock_by_spec
 _dev.get = _mock_get
 _dev._device_by_sn = {sn: FakeDevice(sn, n) for sn, n in _sn_map.items()}
-_dev.hub = None
+# Default: model a hub-equipped bench with a truthy sentinel. Port ops are mocked above, so the
+# object is never actually used. init_hub is stubbed so the real one doesn't probe (absent)
+# hardware and reset this back to None. A scenario test file can patch devices.hub /
+# devices.any_port_powered at import time (before fixtures run) to model a hub-less bench or a
+# port left powered -- see pytest-hubless-setup.py and pytest-port-already-on.py.
+_dev.hub = object()
+_dev.init_hub = lambda: None
 _dev._context = None
 def _mock_query(**kw):
     _tracking["query_kwargs"].append(kw)
     _save_tracking()
 _dev.query = _mock_query
 _dev.map_unknown_ports = lambda: None
 _dev.wait_until_all_ports_disabled = lambda: None
+# Hub hardware port-state probe used by the conftest recycle decision. Default OFF (the previous
+# teardown powered the device down -> setup just enables). A scenario test file patches this to
+# True to model a port left powered by a skipped teardown -> setup recycles it clean.
+_dev.any_port_powered = lambda serials: False
 
-# Track enable_only calls so tests can verify hub port behavior
-def _mock_enable_only(serials, recycle=True):
-    _tracking["enable_only_calls"].append({"serials": list(serials), "recycle": recycle})
+# Track enable_only / disable calls so tests can verify hub port behavior
+def _mock_enable_only(serials, recycle=True, timeout=None, disable_other_ports=False):
+    _tracking["enable_only_calls"].append(
+        {"serials": list(serials), "recycle": recycle, "disable_other_ports": disable_other_ports})
     _save_tracking()
 _dev.enable_only = _mock_enable_only
 
+def _mock_disable(serials, wait=True):
+    _tracking.setdefault("disable_calls", []).append({"serials": list(serials), "wait": wait})
+    _save_tracking()
+_dev.disable = _mock_disable
+
 # exec() the REAL conftest.py
 _conftest_path = os.path.join(_unit_tests_dir, 'conftest.py')
 with open(_conftest_path) as _f:

unit-tests/infra-tests/e2e/e2e_conftest.py

@@ -0,0 +1,13 @@
+# Model a hub-less bench (e.g. Jetson). Patch devices.hub to None at import -- this runs during
+# collection, before module_device_setup -- so the conftest recycle decision takes the no-hub
+# branch: teardown-disable is a no-op there, so setup recycles via enable_only(recycle=True)
+# (which falls back to hardware_reset on a real hub-less machine).
+import rspy.devices as _devices
+_devices.hub = None
+
+import pytest
+
+
+@pytest.mark.device("D455")
+def test_d455(module_device_setup):
+    assert module_device_setup == '111'

unit-tests/infra-tests/e2e/pytest-hubless-setup.py

@@ -0,0 +1,13 @@
+# Model a device whose hub port was left powered by a skipped/crashed teardown. Patch
+# any_port_powered -> True at import (before module_device_setup runs) so the conftest recycle
+# decision takes the "already powered" branch: setup recycles the device clean (recycle=True)
+# instead of reusing a possibly-bad state.
+import rspy.devices as _devices
+_devices.any_port_powered = lambda serials: True
+
+import pytest
+
+
+@pytest.mark.device("D455")
+def test_d455(module_device_setup):
+    assert module_device_setup == '111'

unit-tests/infra-tests/e2e/pytest-port-already-on.py

@@ -138,13 +138,16 @@ def make_device_marker(name, *patterns):
 _E2E_CONFTEST = os.path.join(_E2E_DIR, 'e2e_conftest.py')
 
 
-def run_e2e(test_filename, *extra_pytest_args):
+def run_e2e(test_filename, *extra_pytest_args, env=None):
     """Run a pytest subprocess on a static test file from e2e/.
 
     Copies e2e_conftest.py and the test file to a temp dir for isolation
     from the parent unit-tests/conftest.py. No content is generated — both
     files are static and checked into the repo.
 
+    :param env: optional dict of extra environment variables for the subprocess
+                (e.g. {'E2E_NO_HUB': '1'} to model a hub-less bench).
+
     Returns (returncode, stdout, tracking) where tracking is a dict with:
         - enable_only_calls: list of {serials, recycle} dicts
         - rslog_calls: list of {level} dicts
@@ -156,13 +159,15 @@ def run_e2e(test_filename, *extra_pytest_args):
         shutil.copy(_E2E_CONFTEST, os.path.join(tmpdir, 'conftest.py'))
         shutil.copy(os.path.join(_E2E_DIR, test_filename), os.path.join(tmpdir, test_filename))
 
-        env = os.environ.copy()
-        env['INFRA_UNIT_TESTS_DIR'] = os.path.normpath(os.path.join(_E2E_DIR, '..', '..'))  # unit-tests/
+        sub_env = os.environ.copy()
+        sub_env['INFRA_UNIT_TESTS_DIR'] = os.path.normpath(os.path.join(_E2E_DIR, '..', '..'))  # unit-tests/
+        if env:
+            sub_env.update(env)
 
         p = subprocess.run(
             [sys.executable, "-m", "pytest", test_filename, "-v", *extra_pytest_args],
             cwd=tmpdir,
-            env=env,
+            env=sub_env,
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
             universal_newlines=True,
@@ -176,7 +181,7 @@ def run_e2e(test_filename, *extra_pytest_args):
 
         tracking_file = os.path.join(tmpdir, '_tracking.json')
         tracking = json.loads(open(tracking_file).read()) if os.path.exists(tracking_file) else {
-            "enable_only_calls": [], "rslog_calls": [], "query_kwargs": []
+            "enable_only_calls": [], "rslog_calls": [], "query_kwargs": [], "disable_calls": []
         }
 
         return p.returncode, p.stdout, tracking

unit-tests/infra-tests/helpers.py

@@ -79,7 +79,9 @@ def test_retries(self):
         calls = tracking["enable_only_calls"]
         # Two enable_only calls: initial module-fixture creation + post-retry-teardown re-creation.
         assert len(calls) == 2
-        assert all(c['recycle'] is True for c in calls)
+        assert all(c['recycle'] is False for c in calls)
+        # the device recycle now comes from the teardown-disable between attempts, not recycle=True
+        assert len(tracking["disable_calls"]) >= 1
 
     def test_retries_recreate_module_fixture(self):
         """Module-scoped fixtures must be torn down and re-instantiated between
@@ -101,23 +103,27 @@ def test_retries_on_setup_error(self):
         assert rc == 0
 
     def test_repeat(self):
-        """--repeat 3 should repeat the test 3 times, recycling the device each time."""
+        """--repeat 3 repeats the test 3 times; each pass power-cycles the device via
+        teardown-disable + setup-enable (setup itself uses recycle=False)."""
         rc, out, tracking = run_e2e("pytest-device-setup.py", "-k", "test_d455 and not excluded",
                                      "--repeat", "3")
         assert_outcomes(out, passed=3)
         calls = tracking["enable_only_calls"]
         assert len(calls) == 3
-        assert all(c['recycle'] is True for c in calls)
+        assert all(c['recycle'] is False for c in calls)
+        assert len(tracking["disable_calls"]) >= 1  # teardown disables each pass -> the cycle
 
     def test_repeat_no_reset(self):
         """--repeat 3 --no-reset should repeat without recycling."""
         rc, out, tracking = run_e2e("pytest-device-setup.py", "-k", "test_d455 and not excluded",
                                      "--repeat", "3", "--no-reset")
         assert_outcomes(out, passed=3)
         calls = tracking["enable_only_calls"]
-        # First run enables without recycle, subsequent runs skip enable_only entirely
-        assert len(calls) == 1
-        assert calls[0]['recycle'] is False
+        # --no-reset re-enables each pass but never disables on teardown, so the device is
+        # NOT power-cycled (left on) -- the distinguishing behavior vs the default path.
+        assert len(calls) == 3
+        assert all(c['recycle'] is False for c in calls)
+        assert tracking["disable_calls"] == []
 
     def test_device_nonexistent(self):
         """--device D999 with no matching device should produce 0 parametrized instances."""

unit-tests/infra-tests/test_e2e_cli_options.py

@@ -17,21 +17,40 @@
 class TestDevicePortManagement:
 
     def test_device_marker_enables_correct_port(self):
-        """@device('D455') should call enable_only(['111'], recycle=True)."""
+        """@device('D455') enables the device (recycle=False -- the power cycle is
+        teardown-disable + setup-enable, so setup itself doesn't recycle)."""
         rc, out, tracking = run_e2e("pytest-device-setup.py", "-k", "test_d455 and not excluded")
         assert_outcomes(out, passed=1)
         assert len(tracking["enable_only_calls"]) == 1
         assert tracking["enable_only_calls"][0]['serials'] == ['111']
+        assert tracking["enable_only_calls"][0]['recycle'] is False
+
+    def test_hubless_recycles_via_hw_reset(self):
+        """On a hub-less bench (e.g. Jetson) teardown-disable is a no-op, so setup MUST recycle --
+        enable_only(recycle=True) falls back to hardware_reset. pytest-hubless-setup.py patches
+        devices.hub to None to exercise the no-hub branch of the conftest recycle decision."""
+        rc, out, tracking = run_e2e("pytest-hubless-setup.py")
+        assert_outcomes(out, passed=1)
+        assert len(tracking["enable_only_calls"]) == 1
+        assert tracking["enable_only_calls"][0]['recycle'] is True
+
+    def test_port_already_on_recycles(self):
+        """With a hub, setup expects the device OFF (prev teardown disabled it). If a required port
+        is still powered -- a skipped/crashed teardown -- setup recycles it clean instead of reusing
+        a stale state. pytest-port-already-on.py patches any_port_powered -> True for this branch."""
+        rc, out, tracking = run_e2e("pytest-port-already-on.py")
+        assert_outcomes(out, passed=1)
+        assert len(tracking["enable_only_calls"]) == 1
         assert tracking["enable_only_calls"][0]['recycle'] is True
 
     def test_device_each_enables_one_port_per_test(self):
-        """@device_each('D400*') should call enable_only once per device, each with recycle=True."""
+        """@device_each('D400*') should call enable_only once per device (recycle=False)."""
         rc, out, tracking = run_e2e("pytest-each-setup.py", "-k", "test_d400 and not d999")
         assert_outcomes(out, passed=3)
         assert len(tracking["enable_only_calls"]) == 3
         serials_enabled = [c['serials'][0] for c in tracking["enable_only_calls"]]
         assert set(serials_enabled) == {'111', '222', '777'}
-        assert all(c['recycle'] is True for c in tracking["enable_only_calls"])
+        assert all(c['recycle'] is False for c in tracking["enable_only_calls"])
         assert all(len(c['serials']) == 1 for c in tracking["enable_only_calls"])
 
     def test_second_test_same_device_no_recycle(self):

unit-tests/infra-tests/test_e2e_port_management.py

@@ -332,13 +332,28 @@ def get_port_by_location(self, usb_location):
                     # We don't know how to get the port from these yet!
                     return None  # int(match.group(2))
                 else:
-                    split_location = [int(x) for x in usb_location.split('.')]
+                    # Windows LocationInformation fields are hex (e.g. "0000.000d.0000.003.002...").
+                    # Parse base-16 so fields like "000d" don't raise; only the last two non-zero
+                    # indices (acroname sub-hub port + device port) are used and those are small
+                    # (<=4), so hex vs decimal is identical for them. Unparseable -> defer to
+                    # map_unknown_ports() enumeration.
+                    try:
+                        split_location = [int(x, 16) for x in usb_location.split('.')]
+                    except ValueError:
+                        log.d(f'could not parse usb location {usb_location!r}; deferring to port mapping')
+                        return None
                     # lambda helper to return the last 2 non-zero numbers, used when connecting using an additional hub
                     # ex: laptop -> hub -> acroname
                     get_last_two_digits = lambda array: tuple(
                         reversed(list(reversed([i for i in array if i != 0]))[:2]))
                     # only the last two digits are necessary
-                    first_index, second_index = get_last_two_digits(split_location)
+                    last_two = get_last_two_digits(split_location)
+                    if len(last_two) < 2:
+                        # fewer than two non-zero fields (e.g. all-zero or single-field location):
+                        # not enough to identify a sub-hub+port pair -> defer to port mapping
+                        log.d(f'usb location {usb_location!r} has <2 non-zero fields; deferring to port mapping')
+                        return None
+                    first_index, second_index = last_two
 
                     return get_port_from_usb(first_index, second_index)
     else:
@@ -373,6 +388,9 @@ def get_port_by_location(self, usb_location):
                         match = re.search(r'^(\d+)\.(\d+)', usb_location[len(port) + 1:])
                         if match:
                             return get_port_from_usb(int(match.group(1)), int(match.group(2)))
+                # no known acroname hub prefixes this location (or sub-ports unparseable) -- log so a
+                # misconfigured intermediate hub is diagnosable, then defer to map_unknown_ports()
+                log.d( f'usb location {usb_location!r} not under a known acroname hub; deferring to port mapping' )
 
 specs = None
 def discover(retries = 0):
@@ -414,7 +432,13 @@ def get_port_from_usb(first_usb_index, second_usb_index ):
                              (3, 2): 6,
                              (3, 1): 7,
                              }
-    return acroname_port_usb_map[(first_usb_index, second_usb_index)]
+    # .get() -> None for an unmapped topology so the caller falls back to map_unknown_ports().
+    # Log the miss (pre-PR this raised KeyError which the caller logged) so an operator with a
+    # misconfigured intermediate hub has a signal explaining why a device ended up unmapped.
+    port = acroname_port_usb_map.get( (first_usb_index, second_usb_index) )
+    if port is None:
+        log.d( f'unmapped acroname topology ({first_usb_index},{second_usb_index}); deferring to port mapping' )
+    return port