Implement complete MCP server for parsing Primavera P6 XER files and exposing schedule data through MCP tools. All 4 user stories complete. Tools implemented: - load_xer: Parse XER files into SQLite database - list_activities: Query activities with pagination and filtering - get_activity: Get activity details by ID - list_relationships: Query activity dependencies - get_predecessors/get_successors: Query activity relationships - get_project_summary: Project overview with counts - list_milestones: Query milestone activities - get_critical_path: Query driving path activities Features: - Tab-delimited XER format parsing with pluggable table handlers - In-memory SQLite database for fast queries - Pagination with 100-item default limit - Multi-project file support with project selection - ISO8601 date formatting - NO_FILE_LOADED error handling for all query tools Test coverage: 81 tests (contract, integration, unit)
147 lines
5.7 KiB
Python
147 lines
5.7 KiB
Python
"""Unit tests for XER parser."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
class TestXerParser:
|
|
"""Tests for the XER file parser."""
|
|
|
|
def test_parse_single_project_file(self, sample_xer_single_project: Path) -> None:
|
|
"""Parser should extract project data from single-project XER file."""
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
result = parser.parse(sample_xer_single_project)
|
|
|
|
assert len(result.projects) == 1
|
|
assert result.projects[0]["proj_id"] == "1001"
|
|
assert result.projects[0]["proj_short_name"] == "Test Project"
|
|
|
|
def test_parse_multi_project_file(self, sample_xer_multi_project: Path) -> None:
|
|
"""Parser should extract all projects from multi-project XER file."""
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
result = parser.parse(sample_xer_multi_project)
|
|
|
|
assert len(result.projects) == 2
|
|
project_names = {p["proj_short_name"] for p in result.projects}
|
|
assert project_names == {"Project Alpha", "Project Beta"}
|
|
|
|
def test_parse_activities(self, sample_xer_single_project: Path) -> None:
|
|
"""Parser should extract activities from XER file."""
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
result = parser.parse(sample_xer_single_project)
|
|
|
|
# Single project fixture has 5 activities
|
|
assert len(result.tasks) == 5
|
|
|
|
# Check first milestone
|
|
milestone = next(t for t in result.tasks if t["task_code"] == "A1000")
|
|
assert milestone["task_name"] == "Project Start"
|
|
assert milestone["task_type"] == "TT_Mile"
|
|
|
|
def test_parse_relationships(self, sample_xer_single_project: Path) -> None:
|
|
"""Parser should extract relationships from XER file."""
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
result = parser.parse(sample_xer_single_project)
|
|
|
|
# Single project fixture has 5 relationships
|
|
assert len(result.taskpreds) == 5
|
|
|
|
# Check a FS relationship
|
|
fs_rel = next(r for r in result.taskpreds if r["pred_type"] == "PR_FS")
|
|
assert fs_rel["lag_hr_cnt"] == 0
|
|
|
|
# Check a SS relationship
|
|
ss_rel = next(r for r in result.taskpreds if r["pred_type"] == "PR_SS")
|
|
assert ss_rel["lag_hr_cnt"] == 40
|
|
|
|
def test_parse_wbs(self, sample_xer_single_project: Path) -> None:
|
|
"""Parser should extract WBS hierarchy from XER file."""
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
result = parser.parse(sample_xer_single_project)
|
|
|
|
# Single project fixture has 3 WBS elements
|
|
assert len(result.projwbs) == 3
|
|
|
|
# Check root WBS
|
|
root = next(w for w in result.projwbs if w["wbs_short_name"] == "ROOT")
|
|
assert root["parent_wbs_id"] is None or root["parent_wbs_id"] == ""
|
|
|
|
def test_parse_calendars(self, sample_xer_single_project: Path) -> None:
|
|
"""Parser should extract calendars from XER file."""
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
result = parser.parse(sample_xer_single_project)
|
|
|
|
# Single project fixture has 1 calendar
|
|
assert len(result.calendars) == 1
|
|
cal = result.calendars[0]
|
|
assert cal["clndr_name"] == "Standard 5 Day"
|
|
assert cal["day_hr_cnt"] == 8
|
|
|
|
def test_parse_empty_project(self, sample_xer_empty: Path) -> None:
|
|
"""Parser should handle XER file with no activities."""
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
result = parser.parse(sample_xer_empty)
|
|
|
|
assert len(result.projects) == 1
|
|
assert len(result.tasks) == 0
|
|
assert len(result.taskpreds) == 0
|
|
|
|
def test_parse_invalid_file_raises_error(self, invalid_xer_file: Path) -> None:
|
|
"""Parser should raise ParseError for invalid XER content."""
|
|
from xer_mcp.errors import ParseError
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
with pytest.raises(ParseError):
|
|
parser.parse(invalid_xer_file)
|
|
|
|
def test_parse_nonexistent_file_raises_error(self, nonexistent_xer_path: Path) -> None:
|
|
"""Parser should raise FileNotFoundError for missing file."""
|
|
from xer_mcp.errors import FileNotFoundError
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
with pytest.raises(FileNotFoundError):
|
|
parser.parse(nonexistent_xer_path)
|
|
|
|
def test_parse_dates_converted_to_iso8601(self, sample_xer_single_project: Path) -> None:
|
|
"""Parser should convert XER dates to ISO8601 format."""
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
result = parser.parse(sample_xer_single_project)
|
|
|
|
# Check date conversion (XER: "2026-01-01 07:00" -> ISO: "2026-01-01T07:00:00")
|
|
task = next(t for t in result.tasks if t["task_code"] == "A1000")
|
|
assert "T" in task["target_start_date"]
|
|
|
|
def test_parse_driving_path_flag(self, sample_xer_single_project: Path) -> None:
|
|
"""Parser should correctly parse driving_path_flag as boolean."""
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
result = parser.parse(sample_xer_single_project)
|
|
|
|
# A1000 has driving_path_flag = Y
|
|
critical_task = next(t for t in result.tasks if t["task_code"] == "A1000")
|
|
assert critical_task["driving_path_flag"] is True
|
|
|
|
# A1020 has driving_path_flag = N
|
|
non_critical = next(t for t in result.tasks if t["task_code"] == "A1020")
|
|
assert non_critical["driving_path_flag"] is False
|