test: add pure-Python tests + fix TaskSettings.add_subinfo save_path

ChangLog.md

@@ -1,5 +1,14 @@
 # psyflow change log
 
+## 0.1.23 (2026-04-05)
+
+### Summary
+- Restored `TaskSettings.add_subinfo()` to fall back to `./outputs/human` when `save_path` is unset or empty, while still creating explicit nested output directories.
+- Added pure-Python regression tests for `psyflow.sim.loader`, `psyflow.sim.logging`, `psyflow.utils.trials`, and `TaskSettings.add_subinfo()`.
+
+### Validation
+- `python -m unittest discover -s tests -v` passed.
+
 ## 0.1.22 (2026-04-05)
 
 ### Summary

psyflow/TaskSettings.py

@@ -14,6 +14,8 @@
 from datetime import datetime
 import os
 
+DEFAULT_SAVE_PATH = "./outputs/human"
+
 @dataclass
 class TaskSettings:
     """
@@ -62,7 +64,7 @@ class TaskSettings:
     json_file: Optional[str] = None
 
     # --- File path info ---
-    save_path: Optional[str] = './outputs/human'
+    save_path: Optional[str] = field(default_factory=lambda: DEFAULT_SAVE_PATH)
     task_name: Optional[str] = None
 
     def __post_init__(self):
@@ -176,8 +178,12 @@ def add_subinfo(self, subinfo: Dict[str, Any]) -> None:
             self.overall_seed = int(hashlib.sha256(str(subject_id).encode()).hexdigest(), 16) % (10**8)
             self.set_block_seed(self.overall_seed)
 
-        # Ensure save path exists
-        if self.save_path and not os.path.exists(self.save_path):
+        # Normalize missing output root to the package default.
+        if not self.save_path:
+            self.save_path = DEFAULT_SAVE_PATH
+
+        # Ensure save path exists.
+        if not os.path.exists(self.save_path):
             os.makedirs(self.save_path)
             print(f"[INFO] Created output directory: {self.save_path}")
         else:

tests/test_TaskSettings.py

@@ -0,0 +1,60 @@
+"""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 fall back to the default output directory."""
+
+    def _assert_default_output_dir(self, save_path):
+        with tempfile.TemporaryDirectory() as td:
+            fallback_dir = os.path.join(td, "outputs", "human")
+
+            with patch("psyflow.TaskSettings.DEFAULT_SAVE_PATH", fallback_dir):
+                settings = TaskSettings(save_path=save_path)
+
+                with patch("sys.stdout", new_callable=StringIO) as mock_out:
+                    settings.add_subinfo({"subject_id": "001"})
+
+            self.assertEqual(settings.save_path, fallback_dir)
+            self.assertTrue(os.path.isdir(fallback_dir))
+            self.assertIn("Created", mock_out.getvalue())
+            self.assertTrue(settings.log_file.startswith(fallback_dir))
+            self.assertTrue(settings.res_file.startswith(fallback_dir))
+            self.assertTrue(settings.json_file.startswith(fallback_dir))
+
+    def test_none_save_path_uses_default_output_dir(self):
+        self._assert_default_output_dir(None)
+
+    def test_empty_save_path_uses_default_output_dir(self):
+        self._assert_default_output_dir("")
+
+    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()

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()

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()

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()

website/src/data/generated/changelog.json

@@ -1,4 +1,12 @@
 [
+  {
+    "version": "0.1.23",
+    "date": "2026-04-05",
+    "summary": [
+      "Restored `TaskSettings.add_subinfo()` to fall back to `./outputs/human` when `save_path` is unset or empty, while still creating explicit nested output directories.",
+      "Added pure-Python regression tests for `psyflow.sim.loader`, `psyflow.sim.logging`, `psyflow.utils.trials`, and `TaskSettings.add_subinfo()`."
+    ]
+  },
   {
     "version": "0.1.22",
     "date": "2026-04-05",