Spaces:
Sleeping
Sleeping
| """Tests for core/planner.py Pydantic models and parse functions.""" | |
| import os | |
| import sys | |
| import json | |
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) | |
| from core.planner import ( | |
| Phase, | |
| Risk, | |
| WorkloadWeek, | |
| ProjectPlan, | |
| parse_plan_json, | |
| merge_plan_update, | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Phase validator tests | |
| # --------------------------------------------------------------------------- | |
| def test_phase_duration_days_clamped_to_min_1(): | |
| phase = Phase(id="p1", name="Setup", duration_days=0) | |
| assert phase.duration_days == 1 | |
| def test_phase_duration_days_negative_clamped_to_min_1(): | |
| phase = Phase(id="p1", name="Setup", duration_days=-5) | |
| assert phase.duration_days == 1 | |
| def test_phase_duration_days_positive_unchanged(): | |
| phase = Phase(id="p1", name="Setup", duration_days=7) | |
| assert phase.duration_days == 7 | |
| def test_phase_start_offset_days_clamped_to_min_0(): | |
| phase = Phase(id="p1", name="Setup", start_offset_days=-3) | |
| assert phase.start_offset_days == 0 | |
| def test_phase_start_offset_days_zero_unchanged(): | |
| phase = Phase(id="p1", name="Setup", start_offset_days=0) | |
| assert phase.start_offset_days == 0 | |
| def test_phase_start_offset_days_positive_unchanged(): | |
| phase = Phase(id="p1", name="Setup", start_offset_days=10) | |
| assert phase.start_offset_days == 10 | |
| # --------------------------------------------------------------------------- | |
| # Risk validator tests | |
| # --------------------------------------------------------------------------- | |
| def test_risk_invalid_likelihood_defaults_to_medium(): | |
| risk = Risk(description="Server crash", likelihood="Critical") | |
| assert risk.likelihood == "Medium" | |
| def test_risk_invalid_impact_defaults_to_medium(): | |
| risk = Risk(description="Server crash", impact="Severe") | |
| assert risk.impact == "Medium" | |
| def test_risk_valid_likelihood_preserved(): | |
| for level in ("Low", "Medium", "High"): | |
| risk = Risk(description="Test risk", likelihood=level) | |
| assert risk.likelihood == level | |
| def test_risk_valid_impact_preserved(): | |
| for level in ("Low", "Medium", "High"): | |
| risk = Risk(description="Test risk", impact=level) | |
| assert risk.impact == level | |
| # --------------------------------------------------------------------------- | |
| # WorkloadWeek validator tests | |
| # --------------------------------------------------------------------------- | |
| def test_workload_week_load_score_clamped_above_10(): | |
| week = WorkloadWeek(week_label="Week 1", load_score=15.0) | |
| assert week.load_score == 10.0 | |
| def test_workload_week_load_score_clamped_below_0(): | |
| week = WorkloadWeek(week_label="Week 1", load_score=-2.0) | |
| assert week.load_score == 0.0 | |
| def test_workload_week_load_score_within_range_unchanged(): | |
| week = WorkloadWeek(week_label="Week 1", load_score=7.5) | |
| assert week.load_score == 7.5 | |
| def test_workload_week_load_score_boundary_values(): | |
| week_zero = WorkloadWeek(week_label="W1", load_score=0.0) | |
| week_ten = WorkloadWeek(week_label="W2", load_score=10.0) | |
| assert week_zero.load_score == 0.0 | |
| assert week_ten.load_score == 10.0 | |
| # --------------------------------------------------------------------------- | |
| # parse_plan_json tests | |
| # --------------------------------------------------------------------------- | |
| def test_parse_plan_json_valid(): | |
| data = { | |
| "phases": [ | |
| { | |
| "id": "phase-1", | |
| "name": "Discovery", | |
| "duration_days": 5, | |
| "start_offset_days": 0, | |
| } | |
| ], | |
| "risks": [ | |
| {"description": "Scope creep", "likelihood": "High", "impact": "Medium"} | |
| ], | |
| } | |
| plan = parse_plan_json(json.dumps(data)) | |
| assert isinstance(plan, ProjectPlan) | |
| assert len(plan.phases) == 1 | |
| assert plan.phases[0].name == "Discovery" | |
| assert plan.phases[0].duration_days == 5 | |
| assert len(plan.risks) == 1 | |
| assert plan.risks[0].likelihood == "High" | |
| def test_parse_plan_json_bad_json_returns_empty_plan(): | |
| plan = parse_plan_json("this is not json at all {{{") | |
| assert isinstance(plan, ProjectPlan) | |
| assert plan.phases == [] | |
| assert plan.risks == [] | |
| assert plan.raci == [] | |
| assert plan.workload == [] | |
| def test_parse_plan_json_empty_string_returns_empty_plan(): | |
| plan = parse_plan_json("") | |
| assert isinstance(plan, ProjectPlan) | |
| assert plan.phases == [] | |
| def test_parse_plan_json_strips_markdown_code_fences(): | |
| data = {"phases": [{"id": "p1", "name": "Alpha", "duration_days": 3}]} | |
| raw = "```json\n" + json.dumps(data) + "\n```" | |
| plan = parse_plan_json(raw) | |
| assert isinstance(plan, ProjectPlan) | |
| assert len(plan.phases) == 1 | |
| assert plan.phases[0].name == "Alpha" | |
| def test_parse_plan_json_strips_plain_code_fences(): | |
| data = {"phases": [{"id": "p2", "name": "Beta", "duration_days": 2}]} | |
| raw = "```\n" + json.dumps(data) + "\n```" | |
| plan = parse_plan_json(raw) | |
| assert isinstance(plan, ProjectPlan) | |
| assert len(plan.phases) == 1 | |
| assert plan.phases[0].name == "Beta" | |
| def test_parse_plan_json_invalid_structure_returns_empty_plan(): | |
| plan = parse_plan_json(json.dumps(["not", "a", "dict"])) | |
| assert isinstance(plan, ProjectPlan) | |
| # --------------------------------------------------------------------------- | |
| # merge_plan_update tests | |
| # --------------------------------------------------------------------------- | |
| def test_merge_plan_update_adds_new_field(): | |
| existing = ProjectPlan() | |
| updated = merge_plan_update(existing, {"estimated_start_date": "2026-05-01"}) | |
| assert updated.estimated_start_date == "2026-05-01" | |
| def test_merge_plan_update_replaces_phases(): | |
| existing = ProjectPlan( | |
| phases=[Phase(id="old", name="Old Phase", duration_days=3)] | |
| ) | |
| new_phases = [{"id": "new", "name": "New Phase", "duration_days": 7}] | |
| updated = merge_plan_update(existing, {"phases": new_phases}) | |
| assert len(updated.phases) == 1 | |
| assert updated.phases[0].id == "new" | |
| assert updated.phases[0].duration_days == 7 | |
| def test_merge_plan_update_preserves_unmentioned_fields(): | |
| existing = ProjectPlan( | |
| estimated_start_date="2026-01-01", | |
| timeline_note="On track", | |
| ) | |
| updated = merge_plan_update(existing, {"estimated_end_date": "2026-06-30"}) | |
| assert updated.estimated_start_date == "2026-01-01" | |
| assert updated.timeline_note == "On track" | |
| assert updated.estimated_end_date == "2026-06-30" | |
| def test_merge_plan_update_overwrites_existing_field(): | |
| existing = ProjectPlan(timeline_warning="Behind schedule") | |
| updated = merge_plan_update(existing, {"timeline_warning": "Back on track"}) | |
| assert updated.timeline_warning == "Back on track" | |
| def test_merge_plan_update_returns_project_plan_instance(): | |
| existing = ProjectPlan() | |
| result = merge_plan_update(existing, {}) | |
| assert isinstance(result, ProjectPlan) | |
| # --------------------------------------------------------------------------- | |
| # Entry point for running tests directly | |
| # --------------------------------------------------------------------------- | |
| if __name__ == "__main__": | |
| test_functions = [v for k, v in globals().items() if k.startswith("test_")] | |
| passed = 0 | |
| failed = 0 | |
| for fn in test_functions: | |
| try: | |
| fn() | |
| print(f" PASS {fn.__name__}") | |
| passed += 1 | |
| except Exception as exc: | |
| print(f" FAIL {fn.__name__}: {exc}") | |
| failed += 1 | |
| print(f"\n{passed} passed, {failed} failed") | |