509 lines
19 KiB
Python
509 lines
19 KiB
Python
"""
|
|
Tests for state_manager module - SessionState dataclass.
|
|
Task 1: Test state schema with dataclasses.
|
|
Task 2: Test atomic state save with backup rotation.
|
|
Task 3: Test state load with validation and fallback.
|
|
"""
|
|
import json
|
|
import os
|
|
import pytest
|
|
from pathlib import Path
|
|
from dataclasses import asdict
|
|
from gsd_openclaw.state_manager import SessionState, save_state, load_state, StateManager
|
|
|
|
|
|
class TestSessionStateCreation:
|
|
"""Test 1: SessionState can be created with all fields."""
|
|
|
|
def test_create_with_defaults(self):
|
|
"""SessionState can be created with only required fields."""
|
|
state = SessionState(
|
|
session_id="test-session-123",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test-project",
|
|
tmux_pane_id="%0"
|
|
)
|
|
assert state.session_id == "test-session-123"
|
|
assert state.project_name == "test-project"
|
|
assert state.tmux_session_name == "gsd-test-project"
|
|
assert state.tmux_pane_id == "%0"
|
|
assert state.schema_version == "1.0"
|
|
assert state.command_count == 0
|
|
|
|
def test_create_with_all_fields(self):
|
|
"""SessionState can be created with all fields specified."""
|
|
state = SessionState(
|
|
schema_version="1.0",
|
|
session_id="test-session-456",
|
|
project_name="my-project",
|
|
tmux_session_name="gsd-my-project",
|
|
tmux_pane_id="%1",
|
|
created_at="2026-03-10T10:00:00Z",
|
|
last_active_at="2026-03-10T12:30:00Z",
|
|
command_count=42,
|
|
last_output_snapshot="Last output here...",
|
|
display_name="My Cool Project"
|
|
)
|
|
assert state.session_id == "test-session-456"
|
|
assert state.created_at == "2026-03-10T10:00:00Z"
|
|
assert state.command_count == 42
|
|
assert state.last_output_snapshot == "Last output here..."
|
|
assert state.display_name == "My Cool Project"
|
|
|
|
|
|
class TestSessionStateSerialization:
|
|
"""Test 2 & 3 & 4: SessionState serialization methods."""
|
|
|
|
def test_to_dict_returns_complete_dict(self):
|
|
"""SessionState.to_dict() returns complete dict."""
|
|
state = SessionState(
|
|
session_id="test-session-789",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test-project",
|
|
tmux_pane_id="%2",
|
|
created_at="2026-03-10T10:00:00Z",
|
|
command_count=10
|
|
)
|
|
result = state.to_dict()
|
|
assert isinstance(result, dict)
|
|
assert result["session_id"] == "test-session-789"
|
|
assert result["project_name"] == "test-project"
|
|
assert result["tmux_session_name"] == "gsd-test-project"
|
|
assert result["tmux_pane_id"] == "%2"
|
|
assert result["created_at"] == "2026-03-10T10:00:00Z"
|
|
assert result["command_count"] == 10
|
|
assert result["schema_version"] == "1.0"
|
|
|
|
def test_from_dict_reconstructs_from_dict(self):
|
|
"""SessionState.from_dict() reconstructs from dict."""
|
|
data = {
|
|
"schema_version": "1.0",
|
|
"session_id": "test-session-abc",
|
|
"project_name": "reconstructed-project",
|
|
"tmux_session_name": "gsd-reconstructed",
|
|
"tmux_pane_id": "%3",
|
|
"created_at": "2026-03-10T08:00:00Z",
|
|
"last_active_at": "2026-03-10T09:00:00Z",
|
|
"command_count": 25,
|
|
"last_output_snapshot": "Some output",
|
|
"display_name": "Reconstructed Project"
|
|
}
|
|
state = SessionState.from_dict(data)
|
|
assert state.session_id == "test-session-abc"
|
|
assert state.project_name == "reconstructed-project"
|
|
assert state.tmux_session_name == "gsd-reconstructed"
|
|
assert state.tmux_pane_id == "%3"
|
|
assert state.created_at == "2026-03-10T08:00:00Z"
|
|
assert state.command_count == 25
|
|
assert state.last_output_snapshot == "Some output"
|
|
assert state.display_name == "Reconstructed Project"
|
|
|
|
def test_from_dict_ignores_unknown_fields(self):
|
|
"""SessionState.from_dict() ignores unknown fields in from_dict()."""
|
|
data = {
|
|
"schema_version": "1.0",
|
|
"session_id": "test-session-xyz",
|
|
"project_name": "test-project",
|
|
"tmux_session_name": "gsd-test",
|
|
"tmux_pane_id": "%4",
|
|
"unknown_field": "should be ignored",
|
|
"another_unknown": 12345
|
|
}
|
|
# Should not raise an error
|
|
state = SessionState.from_dict(data)
|
|
assert state.session_id == "test-session-xyz"
|
|
assert not hasattr(state, "unknown_field")
|
|
assert not hasattr(state, "another_unknown")
|
|
|
|
def test_to_dict_and_from_dict_roundtrip(self):
|
|
"""Verify to_dict and from_dict are inverses."""
|
|
original = SessionState(
|
|
session_id="roundtrip-test",
|
|
project_name="roundtrip-project",
|
|
tmux_session_name="gsd-roundtrip",
|
|
tmux_pane_id="%5",
|
|
created_at="2026-03-10T10:00:00Z",
|
|
last_active_at="2026-03-10T11:00:00Z",
|
|
command_count=100,
|
|
last_output_snapshot="Final output",
|
|
display_name="Roundtrip Test"
|
|
)
|
|
data = original.to_dict()
|
|
restored = SessionState.from_dict(data)
|
|
assert restored.session_id == original.session_id
|
|
assert restored.project_name == original.project_name
|
|
assert restored.tmux_session_name == original.tmux_session_name
|
|
assert restored.tmux_pane_id == original.tmux_pane_id
|
|
assert restored.created_at == original.created_at
|
|
assert restored.last_active_at == original.last_active_at
|
|
assert restored.command_count == original.command_count
|
|
assert restored.last_output_snapshot == original.last_output_snapshot
|
|
assert restored.display_name == original.display_name
|
|
|
|
|
|
class TestAtomicWrite:
|
|
"""Test 1 & 4 & 5: save_state creates file atomically."""
|
|
|
|
def test_save_state_creates_file_atomically(self, tmp_path):
|
|
"""save_state creates file atomically (temp file + rename)."""
|
|
state = SessionState(
|
|
session_id="test-atomic",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test",
|
|
tmux_pane_id="%0"
|
|
)
|
|
state_path = tmp_path / "state.json"
|
|
|
|
save_state(state, state_path)
|
|
|
|
assert state_path.exists()
|
|
# Verify content is valid JSON
|
|
with open(state_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
assert data["session_id"] == "test-atomic"
|
|
|
|
def test_temp_file_in_same_directory(self, tmp_path):
|
|
"""Temp file is created in same directory as target (for atomic rename)."""
|
|
state = SessionState(
|
|
session_id="test-same-dir",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test",
|
|
tmux_pane_id="%1"
|
|
)
|
|
state_path = tmp_path / "subdir" / "state.json"
|
|
|
|
save_state(state, state_path)
|
|
|
|
# After save, no temp files should remain
|
|
temp_files = list(tmp_path.rglob("*.tmp"))
|
|
assert len(temp_files) == 0
|
|
# State file should exist
|
|
assert state_path.exists()
|
|
|
|
def test_cleanup_temp_file_on_failure(self, tmp_path):
|
|
"""save_state cleans up temp file on failure."""
|
|
state = SessionState(
|
|
session_id="test-cleanup",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test",
|
|
tmux_pane_id="%2"
|
|
)
|
|
# Create a file that will cause write to fail (permission denied simulation)
|
|
state_path = tmp_path / "state.json"
|
|
|
|
# Make the directory read-only after creating a temp scenario
|
|
# This test verifies the exception handling path cleans up
|
|
# We can't easily test actual failure, but we can verify no temp files remain
|
|
save_state(state, state_path)
|
|
|
|
temp_files = list(tmp_path.glob("*.tmp"))
|
|
assert len(temp_files) == 0
|
|
|
|
|
|
class TestBackupRotation:
|
|
"""Test 2 & 3: save_state creates parent dirs and rotates backup."""
|
|
|
|
def test_save_state_creates_parent_directories(self, tmp_path):
|
|
"""save_state creates parent directories if missing."""
|
|
state = SessionState(
|
|
session_id="test-mkdir",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test",
|
|
tmux_pane_id="%3"
|
|
)
|
|
# Use nested path that doesn't exist
|
|
state_path = tmp_path / "deeply" / "nested" / "dir" / "state.json"
|
|
|
|
save_state(state, state_path)
|
|
|
|
assert state_path.exists()
|
|
assert state_path.parent.is_dir()
|
|
|
|
def test_save_state_rotates_backup(self, tmp_path):
|
|
"""save_state rotates backup (state.json -> state.json.bak)."""
|
|
state1 = SessionState(
|
|
session_id="first-save",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test",
|
|
tmux_pane_id="%4"
|
|
)
|
|
state2 = SessionState(
|
|
session_id="second-save",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test",
|
|
tmux_pane_id="%4"
|
|
)
|
|
state_path = tmp_path / "state.json"
|
|
backup_path = tmp_path / "state.json.bak"
|
|
|
|
# First save - no backup yet
|
|
save_state(state1, state_path)
|
|
assert state_path.exists()
|
|
assert not backup_path.exists()
|
|
|
|
# Second save - backup should be created
|
|
save_state(state2, state_path)
|
|
assert state_path.exists()
|
|
assert backup_path.exists()
|
|
|
|
# Backup should contain first save data
|
|
with open(backup_path, 'r', encoding='utf-8') as f:
|
|
backup_data = json.load(f)
|
|
assert backup_data["session_id"] == "first-save"
|
|
|
|
# Main file should contain second save data
|
|
with open(state_path, 'r', encoding='utf-8') as f:
|
|
main_data = json.load(f)
|
|
assert main_data["session_id"] == "second-save"
|
|
|
|
def test_save_state_overwrites_old_backup(self, tmp_path):
|
|
"""save_state overwrites old backup file."""
|
|
state1 = SessionState(session_id="v1", project_name="p", tmux_session_name="t", tmux_pane_id="%0")
|
|
state2 = SessionState(session_id="v2", project_name="p", tmux_session_name="t", tmux_pane_id="%0")
|
|
state3 = SessionState(session_id="v3", project_name="p", tmux_session_name="t", tmux_pane_id="%0")
|
|
state_path = tmp_path / "state.json"
|
|
backup_path = tmp_path / "state.json.bak"
|
|
|
|
save_state(state1, state_path)
|
|
save_state(state2, state_path)
|
|
save_state(state3, state_path)
|
|
|
|
# Backup should contain v2 (the previous state)
|
|
with open(backup_path, 'r', encoding='utf-8') as f:
|
|
backup_data = json.load(f)
|
|
assert backup_data["session_id"] == "v2"
|
|
|
|
# Main should contain v3
|
|
with open(state_path, 'r', encoding='utf-8') as f:
|
|
main_data = json.load(f)
|
|
assert main_data["session_id"] == "v3"
|
|
|
|
|
|
class TestSaveLoadState:
|
|
"""Test: save and load roundtrip."""
|
|
|
|
def test_save_load_state(self, tmp_path):
|
|
"""save_state and load_state work together."""
|
|
original = SessionState(
|
|
session_id="save-load-test",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test",
|
|
tmux_pane_id="%10",
|
|
created_at="2026-03-10T10:00:00Z",
|
|
command_count=50
|
|
)
|
|
state_path = tmp_path / "state.json"
|
|
|
|
save_state(original, state_path)
|
|
loaded = load_state(state_path)
|
|
|
|
assert loaded is not None
|
|
assert loaded.session_id == original.session_id
|
|
assert loaded.project_name == original.project_name
|
|
assert loaded.tmux_session_name == original.tmux_session_name
|
|
assert loaded.tmux_pane_id == original.tmux_pane_id
|
|
assert loaded.command_count == original.command_count
|
|
|
|
|
|
class TestStateValidation:
|
|
"""Test 3 & 4: load_state validates schema and required fields."""
|
|
|
|
def test_load_state_returns_none_for_missing_file(self, tmp_path):
|
|
"""load_state returns None for missing file."""
|
|
state_path = tmp_path / "nonexistent.json"
|
|
result = load_state(state_path)
|
|
assert result is None
|
|
|
|
def test_load_state_returns_session_state_for_valid_file(self, tmp_path):
|
|
"""load_state returns SessionState for valid file."""
|
|
state = SessionState(
|
|
session_id="valid-test",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test",
|
|
tmux_pane_id="%11"
|
|
)
|
|
state_path = tmp_path / "state.json"
|
|
save_state(state, state_path)
|
|
|
|
loaded = load_state(state_path)
|
|
assert loaded is not None
|
|
assert loaded.session_id == "valid-test"
|
|
|
|
def test_load_state_raises_value_error_for_unsupported_schema(self, tmp_path):
|
|
"""load_state raises ValueError for unsupported schema version."""
|
|
state_path = tmp_path / "state.json"
|
|
# Write a file with unsupported schema version
|
|
invalid_data = {
|
|
"schema_version": "2.0", # Unsupported
|
|
"session_id": "test",
|
|
"project_name": "test",
|
|
"tmux_session_name": "test",
|
|
"tmux_pane_id": "%0"
|
|
}
|
|
with open(state_path, 'w', encoding='utf-8') as f:
|
|
json.dump(invalid_data, f)
|
|
|
|
with pytest.raises(ValueError, match="Unsupported schema version"):
|
|
load_state(state_path)
|
|
|
|
def test_load_state_raises_value_error_for_missing_required_fields(self, tmp_path):
|
|
"""load_state raises ValueError for missing required fields."""
|
|
state_path = tmp_path / "state.json"
|
|
# Write a file with missing required field
|
|
invalid_data = {
|
|
"schema_version": "1.0",
|
|
"session_id": "test",
|
|
# Missing project_name, tmux_session_name
|
|
}
|
|
with open(state_path, 'w', encoding='utf-8') as f:
|
|
json.dump(invalid_data, f)
|
|
|
|
with pytest.raises(ValueError, match="Missing required field"):
|
|
load_state(state_path)
|
|
|
|
|
|
class TestBackupFallback:
|
|
"""Test 5 & 6: load_state falls back to backup when main file is corrupted."""
|
|
|
|
def test_load_state_falls_back_to_backup_on_json_error(self, tmp_path):
|
|
"""load_state falls back to backup when main file is corrupted (JSONDecodeError)."""
|
|
state_path = tmp_path / "state.json"
|
|
backup_path = tmp_path / "state.json.bak"
|
|
|
|
# Create a valid backup
|
|
valid_state = SessionState(
|
|
session_id="backup-valid",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test",
|
|
tmux_pane_id="%12"
|
|
)
|
|
backup_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(backup_path, 'w', encoding='utf-8') as f:
|
|
json.dump(valid_state.to_dict(), f)
|
|
|
|
# Create corrupted main file
|
|
with open(state_path, 'w', encoding='utf-8') as f:
|
|
f.write("{ invalid json }")
|
|
|
|
# Should fall back to backup
|
|
loaded = load_state(state_path)
|
|
assert loaded is not None
|
|
assert loaded.session_id == "backup-valid"
|
|
|
|
def test_load_state_falls_back_to_backup_on_validation_error(self, tmp_path):
|
|
"""load_state falls back to backup when validation fails."""
|
|
state_path = tmp_path / "state.json"
|
|
backup_path = tmp_path / "state.json.bak"
|
|
|
|
# Create a valid backup
|
|
valid_state = SessionState(
|
|
session_id="backup-valid-2",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test",
|
|
tmux_pane_id="%13"
|
|
)
|
|
backup_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(backup_path, 'w', encoding='utf-8') as f:
|
|
json.dump(valid_state.to_dict(), f)
|
|
|
|
# Create invalid main file (bad schema)
|
|
invalid_data = {
|
|
"schema_version": "99.0", # Invalid
|
|
"session_id": "test",
|
|
"project_name": "test",
|
|
"tmux_session_name": "test",
|
|
"tmux_pane_id": "%0"
|
|
}
|
|
with open(state_path, 'w', encoding='utf-8') as f:
|
|
json.dump(invalid_data, f)
|
|
|
|
# Should fall back to backup
|
|
loaded = load_state(state_path)
|
|
assert loaded is not None
|
|
assert loaded.session_id == "backup-valid-2"
|
|
|
|
def test_load_state_tries_backup_when_main_missing(self, tmp_path):
|
|
"""load_state tries backup when main file is missing."""
|
|
state_path = tmp_path / "state.json"
|
|
backup_path = tmp_path / "state.json.bak"
|
|
|
|
# Create a valid backup (no main file)
|
|
valid_state = SessionState(
|
|
session_id="backup-only",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test",
|
|
tmux_pane_id="%14"
|
|
)
|
|
backup_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(backup_path, 'w', encoding='utf-8') as f:
|
|
json.dump(valid_state.to_dict(), f)
|
|
|
|
# Should load from backup
|
|
loaded = load_state(state_path)
|
|
assert loaded is not None
|
|
assert loaded.session_id == "backup-only"
|
|
|
|
|
|
class TestStateManagerClass:
|
|
"""Test StateManager class with workspace path resolution."""
|
|
|
|
def test_get_state_path_returns_correct_path(self, tmp_path):
|
|
"""StateManager.get_state_path returns correct path for project."""
|
|
manager = StateManager("my-project", workspaces_root=tmp_path)
|
|
expected = tmp_path / "my-project" / ".gsd" / "state.json"
|
|
assert manager.get_state_path() == expected
|
|
|
|
def test_state_manager_save_persists_state(self, tmp_path):
|
|
"""StateManager.save persists state to correct location."""
|
|
manager = StateManager("test-project", workspaces_root=tmp_path)
|
|
state = SessionState(
|
|
session_id="manager-test",
|
|
project_name="test-project",
|
|
tmux_session_name="gsd-test-project",
|
|
tmux_pane_id="%20"
|
|
)
|
|
|
|
manager.save(state)
|
|
|
|
# Verify file was created at correct path
|
|
state_path = manager.get_state_path()
|
|
assert state_path.exists()
|
|
|
|
# Verify content
|
|
loaded = load_state(state_path)
|
|
assert loaded is not None
|
|
assert loaded.session_id == "manager-test"
|
|
|
|
def test_state_manager_load_retrieves_state(self, tmp_path):
|
|
"""StateManager.load retrieves state from correct location."""
|
|
manager = StateManager("load-test-project", workspaces_root=tmp_path)
|
|
state = SessionState(
|
|
session_id="load-manager-test",
|
|
project_name="load-test-project",
|
|
tmux_session_name="gsd-load-test",
|
|
tmux_pane_id="%21"
|
|
)
|
|
|
|
# Save using manager
|
|
manager.save(state)
|
|
|
|
# Load using manager
|
|
loaded = manager.load()
|
|
|
|
assert loaded is not None
|
|
assert loaded.session_id == "load-manager-test"
|
|
assert loaded.project_name == "load-test-project"
|
|
|
|
def test_state_manager_load_returns_none_when_no_state(self, tmp_path):
|
|
"""StateManager.load returns None when no state file exists."""
|
|
manager = StateManager("nonexistent-project", workspaces_root=tmp_path)
|
|
loaded = manager.load()
|
|
assert loaded is None
|
|
|
|
def test_state_manager_default_workspaces_root(self):
|
|
"""StateManager uses default ~/gsd-workspaces when not specified."""
|
|
manager = StateManager("default-project")
|
|
expected_root = Path.home() / "gsd-workspaces"
|
|
assert manager.workspaces_root == expected_root
|
|
assert manager.workspace_path == expected_root / "default-project"
|