Private
Public Access
ed
89ce7ad770
project_manager.py (3 sites): - get_all_tracks returns list[dict[str, Any]] where each dict now has an 'errors' field (list[ErrorInfo]) capturing per-track metadata recovery. The 3 SILENT_SWALLOW sites (state.from_dict, metadata.json, plan.md) now append to this list instead of silently passing. orchestrator_pm.py (2 sites): - get_track_history_summary returns Result[str]. The 2 SILENT_SWALLOW sites (metadata.json + spec.md reads) append to a scan_errors list that's threaded through the Result. Tests updated to check result.ok and use result.data.
78 lines
3.3 KiB
Python
78 lines
3.3 KiB
Python
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
import shutil
|
|
import json
|
|
from pathlib import Path
|
|
from src import orchestrator_pm
|
|
from src.result_types import Result
|
|
|
|
class TestOrchestratorPMHistory(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.test_dir = Path("test_conductor")
|
|
self.test_dir.mkdir(exist_ok=True)
|
|
self.archive_dir = self.test_dir / "archive"
|
|
self.tracks_dir = self.test_dir / "tracks"
|
|
self.archive_dir.mkdir(exist_ok=True)
|
|
self.tracks_dir.mkdir(exist_ok=True)
|
|
|
|
def tearDown(self) -> None:
|
|
if self.test_dir.exists():
|
|
shutil.rmtree(self.test_dir)
|
|
|
|
def create_track(self, parent_dir: Path, track_id: str, title: str, status: str, overview: str) -> None:
|
|
track_path = parent_dir / track_id
|
|
track_path.mkdir(exist_ok=True)
|
|
metadata = {"title": title, "status": status}
|
|
with open(track_path / "metadata.json", "w") as f:
|
|
json.dump(metadata, f)
|
|
spec_content = f"# Specification\n\n## Overview\n{overview}"
|
|
with open(track_path / "spec.md", "w") as f:
|
|
f.write(spec_content)
|
|
|
|
@patch('src.paths.get_archive_dir')
|
|
@patch('src.paths.get_tracks_dir')
|
|
def test_get_track_history_summary(self, mock_get_tracks: MagicMock, mock_get_archive: MagicMock) -> None:
|
|
mock_get_archive.return_value = self.archive_dir
|
|
mock_get_tracks.return_value = self.tracks_dir
|
|
self.create_track(self.archive_dir, "track_001", "Initial Setup", "completed", "Setting up the project structure.")
|
|
self.create_track(self.tracks_dir, "track_002", "Feature A", "in_progress", "Implementing Feature A.")
|
|
summary = orchestrator_pm.get_track_history_summary()
|
|
self.assertTrue(summary.ok, f"get_track_history_summary failed: {summary.errors}")
|
|
body = summary.data
|
|
self.assertIn("Initial Setup", body)
|
|
self.assertIn("completed", body)
|
|
self.assertIn("Setting up the project structure.", body)
|
|
self.assertIn("Feature A", body)
|
|
self.assertIn("in_progress", body)
|
|
self.assertIn("Implementing Feature A.", body)
|
|
|
|
@patch('src.paths.get_archive_dir')
|
|
@patch('src.paths.get_tracks_dir')
|
|
def test_get_track_history_summary_missing_files(self, mock_get_tracks: MagicMock, mock_get_archive: MagicMock) -> None:
|
|
mock_get_archive.return_value = self.archive_dir
|
|
mock_get_tracks.return_value = self.tracks_dir
|
|
track_path = self.tracks_dir / "track_003"
|
|
track_path.mkdir(exist_ok=True)
|
|
with open(track_path / "metadata.json", "w") as f:
|
|
json.dump({"title": "Missing Spec", "status": "pending"}, f)
|
|
summary = orchestrator_pm.get_track_history_summary()
|
|
self.assertTrue(summary.ok, f"get_track_history_summary failed: {summary.errors}")
|
|
body = summary.data
|
|
self.assertIn("Missing Spec", body)
|
|
self.assertIn("pending", body)
|
|
self.assertIn("No overview available", body)
|
|
|
|
@patch('src.orchestrator_pm.summarize.build_summary_markdown')
|
|
@patch('src.ai_client.send')
|
|
def test_generate_tracks_with_history(self, mock_send: MagicMock, mock_summarize: MagicMock) -> None:
|
|
mock_summarize.return_value = "REPO_MAP"
|
|
mock_send.return_value = Result(data="[]")
|
|
history_summary = "PAST_HISTORY_SUMMARY"
|
|
orchestrator_pm.generate_tracks("req", {}, [], history_summary=history_summary)
|
|
args, kwargs = mock_send.call_args
|
|
self.assertIn(history_summary, kwargs['user_message'])
|
|
self.assertIn("### TRACK HISTORY:", kwargs['user_message'])
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|