feat(user-sync): add interactive preview and resolution

Author: raederan

src/osw/tools/user_sync/interactive.py

@@ -0,0 +1,199 @@
+"""Interactive preview and conflict resolution for the sync.
+
+Rendering is pure (returns strings). All terminal IO goes through ``Prompter``,
+which wraps ``input`` / ``print`` so tests can script answers and capture output.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Callable, List, Optional, Sequence, Set
+
+from .mapping import ProposedUser
+from .reconcile import (
+    CONFLICT,
+    GAP_FILL,
+    NEW,
+    RECONCILED_FIELDS,
+    UNCHANGED,
+    ReconcilePlan,
+    UserChange,
+    proposed_fields,
+)
+
+_MARKERS = {NEW: "+", GAP_FILL: "~", CONFLICT: "!", UNCHANGED: "="}
+
+
+def _is_empty(value: Any) -> bool:
+    if value is None or value == "":
+        return True
+    return isinstance(value, (list, set, tuple)) and len(value) == 0
+
+
+def nonempty_fields(proposed: ProposedUser) -> Set[str]:
+    """Reconciled field names that carry a value on the proposed user."""
+    pf = proposed_fields(proposed)
+    return {name for name in RECONCILED_FIELDS if not _is_empty(pf[name])}
+
+
+def render_preview(plan: ReconcilePlan) -> List[str]:
+    """Human-readable preview: a header line plus one row per user."""
+    counts = plan.counts()
+    lines = [
+        f"Users: {len(plan.changes)} | "
+        f"new={counts[NEW]} gap_fill={counts[GAP_FILL]} "
+        f"conflict={counts[CONFLICT]} unchanged={counts[UNCHANGED]}"
+    ]
+    for change in plan.changes:
+        marker = _MARKERS[change.category]
+        detail = ""
+        if change.diffs:
+            detail = " (" + ", ".join(d.name for d in change.diffs) + ")"
+        placeholder = " [placeholder name]" if change.proposed.placeholder_name else ""
+        lines.append(
+            f"  {marker} {change.proposed.username} {change.category}{detail}{placeholder}"
+        )
+    return lines
+
+
+def render_conflict(change: UserChange) -> List[str]:
+    """Field-by-field display of one user's differences."""
+    lines = [f"User {change.proposed.username}:"]
+    for diff in change.diffs:
+        lines.append(
+            f"  {diff.name} [{diff.status}]: "
+            f"existing={diff.existing!r} -> proposed={diff.proposed!r}"
+        )
+    return lines
+
+
+class Prompter:
+    """Thin IO seam over input/print for testable prompting."""
+
+    def __init__(
+        self,
+        input_fn: Callable[[str], str] = input,
+        output_fn: Callable[[str], None] = print,
+    ):
+        self._input = input_fn
+        self._output = output_fn
+
+    def write(self, text: str = "") -> None:
+        self._output(text)
+
+    def ask(self, prompt: str, choices: Sequence[str]) -> str:
+        keys = "/".join(choices)
+        while True:
+            answer = self._input(f"{prompt} [{keys}]: ").strip().lower()
+            if answer in choices:
+                return answer
+            self.write(f"Please choose one of: {keys}")
+
+    def confirm(self, prompt: str, default: bool = False) -> bool:
+        suffix = "[Y/n]" if default else "[y/N]"
+        answer = self._input(f"{prompt} {suffix}: ").strip().lower()
+        if not answer:
+            return default
+        return answer in ("y", "yes")
+
+
+@dataclass
+class ResolvedChange:
+    """A user change with the decided action and fields to write."""
+
+    change: UserChange
+    action: str  # "create" | "update" | "skip"
+    apply_fields: Set[str] = field(default_factory=set)
+
+
+@dataclass
+class Resolution:
+    """The outcome of resolving a plan: what to write and whether to proceed."""
+
+    resolved: List[ResolvedChange] = field(default_factory=list)
+    accepted: bool = False
+
+    def creates(self) -> List[ResolvedChange]:
+        return [r for r in self.resolved if r.action == "create"]
+
+    def updates(self) -> List[ResolvedChange]:
+        return [r for r in self.resolved if r.action == "update"]
+
+
+def _review_user(change: UserChange, prompter: Prompter) -> Set[str]:
+    """Resolve one conflicting user's fields, returning accepted conflict fields."""
+    for line in render_conflict(change):
+        prompter.write(line)
+    conflict_diffs = [d for d in change.diffs if d.status == CONFLICT]
+    choice = prompter.ask(
+        "Take all new / Keep all existing / Field-by-field?", ("a", "k", "f")
+    )
+    if choice == "a":
+        return {d.name for d in conflict_diffs}
+    if choice == "k":
+        return set()
+    accepted: Set[str] = set()
+    for diff in conflict_diffs:
+        if prompter.ask(f"{diff.name}: keep or take new?", ("k", "n")) == "n":
+            accepted.add(diff.name)
+    return accepted
+
+
+def resolve_plan(
+    plan: ReconcilePlan,
+    prompter: Prompter,
+    dry_run: bool = False,
+    assume_yes: bool = False,
+    summary_lines: Optional[Sequence[str]] = None,
+) -> Resolution:
+    """Preview the plan and resolve conflicts into a write decision."""
+    for line in render_preview(plan):
+        prompter.write(line)
+    for line in summary_lines or []:
+        prompter.write(line)
+
+    conflict_mode: Optional[str] = None
+    if plan.conflicts and not dry_run:
+        if assume_yes:
+            conflict_mode = "keep"
+        else:
+            choice = prompter.ask(
+                "Conflicts found: Apply all / Keep existing / Review each / Abort?",
+                ("a", "k", "r", "x"),
+            )
+            if choice == "x":
+                return Resolution(resolved=[], accepted=False)
+            conflict_mode = {"a": "all", "k": "keep", "r": "review"}[choice]
+
+    resolved: List[ResolvedChange] = []
+    for change in plan.changes:
+        if change.category == NEW:
+            resolved.append(
+                ResolvedChange(change, "create", nonempty_fields(change.proposed))
+            )
+        elif change.category == UNCHANGED:
+            resolved.append(ResolvedChange(change, "skip"))
+        elif change.category == GAP_FILL:
+            resolved.append(
+                ResolvedChange(change, "update", {d.name for d in change.diffs})
+            )
+        else:  # CONFLICT
+            accepted = {d.name for d in change.diffs if d.status == GAP_FILL}
+            if conflict_mode == "all":
+                accepted |= {d.name for d in change.diffs if d.status == CONFLICT}
+            elif conflict_mode == "review":
+                accepted |= _review_user(change, prompter)
+            action = "update" if accepted else "skip"
+            resolved.append(ResolvedChange(change, action, accepted))
+
+    if dry_run:
+        accepted_run = False
+    elif assume_yes:
+        accepted_run = True
+    else:
+        n_create = sum(1 for r in resolved if r.action == "create")
+        n_update = sum(1 for r in resolved if r.action == "update")
+        accepted_run = prompter.confirm(
+            f"Proceed to create {n_create} and update {n_update} user items?"
+        )
+    return Resolution(resolved=resolved, accepted=accepted_run)

tests/tools/test_user_sync_interactive.py

@@ -0,0 +1,196 @@
+"""Unit tests for the interactive preview and conflict resolution."""
+
+import uuid
+
+from osw.tools.user_sync.interactive import (
+    Prompter,
+    nonempty_fields,
+    render_preview,
+    resolve_plan,
+)
+from osw.tools.user_sync.mapping import ProposedUser
+from osw.tools.user_sync.reconcile import (
+    CONFLICT,
+    GAP_FILL,
+    NEW,
+    UNCHANGED,
+    FieldDiff,
+    ReconcilePlan,
+    UserChange,
+)
+
+
+class FakeIO:
+    def __init__(self, answers):
+        self.answers = list(answers)
+        self.outputs = []
+
+    def input(self, prompt=""):
+        return self.answers.pop(0)
+
+    def output(self, text=""):
+        self.outputs.append(text)
+
+
+def _proposed(username="u", **kw):
+    base = dict(
+        uuid=uuid.uuid4(),
+        username=username,
+        first_name="F",
+        surname="S",
+        label="F S",
+        orcid=None,
+        emails=[],
+        websites=[],
+        organizations=[],
+    )
+    base.update(kw)
+    return ProposedUser(**base)
+
+
+def _change(category, username="u", diffs=None, placeholder=False):
+    proposed = _proposed(username=username)
+    proposed.placeholder_name = placeholder
+    return UserChange(
+        proposed=proposed,
+        existing=None if category == NEW else object(),
+        category=category,
+        diffs=diffs or [],
+    )
+
+
+def _prompter(answers):
+    io = FakeIO(answers)
+    return Prompter(io.input, io.output), io
+
+
+def test_render_preview():
+    plan = ReconcilePlan(changes=[_change(NEW, "alice", placeholder=True)])
+    lines = render_preview(plan)
+    assert lines[0].startswith("Users: 1")
+    assert "new=1" in lines[0]
+    assert "+ alice new" in lines[1]
+    assert "[placeholder name]" in lines[1]
+
+
+def test_nonempty_fields():
+    proposed = _proposed(orcid="https://orcid.org/x", emails=["e@x.org"])
+    assert nonempty_fields(proposed) == {
+        "label",
+        "first_name",
+        "surname",
+        "orcid",
+        "emails",
+    }
+
+
+def _mixed_plan():
+    return ReconcilePlan(
+        changes=[
+            _change(NEW, "new1"),
+            _change(GAP_FILL, "gap1", [FieldDiff("emails", set(), {"e"}, GAP_FILL)]),
+            _change(CONFLICT, "con1", [FieldDiff("surname", "Old", "New", CONFLICT)]),
+            _change(
+                CONFLICT,
+                "mix1",
+                [
+                    FieldDiff("emails", set(), {"e"}, GAP_FILL),
+                    FieldDiff("surname", "Old", "New", CONFLICT),
+                ],
+            ),
+            _change(UNCHANGED, "same1"),
+        ]
+    )
+
+
+def test_resolve_keep_existing():
+    prompter, _ = _prompter(["k", "y"])
+    res = resolve_plan(_mixed_plan(), prompter)
+    assert [r.change.proposed.username for r in res.creates()] == ["new1"]
+    updates = {r.change.proposed.username: r.apply_fields for r in res.updates()}
+    assert updates == {"gap1": {"emails"}, "mix1": {"emails"}}
+    skipped = {r.change.proposed.username for r in res.resolved if r.action == "skip"}
+    assert skipped == {"con1", "same1"}
+    assert res.accepted is True
+
+
+def test_resolve_apply_all():
+    prompter, _ = _prompter(["a", "y"])
+    res = resolve_plan(_mixed_plan(), prompter)
+    updates = {r.change.proposed.username: r.apply_fields for r in res.updates()}
+    assert updates["con1"] == {"surname"}
+    assert updates["mix1"] == {"emails", "surname"}
+
+
+def test_resolve_abort():
+    prompter, _ = _prompter(["x"])
+    res = resolve_plan(_mixed_plan(), prompter)
+    assert res.resolved == []
+    assert res.accepted is False
+
+
+def test_resolve_dry_run_makes_no_writes_and_no_prompts():
+    prompter, io = _prompter([])  # no answers available
+    res = resolve_plan(_mixed_plan(), prompter, dry_run=True)
+    assert res.accepted is False
+    assert io.answers == []  # nothing was asked
+
+
+def test_resolve_assume_yes_keeps_conflicts():
+    prompter, io = _prompter([])
+    res = resolve_plan(_mixed_plan(), prompter, assume_yes=True)
+    assert res.accepted is True
+    updates = {r.change.proposed.username: r.apply_fields for r in res.updates()}
+    assert updates == {"gap1": {"emails"}, "mix1": {"emails"}}
+    assert io.answers == []
+
+
+def test_resolve_review_field_by_field():
+    plan = ReconcilePlan(
+        changes=[
+            _change(
+                CONFLICT,
+                "con1",
+                [
+                    FieldDiff("surname", "Old", "New", CONFLICT),
+                    FieldDiff("label", "L1", "L2", CONFLICT),
+                ],
+            )
+        ]
+    )
+    prompter, _ = _prompter(["r", "f", "n", "k", "y"])
+    res = resolve_plan(plan, prompter)
+    assert res.updates()[0].apply_fields == {"surname"}
+    assert res.accepted is True
+
+
+def test_resolve_review_take_all_for_user():
+    plan = ReconcilePlan(
+        changes=[
+            _change(
+                CONFLICT,
+                "con1",
+                [
+                    FieldDiff("surname", "Old", "New", CONFLICT),
+                    FieldDiff("label", "L1", "L2", CONFLICT),
+                ],
+            )
+        ]
+    )
+    prompter, _ = _prompter(["r", "a", "y"])
+    res = resolve_plan(plan, prompter)
+    assert res.updates()[0].apply_fields == {"surname", "label"}
+
+
+def test_final_confirm_no():
+    plan = ReconcilePlan(changes=[_change(NEW, "new1")])
+    prompter, _ = _prompter(["n"])
+    res = resolve_plan(plan, prompter)
+    assert res.accepted is False
+    assert res.creates()[0].change.proposed.username == "new1"
+
+
+def test_prompter_reprompts_on_invalid():
+    prompter, io = _prompter(["z", "k"])
+    assert prompter.ask("q", ("k", "n")) == "k"
+    assert any("Please choose one of" in o for o in io.outputs)