more adjustments
This commit is contained in:
@@ -4,33 +4,38 @@ from src import paths
|
||||
|
||||
# We mock App to avoid the heavy initialization logic
|
||||
class MockApp:
|
||||
def __init__(self):
|
||||
self.ui_conductor_dir = '/mock/conductor'
|
||||
self.ui_logs_dir = '/mock/logs'
|
||||
self.ui_scripts_dir = '/mock/scripts'
|
||||
self.config = {"paths": {}}
|
||||
self.ai_status = ""
|
||||
|
||||
from src.gui_2 import App
|
||||
_save_paths = App._save_paths
|
||||
def __init__(self):
|
||||
self.ui_conductor_dir = '/mock/conductor'
|
||||
self.ui_logs_dir = '/mock/logs'
|
||||
self.ui_scripts_dir = '/mock/scripts'
|
||||
self.config = {"paths": {}}
|
||||
self.ai_status = ""
|
||||
|
||||
def init_state(self):
|
||||
pass
|
||||
|
||||
from src.gui_2 import App
|
||||
_save_paths = App._save_paths
|
||||
|
||||
def test_save_paths():
|
||||
mock_app = MockApp()
|
||||
|
||||
with patch('src.models.save_config') as mock_save, \
|
||||
patch('shutil.copy') as mock_copy, \
|
||||
patch('src.paths.get_config_path') as mock_get_cfg, \
|
||||
patch('src.paths.reset_resolved') as mock_reset:
|
||||
|
||||
mock_get_cfg.return_value = MagicMock()
|
||||
mock_get_cfg.return_value.exists.return_value = True
|
||||
|
||||
mock_app.ui_conductor_dir = '/new/conductor'
|
||||
mock_app._save_paths()
|
||||
|
||||
# Verify config update
|
||||
assert mock_app.config['paths']['conductor_dir'] == '/new/conductor'
|
||||
mock_save.assert_called_once()
|
||||
mock_copy.assert_called_once()
|
||||
assert 'restart required' in mock_app.ai_status
|
||||
mock_reset.assert_called_once()
|
||||
mock_app = MockApp()
|
||||
|
||||
with patch('src.models.save_config') as mock_save, \
|
||||
patch('shutil.copy') as mock_copy, \
|
||||
patch('src.paths.get_config_path') as mock_get_cfg, \
|
||||
patch('src.paths.reset_resolved') as mock_reset, \
|
||||
patch.object(MockApp, 'init_state') as mock_init:
|
||||
|
||||
mock_get_cfg.return_value = MagicMock()
|
||||
mock_get_cfg.return_value.exists.return_value = True
|
||||
|
||||
mock_app.ui_conductor_dir = '/new/conductor'
|
||||
mock_app._save_paths()
|
||||
|
||||
# Verify config update
|
||||
assert 'conductor_dir' not in mock_app.config['paths']
|
||||
mock_save.assert_called_once()
|
||||
mock_copy.assert_called_once()
|
||||
assert 'applied' in mock_app.ai_status
|
||||
mock_reset.assert_called_once()
|
||||
mock_init.assert_called_once()
|
||||
|
||||
@@ -9,25 +9,15 @@ def reset_paths():
|
||||
yield
|
||||
paths.reset_resolved()
|
||||
|
||||
def test_default_paths():
|
||||
def test_default_paths(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("SLOP_CONFIG", str(tmp_path / "non_existent.toml"))
|
||||
root_dir = Path(paths.__file__).resolve().parent.parent
|
||||
assert paths.get_conductor_dir() == root_dir / "conductor"
|
||||
assert paths.get_logs_dir() == root_dir / "logs/sessions"
|
||||
assert paths.get_scripts_dir() == root_dir / "scripts/generated"
|
||||
# config path should now be an absolute path relative to src/paths.py
|
||||
assert paths.get_config_path() == root_dir / "config.toml"
|
||||
assert paths.get_tracks_dir() == root_dir / "conductor/tracks"
|
||||
assert paths.get_archive_dir() == root_dir / "conductor/archive"
|
||||
# config path should be what we set in env
|
||||
assert paths.get_config_path() == tmp_path / "non_existent.toml"
|
||||
|
||||
def test_env_var_overrides(tmp_path, monkeypatch):
|
||||
root_dir = Path(paths.__file__).resolve().parent.parent
|
||||
|
||||
# Relative env var (resolved against root_dir)
|
||||
monkeypatch.setenv("SLOP_CONDUCTOR_DIR", "custom_conductor")
|
||||
assert paths.get_conductor_dir() == (root_dir / "custom_conductor").resolve()
|
||||
|
||||
paths.reset_resolved()
|
||||
|
||||
# Absolute env var
|
||||
abs_logs = (tmp_path / "abs_logs").resolve()
|
||||
monkeypatch.setenv("SLOP_LOGS_DIR", str(abs_logs))
|
||||
@@ -38,14 +28,12 @@ def test_config_overrides(tmp_path, monkeypatch):
|
||||
config_file = tmp_path / "custom_config.toml"
|
||||
content = """
|
||||
[paths]
|
||||
conductor_dir = "cfg_conductor"
|
||||
logs_dir = "cfg_logs"
|
||||
scripts_dir = "cfg_scripts"
|
||||
"""
|
||||
config_file.write_text(content)
|
||||
monkeypatch.setenv("SLOP_CONFIG", str(config_file))
|
||||
|
||||
assert paths.get_conductor_dir() == (root_dir / "cfg_conductor").resolve()
|
||||
assert paths.get_logs_dir() == root_dir / "cfg_logs"
|
||||
assert paths.get_scripts_dir() == root_dir / "cfg_scripts"
|
||||
|
||||
@@ -54,11 +42,20 @@ def test_precedence(tmp_path, monkeypatch):
|
||||
config_file = tmp_path / "custom_config.toml"
|
||||
content = """
|
||||
[paths]
|
||||
conductor_dir = "cfg_conductor"
|
||||
logs_dir = "cfg_logs"
|
||||
"""
|
||||
config_file.write_text(content)
|
||||
monkeypatch.setenv("SLOP_CONFIG", str(config_file))
|
||||
monkeypatch.setenv("SLOP_CONDUCTOR_DIR", "env_conductor")
|
||||
monkeypatch.setenv("SLOP_LOGS_DIR", "env_logs")
|
||||
|
||||
# Env var should take precedence over config
|
||||
assert paths.get_conductor_dir() == (root_dir / "env_conductor").resolve()
|
||||
assert paths.get_logs_dir() == (root_dir / "env_logs").resolve()
|
||||
|
||||
def test_conductor_dir_project_relative(tmp_path):
|
||||
# Should default to tmp_path/conductor
|
||||
project_path = str(tmp_path)
|
||||
base = (tmp_path / 'conductor').resolve()
|
||||
assert paths.get_conductor_dir(project_path) == base
|
||||
assert paths.get_tracks_dir(project_path) == base / "tracks"
|
||||
assert paths.get_archive_dir(project_path) == base / "archive"
|
||||
assert paths.get_track_state_dir("test_track", project_path) == base / "tracks" / "test_track"
|
||||
|
||||
@@ -7,82 +7,53 @@ from src import paths
|
||||
from src import project_manager
|
||||
|
||||
def test_get_conductor_dir_default():
|
||||
paths.reset_resolved()
|
||||
# Should return absolute path to "conductor" in project root
|
||||
expected = Path(__file__).resolve().parent.parent / "conductor"
|
||||
assert paths.get_conductor_dir() == expected
|
||||
paths.reset_resolved()
|
||||
# Should return absolute path to "conductor" in project root
|
||||
expected = Path(__file__).resolve().parent.parent / "conductor"
|
||||
assert paths.get_conductor_dir() == expected
|
||||
|
||||
def test_get_conductor_dir_project_specific_with_toml(tmp_path):
|
||||
paths.reset_resolved()
|
||||
project_root = tmp_path / "my_project"
|
||||
project_root.mkdir()
|
||||
|
||||
# Create manual_slop.toml with custom conductor dir
|
||||
toml_path = project_root / "manual_slop.toml"
|
||||
config = {
|
||||
"conductor": {
|
||||
"dir": "custom_tracks"
|
||||
}
|
||||
}
|
||||
with open(toml_path, "wb") as f:
|
||||
f.write(tomli_w.dumps(config).encode())
|
||||
|
||||
res = paths.get_conductor_dir(project_path=str(project_root))
|
||||
assert res == project_root / "custom_tracks"
|
||||
paths.reset_resolved()
|
||||
project_root = tmp_path / "my_project"
|
||||
project_root.mkdir()
|
||||
|
||||
# Create manual_slop.toml with custom conductor dir
|
||||
toml_path = project_root / "manual_slop.toml"
|
||||
config = {
|
||||
"conductor": {
|
||||
"dir": "custom_tracks"
|
||||
}
|
||||
}
|
||||
with open(toml_path, "wb") as f:
|
||||
f.write(tomli_w.dumps(config).encode())
|
||||
|
||||
res = paths.get_conductor_dir(project_path=str(project_root))
|
||||
assert res == project_root / "custom_tracks"
|
||||
|
||||
def test_get_all_tracks_project_specific(tmp_path):
|
||||
paths.reset_resolved()
|
||||
project_root = tmp_path / "my_project"
|
||||
project_root.mkdir()
|
||||
|
||||
# Custom conductor dir
|
||||
custom_dir = project_root / "my_conductor"
|
||||
custom_dir.mkdir()
|
||||
tracks_dir = custom_dir / "tracks"
|
||||
tracks_dir.mkdir()
|
||||
|
||||
# Create a dummy track
|
||||
track_dir = tracks_dir / "test_track_20260312"
|
||||
track_dir.mkdir()
|
||||
with open(track_dir / "metadata.json", "w") as f:
|
||||
json.dump({"id": "test_track", "title": "Test Track"}, f)
|
||||
|
||||
# Setup manual_slop.toml
|
||||
toml_path = project_root / "manual_slop.toml"
|
||||
config = {"conductor": {"dir": "my_conductor"}}
|
||||
with open(toml_path, "wb") as f:
|
||||
f.write(tomli_w.dumps(config).encode())
|
||||
|
||||
# project_manager.get_all_tracks(base_dir) should now find it
|
||||
tracks = project_manager.get_all_tracks(str(project_root))
|
||||
assert len(tracks) == 1
|
||||
assert tracks[0]["title"] == "Test Track"
|
||||
|
||||
def test_get_all_tracks_global_fallback(tmp_path):
|
||||
paths.reset_resolved()
|
||||
|
||||
# Create a directory without manual_slop.toml
|
||||
empty_dir = tmp_path / "empty_project"
|
||||
empty_dir.mkdir()
|
||||
|
||||
# Setup a fake global conductor
|
||||
global_conductor = tmp_path / "global_conductor"
|
||||
global_conductor.mkdir()
|
||||
global_tracks = global_conductor / "tracks"
|
||||
global_tracks.mkdir()
|
||||
|
||||
track_dir = global_tracks / "global_track"
|
||||
track_dir.mkdir()
|
||||
with open(track_dir / "metadata.json", "w") as f:
|
||||
json.dump({"id": "global_track", "title": "Global Track"}, f)
|
||||
|
||||
# Override global conductor dir via env var
|
||||
os.environ["SLOP_CONDUCTOR_DIR"] = str(global_conductor)
|
||||
try:
|
||||
paths.reset_resolved()
|
||||
# Pass project_path pointing to a dir without TOML
|
||||
tracks = project_manager.get_all_tracks(str(empty_dir))
|
||||
# paths.get_conductor_dir(str(empty_dir)) should fall back to global
|
||||
assert any(t["id"] == "global_track" for t in tracks)
|
||||
finally:
|
||||
del os.environ["SLOP_CONDUCTOR_DIR"]
|
||||
paths.reset_resolved()
|
||||
project_root = tmp_path / "my_project"
|
||||
project_root.mkdir()
|
||||
|
||||
# Custom conductor dir
|
||||
custom_dir = project_root / "my_conductor"
|
||||
custom_dir.mkdir()
|
||||
tracks_dir = custom_dir / "tracks"
|
||||
tracks_dir.mkdir()
|
||||
|
||||
# Create a dummy track
|
||||
track_dir = tracks_dir / "test_track_20260312"
|
||||
track_dir.mkdir()
|
||||
with open(track_dir / "metadata.json", "w") as f:
|
||||
json.dump({"id": "test_track", "title": "Test Track"}, f)
|
||||
|
||||
# Setup manual_slop.toml
|
||||
toml_path = project_root / "manual_slop.toml"
|
||||
config = {"conductor": {"dir": "my_conductor"}}
|
||||
with open(toml_path, "wb") as f:
|
||||
f.write(tomli_w.dumps(config).encode())
|
||||
|
||||
# project_manager.get_all_tracks(base_dir) should now find it
|
||||
tracks = project_manager.get_all_tracks(str(project_root))
|
||||
assert len(tracks) == 1
|
||||
assert tracks[0]["title"] == "Test Track"
|
||||
|
||||
51
tests/test_session_logger_reset.py
Normal file
51
tests/test_session_logger_reset.py
Normal file
@@ -0,0 +1,51 @@
|
||||
import pytest
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
from typing import Generator
|
||||
from src import session_logger
|
||||
import time
|
||||
|
||||
@pytest.fixture
|
||||
def temp_logs(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[Path, None, None]:
|
||||
# Ensure closed before starting
|
||||
session_logger.close_session()
|
||||
monkeypatch.setattr(session_logger, "_comms_fh", None)
|
||||
|
||||
log_dir = tmp_path / "logs"
|
||||
scripts_dir = tmp_path / "scripts" / "generated"
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
scripts_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
from src import paths
|
||||
monkeypatch.setattr(paths, "get_logs_dir", lambda: log_dir)
|
||||
monkeypatch.setattr(paths, "get_scripts_dir", lambda: scripts_dir)
|
||||
|
||||
yield log_dir
|
||||
# Cleanup: Close handles if open
|
||||
session_logger.close_session()
|
||||
|
||||
def test_reset_session(temp_logs: Path) -> None:
|
||||
# 1. Open first session
|
||||
label1 = "label1"
|
||||
session_logger.open_session(label=label1)
|
||||
|
||||
subdirs = [d for d in temp_logs.iterdir() if d.is_dir()]
|
||||
assert len(subdirs) == 1
|
||||
session1_dir = subdirs[0]
|
||||
assert session1_dir.name.endswith(f"_{label1}")
|
||||
assert session_logger._comms_fh is not None
|
||||
|
||||
# 2. Reset to second session
|
||||
time.sleep(1.1)
|
||||
|
||||
label2 = "label2"
|
||||
session_logger.reset_session(label=label2)
|
||||
|
||||
subdirs = sorted([d for d in temp_logs.iterdir() if d.is_dir()], key=lambda x: x.name)
|
||||
assert len(subdirs) == 2
|
||||
session2_dir = subdirs[1]
|
||||
assert session2_dir.name.endswith(f"_{label2}")
|
||||
assert session_logger._comms_fh is not None
|
||||
|
||||
# Verify new handle points to new dir
|
||||
assert str(session2_dir) in str(Path(session_logger._comms_fh.name).resolve())
|
||||
Reference in New Issue
Block a user