gentl: reset harvester and CTI failure tests

Add a _reset_harvester helper on GenTLCameraBackend and call it where the harvester must be torn down on failure to ensure proper cleanup. Enhance FakeHarvester to support deterministic add_file failures and to record reset/add/update/create calls; keep create_image_acquirer for compatibility. Update test fixtures (conftest) to expose gentl_fail_add_file_for control, inject a dummy CTI only when none are explicitly provided, and expose gb.fail_add_file_for from patch_gentl_sdk. Add test helpers and new tests to isolate GENICAM env vars and verify CTI load diagnostics for all-success, partial-failure, and complete-failure scenarios. Also adjust some discovery tests to explicitly isolate the environment and tighten assertions.

Author: C-Achard

dlclivegui/cameras/backends/gentl_backend.py

@@ -372,7 +372,7 @@ def open(self) -> None:
             ns["cti_file"] = str(cti_files[0])  # best effort
 
         if not loaded:
-            self._harvester = None
+            self._reset_harvester()
             raise RuntimeError(
                 "No GenTL producer (.cti) could be loaded.\n\n"
                 f"Resolved CTIs: {cti_files}\n"
@@ -385,7 +385,7 @@ def open(self) -> None:
         self._harvester.update()
 
         if not self._harvester.device_info_list:
-            self._harvester = None
+            self._reset_harvester()
             raise RuntimeError(
                 "No GenTL cameras detected via Harvesters after loading producers.\n\n"
                 f"Loaded CTIs: {loaded}\n"
@@ -797,6 +797,7 @@ def rebind_settings(cls, settings):
                         continue
 
                 if not loaded:
+                    cls._reset_select_harvester(harvester)
                     return settings
 
                 harvester.update()
@@ -936,6 +937,20 @@ def stop(self) -> None:
             except Exception:
                 pass
 
+    @staticmethod
+    def _reset_select_harvester(harvester) -> None:
+        if harvester is not None:
+            try:
+                harvester.reset()
+            except Exception:
+                pass
+
+    def _reset_harvester(self) -> None:
+        try:
+            self._reset_select_harvester(self._harvester)
+        finally:
+            self._harvester = None
+
     def close(self) -> None:
         if self._acquirer is not None:
             try:

tests/cameras/backends/conftest.py

@@ -2,13 +2,16 @@
 from __future__ import annotations
 
 import importlib
+import logging
 import os
 from dataclasses import dataclass
 from typing import Any
 
 import numpy as np
 import pytest
 
+logger = logging.getLogger(__name__)
+
 
 # -----------------------------
 # Dependency detection helpers
@@ -720,21 +723,20 @@ class FakeHarvester:
     Inventory-driven so tests can control enumeration.
     """
 
-    def __init__(self, inventory: list[dict[str, Any]] | None = None):
+    def __init__(self, inventory: list[dict[str, Any]] | None = None, *, fail_add_file_for: set[str] | None = None):
         self._files: list[str] = []
         self._inventory: list[dict[str, Any]] = list(inventory or [])
         self.device_info_list: list[Any] = []
 
+        # NEW: failure control
+        self._fail_add_file_for = set(fail_add_file_for or [])
+
         # Call tracing
         self.add_file_calls: list[str] = []
         self.update_calls = 0
         self.reset_calls = 0
         self.create_calls: list[Any] = []
 
-    def add_file(self, file_path: str):
-        self._files.append(str(file_path))
-        self.add_file_calls.append(str(file_path))
-
     def update(self):
         self.update_calls += 1
         # If not provided, default to a single fake device
@@ -788,6 +790,16 @@ def create(self, selector=None, index: int | None = None, *args, **kwargs):
     def create_image_acquirer(self, *args, **kwargs):
         return self.create(*args, **kwargs)
 
+    def add_file(self, file_path: str):
+        p = str(file_path)
+        self.add_file_calls.append(p)
+
+        # NEW: fail deterministically if requested
+        if p in self._fail_add_file_for:
+            raise RuntimeError(f"Simulated CTI load failure for: {p}")
+
+        self._files.append(p)
+
 
 # -----------------------------------------------------------------------------
 # GentL fixtures: inventory, patching, settings factory
@@ -817,41 +829,53 @@ def gentl_inventory():
 
 
 @pytest.fixture()
-def fake_harvester_factory(gentl_inventory):
+def fake_harvester_factory(gentl_inventory, gentl_fail_add_file_for):
     """
-    Factory that returns a FakeHarvester bound to the current gentl_inventory.
-    Allows tests to mutate gentl_inventory before calling backend.open().
+    Factory that returns a FakeHarvester bound to the current gentl_inventory and
+    gentl_fail_add_file_for. Tests can mutate both before calling backend.open().
     """
 
     def _make():
-        return FakeHarvester(inventory=gentl_inventory)
+        return FakeHarvester(inventory=gentl_inventory, fail_add_file_for=gentl_fail_add_file_for)
 
     return _make
 
 
 @pytest.fixture()
-def patch_gentl_sdk(monkeypatch, fake_harvester_factory, tmp_path):
+def gentl_fail_add_file_for():
+    """
+    Mutable set of CTI file paths that FakeHarvester.add_file should fail on.
+    Tests can add/remove paths to simulate partial/complete CTI load failures.
+    """
+    return set()
+
+
+@pytest.fixture()
+def patch_gentl_sdk(monkeypatch, fake_harvester_factory, gentl_fail_add_file_for, tmp_path):
     """
     Patch dlclivegui.cameras.backends.gentl_backend to use FakeHarvester + Fake timeout.
-    Also ensure CTI discovery succeeds for classmethods (discover_devices/quick_ping)
-    by creating a real dummy .cti and exposing it via GENICAM_GENTL64_PATH.
+    Ensure CTI discovery succeeds for classmethods by creating a real dummy .cti and
+    exposing it via GENICAM_GENTL64_PATH.
     """
     import dlclivegui.cameras.backends.gentl_backend as gb
 
     # Patch Harvester symbol (the backend calls Harvester() directly)
     monkeypatch.setattr(gb, "Harvester", lambda: fake_harvester_factory(), raising=False)
 
-    # Keep your backend timeout contract as-is: it catches HarvesterTimeoutError
+    # Keep timeout contract
     monkeypatch.setattr(gb, "HarvesterTimeoutError", FakeGenTLTimeoutException, raising=False)
 
-    # Create a real CTI file and advertise it via env var (cross-platform via os.pathsep)
+    # Create a real CTI file and advertise it via env var
     cti_file = tmp_path / "dummy.cti"
     if not cti_file.exists():
         cti_file.write_text("fake", encoding="utf-8")
 
     monkeypatch.setenv("GENICAM_GENTL64_PATH", str(tmp_path))
     monkeypatch.delenv("GENICAM_GENTL32_PATH", raising=False)
 
+    # OPTIONAL: expose failure control so tests can do gb.fail_add_file_for.add(...)
+    gb.fail_add_file_for = gentl_fail_add_file_for
+
     return gb
 
 
@@ -877,10 +901,22 @@ def _make(
         cti = tmp_path / "dummy.cti"
         if not cti.exists():
             cti.write_text("fake", encoding="utf-8")
+
         props = properties if isinstance(properties, dict) else {}
         props.setdefault("gentl", {})
-        props["gentl"] = dict(props["gentl"])
-        props["gentl"].setdefault("cti_file", str(cti))
+        props["gentl"] = dict(props["gentl"])  # copy to avoid mutating caller dict unexpectedly
+
+        ns = props["gentl"]
+
+        # Detect whether CTIs were explicitly provided in *either* namespace or legacy keys
+        explicit_ns = bool(ns.get("cti_file") or ns.get("cti_files"))
+        explicit_legacy = bool(props.get("cti_file") or props.get("cti_files"))
+
+        # Only inject a default dummy.cti if nothing explicit was provided
+        if not explicit_ns and not explicit_legacy:
+            logger.debug("No CTI file(s) explicitly provided in settings; injecting dummy CTI for gentl tests.")
+            ns.setdefault("cti_file", str(cti))
+
         return CameraSettings(
             name=name,
             index=index,

tests/cameras/backends/test_gentl_backend.py

@@ -14,6 +14,31 @@
     discover_cti_files,
 )
 
+# ----------------------------------------------------------------------
+# Helper functions
+# ----------------------------------------------------------------------
+
+
+@pytest.fixture
+def isolate_gentl_env(monkeypatch):
+    monkeypatch.delenv("GENICAM_GENTL64_PATH", raising=False)
+    monkeypatch.delenv("GENICAM_GENTL32_PATH", raising=False)
+    yield
+
+
+def _force_only_these_ctis(settings, ctis: list[str]) -> None:
+    # Ensure namespace exists
+    props = settings.properties
+    props.setdefault("gentl", {})
+    ns = props["gentl"]
+
+    # Make sure no default single-cti sneaks in (dummy.cti)
+    ns.pop("cti_file", None)
+    props.pop("cti_file", None)
+
+    # Explicit list
+    ns["cti_files"] = ctis
+
 
 # ---------------------------------------------------------------------
 # Core lifecycle + strict transaction model
@@ -582,7 +607,7 @@ def test_discover_glob_patterns(tmp_path):
     assert pattern in diag.glob_patterns
 
 
-def test_discover_env_var_directory(monkeypatch, tmp_path):
+def test_discover_env_var_directory(monkeypatch, tmp_path, isolate_gentl_env):
     _make_cti(tmp_path, "Env.cti")
 
     monkeypatch.setenv("GENICAM_GENTL64_PATH", str(tmp_path))
@@ -593,18 +618,19 @@ def test_discover_env_var_directory(monkeypatch, tmp_path):
     assert "GENICAM_GENTL64_PATH" in diag.env_vars_used
 
 
-def test_discover_env_var_direct_file(monkeypatch, tmp_path):
+def test_discover_env_var_direct_file(monkeypatch, tmp_path, isolate_gentl_env):
     cti = _make_cti(tmp_path, "Direct.cti")
 
     monkeypatch.setenv("GENICAM_GENTL64_PATH", str(cti))
 
     candidates, diag = discover_cti_files(include_env=True, must_exist=True)
 
-    assert candidates == [str(cti.resolve())] or Path(candidates[0]).name == "Direct.cti"
+    assert len(candidates) == 1
+    assert Path(candidates[0]).name == "Direct.cti"
     assert diag.env_paths_expanded  # should include the raw env entry
 
 
-def test_discover_env_var_multiple_entries(monkeypatch, tmp_path):
+def test_discover_env_var_multiple_entries(monkeypatch, tmp_path, isolate_gentl_env):
     d1 = tmp_path / "d1"
     d2 = tmp_path / "d2"
     d1.mkdir()
@@ -622,7 +648,7 @@ def test_discover_env_var_multiple_entries(monkeypatch, tmp_path):
     assert names == ["A.cti", "B.cti"]
 
 
-def test_discover_deduplicates_same_file_from_multiple_sources(monkeypatch, tmp_path):
+def test_discover_deduplicates_same_file_from_multiple_sources(monkeypatch, tmp_path, isolate_gentl_env):
     cti = _make_cti(tmp_path, "Dup.cti")
 
     # Discover it twice: explicit + env dir
@@ -664,3 +690,81 @@ def test_choose_cti_files_newest_policy(tmp_path):
     selected = choose_cti_files([str(old), str(new)], policy=GenTLDiscoveryPolicy.NEWEST, max_files=1)
     assert len(selected) == 1
     assert Path(selected[0]).name == "New.cti"
+
+
+def test_open_persists_cti_load_diagnostics_all_success(patch_gentl_sdk, gentl_settings_factory, tmp_path):
+    gb = patch_gentl_sdk
+
+    c1 = _make_cti(tmp_path, "A.cti")
+    c2 = _make_cti(tmp_path, "B.cti")
+
+    # Provide multiple CTIs (how your backend reads these may vary)
+    settings = gentl_settings_factory(properties={"gentl": {"cti_files": [str(c1), str(c2)]}})
+    _force_only_these_ctis(settings, [str(c1), str(c2)])
+    be = gb.GenTLCameraBackend(settings)
+
+    be.open()
+
+    ns = settings.properties["gentl"]
+    assert ns["cti_files"] == [str(c1), str(c2)]
+    assert ns["cti_files_loaded"] == [str(c1), str(c2)]
+    # If dict:
+    assert ns["cti_files_failed"] == []
+
+    be.close()
+
+
+def test_open_persists_cti_load_diagnostics_partial_failure(patch_gentl_sdk, gentl_settings_factory, tmp_path):
+    gb = patch_gentl_sdk
+
+    ok = _make_cti(tmp_path, "OK.cti")
+    bad = _make_cti(tmp_path, "BAD.cti")
+
+    gb.fail_add_file_for.clear()
+    gb.fail_add_file_for.add(str(bad))
+
+    settings = gentl_settings_factory(properties={"gentl": {"cti_files": [str(ok), str(bad)]}})
+    _force_only_these_ctis(settings, [str(ok), str(bad)])
+
+    be = gb.GenTLCameraBackend(settings)
+    be.open()
+
+    ns = settings.properties["gentl"]
+    assert ns["cti_files"] == [str(ok), str(bad)]
+    assert ns["cti_files_loaded"] == [str(ok)]
+
+    failed = ns["cti_files_failed"]
+    assert isinstance(failed, list)
+    assert len(failed) == 1
+    assert failed[0]["cti"] == str(bad)
+    assert isinstance(failed[0]["error"], str) and failed[0]["error"]
+
+    be.close()
+
+
+def test_open_persists_cti_load_diagnostics_complete_failure(patch_gentl_sdk, gentl_settings_factory, tmp_path):
+    gb = patch_gentl_sdk
+
+    b1 = _make_cti(tmp_path, "B1.cti")
+    b2 = _make_cti(tmp_path, "B2.cti")
+
+    gb.fail_add_file_for.clear()
+    gb.fail_add_file_for.update({str(b1), str(b2)})
+
+    settings = gentl_settings_factory(properties={"gentl": {"cti_files": [str(b1), str(b2)]}})
+    _force_only_these_ctis(settings, [str(b1), str(b2)])
+    be = gb.GenTLCameraBackend(settings)
+
+    with pytest.raises(RuntimeError):
+        be.open()
+
+    # Keys should still be persisted for debugging even though open failed.
+    ns = settings.properties.get("gentl", {})
+    assert ns.get("cti_files") == [str(b1), str(b2)]
+    assert ns.get("cti_files_loaded") == []
+
+    failed = ns.get("cti_files_failed")
+    assert isinstance(failed, list)
+    assert sorted(d["cti"] for d in failed) == sorted([str(b1), str(b2)])
+    for d in failed:
+        assert isinstance(d.get("error"), str) and d["error"]