Bill Ballou
af8cdc1d31
Add computed driving flag to all relationship queries (list_relationships, get_predecessors, get_successors). A relationship is marked as driving when the predecessor's early end date plus lag determines the successor's early start date. Changes: - Add early_start_date and early_end_date columns to activities schema - Parse early dates from TASK table in XER files - Implement is_driving_relationship() helper with 24hr tolerance for calendar gaps - Update all relationship queries to compute and return driving flag - Add contract and unit tests for driving flag functionality - Update spec, contracts, and documentation
177 lines
6.2 KiB
Python
177 lines
6.2 KiB
Python
"""Unit tests for database query functions."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from xer_mcp.db import db
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_db():
|
|
"""Initialize and clear database for each test."""
|
|
db.initialize()
|
|
yield
|
|
db.clear()
|
|
|
|
|
|
class TestActivityQueries:
|
|
"""Tests for activity query functions."""
|
|
|
|
def test_query_activities_with_pagination(self, sample_xer_single_project: Path) -> None:
|
|
"""Should return paginated activity results."""
|
|
from xer_mcp.db.loader import load_parsed_data
|
|
from xer_mcp.db.queries import query_activities
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
parsed = parser.parse(sample_xer_single_project)
|
|
load_parsed_data(parsed, project_id="1001")
|
|
|
|
activities, total = query_activities(limit=2, offset=0)
|
|
|
|
assert len(activities) == 2
|
|
assert total == 5
|
|
|
|
def test_query_activities_filter_by_type(self, sample_xer_single_project: Path) -> None:
|
|
"""Should filter activities by task type."""
|
|
from xer_mcp.db.loader import load_parsed_data
|
|
from xer_mcp.db.queries import query_activities
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
parsed = parser.parse(sample_xer_single_project)
|
|
load_parsed_data(parsed, project_id="1001")
|
|
|
|
activities, total = query_activities(activity_type="TT_Mile")
|
|
|
|
assert total == 2
|
|
for act in activities:
|
|
assert act["task_type"] == "TT_Mile"
|
|
|
|
def test_query_activities_filter_by_date_range(self, sample_xer_single_project: Path) -> None:
|
|
"""Should filter activities by date range."""
|
|
from xer_mcp.db.loader import load_parsed_data
|
|
from xer_mcp.db.queries import query_activities
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
parsed = parser.parse(sample_xer_single_project)
|
|
load_parsed_data(parsed, project_id="1001")
|
|
|
|
# Filter to very narrow range
|
|
activities, total = query_activities(start_date="2026-01-01", end_date="2026-01-01")
|
|
|
|
# Only activities starting on 2026-01-01
|
|
for act in activities:
|
|
assert "2026-01-01" in act["target_start_date"]
|
|
|
|
def test_query_activities_filter_by_wbs(self, sample_xer_single_project: Path) -> None:
|
|
"""Should filter activities by WBS ID."""
|
|
from xer_mcp.db.loader import load_parsed_data
|
|
from xer_mcp.db.queries import query_activities
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
parsed = parser.parse(sample_xer_single_project)
|
|
load_parsed_data(parsed, project_id="1001")
|
|
|
|
activities, total = query_activities(wbs_id="102")
|
|
|
|
# WBS 102 has 2 activities
|
|
assert total == 2
|
|
|
|
def test_get_activity_by_id(self, sample_xer_single_project: Path) -> None:
|
|
"""Should return single activity by ID."""
|
|
from xer_mcp.db.loader import load_parsed_data
|
|
from xer_mcp.db.queries import get_activity_by_id
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
parsed = parser.parse(sample_xer_single_project)
|
|
load_parsed_data(parsed, project_id="1001")
|
|
|
|
activity = get_activity_by_id("2002")
|
|
|
|
assert activity is not None
|
|
assert activity["task_code"] == "A1010"
|
|
|
|
def test_get_activity_by_id_not_found(self, sample_xer_single_project: Path) -> None:
|
|
"""Should return None for non-existent activity."""
|
|
from xer_mcp.db.loader import load_parsed_data
|
|
from xer_mcp.db.queries import get_activity_by_id
|
|
from xer_mcp.parser.xer_parser import XerParser
|
|
|
|
parser = XerParser()
|
|
parsed = parser.parse(sample_xer_single_project)
|
|
load_parsed_data(parsed, project_id="1001")
|
|
|
|
activity = get_activity_by_id("nonexistent")
|
|
|
|
assert activity is None
|
|
|
|
|
|
class TestDrivingRelationship:
|
|
"""Tests for driving relationship computation."""
|
|
|
|
def test_is_driving_relationship_fs_driving(self) -> None:
|
|
"""FS relationship is driving when pred_end + lag = succ_start."""
|
|
from xer_mcp.db.queries import is_driving_relationship
|
|
|
|
# Pred ends at 2026-01-08T15:00, succ starts at 2026-01-09T07:00
|
|
# With 0 lag and overnight gap, this is driving
|
|
result = is_driving_relationship(
|
|
pred_early_end="2026-01-08T15:00:00",
|
|
succ_early_start="2026-01-09T07:00:00",
|
|
lag_hours=0.0,
|
|
pred_type="FS",
|
|
)
|
|
assert result is True
|
|
|
|
def test_is_driving_relationship_fs_not_driving(self) -> None:
|
|
"""FS relationship is not driving when there's float."""
|
|
from xer_mcp.db.queries import is_driving_relationship
|
|
|
|
# Pred ends much earlier than succ starts (has float)
|
|
result = is_driving_relationship(
|
|
pred_early_end="2026-01-01T15:00:00",
|
|
succ_early_start="2026-01-10T07:00:00",
|
|
lag_hours=0.0,
|
|
pred_type="FS",
|
|
)
|
|
assert result is False
|
|
|
|
def test_is_driving_relationship_with_lag(self) -> None:
|
|
"""FS relationship with lag is driving when pred_end + lag = succ_start."""
|
|
from xer_mcp.db.queries import is_driving_relationship
|
|
|
|
# Pred ends at 2026-01-08T15:00, 16hr lag, succ starts at 2026-01-09T07:00
|
|
# 15:00 + 16hrs = next day 07:00 (exactly matches)
|
|
result = is_driving_relationship(
|
|
pred_early_end="2026-01-08T15:00:00",
|
|
succ_early_start="2026-01-09T07:00:00",
|
|
lag_hours=16.0,
|
|
pred_type="FS",
|
|
)
|
|
assert result is True
|
|
|
|
def test_is_driving_relationship_missing_dates(self) -> None:
|
|
"""Relationship is not driving when dates are missing."""
|
|
from xer_mcp.db.queries import is_driving_relationship
|
|
|
|
result = is_driving_relationship(
|
|
pred_early_end=None,
|
|
succ_early_start="2026-01-09T07:00:00",
|
|
lag_hours=0.0,
|
|
pred_type="FS",
|
|
)
|
|
assert result is False
|
|
|
|
result = is_driving_relationship(
|
|
pred_early_end="2026-01-08T15:00:00",
|
|
succ_early_start=None,
|
|
lag_hours=0.0,
|
|
pred_type="FS",
|
|
)
|
|
assert result is False
|