|
| 1 | +"""Unit tests for sync plan output helpers.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +import io |
| 6 | +import json |
| 7 | +import typing as t |
| 8 | +from contextlib import redirect_stdout |
| 9 | + |
| 10 | +import pytest |
| 11 | + |
| 12 | +from vcspull.cli._colors import ColorMode, Colors |
| 13 | +from vcspull.cli._output import ( |
| 14 | + OutputFormatter, |
| 15 | + OutputMode, |
| 16 | + PlanAction, |
| 17 | + PlanEntry, |
| 18 | + PlanResult, |
| 19 | + PlanSummary, |
| 20 | +) |
| 21 | +from vcspull.cli.sync import PlanProgressPrinter |
| 22 | + |
| 23 | + |
| 24 | +class PlanEntryPayloadFixture(t.NamedTuple): |
| 25 | + """Fixture for PlanEntry payload serialization.""" |
| 26 | + |
| 27 | + test_id: str |
| 28 | + kwargs: dict[str, t.Any] |
| 29 | + expected_keys: dict[str, t.Any] |
| 30 | + unexpected_keys: set[str] |
| 31 | + |
| 32 | + |
| 33 | +PLAN_ENTRY_PAYLOAD_FIXTURES: list[PlanEntryPayloadFixture] = [ |
| 34 | + PlanEntryPayloadFixture( |
| 35 | + test_id="clone-with-url", |
| 36 | + kwargs={ |
| 37 | + "name": "repo-one", |
| 38 | + "path": "/tmp/repo-one", |
| 39 | + "workspace_root": "~/code/", |
| 40 | + "action": PlanAction.CLONE, |
| 41 | + "detail": "missing", |
| 42 | + "url": "git+https://example.com/repo-one.git", |
| 43 | + }, |
| 44 | + expected_keys={ |
| 45 | + "type": "operation", |
| 46 | + "action": "clone", |
| 47 | + "detail": "missing", |
| 48 | + "url": "git+https://example.com/repo-one.git", |
| 49 | + }, |
| 50 | + unexpected_keys={"branch", "ahead", "behind", "dirty", "error"}, |
| 51 | + ), |
| 52 | + PlanEntryPayloadFixture( |
| 53 | + test_id="update-with-status", |
| 54 | + kwargs={ |
| 55 | + "name": "repo-two", |
| 56 | + "path": "/tmp/repo-two", |
| 57 | + "workspace_root": "~/code/", |
| 58 | + "action": PlanAction.UPDATE, |
| 59 | + "detail": "behind 2", |
| 60 | + "branch": "main", |
| 61 | + "remote_branch": "origin/main", |
| 62 | + "current_rev": "abc1234", |
| 63 | + "target_rev": "def5678", |
| 64 | + "ahead": 0, |
| 65 | + "behind": 2, |
| 66 | + "dirty": False, |
| 67 | + }, |
| 68 | + expected_keys={ |
| 69 | + "branch": "main", |
| 70 | + "remote_branch": "origin/main", |
| 71 | + "current_rev": "abc1234", |
| 72 | + "target_rev": "def5678", |
| 73 | + "ahead": 0, |
| 74 | + "behind": 2, |
| 75 | + "dirty": False, |
| 76 | + }, |
| 77 | + unexpected_keys={"url", "error"}, |
| 78 | + ), |
| 79 | +] |
| 80 | + |
| 81 | + |
| 82 | +@pytest.mark.parametrize( |
| 83 | + list(PlanEntryPayloadFixture._fields), |
| 84 | + PLAN_ENTRY_PAYLOAD_FIXTURES, |
| 85 | + ids=[fixture.test_id for fixture in PLAN_ENTRY_PAYLOAD_FIXTURES], |
| 86 | +) |
| 87 | +def test_plan_entry_to_payload( |
| 88 | + test_id: str, |
| 89 | + kwargs: dict[str, t.Any], |
| 90 | + expected_keys: dict[str, t.Any], |
| 91 | + unexpected_keys: set[str], |
| 92 | +) -> None: |
| 93 | + """Ensure PlanEntry serialises optional fields correctly.""" |
| 94 | + entry = PlanEntry(**kwargs) |
| 95 | + payload = entry.to_payload() |
| 96 | + |
| 97 | + for key, value in expected_keys.items(): |
| 98 | + assert payload[key] == value |
| 99 | + |
| 100 | + for key in unexpected_keys: |
| 101 | + assert key not in payload |
| 102 | + |
| 103 | + assert payload["format_version"] == "1" |
| 104 | + assert payload["type"] == "operation" |
| 105 | + assert payload["name"] == kwargs["name"] |
| 106 | + assert payload["path"] == kwargs["path"] |
| 107 | + assert payload["workspace_root"] == kwargs["workspace_root"] |
| 108 | + |
| 109 | + |
| 110 | +class PlanSummaryPayloadFixture(t.NamedTuple): |
| 111 | + """Fixture for PlanSummary payload serialization.""" |
| 112 | + |
| 113 | + test_id: str |
| 114 | + summary: PlanSummary |
| 115 | + expected_total: int |
| 116 | + |
| 117 | + |
| 118 | +PLAN_SUMMARY_PAYLOAD_FIXTURES: list[PlanSummaryPayloadFixture] = [ |
| 119 | + PlanSummaryPayloadFixture( |
| 120 | + test_id="basic-counts", |
| 121 | + summary=PlanSummary(clone=1, update=2, unchanged=3, blocked=4, errors=5), |
| 122 | + expected_total=15, |
| 123 | + ), |
| 124 | + PlanSummaryPayloadFixture( |
| 125 | + test_id="with-duration", |
| 126 | + summary=PlanSummary( |
| 127 | + clone=0, update=1, unchanged=0, blocked=0, errors=0, duration_ms=120 |
| 128 | + ), |
| 129 | + expected_total=1, |
| 130 | + ), |
| 131 | +] |
| 132 | + |
| 133 | + |
| 134 | +@pytest.mark.parametrize( |
| 135 | + list(PlanSummaryPayloadFixture._fields), |
| 136 | + PLAN_SUMMARY_PAYLOAD_FIXTURES, |
| 137 | + ids=[fixture.test_id for fixture in PLAN_SUMMARY_PAYLOAD_FIXTURES], |
| 138 | +) |
| 139 | +def test_plan_summary_to_payload( |
| 140 | + test_id: str, |
| 141 | + summary: PlanSummary, |
| 142 | + expected_total: int, |
| 143 | +) -> None: |
| 144 | + """Validate PlanSummary total and serialization behaviour.""" |
| 145 | + payload = summary.to_payload() |
| 146 | + assert payload["total"] == expected_total |
| 147 | + assert payload["clone"] == summary.clone |
| 148 | + assert payload["update"] == summary.update |
| 149 | + assert payload["unchanged"] == summary.unchanged |
| 150 | + assert payload["blocked"] == summary.blocked |
| 151 | + assert payload["errors"] == summary.errors |
| 152 | + if summary.duration_ms is not None: |
| 153 | + assert payload["duration_ms"] == summary.duration_ms |
| 154 | + else: |
| 155 | + assert "duration_ms" not in payload |
| 156 | + |
| 157 | + |
| 158 | +def test_plan_result_grouping_and_json_output() -> None: |
| 159 | + """PlanResult should group entries and produce stable JSON.""" |
| 160 | + entries = [ |
| 161 | + PlanEntry( |
| 162 | + name="repo-a", |
| 163 | + path="/tmp/workspace-a/repo-a", |
| 164 | + workspace_root="~/workspace-a/", |
| 165 | + action=PlanAction.CLONE, |
| 166 | + ), |
| 167 | + PlanEntry( |
| 168 | + name="repo-b", |
| 169 | + path="/tmp/workspace-b/repo-b", |
| 170 | + workspace_root="~/workspace-b/", |
| 171 | + action=PlanAction.UPDATE, |
| 172 | + ), |
| 173 | + PlanEntry( |
| 174 | + name="repo-c", |
| 175 | + path="/tmp/workspace-a/repo-c", |
| 176 | + workspace_root="~/workspace-a/", |
| 177 | + action=PlanAction.UNCHANGED, |
| 178 | + ), |
| 179 | + ] |
| 180 | + summary = PlanSummary(clone=1, update=1, unchanged=1) |
| 181 | + result = PlanResult(entries=entries, summary=summary) |
| 182 | + |
| 183 | + mapping = result.to_workspace_mapping() |
| 184 | + assert set(mapping.keys()) == {"~/workspace-a/", "~/workspace-b/"} |
| 185 | + assert {entry.name for entry in mapping["~/workspace-a/"]} == {"repo-a", "repo-c"} |
| 186 | + assert {entry.name for entry in mapping["~/workspace-b/"]} == {"repo-b"} |
| 187 | + |
| 188 | + json_object = result.to_json_object() |
| 189 | + assert json_object["summary"]["total"] == 3 |
| 190 | + workspaces = { |
| 191 | + workspace["path"]: workspace for workspace in json_object["workspaces"] |
| 192 | + } |
| 193 | + assert set(workspaces) == {"~/workspace-a/", "~/workspace-b/"} |
| 194 | + assert len(workspaces["~/workspace-a/"]["operations"]) == 2 |
| 195 | + assert workspaces["~/workspace-b/"]["operations"][0]["name"] == "repo-b" |
| 196 | + |
| 197 | + |
| 198 | +def test_output_formatter_json_mode_finalises_buffer() -> None: |
| 199 | + """OutputFormatter should flush buffered JSON payloads on finalize.""" |
| 200 | + entry = PlanEntry( |
| 201 | + name="repo-buffer", |
| 202 | + path="/tmp/repo-buffer", |
| 203 | + workspace_root="~/code/", |
| 204 | + action=PlanAction.CLONE, |
| 205 | + ) |
| 206 | + formatter = OutputFormatter(mode=OutputMode.JSON) |
| 207 | + captured = io.StringIO() |
| 208 | + with redirect_stdout(captured): |
| 209 | + formatter.emit(entry) |
| 210 | + formatter.emit(PlanSummary(clone=1)) |
| 211 | + formatter.finalize() |
| 212 | + |
| 213 | + output = json.loads(captured.getvalue()) |
| 214 | + assert len(output) == 2 |
| 215 | + assert output[0]["name"] == "repo-buffer" |
| 216 | + assert output[1]["type"] == "summary" |
| 217 | + |
| 218 | + |
| 219 | +def test_plan_progress_printer_updates_and_finishes() -> None: |
| 220 | + """Progress printer should render a single line and terminate cleanly.""" |
| 221 | + colors = Colors(mode=ColorMode.NEVER) |
| 222 | + printer = PlanProgressPrinter(total=3, colors=colors, enabled=True) |
| 223 | + buffer = io.StringIO() |
| 224 | + printer._stream = buffer |
| 225 | + |
| 226 | + summary = PlanSummary(clone=1) |
| 227 | + printer.update(summary, processed=1) |
| 228 | + assert "Progress: 1/3" in buffer.getvalue() |
| 229 | + |
| 230 | + printer.finish() |
| 231 | + assert buffer.getvalue().endswith("\n") |
0 commit comments