From d8c673a8196f457efd7785ba84391ac02c7468bf Mon Sep 17 00:00:00 2001 From: Hugo Farajallah Date: Wed, 1 Apr 2026 14:43:01 +0200 Subject: [PATCH 1/2] test: add pure-Python tests for sim/loader, sim/logging, and utils/trials These tests run in CI without psychopy/numpy. They cover: - sim/loader: _deep_get, _import_attr, _resolve_spec helpers - sim/logging: _to_jsonable serialization, JSONL logger roundtrip - utils/trials: trial counter, resolve_deadline, resolve_trial_id --- tests/test_sim_loader.py | 78 +++++++++++++++++++++++++++++++ tests/test_sim_logging.py | 95 +++++++++++++++++++++++++++++++++++++ tests/test_utils_trials.py | 96 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 269 insertions(+) create mode 100644 tests/test_sim_loader.py create mode 100644 tests/test_sim_logging.py create mode 100644 tests/test_utils_trials.py diff --git a/tests/test_sim_loader.py b/tests/test_sim_loader.py new file mode 100644 index 0000000..2346e64 --- /dev/null +++ b/tests/test_sim_loader.py @@ -0,0 +1,78 @@ +"""Tests for psyflow.sim.loader — responder resolution and import helpers.""" + +import unittest + +from psyflow.sim.loader import _deep_get, _import_attr, _resolve_spec + + +class TestDeepGet(unittest.TestCase): + """_deep_get() nested dictionary traversal.""" + + def test_single_key(self): + self.assertEqual(_deep_get({"a": 1}, ("a",)), 1) + + def test_nested_keys(self): + self.assertEqual(_deep_get({"a": {"b": {"c": 3}}}, ("a", "b", "c")), 3) + + def test_missing_key_returns_default(self): + self.assertIsNone(_deep_get({"a": 1}, ("x",))) + self.assertEqual(_deep_get({"a": 1}, ("x",), "fallback"), "fallback") + + def test_none_mapping(self): + self.assertEqual(_deep_get(None, ("a",), "d"), "d") + + def test_non_dict_intermediate(self): + self.assertEqual(_deep_get({"a": 42}, ("a", "b"), "d"), "d") + + +class TestImportAttr(unittest.TestCase): + """_import_attr() dynamic import from dotted or colon-separated paths.""" + + def test_colon_syntax(self): + cls = _import_attr("collections:OrderedDict") + from collections import OrderedDict + self.assertIs(cls, OrderedDict) + + def test_dot_syntax(self): + cls = _import_attr("collections.OrderedDict") + from collections import OrderedDict + self.assertIs(cls, OrderedDict) + + def test_invalid_module_raises(self): + with self.assertRaises(ModuleNotFoundError): + _import_attr("nonexistent_module_xyz:Foo") + + +class TestResolveSpec(unittest.TestCase): + """_resolve_spec() config → (type, kwargs, source) resolution.""" + + def test_human_mode_returns_none(self): + spec, kwargs, source = _resolve_spec("human", {}) + self.assertIsNone(spec) + self.assertEqual(source, "disabled") + + def test_default_is_scripted(self): + spec, kwargs, source = _resolve_spec("sim", {}) + self.assertEqual(spec, "scripted") + self.assertEqual(source, "default") + + def test_config_type_used(self): + cfg = {"responder": {"type": "my_module:MyResponder", "kwargs": {"rt": 0.5}}} + spec, kwargs, source = _resolve_spec("qa", cfg) + self.assertEqual(spec, "my_module:MyResponder") + self.assertEqual(kwargs, {"rt": 0.5}) + self.assertEqual(source, "config.type") + + def test_empty_type_falls_back_to_scripted(self): + cfg = {"responder": {"type": " "}} + spec, kwargs, source = _resolve_spec("sim", cfg) + self.assertEqual(spec, "scripted") + + def test_non_dict_responder_ignored(self): + cfg = {"responder": "not_a_dict"} + spec, kwargs, source = _resolve_spec("sim", cfg) + self.assertEqual(spec, "scripted") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sim_logging.py b/tests/test_sim_logging.py new file mode 100644 index 0000000..98c0313 --- /dev/null +++ b/tests/test_sim_logging.py @@ -0,0 +1,95 @@ +"""Tests for psyflow.sim.logging — JSONL serialization and roundtrip.""" + +import json +import tempfile +import unittest +from dataclasses import dataclass +from pathlib import Path + +from psyflow.sim.logging import _to_jsonable, iter_sim_events, make_sim_jsonl_logger + + +class TestToJsonable(unittest.TestCase): + """_to_jsonable() should flatten dataclasses, dicts, and lists.""" + + def test_plain_values_pass_through(self): + self.assertEqual(_to_jsonable(42), 42) + self.assertEqual(_to_jsonable("hi"), "hi") + self.assertIsNone(_to_jsonable(None)) + + def test_nested_dict(self): + result = _to_jsonable({"a": {"b": [1, 2]}}) + self.assertEqual(result, {"a": {"b": [1, 2]}}) + + def test_dataclass_flattened(self): + @dataclass + class Pt: + x: int + y: int + + result = _to_jsonable(Pt(1, 2)) + self.assertEqual(result, {"x": 1, "y": 2}) + + def test_nested_dataclass_in_dict(self): + @dataclass + class Inner: + val: str + + result = _to_jsonable({"key": Inner("hello")}) + self.assertEqual(result, {"key": {"val": "hello"}}) + + def test_list_of_mixed(self): + @dataclass + class Tag: + name: str + + result = _to_jsonable([Tag("a"), 1, "b"]) + self.assertEqual(result, [{"name": "a"}, 1, "b"]) + + +class TestLoggerRoundtrip(unittest.TestCase): + """make_sim_jsonl_logger → iter_sim_events roundtrip.""" + + def test_write_and_read_back(self): + with tempfile.TemporaryDirectory() as td: + path = Path(td) / "events.jsonl" + logger = make_sim_jsonl_logger(path) + + logger({"type": "test", "value": 1}) + logger({"type": "test", "value": 2}) + + events = list(iter_sim_events(path)) + self.assertEqual(len(events), 2) + self.assertEqual(events[0]["type"], "test") + self.assertEqual(events[1]["value"], 2) + # Auto-injected timestamps + self.assertIn("t", events[0]) + self.assertIn("t_utc", events[0]) + + def test_iter_nonexistent_file_yields_nothing(self): + events = list(iter_sim_events("/tmp/does_not_exist_xyz.jsonl")) + self.assertEqual(events, []) + + def test_iter_skips_blank_and_invalid_lines(self): + with tempfile.TemporaryDirectory() as td: + path = Path(td) / "messy.jsonl" + path.write_text( + '{"ok": true}\n' + '\n' + 'not json\n' + '{"also": "ok"}\n', + encoding="utf-8", + ) + events = list(iter_sim_events(path)) + self.assertEqual(len(events), 2) + + def test_logger_creates_parent_dirs(self): + with tempfile.TemporaryDirectory() as td: + path = Path(td) / "sub" / "dir" / "events.jsonl" + logger = make_sim_jsonl_logger(path) + logger({"type": "init"}) + self.assertTrue(path.exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_utils_trials.py b/tests/test_utils_trials.py new file mode 100644 index 0000000..88b1b12 --- /dev/null +++ b/tests/test_utils_trials.py @@ -0,0 +1,96 @@ +"""Tests for psyflow.utils.trials — trial ID and deadline helpers.""" + +import importlib.util +import unittest + +# Load the module directly from its file path to avoid the psyflow.utils +# __init__.py which eagerly imports psychopy-dependent modules. +_spec = importlib.util.spec_from_file_location( + "psyflow.utils.trials", + "psyflow/utils/trials.py", +) +_mod = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_mod) + +next_trial_id = _mod.next_trial_id +reset_trial_counter = _mod.reset_trial_counter +resolve_deadline = _mod.resolve_deadline +resolve_trial_id = _mod.resolve_trial_id + + +class TestTrialCounter(unittest.TestCase): + """Global trial counter increment and reset.""" + + def setUp(self): + reset_trial_counter(0) + + def test_increments(self): + self.assertEqual(next_trial_id(), 1) + self.assertEqual(next_trial_id(), 2) + self.assertEqual(next_trial_id(), 3) + + def test_reset_to_custom_start(self): + reset_trial_counter(100) + self.assertEqual(next_trial_id(), 101) + + +class TestResolveDeadline(unittest.TestCase): + """resolve_deadline() scalar/list/tuple → float | None.""" + + def test_int(self): + self.assertEqual(resolve_deadline(5), 5.0) + + def test_float(self): + self.assertEqual(resolve_deadline(1.5), 1.5) + + def test_list_returns_max(self): + self.assertEqual(resolve_deadline([0.2, 0.5, 0.3]), 0.5) + + def test_tuple_returns_max(self): + self.assertEqual(resolve_deadline((1, 3, 2)), 3.0) + + def test_empty_list_returns_none(self): + self.assertIsNone(resolve_deadline([])) + + def test_none_returns_none(self): + self.assertIsNone(resolve_deadline(None)) + + def test_string_returns_none(self): + self.assertIsNone(resolve_deadline("fast")) + + +class TestResolveTrialId(unittest.TestCase): + """resolve_trial_id() from various input types.""" + + def test_none_passthrough(self): + self.assertIsNone(resolve_trial_id(None)) + + def test_int_passthrough(self): + self.assertEqual(resolve_trial_id(42), 42) + + def test_str_passthrough(self): + self.assertEqual(resolve_trial_id("trial_1"), "trial_1") + + def test_callable(self): + self.assertEqual(resolve_trial_id(lambda: 7), 7) + + def test_callable_returning_non_int_str_coerced(self): + result = resolve_trial_id(lambda: 3.14) + self.assertEqual(result, "3.14") + + def test_callable_exception_returns_none(self): + def bad(): + raise ValueError("boom") + self.assertIsNone(resolve_trial_id(bad)) + + def test_object_with_histories(self): + class FakeController: + histories = {"A": [1, 2], "B": [3]} + self.assertEqual(resolve_trial_id(FakeController()), 4) + + def test_unknown_type_coerced_to_str(self): + self.assertEqual(resolve_trial_id(42.0), "42.0") + + +if __name__ == "__main__": + unittest.main() From f0da889b2afd6454bc9b6e5ed761d252a17e0f3b Mon Sep 17 00:00:00 2001 From: Hugo Farajallah Date: Wed, 1 Apr 2026 14:44:04 +0200 Subject: [PATCH 2/2] fix: guard add_subinfo() when save_path is None or empty The else branch incorrectly printed "Output directory already exists: None" when save_path was falsy. Also guard os.path.join calls that would crash with TypeError on None save_path. --- psyflow/TaskSettings.py | 18 +++++++------ tests/test_TaskSettings.py | 55 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 8 deletions(-) create mode 100644 tests/test_TaskSettings.py diff --git a/psyflow/TaskSettings.py b/psyflow/TaskSettings.py index faf119b..aab87dd 100644 --- a/psyflow/TaskSettings.py +++ b/psyflow/TaskSettings.py @@ -177,11 +177,12 @@ def add_subinfo(self, subinfo: Dict[str, Any]) -> None: self.set_block_seed(self.overall_seed) # Ensure save path exists - if self.save_path and not os.path.exists(self.save_path): - os.makedirs(self.save_path) - print(f"[INFO] Created output directory: {self.save_path}") - else: - print(f"[INFO] Output directory already exists: {self.save_path}") + if self.save_path: + if not os.path.exists(self.save_path): + os.makedirs(self.save_path) + print(f"[INFO] Created output directory: {self.save_path}") + else: + print(f"[INFO] Output directory already exists: {self.save_path}") # Construct log/result filenames timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') @@ -190,9 +191,10 @@ def add_subinfo(self, subinfo: Dict[str, Any]) -> None: else: basename = f"sub-{subject_id}_{timestamp}" - self.log_file = os.path.join(self.save_path, f"{basename}.log") - self.res_file = os.path.join(self.save_path, f"{basename}.csv") - self.json_file = os.path.join(self.save_path, f"{basename}.json") + if self.save_path: + self.log_file = os.path.join(self.save_path, f"{basename}.log") + self.res_file = os.path.join(self.save_path, f"{basename}.csv") + self.json_file = os.path.join(self.save_path, f"{basename}.json") def __repr__(self) -> str: """ diff --git a/tests/test_TaskSettings.py b/tests/test_TaskSettings.py new file mode 100644 index 0000000..237a0bc --- /dev/null +++ b/tests/test_TaskSettings.py @@ -0,0 +1,55 @@ +"""Tests for psyflow.TaskSettings — add_subinfo edge cases.""" + +import os +import tempfile +import unittest +from io import StringIO +from unittest.mock import patch + +from psyflow.TaskSettings import TaskSettings + + +class TestAddSubinfoSavePath(unittest.TestCase): + """add_subinfo() should not claim a directory exists when save_path is None.""" + + def test_none_save_path_does_not_print_exists(self): + settings = TaskSettings(save_path=None) + + with patch("sys.stdout", new_callable=StringIO) as mock_out: + settings.add_subinfo({"subject_id": "001"}) + + output = mock_out.getvalue() + self.assertNotIn("already exists", output) + + def test_empty_save_path_does_not_print_exists(self): + settings = TaskSettings(save_path="") + + with patch("sys.stdout", new_callable=StringIO) as mock_out: + settings.add_subinfo({"subject_id": "001"}) + + output = mock_out.getvalue() + self.assertNotIn("already exists", output) + + def test_existing_dir_prints_exists(self): + with tempfile.TemporaryDirectory() as td: + settings = TaskSettings(save_path=td) + + with patch("sys.stdout", new_callable=StringIO) as mock_out: + settings.add_subinfo({"subject_id": "001"}) + + self.assertIn("already exists", mock_out.getvalue()) + + def test_new_dir_is_created(self): + with tempfile.TemporaryDirectory() as td: + new_dir = os.path.join(td, "outputs", "human") + settings = TaskSettings(save_path=new_dir) + + with patch("sys.stdout", new_callable=StringIO) as mock_out: + settings.add_subinfo({"subject_id": "001"}) + + self.assertTrue(os.path.isdir(new_dir)) + self.assertIn("Created", mock_out.getvalue()) + + +if __name__ == "__main__": + unittest.main()