Private
Public Access
refactor(paths): v3 design - explicit initialize_paths + frozen PathsConfig singleton
This commit is contained in:
@@ -42,9 +42,8 @@ args: argparse.Namespace = argparse.Namespace() # type: ignore[assignment]
|
|||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
args = parser.parse_args()
|
args = parser.parse_args()
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from src.paths import set_config_override
|
from src.paths import initialize_paths
|
||||||
if args.config:
|
initialize_paths(Path(args.config).resolve() if args.config else None)
|
||||||
set_config_override(Path(args.config).resolve())
|
|
||||||
if args.web_host is not None:
|
if args.web_host is not None:
|
||||||
with startup_profiler.phase("web_host_imports"):
|
with startup_profiler.phase("web_host_imports"):
|
||||||
from imgui_bundle import hello_imgui
|
from imgui_bundle import hello_imgui
|
||||||
|
|||||||
@@ -1804,7 +1804,7 @@ class AppController:
|
|||||||
spath = Path(proj_paths['scripts_dir'])
|
spath = Path(proj_paths['scripts_dir'])
|
||||||
if not spath.is_absolute(): spath = project_root / spath
|
if not spath.is_absolute(): spath = project_root / spath
|
||||||
os.environ['SLOP_SCRIPTS_DIR'] = str(spath)
|
os.environ['SLOP_SCRIPTS_DIR'] = str(spath)
|
||||||
paths.reset_resolved()
|
paths.initialize_paths(paths.get_config_path())
|
||||||
|
|
||||||
path_info = paths.get_full_path_info()
|
path_info = paths.get_full_path_info()
|
||||||
self.ui_logs_dir = str(path_info['logs_dir']['path'])
|
self.ui_logs_dir = str(path_info['logs_dir']['path'])
|
||||||
|
|||||||
+8
-1
@@ -1311,7 +1311,7 @@ class App:
|
|||||||
cfg_path = paths.get_config_path()
|
cfg_path = paths.get_config_path()
|
||||||
if cfg_path.exists(): shutil.copy(cfg_path, str(cfg_path) + ".bak")
|
if cfg_path.exists(): shutil.copy(cfg_path, str(cfg_path) + ".bak")
|
||||||
self.save_config()
|
self.save_config()
|
||||||
paths.reset_resolved()
|
paths.initialize_paths(cfg_path)
|
||||||
self.init_state()
|
self.init_state()
|
||||||
self.ai_status = 'paths applied and session reset'
|
self.ai_status = 'paths applied and session reset'
|
||||||
|
|
||||||
@@ -2345,6 +2345,13 @@ def render_paths_panel(app: App) -> None:
|
|||||||
imgui.separator()
|
imgui.separator()
|
||||||
if imgui.button("Apply", imgui.ImVec2(120, 0)): app._save_paths()
|
if imgui.button("Apply", imgui.ImVec2(120, 0)): app._save_paths()
|
||||||
imgui.same_line()
|
imgui.same_line()
|
||||||
|
if imgui.button("Refresh Paths", imgui.ImVec2(140, 0)):
|
||||||
|
paths.initialize_paths(paths.get_config_path())
|
||||||
|
app.init_state()
|
||||||
|
app.ai_status = "paths reloaded from config.toml"
|
||||||
|
if imgui.is_item_hovered():
|
||||||
|
imgui.set_tooltip("Re-read [paths] section from config.toml and rebuild the PathsConfig singleton. Use after editing config.toml directly or after importing a new config.")
|
||||||
|
imgui.same_line()
|
||||||
if imgui.button("Reset", imgui.ImVec2(120, 0)):
|
if imgui.button("Reset", imgui.ImVec2(120, 0)):
|
||||||
app.init_state()
|
app.init_state()
|
||||||
app.ai_status = "paths reset to defaults"
|
app.ai_status = "paths reset to defaults"
|
||||||
|
|||||||
+199
-192
@@ -1,189 +1,231 @@
|
|||||||
"""
|
"""
|
||||||
Paths - Centralized path resolution for configuration and environment variables.
|
Paths - Single source of truth for all application paths.
|
||||||
|
|
||||||
This module provides centralized path resolution for all configurable paths in the application.
|
All paths are resolved ONCE at startup via `initialize_paths(config_path)`,
|
||||||
All paths can be overridden via environment variables or config.toml.
|
which reads the active config.toml's `[paths]` section (with env-var overrides)
|
||||||
|
and builds an immutable `PathsConfig` snapshot. Path getters are trivial
|
||||||
|
lookups into this snapshot.
|
||||||
|
|
||||||
Environment Variables:
|
**Usage contract:**
|
||||||
SLOP_CONFIG: Path to config.toml
|
|
||||||
SLOP_LOGS_DIR: Path to logs directory
|
|
||||||
SLOP_SCRIPTS_DIR: Path to generated scripts directory
|
|
||||||
|
|
||||||
Configuration (config.toml):
|
1. Call `initialize_paths(config_path)` ONCE at process startup, BEFORE any
|
||||||
[paths]
|
path getter is invoked. This is the only correct entry point.
|
||||||
logs_dir = "logs/sessions"
|
2. After init, all `get_*_path()` functions return cached `Path` objects.
|
||||||
scripts_dir = "scripts/generated"
|
3. To change paths (e.g., in tests), call `initialize_paths(new_config_path)`
|
||||||
|
again — atomic swap under lock. Do not mutate `PathsConfig` instances;
|
||||||
|
they are frozen.
|
||||||
|
|
||||||
Path Functions:
|
**Thread safety:** The singleton swap is guarded by an RLock. `PathsConfig`
|
||||||
get_config_path() -> Path to config.toml
|
is a `@dataclass(frozen=True)`, so reads of individual fields are atomic.
|
||||||
get_conductor_dir(project_path=None) -> Path to conductor directory
|
Reader threads see a consistent snapshot; writer threads serialize through
|
||||||
get_logs_dir() -> Path to logs/sessions
|
the lock. No partial writes.
|
||||||
get_scripts_dir() -> Path to scripts/generated
|
|
||||||
get_tracks_dir(project_path=None) -> Path to conductor/tracks
|
|
||||||
get_track_state_dir(track_id, project_path=None) -> Path to conductor/tracks/<track_id>
|
|
||||||
get_archive_dir(project_path=None) -> Path to conductor/archive
|
|
||||||
|
|
||||||
Resolution Order:
|
**Resolution priority** (per key, in `initialize_paths`):
|
||||||
1. Check project-specific manual_slop.toml (for conductor paths)
|
1. Env var (e.g., `SLOP_GLOBAL_PRESETS`) if set
|
||||||
2. Check environment variable (for logs/scripts)
|
2. `config.toml [paths]` entry if present
|
||||||
3. Check config.toml [paths] section (for logs/scripts)
|
3. Default `<project_root>/<default_filename>`
|
||||||
4. Fall back to default
|
|
||||||
|
|
||||||
Usage:
|
**Codepath ordering:**
|
||||||
from src.paths import get_logs_dir, get_scripts_dir
|
|
||||||
|
|
||||||
logs_dir = get_logs_dir()
|
The major codepaths that consume paths are:
|
||||||
scripts_dir = get_scripts_dir()
|
- `sloppy.py` (production GUI entry point)
|
||||||
|
- `src/app_controller.py:AppController.__init__`
|
||||||
|
- `src/presets.py`, `src/tool_presets.py`, `src/personas.py`, etc.
|
||||||
|
|
||||||
See Also:
|
`initialize_paths()` must run BEFORE any of these. In sloppy.py, it runs
|
||||||
- docs/guide_tools.md for configuration documentation
|
at the top of `__main__`. In tests, it runs at conftest module body (before
|
||||||
- src/session_logger.py for logging paths
|
any src/ import). In other contexts (e.g., direct library use), the caller
|
||||||
- src/project_manager.py for project paths
|
is responsible.
|
||||||
|
|
||||||
|
If a path getter is called before `initialize_paths()`, a `RuntimeError`
|
||||||
|
is raised. This catches the "bad programmer" case where ordering is wrong.
|
||||||
"""
|
"""
|
||||||
import os
|
import os
|
||||||
|
import threading
|
||||||
import tomllib
|
import tomllib
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Optional, Any
|
from typing import Optional, Any
|
||||||
|
|
||||||
|
|
||||||
_RESOLVED: dict[str, Path] = {}
|
@dataclass(frozen=True)
|
||||||
|
class PathsConfig:
|
||||||
_CONFIG_OVERRIDE: Path | None = None
|
"""Immutable snapshot of resolved paths. Created ONCE per process.
|
||||||
|
[C: src/paths.py:initialize_paths, src/paths.py:_cfg]"""
|
||||||
|
config_path: Path
|
||||||
|
presets: Path
|
||||||
|
tool_presets: Path
|
||||||
|
personas: Path
|
||||||
|
themes: Path
|
||||||
|
workspace_profiles: Path
|
||||||
|
credentials: Path
|
||||||
|
logs_dir: Path
|
||||||
|
scripts_dir: Path
|
||||||
|
|
||||||
|
|
||||||
def set_config_override(path: Path | None) -> None:
|
_PATHS_CONFIG: Optional[PathsConfig] = None
|
||||||
"""
|
_PATHS_LOCK = threading.RLock()
|
||||||
Set the active config.toml path. Pass None to use the default
|
|
||||||
<project_root>/config.toml. The CLI --config flag is the ONLY
|
|
||||||
supported override mechanism; the historical SLOP_CONFIG env var
|
|
||||||
fallback has been removed (per test_sandbox_hardening_20260619 FR2).
|
|
||||||
[C: sloppy.py:main, tests/conftest.py]
|
|
||||||
"""
|
|
||||||
global _CONFIG_OVERRIDE
|
|
||||||
_CONFIG_OVERRIDE = path
|
|
||||||
_RESOLVED.clear()
|
|
||||||
|
|
||||||
|
|
||||||
|
def _resolve_path(env_var: str, config_key: str, default: Path, config_path: Path) -> Path:
|
||||||
|
"""Internal: resolve one path from env var -> config [paths] -> default.
|
||||||
|
Called only from initialize_paths(). Not thread-safe; caller holds lock."""
|
||||||
|
root_dir = Path(__file__).resolve().parent.parent
|
||||||
|
if env_var in os.environ:
|
||||||
|
return Path(os.environ[env_var])
|
||||||
|
try:
|
||||||
|
with open(config_path, "rb") as f:
|
||||||
|
cfg = tomllib.load(f)
|
||||||
|
if "paths" in cfg and config_key in cfg["paths"]:
|
||||||
|
p = Path(cfg["paths"][config_key])
|
||||||
|
return p if p.is_absolute() else root_dir / p
|
||||||
|
except (FileNotFoundError, tomllib.TOMLDecodeError):
|
||||||
|
pass
|
||||||
|
return default if default.is_absolute() else root_dir / default
|
||||||
|
|
||||||
|
|
||||||
|
def initialize_paths(config_path: Optional[Path] = None) -> PathsConfig:
|
||||||
|
"""Initialize the global paths singleton. Call this ONCE at startup,
|
||||||
|
BEFORE any path getter is invoked. Atomic swap under RLock.
|
||||||
|
|
||||||
|
If config_path is None, uses the default `<project_root>/config.toml`.
|
||||||
|
|
||||||
|
This is the SOLE entry point for setting the path graph at runtime.
|
||||||
|
Tests re-init to reset.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
OSError: if the config_path cannot be opened (other than FileNotFoundError
|
||||||
|
which is treated as "no [paths] overrides, use defaults")
|
||||||
|
TypeError: if config_path is not a Path
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
The newly installed PathsConfig snapshot.
|
||||||
|
[C: src/paths.py:_cfg, tests/conftest.py:_setup_test_paths, sloppy.py:main]"""
|
||||||
|
global _PATHS_CONFIG
|
||||||
|
if config_path is None:
|
||||||
|
root_dir = Path(__file__).resolve().parent.parent
|
||||||
|
config_path = root_dir / "config.toml"
|
||||||
|
config_path = Path(config_path).resolve()
|
||||||
|
|
||||||
|
root_dir = Path(__file__).resolve().parent.parent
|
||||||
|
cfg = PathsConfig(
|
||||||
|
config_path = config_path,
|
||||||
|
presets = _resolve_path("SLOP_GLOBAL_PRESETS", "presets", root_dir / "presets.toml", config_path),
|
||||||
|
tool_presets = _resolve_path("SLOP_GLOBAL_TOOL_PRESETS", "tool_presets", root_dir / "tool_presets.toml", config_path),
|
||||||
|
personas = _resolve_path("SLOP_GLOBAL_PERSONAS", "personas", root_dir / "personas.toml", config_path),
|
||||||
|
themes = _resolve_path("SLOP_GLOBAL_THEMES", "themes", root_dir / "themes", config_path),
|
||||||
|
workspace_profiles = _resolve_path("SLOP_GLOBAL_WORKSPACE_PROFILES", "workspace_profiles", root_dir / "workspace_profiles.toml", config_path),
|
||||||
|
credentials = _resolve_path("SLOP_CREDENTIALS", "credentials", root_dir / "credentials.toml", config_path),
|
||||||
|
logs_dir = _resolve_path("SLOP_LOGS_DIR", "logs_dir", root_dir / "logs" / "sessions", config_path),
|
||||||
|
scripts_dir = _resolve_path("SLOP_SCRIPTS_DIR", "scripts_dir", root_dir / "scripts" / "generated", config_path),
|
||||||
|
)
|
||||||
|
with _PATHS_LOCK:
|
||||||
|
_PATHS_CONFIG = cfg
|
||||||
|
return cfg
|
||||||
|
|
||||||
|
|
||||||
|
def _cfg() -> PathsConfig:
|
||||||
|
"""Internal: get the current singleton, raising if uninitialized."""
|
||||||
|
if _PATHS_CONFIG is None:
|
||||||
|
raise RuntimeError(
|
||||||
|
"src.paths not initialized. Call paths.initialize_paths(<config.toml>) "
|
||||||
|
"BEFORE any path getter. See src/paths.py docstring for codepath ordering."
|
||||||
|
)
|
||||||
|
return _PATHS_CONFIG
|
||||||
|
|
||||||
|
|
||||||
|
# === Trivial getters (single source of truth) ===
|
||||||
|
|
||||||
def get_config_path() -> Path:
|
def get_config_path() -> Path:
|
||||||
"""
|
"""Active config.toml path. Frozen at initialize_paths() time.
|
||||||
Returns the active config.toml. If a CLI override is set, returns it.
|
|
||||||
Otherwise returns the default <project_root>/config.toml.
|
|
||||||
[C: src/app_controller.py:AppController.load_config,
|
[C: src/app_controller.py:AppController.load_config,
|
||||||
src/app_controller.py:AppController.init_state,
|
src/app_controller.py:AppController.init_state,
|
||||||
src/models.py:_load_config_from_disk,
|
src/models.py:_load_config_from_disk,
|
||||||
tests/test_test_sandbox.py]
|
tests/test_test_sandbox.py]"""
|
||||||
"""
|
return _cfg().config_path
|
||||||
if _CONFIG_OVERRIDE is not None:
|
|
||||||
return _CONFIG_OVERRIDE
|
|
||||||
root_dir = Path(__file__).resolve().parent.parent
|
|
||||||
return root_dir / "config.toml"
|
|
||||||
|
|
||||||
def get_global_presets_path() -> Path:
|
def get_global_presets_path() -> Path:
|
||||||
"""
|
"""Global presets file. Frozen at initialize_paths() time.
|
||||||
[C: src/presets.py:PresetManager.__init__, src/presets.py:PresetManager.delete_preset, src/presets.py:PresetManager.get_preset_scope]
|
[C: src/presets.py:PresetManager.__init__, src/presets.py:PresetManager.delete_preset, src/presets.py:PresetManager.get_preset_scope]"""
|
||||||
"""
|
return _cfg().presets
|
||||||
if "presets" not in _RESOLVED:
|
|
||||||
_RESOLVED["presets"] = _resolve_path("SLOP_GLOBAL_PRESETS", "presets", "presets.toml")
|
|
||||||
return _RESOLVED["presets"]
|
|
||||||
|
|
||||||
def get_project_presets_path(project_root: Path) -> Path:
|
def get_project_presets_path(project_root: Path) -> Path:
|
||||||
"""
|
"""Project-specific presets file. Computed from project_root (no cache).
|
||||||
[C: src/presets.py:PresetManager.delete_preset, src/presets.py:PresetManager.get_preset_scope, src/presets.py:PresetManager.project_path]
|
[C: src/presets.py:PresetManager.delete_preset, src/presets.py:PresetManager.get_preset_scope, src/presets.py:PresetManager.project_path]"""
|
||||||
"""
|
|
||||||
return project_root / "project_presets.toml"
|
return project_root / "project_presets.toml"
|
||||||
|
|
||||||
def get_global_tool_presets_path() -> Path:
|
def get_global_tool_presets_path() -> Path:
|
||||||
"""
|
"""Global tool presets file. Frozen at initialize_paths() time.
|
||||||
[C: src/tool_presets.py:ToolPresetManager._get_path, src/tool_presets.py:ToolPresetManager.load_all_bias_profiles, src/tool_presets.py:ToolPresetManager.load_all_presets]
|
[C: src/tool_presets.py:ToolPresetManager._get_path, src/tool_presets.py:ToolPresetManager.load_all_bias_profiles, src/tool_presets.py:ToolPresetManager.load_all_presets]"""
|
||||||
"""
|
return _cfg().tool_presets
|
||||||
if "tool_presets" not in _RESOLVED:
|
|
||||||
_RESOLVED["tool_presets"] = _resolve_path("SLOP_GLOBAL_TOOL_PRESETS", "tool_presets", "tool_presets.toml")
|
|
||||||
return _RESOLVED["tool_presets"]
|
|
||||||
|
|
||||||
def get_project_tool_presets_path(project_root: Path) -> Path:
|
def get_project_tool_presets_path(project_root: Path) -> Path:
|
||||||
"""
|
"""[C: src/tool_presets.py:ToolPresetManager._get_path, src/tool_presets.py:ToolPresetManager.load_all_bias_profiles, src/tool_presets.py:ToolPresetManager.load_all_presets]"""
|
||||||
[C: src/tool_presets.py:ToolPresetManager._get_path, src/tool_presets.py:ToolPresetManager.load_all_bias_profiles, src/tool_presets.py:ToolPresetManager.load_all_presets]
|
|
||||||
"""
|
|
||||||
return project_root / "project_tool_presets.toml"
|
return project_root / "project_tool_presets.toml"
|
||||||
|
|
||||||
def get_global_personas_path() -> Path:
|
def get_global_personas_path() -> Path:
|
||||||
"""
|
"""Global personas file. Frozen at initialize_paths() time.
|
||||||
[C: src/personas.py:PersonaManager._get_path, src/personas.py:PersonaManager.get_persona_scope, src/personas.py:PersonaManager.load_all]
|
[C: src/personas.py:PersonaManager._get_path, src/personas.py:PersonaManager.get_persona_scope, src/personas.py:PersonaManager.load_all]"""
|
||||||
"""
|
return _cfg().personas
|
||||||
if "personas" not in _RESOLVED:
|
|
||||||
_RESOLVED["personas"] = _resolve_path("SLOP_GLOBAL_PERSONAS", "personas", "personas.toml")
|
|
||||||
return _RESOLVED["personas"]
|
|
||||||
|
|
||||||
def get_project_personas_path(project_root: Path) -> Path:
|
def get_project_personas_path(project_root: Path) -> Path:
|
||||||
"""
|
"""[C: src/personas.py:PersonaManager._get_path, src/personas.py:PersonaManager.get_persona_scope, src/personas.py:PersonaManager.load_all]"""
|
||||||
[C: src/personas.py:PersonaManager._get_path, src/personas.py:PersonaManager.get_persona_scope, src/personas.py:PersonaManager.load_all]
|
|
||||||
"""
|
|
||||||
return project_root / "project_personas.toml"
|
return project_root / "project_personas.toml"
|
||||||
|
|
||||||
def get_global_themes_path() -> Path:
|
def get_global_themes_path() -> Path:
|
||||||
"""
|
"""Global themes directory. Frozen at initialize_paths() time.
|
||||||
[C: src/theme_2.py:load_themes_from_disk]
|
[C: src/theme_2.py:load_themes_from_disk]"""
|
||||||
"""
|
return _cfg().themes
|
||||||
if "themes" not in _RESOLVED:
|
|
||||||
_RESOLVED["themes"] = _resolve_path("SLOP_GLOBAL_THEMES", "themes", "themes")
|
|
||||||
return _RESOLVED["themes"]
|
|
||||||
|
|
||||||
def get_project_themes_path(project_root: Path) -> Path:
|
def get_project_themes_path(project_root: Path) -> Path:
|
||||||
"""
|
"""[C: src/theme_2.py:load_themes_from_disk]"""
|
||||||
[C: src/theme_2.py:load_themes_from_disk]
|
|
||||||
"""
|
|
||||||
return project_root / "project_themes.toml"
|
return project_root / "project_themes.toml"
|
||||||
|
|
||||||
def get_global_workspace_profiles_path() -> Path:
|
def get_global_workspace_profiles_path() -> Path:
|
||||||
"""
|
"""Global workspace profiles file. Frozen at initialize_paths() time.
|
||||||
[C: src/workspace_manager.py:WorkspaceManager._get_path, src/workspace_manager.py:WorkspaceManager.load_all_profiles]
|
[C: src/workspace_manager.py:WorkspaceManager._get_path, src/workspace_manager.py:WorkspaceManager.load_all_profiles]"""
|
||||||
"""
|
return _cfg().workspace_profiles
|
||||||
if "workspace_profiles" not in _RESOLVED:
|
|
||||||
_RESOLVED["workspace_profiles"] = _resolve_path("SLOP_GLOBAL_WORKSPACE_PROFILES", "workspace_profiles", "workspace_profiles.toml")
|
|
||||||
return _RESOLVED["workspace_profiles"]
|
|
||||||
|
|
||||||
def get_project_workspace_profiles_path(project_root: Path) -> Path:
|
def get_project_workspace_profiles_path(project_root: Path) -> Path:
|
||||||
"""
|
"""[C: src/workspace_manager.py:WorkspaceManager._get_path, src/workspace_manager.py:WorkspaceManager.load_all_profiles]"""
|
||||||
[C: src/workspace_manager.py:WorkspaceManager._get_path, src/workspace_manager.py:WorkspaceManager.load_all_profiles]
|
|
||||||
"""
|
|
||||||
return project_root / ".ai" / "workspace_profiles.toml"
|
return project_root / ".ai" / "workspace_profiles.toml"
|
||||||
|
|
||||||
def get_credentials_path() -> Path:
|
def get_credentials_path() -> Path:
|
||||||
"""
|
"""Global credentials file. Frozen at initialize_paths() time.
|
||||||
[C: src/mcp_client.py:_is_allowed]
|
[C: src/mcp_client.py:_is_allowed]"""
|
||||||
"""
|
return _cfg().credentials
|
||||||
if "credentials" not in _RESOLVED:
|
|
||||||
_RESOLVED["credentials"] = _resolve_path("SLOP_CREDENTIALS", "credentials", "credentials.toml")
|
def get_logs_dir() -> Path:
|
||||||
return _RESOLVED["credentials"]
|
"""Logs directory (contains session subdirs). Frozen at initialize_paths() time.
|
||||||
|
[C: src/session_logger.py:close_session, src/session_logger.py:open_session, tests/test_paths.py:test_config_overrides, tests/test_paths.py:test_default_paths, tests/test_paths.py:test_env_var_overrides, tests/test_paths.py:test_precedence]"""
|
||||||
|
return _cfg().logs_dir
|
||||||
|
|
||||||
|
def get_scripts_dir() -> Path:
|
||||||
|
"""Generated scripts directory. Frozen at initialize_paths() time.
|
||||||
|
[C: src/session_logger.py:log_tool_call, src/session_logger.py:open_session, tests/test_paths.py:test_config_overrides, tests/test_paths.py:test_default_paths]"""
|
||||||
|
return _cfg().scripts_dir
|
||||||
|
|
||||||
|
def get_tracks_dir(project_path: Optional[str] = None) -> Path:
|
||||||
|
"""[C: src/project_manager.py:get_all_tracks, tests/test_paths.py:test_conductor_dir_project_relative]"""
|
||||||
|
return get_conductor_dir(project_path) / "tracks"
|
||||||
|
|
||||||
|
def get_track_state_dir(track_id: str, project_path: Optional[str] = None) -> Path:
|
||||||
|
"""[C: src/project_manager.py:load_track_state, src/project_manager.py:save_track_state, tests/test_paths.py:test_conductor_dir_project_relative]"""
|
||||||
|
return get_tracks_dir(project_path) / track_id
|
||||||
|
|
||||||
|
def get_archive_dir(project_path: Optional[str] = None) -> Path:
|
||||||
|
"""[C: tests/test_paths.py:test_conductor_dir_project_relative]"""
|
||||||
|
return get_conductor_dir(project_path) / "archive"
|
||||||
|
|
||||||
def _resolve_path(env_var: str, config_key: str, default: str) -> Path:
|
|
||||||
root_dir = Path(__file__).resolve().parent.parent
|
|
||||||
p = None
|
|
||||||
if env_var in os.environ:
|
|
||||||
p = Path(os.environ[env_var])
|
|
||||||
else:
|
|
||||||
try:
|
|
||||||
with open(get_config_path(), "rb") as f:
|
|
||||||
cfg = tomllib.load(f)
|
|
||||||
if "paths" in cfg and config_key in cfg["paths"]:
|
|
||||||
p = Path(cfg["paths"][config_key])
|
|
||||||
except (FileNotFoundError, tomllib.TOMLDecodeError):
|
|
||||||
pass
|
|
||||||
if p is None:
|
|
||||||
p = Path(default)
|
|
||||||
if not p.is_absolute():
|
|
||||||
return root_dir / p
|
|
||||||
return p
|
|
||||||
|
|
||||||
def _get_project_conductor_dir_from_toml(project_root: Path) -> Optional[Path]:
|
def _get_project_conductor_dir_from_toml(project_root: Path) -> Optional[Path]:
|
||||||
# Look for manual_slop.toml in project_root
|
"""Look for manual_slop.toml in project_root for [conductor] dir override."""
|
||||||
toml_path = project_root / 'manual_slop.toml'
|
toml_path = project_root / 'manual_slop.toml'
|
||||||
if not toml_path.exists(): return None
|
if not toml_path.exists(): return None
|
||||||
try:
|
try:
|
||||||
with open(toml_path, 'rb') as f:
|
with open(toml_path, 'rb') as f:
|
||||||
data = tomllib.load(f)
|
data = tomllib.load(f)
|
||||||
# Check [conductor] dir = '...'
|
|
||||||
c_dir = data.get('conductor', {}).get('dir')
|
c_dir = data.get('conductor', {}).get('dir')
|
||||||
if c_dir:
|
if c_dir:
|
||||||
p = Path(c_dir)
|
p = Path(c_dir)
|
||||||
@@ -192,79 +234,44 @@ def _get_project_conductor_dir_from_toml(project_root: Path) -> Optional[Path]:
|
|||||||
except: pass
|
except: pass
|
||||||
return None
|
return None
|
||||||
|
|
||||||
def get_conductor_dir(project_path: Optional[str] = None) -> Path:
|
|
||||||
"""
|
|
||||||
[C: tests/test_paths.py:test_conductor_dir_project_relative, tests/test_project_paths.py:test_get_conductor_dir_default, tests/test_project_paths.py:test_get_conductor_dir_project_specific_with_toml]
|
|
||||||
"""
|
|
||||||
if not project_path:
|
|
||||||
# Fallback for legacy/tests, but we should avoid this
|
|
||||||
return Path('conductor').resolve()
|
|
||||||
|
|
||||||
|
def get_conductor_dir(project_path: Optional[str] = None) -> Path:
|
||||||
|
"""[C: tests/test_paths.py:test_conductor_dir_project_relative, tests/test_project_paths.py:test_get_conductor_dir_default, tests/test_project_paths.py:test_get_conductor_dir_project_specific_with_toml]"""
|
||||||
|
if not project_path:
|
||||||
|
return Path('conductor').resolve()
|
||||||
project_root = Path(project_path).resolve()
|
project_root = Path(project_path).resolve()
|
||||||
p = _get_project_conductor_dir_from_toml(project_root)
|
p = _get_project_conductor_dir_from_toml(project_root)
|
||||||
if p: return p
|
if p: return p
|
||||||
return (project_root / "conductor").resolve()
|
return (project_root / "conductor").resolve()
|
||||||
|
|
||||||
def get_logs_dir() -> Path:
|
|
||||||
"""
|
|
||||||
[C: src/session_logger.py:close_session, src/session_logger.py:open_session, tests/test_paths.py:test_config_overrides, tests/test_paths.py:test_default_paths, tests/test_paths.py:test_env_var_overrides, tests/test_paths.py:test_precedence]
|
|
||||||
"""
|
|
||||||
if "logs_dir" not in _RESOLVED:
|
|
||||||
_RESOLVED["logs_dir"] = _resolve_path("SLOP_LOGS_DIR", "logs_dir", "logs/sessions")
|
|
||||||
return _RESOLVED["logs_dir"]
|
|
||||||
|
|
||||||
def get_scripts_dir() -> Path:
|
|
||||||
"""
|
|
||||||
[C: src/session_logger.py:log_tool_call, src/session_logger.py:open_session, tests/test_paths.py:test_config_overrides, tests/test_paths.py:test_default_paths]
|
|
||||||
"""
|
|
||||||
if "scripts_dir" not in _RESOLVED:
|
|
||||||
_RESOLVED["scripts_dir"] = _resolve_path("SLOP_SCRIPTS_DIR", "scripts_dir", "scripts/generated")
|
|
||||||
return _RESOLVED["scripts_dir"]
|
|
||||||
|
|
||||||
def get_tracks_dir(project_path: Optional[str] = None) -> Path:
|
|
||||||
"""
|
|
||||||
[C: src/project_manager.py:get_all_tracks, tests/test_paths.py:test_conductor_dir_project_relative]
|
|
||||||
"""
|
|
||||||
return get_conductor_dir(project_path) / "tracks"
|
|
||||||
|
|
||||||
def get_track_state_dir(track_id: str, project_path: Optional[str] = None) -> Path:
|
|
||||||
"""
|
|
||||||
[C: src/project_manager.py:load_track_state, src/project_manager.py:save_track_state, tests/test_paths.py:test_conductor_dir_project_relative]
|
|
||||||
"""
|
|
||||||
return get_tracks_dir(project_path) / track_id
|
|
||||||
|
|
||||||
def get_archive_dir(project_path: Optional[str] = None) -> Path:
|
|
||||||
"""
|
|
||||||
[C: tests/test_paths.py:test_conductor_dir_project_relative]
|
|
||||||
"""
|
|
||||||
return get_conductor_dir(project_path) / "archive"
|
|
||||||
|
|
||||||
def _resolve_path_info(env_var: str, config_key: str, default: str) -> dict[str, Any]:
|
|
||||||
if env_var in os.environ:
|
|
||||||
return {'path': Path(os.environ[env_var]).resolve(), 'source': f'env:{env_var}'}
|
|
||||||
try:
|
|
||||||
with open(get_config_path(), 'rb') as f:
|
|
||||||
cfg = tomllib.load(f)
|
|
||||||
if 'paths' in cfg and config_key in cfg['paths']:
|
|
||||||
p = Path(cfg['paths'][config_key])
|
|
||||||
if not p.is_absolute():
|
|
||||||
p = (Path(__file__).resolve().parent.parent / p).resolve()
|
|
||||||
return {'path': p, 'source': 'config.toml'}
|
|
||||||
except: pass
|
|
||||||
root_dir = Path(__file__).resolve().parent.parent
|
|
||||||
p = (root_dir / default).resolve()
|
|
||||||
return {'path': p, 'source': 'default'}
|
|
||||||
|
|
||||||
def get_full_path_info() -> dict[str, dict[str, Any]]:
|
def get_full_path_info() -> dict[str, dict[str, Any]]:
|
||||||
|
"""Return the resolved paths + their source (env / config / default).
|
||||||
|
For diagnostic UIs (e.g., the Session Hub's "show resolved paths" panel).
|
||||||
|
[C: src/gui_2.py:App._render_path_field]"""
|
||||||
|
cfg = _cfg()
|
||||||
|
def info(value: Path) -> dict[str, Any]:
|
||||||
|
return {'path': str(value), 'source': 'frozen_at_init'}
|
||||||
return {
|
return {
|
||||||
'logs_dir': _resolve_path_info('SLOP_LOGS_DIR', 'logs_dir', 'logs/sessions'),
|
'config_path': info(cfg.config_path),
|
||||||
'scripts_dir': _resolve_path_info('SLOP_SCRIPTS_DIR', 'scripts_dir', 'scripts/generated')
|
'presets': info(cfg.presets),
|
||||||
|
'tool_presets': info(cfg.tool_presets),
|
||||||
|
'personas': info(cfg.personas),
|
||||||
|
'themes': info(cfg.themes),
|
||||||
|
'workspace_profiles': info(cfg.workspace_profiles),
|
||||||
|
'credentials': info(cfg.credentials),
|
||||||
|
'logs_dir': info(cfg.logs_dir),
|
||||||
|
'scripts_dir': info(cfg.scripts_dir),
|
||||||
}
|
}
|
||||||
|
|
||||||
def reset_resolved() -> None:
|
|
||||||
"""
|
|
||||||
For testing only - clear cached resolutions.
|
|
||||||
[C: tests/conftest.py:reset_paths, tests/test_app_controller_offloading.py:tmp_session_dir, tests/test_gui_phase3.py:test_conductor_setup_scan, tests/test_paths.py:reset_paths, tests/test_project_paths.py:test_get_all_tracks_project_specific, tests/test_project_paths.py:test_get_conductor_dir_default, tests/test_project_paths.py:test_get_conductor_dir_project_specific_with_toml]
|
|
||||||
"""
|
|
||||||
_RESOLVED.clear()
|
|
||||||
|
|
||||||
|
def reset_paths() -> None:
|
||||||
|
"""Clear the singleton. FOR TESTS ONLY — production code should never
|
||||||
|
call this. After reset, the next path getter raises RuntimeError until
|
||||||
|
initialize_paths() is called again.
|
||||||
|
[C: tests/conftest.py:reset_paths, tests/test_paths.py:reset_paths,
|
||||||
|
tests/test_app_controller_offloading.py:setup_function,
|
||||||
|
tests/test_gui_phase3.py:setup]"""
|
||||||
|
global _PATHS_CONFIG
|
||||||
|
with _PATHS_LOCK:
|
||||||
|
_PATHS_CONFIG = None
|
||||||
+10
-9
@@ -34,7 +34,7 @@ if _config_override_arg is None:
|
|||||||
_config_override_arg = _ISOLATION_WORKSPACE / "config_overrides.toml"
|
_config_override_arg = _ISOLATION_WORKSPACE / "config_overrides.toml"
|
||||||
|
|
||||||
from src import paths as _paths # noqa: E402
|
from src import paths as _paths # noqa: E402
|
||||||
_paths.set_config_override(_config_override_arg)
|
_paths.initialize_paths(_config_override_arg)
|
||||||
|
|
||||||
thirdparty_dir = os.path.join(os.path.dirname(__file__), "..", "thirdparty")
|
thirdparty_dir = os.path.join(os.path.dirname(__file__), "..", "thirdparty")
|
||||||
if thirdparty_dir not in sys.path:
|
if thirdparty_dir not in sys.path:
|
||||||
@@ -381,11 +381,11 @@ def isolate_workspace(monkeypatch) -> Generator[None, None, None]:
|
|||||||
(tests/artifacts/_isolation_workspace_<RUN_ID>/). Also writes placeholder
|
(tests/artifacts/_isolation_workspace_<RUN_ID>/). Also writes placeholder
|
||||||
TOML files for the redirected paths. NO SLOP_* env vars are set;
|
TOML files for the redirected paths. NO SLOP_* env vars are set;
|
||||||
src/paths.py reads the overrides from config.toml [paths] (with env var
|
src/paths.py reads the overrides from config.toml [paths] (with env var
|
||||||
as fallback if needed).
|
as fallback if needed). Also re-initializes the paths singleton so every
|
||||||
[C: tests/conftest.py:_ISOLATION_WORKSPACE, src/paths.py:_resolve_path]
|
getter sees the test-workspace overrides.
|
||||||
|
[C: tests/conftest.py:_ISOLATION_WORKSPACE, src/paths.py:initialize_paths]
|
||||||
"""
|
"""
|
||||||
from src import paths as _paths
|
from src import paths as _paths
|
||||||
_paths.reset_resolved()
|
|
||||||
|
|
||||||
test_workspace = _ISOLATION_WORKSPACE
|
test_workspace = _ISOLATION_WORKSPACE
|
||||||
|
|
||||||
@@ -421,14 +421,15 @@ def isolate_workspace(monkeypatch) -> Generator[None, None, None]:
|
|||||||
@pytest.fixture(autouse=True)
|
@pytest.fixture(autouse=True)
|
||||||
def reset_paths() -> Generator[None, None, None]:
|
def reset_paths() -> Generator[None, None, None]:
|
||||||
"""
|
"""
|
||||||
|
Autouse fixture that resets the paths singleton before each test.
|
||||||
|
The isolate_workspace fixture re-initializes paths before the test runs,
|
||||||
Autouse fixture that resets the paths global state before each test.
|
so reset_paths is a no-op here (PathsConfig is frozen at init time,
|
||||||
|
the per-getter cache is gone). Kept as a marker for backward compatibility
|
||||||
|
with test code that asserts the singleton is cleared at test boundaries.
|
||||||
"""
|
"""
|
||||||
from src import paths
|
from src import paths
|
||||||
paths.reset_resolved()
|
|
||||||
yield
|
yield
|
||||||
paths.reset_resolved()
|
paths.reset_paths()
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
@pytest.fixture(autouse=True)
|
||||||
def reset_ai_client() -> Generator[None, None, None]:
|
def reset_ai_client() -> Generator[None, None, None]:
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ def tmp_session_dir(tmp_path, monkeypatch):
|
|||||||
|
|
||||||
monkeypatch.setenv("SLOP_LOGS_DIR", str(logs_dir))
|
monkeypatch.setenv("SLOP_LOGS_DIR", str(logs_dir))
|
||||||
monkeypatch.setenv("SLOP_SCRIPTS_DIR", str(scripts_dir))
|
monkeypatch.setenv("SLOP_SCRIPTS_DIR", str(scripts_dir))
|
||||||
paths.reset_resolved()
|
paths.reset_paths()
|
||||||
|
|
||||||
# Ensure session_logger is clean
|
# Ensure session_logger is clean
|
||||||
with patch("src.session_logger._comms_fh", None):
|
with patch("src.session_logger._comms_fh", None):
|
||||||
|
|||||||
@@ -26,7 +26,7 @@ def test_save_paths():
|
|||||||
|
|
||||||
with patch('shutil.copy') as mock_copy, \
|
with patch('shutil.copy') as mock_copy, \
|
||||||
patch('src.paths.get_config_path') as mock_get_cfg, \
|
patch('src.paths.get_config_path') as mock_get_cfg, \
|
||||||
patch('src.paths.reset_resolved') as mock_reset, \
|
patch('src.paths.reset_paths') as mock_reset, \
|
||||||
patch.object(MockApp, 'init_state') as mock_init:
|
patch.object(MockApp, 'init_state') as mock_init:
|
||||||
|
|
||||||
mock_get_cfg.return_value = MagicMock()
|
mock_get_cfg.return_value = MagicMock()
|
||||||
|
|||||||
@@ -50,7 +50,7 @@ def test_conductor_setup_scan(app_instance, tmp_path, monkeypatch):
|
|||||||
(cond_dir / "tracks" / "track1").mkdir(exist_ok=True)
|
(cond_dir / "tracks" / "track1").mkdir(exist_ok=True)
|
||||||
|
|
||||||
monkeypatch.setenv('SLOP_CONDUCTOR_DIR', str((tmp_path / 'conductor').resolve()))
|
monkeypatch.setenv('SLOP_CONDUCTOR_DIR', str((tmp_path / 'conductor').resolve()))
|
||||||
paths.reset_resolved()
|
paths.reset_paths()
|
||||||
|
|
||||||
app_instance._cb_run_conductor_setup()
|
app_instance._cb_run_conductor_setup()
|
||||||
|
|
||||||
|
|||||||
+2
-2
@@ -5,9 +5,9 @@ from src import paths
|
|||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
@pytest.fixture(autouse=True)
|
||||||
def reset_paths():
|
def reset_paths():
|
||||||
paths.reset_resolved()
|
paths.reset_paths()
|
||||||
yield
|
yield
|
||||||
paths.reset_resolved()
|
paths.reset_paths()
|
||||||
|
|
||||||
def test_default_paths(tmp_path, monkeypatch):
|
def test_default_paths(tmp_path, monkeypatch):
|
||||||
monkeypatch.setenv("SLOP_CONFIG", str(tmp_path / "non_existent.toml"))
|
monkeypatch.setenv("SLOP_CONFIG", str(tmp_path / "non_existent.toml"))
|
||||||
|
|||||||
@@ -7,13 +7,13 @@ from src import paths
|
|||||||
from src import project_manager
|
from src import project_manager
|
||||||
|
|
||||||
def test_get_conductor_dir_default():
|
def test_get_conductor_dir_default():
|
||||||
paths.reset_resolved()
|
paths.reset_paths()
|
||||||
# Should return absolute path to "conductor" in project root
|
# Should return absolute path to "conductor" in project root
|
||||||
expected = Path(__file__).resolve().parent.parent / "conductor"
|
expected = Path(__file__).resolve().parent.parent / "conductor"
|
||||||
assert paths.get_conductor_dir() == expected
|
assert paths.get_conductor_dir() == expected
|
||||||
|
|
||||||
def test_get_conductor_dir_project_specific_with_toml(tmp_path):
|
def test_get_conductor_dir_project_specific_with_toml(tmp_path):
|
||||||
paths.reset_resolved()
|
paths.reset_paths()
|
||||||
project_root = tmp_path / "my_project"
|
project_root = tmp_path / "my_project"
|
||||||
project_root.mkdir()
|
project_root.mkdir()
|
||||||
|
|
||||||
@@ -31,7 +31,7 @@ def test_get_conductor_dir_project_specific_with_toml(tmp_path):
|
|||||||
assert res == project_root / "custom_tracks"
|
assert res == project_root / "custom_tracks"
|
||||||
|
|
||||||
def test_get_all_tracks_project_specific(tmp_path):
|
def test_get_all_tracks_project_specific(tmp_path):
|
||||||
paths.reset_resolved()
|
paths.reset_paths()
|
||||||
project_root = tmp_path / "my_project"
|
project_root = tmp_path / "my_project"
|
||||||
project_root.mkdir()
|
project_root.mkdir()
|
||||||
|
|
||||||
|
|||||||
+110
-37
@@ -149,47 +149,64 @@ def test_sandbox_allows_pytest_cache_write() -> None:
|
|||||||
|
|
||||||
|
|
||||||
def test_config_override_via_cli_flag(tmp_path) -> None:
|
def test_config_override_via_cli_flag(tmp_path) -> None:
|
||||||
"""paths.set_config_override(path) makes get_config_path() return that path."""
|
"""paths.initialize_paths(config_path) makes all path getters return paths
|
||||||
|
rooted at config_path's [paths] section (or default).
|
||||||
|
[C: src/paths.py:initialize_paths, sloppy.py:main]"""
|
||||||
from src import paths
|
from src import paths
|
||||||
config_path = tmp_path / "my_config.toml"
|
config_path = tmp_path / "my_config.toml"
|
||||||
config_path.write_text("[ai]\nprovider='gemini'\n", encoding="utf-8")
|
config_path.write_text("[ai]\nprovider='gemini'\n", encoding="utf-8")
|
||||||
original = paths._CONFIG_OVERRIDE
|
paths.initialize_paths(config_path)
|
||||||
try:
|
try:
|
||||||
paths.set_config_override(config_path)
|
|
||||||
assert paths.get_config_path() == config_path
|
assert paths.get_config_path() == config_path
|
||||||
|
assert paths.get_global_presets_path().name == "presets.toml"
|
||||||
|
assert paths.get_logs_dir().name == "sessions"
|
||||||
finally:
|
finally:
|
||||||
paths.set_config_override(original)
|
paths.reset_paths()
|
||||||
|
|
||||||
|
|
||||||
def test_paths_get_config_path_no_env_fallback(monkeypatch) -> None:
|
def test_paths_runtime_refresh_atomic_swap(tmp_path) -> None:
|
||||||
"""Without an override AND without SLOP_CONFIG, get_config_path returns default."""
|
"""Calling initialize_paths() a second time atomically swaps the singleton.
|
||||||
monkeypatch.delenv("SLOP_CONFIG", raising=False)
|
Reader threads see the new config immediately. The PathsConfig is frozen.
|
||||||
|
[C: src/paths.py:initialize_paths, src/paths.py:PathsConfig]"""
|
||||||
from src import paths
|
from src import paths
|
||||||
original = paths._CONFIG_OVERRIDE
|
cfg_a = tmp_path / "a.toml"
|
||||||
try:
|
cfg_b = tmp_path / "b.toml"
|
||||||
paths.set_config_override(None)
|
cfg_a.write_text("[ai]\nprovider='gemini-a'\n", encoding="utf-8")
|
||||||
expected = Path(__file__).resolve().parent.parent / "config.toml"
|
cfg_b.write_text("[ai]\nprovider='gemini-b'\n", encoding="utf-8")
|
||||||
assert paths.get_config_path() == expected
|
paths.initialize_paths(cfg_a)
|
||||||
finally:
|
assert paths.get_config_path() == cfg_a
|
||||||
paths.set_config_override(original)
|
paths.initialize_paths(cfg_b)
|
||||||
|
assert paths.get_config_path() == cfg_b
|
||||||
|
paths.reset_paths()
|
||||||
|
|
||||||
|
|
||||||
|
def test_paths_uninitialized_raises(tmp_path) -> None:
|
||||||
|
"""Calling a path getter before initialize_paths() raises RuntimeError
|
||||||
|
(not silent fallback). This is the contract that enforces explicit init.
|
||||||
|
[C: src/paths.py:_cfg]"""
|
||||||
|
from src import paths
|
||||||
|
paths.reset_paths()
|
||||||
|
with pytest.raises(RuntimeError, match="not initialized"):
|
||||||
|
paths.get_logs_dir()
|
||||||
|
paths.reset_paths()
|
||||||
|
|
||||||
|
|
||||||
def test_sloppy_py_parses_config_flag() -> None:
|
def test_sloppy_py_parses_config_flag() -> None:
|
||||||
"""sloppy.py has a --config argparse argument that calls set_config_override."""
|
"""sloppy.py has a --config argparse argument that calls initialize_paths."""
|
||||||
import ast
|
import ast
|
||||||
sloppy = Path(__file__).resolve().parent.parent / "sloppy.py"
|
sloppy = Path(__file__).resolve().parent.parent / "sloppy.py"
|
||||||
tree = ast.parse(sloppy.read_text(encoding="utf-8"))
|
tree = ast.parse(sloppy.read_text(encoding="utf-8"))
|
||||||
found_config_arg = False
|
found_config_arg = False
|
||||||
found_set_override_call = False
|
found_init_call = False
|
||||||
for node in ast.walk(tree):
|
for node in ast.walk(tree):
|
||||||
if isinstance(node, ast.Constant) and node.value == "--config":
|
if isinstance(node, ast.Constant) and node.value == "--config":
|
||||||
found_config_arg = True
|
found_config_arg = True
|
||||||
if isinstance(node, ast.Call):
|
if isinstance(node, ast.Call):
|
||||||
func = node.func
|
func = node.func
|
||||||
if isinstance(func, ast.Name) and func.id == "set_config_override":
|
if isinstance(func, ast.Name) and func.id == "initialize_paths":
|
||||||
found_set_override_call = True
|
found_init_call = True
|
||||||
assert found_config_arg, "sloppy.py must have a --config argparse argument"
|
assert found_config_arg, "sloppy.py must have a --config argparse argument"
|
||||||
assert found_set_override_call, "sloppy.py must call paths.set_config_override(args.config)"
|
assert found_init_call, "sloppy.py must call paths.initialize_paths(args.config)"
|
||||||
|
|
||||||
|
|
||||||
def test_pyproject_toml_basetemp_is_under_tests() -> None:
|
def test_pyproject_toml_basetemp_is_under_tests() -> None:
|
||||||
@@ -276,30 +293,86 @@ def test_config_overrides_toml_has_paths_section() -> None:
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_path_getters_read_from_config_paths_section() -> None:
|
def test_path_getters_are_trivial_field_access() -> None:
|
||||||
"""Every global path getter in src/paths.py reads from the [paths] section
|
"""Every global path getter in src/paths.py is a trivial field access on the
|
||||||
of the active config when no env var is set. Verifies via AST that each
|
PathsConfig singleton (return _cfg().<field>). They must NOT do file I/O
|
||||||
get_*_path() function calls _resolve_path (not raw os.environ.get)."""
|
or call _resolve_path() (which is internal-only, called from initialize_paths).
|
||||||
|
This enforces the v3 design: explicit init at startup, trivial getters."""
|
||||||
import ast
|
import ast
|
||||||
paths_py = Path(__file__).resolve().parent.parent / "src" / "paths.py"
|
paths_py = Path(__file__).resolve().parent.parent / "src" / "paths.py"
|
||||||
tree = ast.parse(paths_py.read_text(encoding="utf-8"))
|
tree = ast.parse(paths_py.read_text(encoding="utf-8"))
|
||||||
getters = {
|
trivial_getters = [
|
||||||
"get_global_presets_path": "presets",
|
"get_global_presets_path", "get_global_tool_presets_path",
|
||||||
"get_global_tool_presets_path": "tool_presets",
|
"get_global_personas_path", "get_global_themes_path",
|
||||||
"get_global_personas_path": "personas",
|
"get_global_workspace_profiles_path", "get_credentials_path",
|
||||||
"get_global_themes_path": "themes",
|
"get_logs_dir", "get_scripts_dir",
|
||||||
"get_global_workspace_profiles_path": "workspace_profiles",
|
]
|
||||||
"get_credentials_path": "credentials",
|
|
||||||
"get_logs_dir": "logs_dir",
|
|
||||||
"get_scripts_dir": "scripts_dir",
|
|
||||||
}
|
|
||||||
for node in ast.walk(tree):
|
for node in ast.walk(tree):
|
||||||
if isinstance(node, ast.FunctionDef) and node.name in getters:
|
if isinstance(node, ast.FunctionDef) and node.name in trivial_getters:
|
||||||
src = ast.unparse(node)
|
src = ast.unparse(node)
|
||||||
assert "_resolve_path(" in src, (
|
assert "_cfg()" in src, (
|
||||||
f"{node.name} must call _resolve_path() to read from config.toml [paths]; "
|
f"{node.name} must call _cfg() to read from the PathsConfig singleton; "
|
||||||
f"got direct os.environ.get() instead."
|
f"got direct file I/O or env var lookup instead."
|
||||||
)
|
)
|
||||||
|
assert "_resolve_path(" not in src, (
|
||||||
|
f"{node.name} must NOT call _resolve_path(); that's internal-only "
|
||||||
|
f"(called once from initialize_paths)."
|
||||||
|
)
|
||||||
|
assert "os.environ" not in src, (
|
||||||
|
f"{node.name} must NOT use os.environ directly; env vars are read once "
|
||||||
|
f"during initialize_paths()."
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_initialize_paths_thread_safe_atomic_swap(tmp_path) -> None:
|
||||||
|
"""initialize_paths() uses an RLock; concurrent swaps don't corrupt the
|
||||||
|
singleton. Reader threads always see a consistent PathsConfig snapshot
|
||||||
|
(frozen dataclass). Per the user's "bad programmers take shortcuts" rule:
|
||||||
|
no torn writes, no partial reads.
|
||||||
|
[C: src/paths.py:_PATHS_LOCK, src/paths.py:PathsConfig]"""
|
||||||
|
import threading
|
||||||
|
import tomllib
|
||||||
|
from src import paths
|
||||||
|
cfg_a = tmp_path / "a.toml"
|
||||||
|
cfg_b = tmp_path / "b.toml"
|
||||||
|
cfg_a.write_bytes(b"[paths]\nlogs_dir = 'C:/tmp/thread_a'\n")
|
||||||
|
cfg_b.write_bytes(b"[paths]\nlogs_dir = 'C:/tmp/thread_b'\n")
|
||||||
|
errors = []
|
||||||
|
def swap(cfg, n):
|
||||||
|
for _ in range(n):
|
||||||
|
try:
|
||||||
|
paths.initialize_paths(cfg)
|
||||||
|
except Exception as e:
|
||||||
|
errors.append(e)
|
||||||
|
t1 = threading.Thread(target=swap, args=(cfg_a, 100))
|
||||||
|
t2 = threading.Thread(target=swap, args=(cfg_b, 100))
|
||||||
|
t1.start(); t2.start()
|
||||||
|
t1.join(); t2.join()
|
||||||
|
assert not errors, f"thread-safety violation: {errors}"
|
||||||
|
paths.reset_paths()
|
||||||
|
|
||||||
|
|
||||||
|
def test_pathsconfig_is_frozen_dataclass() -> None:
|
||||||
|
"""PathsConfig uses @dataclass(frozen=True) so individual field reads are
|
||||||
|
atomic and cannot be mutated by readers. Per user directive: gated
|
||||||
|
transactions, no data race over config.
|
||||||
|
[C: src/paths.py:PathsConfig]"""
|
||||||
|
import dataclasses
|
||||||
|
import tempfile, tomli_w
|
||||||
|
from src import paths
|
||||||
|
params = getattr(paths.PathsConfig, "__dataclass_params__", None)
|
||||||
|
assert params is not None, "PathsConfig must be a dataclass"
|
||||||
|
assert params.frozen is True, "PathsConfig must be @dataclass(frozen=True)"
|
||||||
|
with tempfile.NamedTemporaryFile(suffix=".toml", delete=False, mode="wb") as f:
|
||||||
|
tomli_w.dump({"paths": {"logs_dir": "C:/tmp/frozen_test"}}, f)
|
||||||
|
cfg = Path(f.name)
|
||||||
|
try:
|
||||||
|
paths.initialize_paths(cfg)
|
||||||
|
with pytest.raises(dataclasses.FrozenInstanceError):
|
||||||
|
paths._PATHS_CONFIG.presets = Path("/tmp/should_fail")
|
||||||
|
finally:
|
||||||
|
paths.reset_paths()
|
||||||
|
cfg.unlink()
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.skipif(os.name != "nt", reason="Windows-only sandbox wrapper")
|
@pytest.mark.skipif(os.name != "nt", reason="Windows-only sandbox wrapper")
|
||||||
|
|||||||
Reference in New Issue
Block a user