Skip to content
10 changes: 9 additions & 1 deletion dlclivegui/gui/camera_config/trigger_config_dialog.py
Original file line number Diff line number Diff line change
Expand Up @@ -455,6 +455,14 @@ def _accept(self) -> None:
return

ns = _backend_namespace(self._cam)
ns["trigger"] = trigger.to_properties()
trigger_props = trigger.to_properties()

if not self._profile.show_strobe_fields:
trigger_props.pop("strobe_polarity", None)
trigger_props.pop("strobe_operation", None)
trigger_props.pop("strobe_duration", None)
trigger_props.pop("strobe_delay", None)

ns["trigger"] = trigger_props

self.accept()
6 changes: 6 additions & 0 deletions dlclivegui/gui/main_window.py
Original file line number Diff line number Diff line change
Expand Up @@ -1123,6 +1123,12 @@ def _save_config_to_path(self, path: Path) -> bool:
config.save(path)
self._settings_store.set_last_config_path(str(path))
self._settings_store.save_full_config_snapshot(config)

try:
self.settings.sync()
except Exception:
logger.debug("Failed to sync settings after saving config", exc_info=True)

except Exception as exc: # pragma: no cover - GUI interaction
self._show_error(str(exc))
return False
Expand Down
361 changes: 361 additions & 0 deletions tests/cameras/backends/utils/test_gentl_discovery.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,361 @@
from __future__ import annotations

import os
from pathlib import Path

import pytest

from dlclivegui.cameras.backends.utils import gentl_discovery as gd

pytestmark = pytest.mark.unit


@pytest.fixture(autouse=True)
def clear_shared_harvester_pool():
gd.SharedHarvesterPool._entries.clear()
yield
gd.SharedHarvesterPool._entries.clear()


def test_cti_files_as_list_handles_none_strings_and_sequences():
assert gd.cti_files_as_list(None) == []
assert gd.cti_files_as_list("") == []
assert gd.cti_files_as_list(" ") == []
assert gd.cti_files_as_list("camera.cti") == ["camera.cti"]
assert gd.cti_files_as_list(["a.cti", None, "", " ", Path("b.cti")]) == ["a.cti", "b.cti"]


def test_discover_explicit_cti_file_without_harvester(tmp_path: Path):
cti = tmp_path / "producer.cti"
cti.write_text("", encoding="utf-8")

candidates, diag = gd.discover_cti_files(
cti_file=str(cti),
include_env=False,
)

assert candidates == [str(cti.resolve())]
assert diag.explicit_files == [str(cti)]
assert diag.candidates == [str(cti.resolve())]
assert diag.rejected == []


def test_discover_rejects_missing_explicit_file(tmp_path: Path):
missing = tmp_path / "missing.cti"

candidates, diag = gd.discover_cti_files(
cti_file=str(missing),
include_env=False,
)

assert candidates == []
assert diag.rejected == [(str(missing.resolve()), "not a file (explicit)")]


def test_discover_rejects_non_cti_file(tmp_path: Path):
not_cti = tmp_path / "producer.txt"
not_cti.write_text("", encoding="utf-8")

candidates, diag = gd.discover_cti_files(
cti_file=str(not_cti),
include_env=False,
)

assert candidates == []
assert diag.rejected == [(str(not_cti.resolve()), "not a .cti (explicit)")]


def test_discover_accepts_missing_cti_when_must_exist_false(tmp_path: Path):
missing = tmp_path / "missing.cti"

candidates, diag = gd.discover_cti_files(
cti_file=str(missing),
include_env=False,
must_exist=False,
)

assert candidates == [str(missing.resolve())]
assert diag.rejected == []


def test_discover_extra_dir_collects_cti_files_non_recursive(tmp_path: Path):
root = tmp_path / "ctis"
root.mkdir()

a = root / "a.cti"
b = root / "b.cti"
ignored = root / "ignored.txt"
nested = root / "nested"
nested.mkdir()
nested_cti = nested / "nested.cti"

a.write_text("", encoding="utf-8")
b.write_text("", encoding="utf-8")
ignored.write_text("", encoding="utf-8")
nested_cti.write_text("", encoding="utf-8")

candidates, diag = gd.discover_cti_files(
include_env=False,
extra_dirs=[str(root)],
recursive_extra_search=False,
)

assert candidates == [str(a.resolve()), str(b.resolve())]
assert diag.extra_dirs == [str(root)]


def test_discover_extra_dir_collects_cti_files_recursive(tmp_path: Path):
root = tmp_path / "ctis"
nested = root / "nested"
nested.mkdir(parents=True)

top = root / "top.cti"
child = nested / "child.cti"

top.write_text("", encoding="utf-8")
child.write_text("", encoding="utf-8")

candidates, _diag = gd.discover_cti_files(
include_env=False,
extra_dirs=[str(root)],
recursive_extra_search=True,
)

assert candidates == [str(top.resolve()), str(child.resolve())]


def test_discover_deduplicates_candidates_preserving_order(tmp_path: Path):
cti = tmp_path / "producer.cti"
cti.write_text("", encoding="utf-8")

candidates, diag = gd.discover_cti_files(
cti_file=str(cti),
cti_files=[str(cti)],
extra_dirs=[str(tmp_path)],
include_env=False,
)

assert candidates == [str(cti.resolve())]
assert diag.candidates == [str(cti.resolve())]


def test_discover_env_var_direct_file(monkeypatch, tmp_path: Path):
cti = tmp_path / "producer.cti"
cti.write_text("", encoding="utf-8")

monkeypatch.setenv("MY_GENTL_PATH", str(cti))

candidates, diag = gd.discover_cti_files(
include_env=True,
env_vars=("MY_GENTL_PATH",),
)

assert candidates == [str(cti.resolve())]
assert diag.env_vars_used == {"MY_GENTL_PATH": str(cti)}
assert diag.env_paths_expanded == [str(cti)]


def test_discover_env_var_directory(monkeypatch, tmp_path: Path):
cti = tmp_path / "producer.cti"
cti.write_text("", encoding="utf-8")

monkeypatch.setenv("MY_GENTL_PATH", str(tmp_path))

candidates, diag = gd.discover_cti_files(
include_env=True,
env_vars=("MY_GENTL_PATH",),
)

assert candidates == [str(cti.resolve())]
assert diag.env_vars_used == {"MY_GENTL_PATH": str(tmp_path)}
assert diag.env_paths_expanded == [str(tmp_path)]


def test_discover_env_var_multiple_entries(monkeypatch, tmp_path: Path):
d1 = tmp_path / "one"
d2 = tmp_path / "two"
d1.mkdir()
d2.mkdir()

cti1 = d1 / "one.cti"
cti2 = d2 / "two.cti"
cti1.write_text("", encoding="utf-8")
cti2.write_text("", encoding="utf-8")

monkeypatch.setenv("MY_GENTL_PATH", os.pathsep.join([str(d1), str(d2)]))

candidates, _diag = gd.discover_cti_files(
include_env=True,
env_vars=("MY_GENTL_PATH",),
)

assert candidates == [str(cti1.resolve()), str(cti2.resolve())]


def test_validate_glob_pattern_rejects_empty_pattern():
ok, reason = gd._validate_glob_pattern("")

assert ok is False
assert reason == "empty glob pattern"


def test_validate_glob_pattern_rejects_traversal(tmp_path: Path):
pattern = str(tmp_path / ".." / "*.cti")

ok, reason = gd._validate_glob_pattern(pattern)

assert ok is False
assert reason == "glob pattern contains '..' traversal"


def test_validate_glob_pattern_rejects_non_cti_pattern(tmp_path: Path):
pattern = str(tmp_path / "*.txt")

ok, reason = gd._validate_glob_pattern(pattern)

assert ok is False
assert reason == "glob pattern does not target .cti files"


def test_validate_glob_pattern_rejects_outside_allowed_roots(tmp_path: Path):
allowed = tmp_path / "allowed"
outside = tmp_path / "outside"
allowed.mkdir()
outside.mkdir()

pattern = str(outside / "*.cti")

ok, reason = gd._validate_glob_pattern(
pattern,
allowed_roots=[str(allowed)],
)

assert ok is False
assert reason == "glob pattern base is outside allowed roots"


def test_discover_glob_pattern_with_allowed_root(tmp_path: Path):
cti_dir = tmp_path / "ctis"
cti_dir.mkdir()

cti = cti_dir / "producer.cti"
cti.write_text("", encoding="utf-8")

candidates, diag = gd.discover_cti_files(
cti_search_paths=[str(cti_dir / "*.cti")],
include_env=False,
root_globs_allowed=[str(tmp_path)],
)

assert candidates == [str(cti.resolve())]
assert diag.rejected == []


def test_choose_cti_files_first_policy():
assert gd.choose_cti_files(
["a.cti", "b.cti", "c.cti"],
policy=gd.GenTLDiscoveryPolicy.FIRST,
max_files=2,
) == ["a.cti", "b.cti"]


def test_choose_cti_files_raise_if_multiple_policy_raises():
with pytest.raises(RuntimeError, match="Multiple GenTL producers"):
gd.choose_cti_files(
["a.cti", "b.cti"],
policy=gd.GenTLDiscoveryPolicy.RAISE_IF_MULTIPLE,
max_files=1,
)


def test_choose_cti_files_raise_if_multiple_policy_allows_within_limit():
assert gd.choose_cti_files(
["a.cti"],
policy=gd.GenTLDiscoveryPolicy.RAISE_IF_MULTIPLE,
max_files=1,
) == ["a.cti"]


def test_choose_cti_files_newest_policy(tmp_path: Path):
old = tmp_path / "old.cti"
new = tmp_path / "new.cti"

old.write_text("", encoding="utf-8")
new.write_text("", encoding="utf-8")

os.utime(old, (1000, 1000))
os.utime(new, (2000, 2000))

assert gd.choose_cti_files(
[str(old), str(new)],
policy=gd.GenTLDiscoveryPolicy.NEWEST,
max_files=1,
) == [str(new)]


def test_choose_cti_files_empty_candidates():
assert gd.choose_cti_files([]) == []


def test_choose_cti_files_unknown_policy_raises():
with pytest.raises(ValueError, match="Unknown policy"):
gd.choose_cti_files(
["a.cti"],
policy=object(), # type: ignore[arg-type]
)


def test_shared_harvester_pool_reuses_entry_and_refcounts(monkeypatch, tmp_path: Path):
calls: list[tuple[str, str | None]] = []

class FakeHarvester:
def __init__(self):
calls.append(("init", None))

def add_file(self, path: str) -> None:
calls.append(("add_file", path))

def update(self) -> None:
calls.append(("update", None))

def reset(self) -> None:
calls.append(("reset", None))

cti = tmp_path / "producer.cti"
cti.write_text("", encoding="utf-8")

monkeypatch.setattr(gd, "Harvester", FakeHarvester)
gd.SharedHarvesterPool._entries.clear()

entry1 = gd.SharedHarvesterPool.acquire([str(cti)])
entry2 = gd.SharedHarvesterPool.acquire([str(cti)])

assert entry1 is entry2
assert gd.SharedHarvesterPool.get_refcount(entry1) == 2

gd.SharedHarvesterPool.release(entry1)
assert gd.SharedHarvesterPool.get_refcount(entry2) == 1

gd.SharedHarvesterPool.release(entry2)
assert gd.SharedHarvesterPool.get_refcount(entry2) == 0
assert ("reset", None) in calls


def test_shared_harvester_entry_reports_failed_files(monkeypatch, tmp_path: Path):
class FakeHarvester:
def add_file(self, path: str) -> None:
raise RuntimeError("load failed")

def update(self) -> None:
raise AssertionError("update should not be called")

def reset(self) -> None:
pass

cti = tmp_path / "bad.cti"
cti.write_text("", encoding="utf-8")

monkeypatch.setattr(gd, "Harvester", FakeHarvester)

with pytest.raises(RuntimeError, match="No GenTL producer"):
gd.SharedHarvesterEntry([str(cti)])
Loading
Loading