tegwick
f88e9a28fe
Add MappingService (work_record_uuid ↔ external id), project/map CLI with outward push-status, optional TaskSpec work_record fields, boundary-sync policy, and packaging/capability framing cleanup. Mark WP-0005 finished.
129 lines
4.1 KiB
Python
129 lines
4.1 KiB
Python
"""Unit tests for MappingService and status mapping."""
|
|
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from issue_core.core.mapping import (
|
|
MappingService,
|
|
map_work_record_status_to_issue_state,
|
|
)
|
|
from issue_core.core.models import Issue, IssueState, Label
|
|
from issue_core.backends.local import LocalSQLiteBackend
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_status_mapping_table():
|
|
assert map_work_record_status_to_issue_state("todo") == "open"
|
|
assert map_work_record_status_to_issue_state("progress") == "in_progress"
|
|
assert map_work_record_status_to_issue_state("done") == "closed"
|
|
assert map_work_record_status_to_issue_state("cancel") == "closed"
|
|
assert map_work_record_status_to_issue_state("blocked") == "blocked"
|
|
with pytest.raises(ValueError):
|
|
map_work_record_status_to_issue_state("nope")
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_upsert_idempotent(tmp_path):
|
|
db = str(tmp_path / "mappings.db")
|
|
svc = MappingService()
|
|
svc.connect(db)
|
|
wr = str(uuid.uuid4())
|
|
m1 = svc.upsert(
|
|
work_record_uuid=wr,
|
|
work_record_id="ISSUE-WP-0005-T04",
|
|
work_record_kind="task",
|
|
backend="sqlite",
|
|
external_id="1",
|
|
direction="outward",
|
|
)
|
|
m2 = svc.upsert(
|
|
work_record_uuid=wr,
|
|
work_record_id="ISSUE-WP-0005-T04",
|
|
work_record_kind="task",
|
|
backend="sqlite",
|
|
external_id="1",
|
|
direction="outward",
|
|
)
|
|
assert m1.id == m2.id
|
|
assert m2.external_id == "1"
|
|
assert svc.get_active_by_canonical_id("ISSUE-WP-0005-T04").work_record_uuid == wr
|
|
svc.disconnect()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_unique_external_conflict(tmp_path):
|
|
db = str(tmp_path / "mappings.db")
|
|
svc = MappingService()
|
|
svc.connect(db)
|
|
a = str(uuid.uuid4())
|
|
b = str(uuid.uuid4())
|
|
svc.upsert(work_record_uuid=a, backend="sqlite", external_id="9")
|
|
with pytest.raises(ValueError, match="already mapped"):
|
|
svc.upsert(work_record_uuid=b, backend="sqlite", external_id="9")
|
|
svc.disconnect()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_detach(tmp_path):
|
|
db = str(tmp_path / "mappings.db")
|
|
svc = MappingService()
|
|
svc.connect(db)
|
|
wr = str(uuid.uuid4())
|
|
svc.upsert(work_record_uuid=wr, backend="gitea", external_id="42")
|
|
detached = svc.detach(work_record_uuid=wr, backend="gitea")
|
|
assert detached is not None
|
|
assert detached.status == "detached"
|
|
assert svc.get_active_by_uuid(wr) is None
|
|
# Can re-map after detach
|
|
again = svc.upsert(work_record_uuid=wr, backend="gitea", external_id="43")
|
|
assert again.status == "active"
|
|
assert again.external_id == "43"
|
|
svc.disconnect()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_project_flow_local_backend(tmp_path, monkeypatch):
|
|
"""End-to-end: create issue on local backend + mapping row."""
|
|
from datetime import datetime, timezone
|
|
from issue_core.core.mapping import MappingService
|
|
|
|
issues_db = str(tmp_path / "issues.db")
|
|
map_db = str(tmp_path / "mappings.db")
|
|
backend = LocalSQLiteBackend()
|
|
backend.connect({"db_path": issues_db})
|
|
wr = str(uuid.uuid4())
|
|
now = datetime.now(timezone.utc)
|
|
issue = backend.create_issue(
|
|
Issue(
|
|
id="",
|
|
number=0,
|
|
title="Projected task",
|
|
description="",
|
|
state=IssueState.OPEN,
|
|
created_at=now,
|
|
updated_at=now,
|
|
labels=[Label(name="work_record:ISSUE-WP-0005-T04")],
|
|
backend_type="local",
|
|
)
|
|
)
|
|
svc = MappingService()
|
|
svc.connect(map_db)
|
|
mapping = svc.upsert(
|
|
work_record_uuid=wr,
|
|
work_record_id="ISSUE-WP-0005-T04",
|
|
work_record_kind="task",
|
|
backend="sqlite",
|
|
external_id=issue.id or str(issue.number),
|
|
direction="outward",
|
|
)
|
|
# push-status simulation
|
|
target = map_work_record_status_to_issue_state("progress")
|
|
issue.state = IssueState.from_string(target)
|
|
updated = backend.update_issue(issue)
|
|
assert updated.state == IssueState.IN_PROGRESS
|
|
svc.mark_pushed(mapping.id)
|
|
assert svc.get_active_by_uuid(wr).last_pushed_at is not None
|
|
svc.disconnect()
|
|
backend.disconnect()
|