issue-core/tests/test_mapping.py

"""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()