82 lines
3.6 KiB
Python
82 lines
3.6 KiB
Python
import pytest
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
import os
|
|
|
|
# Import the real models
|
|
from models import TrackState, Metadata, Ticket
|
|
# Import the persistence functions from project_manager
|
|
from project_manager import save_track_state, load_track_state
|
|
|
|
def test_track_state_persistence(tmp_path):
|
|
"""
|
|
Tests saving and loading a TrackState object to/from a TOML file.
|
|
1. Create a TrackState object with sample metadata, discussion, and tasks.
|
|
2. Call save_track_state('test_track', state, base_dir).
|
|
3. Verify that base_dir/conductor/tracks/test_track/state.toml exists.
|
|
4. Call load_track_state('test_track', base_dir) and verify it returns an identical TrackState object.
|
|
"""
|
|
base_dir = tmp_path
|
|
track_id = "test-track-999" # Metadata internal ID
|
|
track_folder_name = "test_track" # Folder name used in persistence
|
|
|
|
# 1. Create a TrackState object with sample data
|
|
metadata = Metadata(
|
|
id=track_id,
|
|
name="Test Track",
|
|
status="in_progress",
|
|
created_at=datetime(2023, 1, 1, 12, 0, 0),
|
|
updated_at=datetime(2023, 1, 2, 13, 0, 0)
|
|
)
|
|
|
|
discussion = [
|
|
{"role": "User", "content": "Hello", "ts": datetime(2023, 1, 1, 12, 0, 0)},
|
|
{"role": "AI", "content": "Hi there!", "ts": datetime(2023, 1, 1, 12, 0, 5)}
|
|
]
|
|
|
|
tasks = [
|
|
Ticket(id="task-1", description="First task", status="completed", assigned_to="worker-1"),
|
|
Ticket(id="task-2", description="Second task", status="todo", assigned_to="worker-2")
|
|
]
|
|
|
|
original_state = TrackState(
|
|
metadata=metadata,
|
|
discussion=discussion,
|
|
tasks=tasks
|
|
)
|
|
|
|
# 2. Call save_track_state('test_track', state, base_dir)
|
|
save_track_state(track_folder_name, original_state, base_dir)
|
|
|
|
# 3. Verify that base_dir/conductor/tracks/test_track/state.toml exists
|
|
state_file_path = base_dir / "conductor" / "tracks" / track_folder_name / "state.toml"
|
|
assert state_file_path.exists(), f"State file should exist at {state_file_path}"
|
|
|
|
# 4. Call load_track_state('test_track', base_dir) and verify it returns an identical TrackState object
|
|
loaded_state = load_track_state(track_folder_name, base_dir)
|
|
|
|
assert loaded_state is not None, "load_track_state returned None"
|
|
|
|
# Verify equality
|
|
assert loaded_state.metadata.id == original_state.metadata.id
|
|
assert loaded_state.metadata.name == original_state.metadata.name
|
|
assert loaded_state.metadata.status == original_state.metadata.status
|
|
assert loaded_state.metadata.created_at == original_state.metadata.created_at
|
|
assert loaded_state.metadata.updated_at == original_state.metadata.updated_at
|
|
|
|
assert len(loaded_state.tasks) == len(original_state.tasks)
|
|
for i in range(len(original_state.tasks)):
|
|
assert loaded_state.tasks[i].id == original_state.tasks[i].id
|
|
assert loaded_state.tasks[i].description == original_state.tasks[i].description
|
|
assert loaded_state.tasks[i].status == original_state.tasks[i].status
|
|
assert loaded_state.tasks[i].assigned_to == original_state.tasks[i].assigned_to
|
|
|
|
assert len(loaded_state.discussion) == len(original_state.discussion)
|
|
for i in range(len(original_state.discussion)):
|
|
assert loaded_state.discussion[i]["role"] == original_state.discussion[i]["role"]
|
|
assert loaded_state.discussion[i]["content"] == original_state.discussion[i]["content"]
|
|
assert loaded_state.discussion[i]["ts"] == original_state.discussion[i]["ts"]
|
|
|
|
# Final check: deep equality of dataclasses
|
|
assert loaded_state == original_state
|