ThinkStorm/tests/test_gitea_claimant_publishing.py

import asyncio

import pytest
from fastapi import HTTPException

from thinkstorm import database
from thinkstorm import auth
from thinkstorm.api import ideas as ideas_api
from thinkstorm.models import User, UserRole
from thinkstorm.services import gitea as gitea_module


@pytest.fixture
def isolated_db(tmp_path, monkeypatch):
    monkeypatch.setattr(database, "DB_PATH", str(tmp_path / "claimant-gitea.db"))
    database.init_db()
    return tmp_path


def _insert_idea(idea_id, claimed_by=None):
    now = database.get_utc_now()
    with database.get_db() as conn:
        conn.execute(
            """
            INSERT INTO ideas
                (id, original_text, submitted_at, title, summary, lifecycle_state,
                 processing_state, claimed_by, claimed_at, created_at, updated_at)
            VALUES (?, 'Original idea', ?, 'Claimant Publishing', 'Summary', ?,
                    'IDLE', ?, ?, ?, ?)
            """,
            (
                idea_id,
                now,
                "CLAIMED" if claimed_by else "AVAILABLE",
                claimed_by,
                now if claimed_by else None,
                now,
                now,
            ),
        )


def _user(username, role=UserRole.USER):
    return User(id=10, username=username, password_hash="", role=role)


def test_unclaimed_idea_cannot_be_published(isolated_db):
    _insert_idea("TS-PUB1")

    with pytest.raises(HTTPException) as exc:
        asyncio.run(
            ideas_api.sync_idea_to_gitea(
                "TS-PUB1", current_user=_user("researcher")
            )
        )

    assert exc.value.status_code == 409


def test_claimant_must_have_verified_gitea_identity(isolated_db):
    _insert_idea("TS-PUB2", claimed_by="researcher")

    with pytest.raises(HTTPException) as exc:
        asyncio.run(
            ideas_api.sync_idea_to_gitea(
                "TS-PUB2", current_user=_user("researcher")
            )
        )

    assert exc.value.status_code == 403
    assert "sign in with Gitea" in exc.value.detail


def test_only_claimant_can_publish(isolated_db):
    _insert_idea("TS-PUB3", claimed_by="researcher")
    with database.get_db() as conn:
        conn.execute(
            "UPDATE users SET gitea_id = 1234 WHERE username = 'researcher'"
        )

    with pytest.raises(HTTPException) as exc:
        asyncio.run(
            ideas_api.sync_idea_to_gitea(
                "TS-PUB3", current_user=_user("someone-else")
            )
        )

    assert exc.value.status_code == 403


def test_claimant_publish_is_idempotent_and_grants_write_access(
    isolated_db, monkeypatch
):
    _insert_idea("TS-PUB4", claimed_by="researcher")
    with database.get_db() as conn:
        conn.execute(
            "UPDATE users SET gitea_id = 1234 WHERE username = 'researcher'"
        )

    calls = []

    async def fake_publish(**kwargs):
        calls.append(kwargs)
        return {
            "repo_name": "thinkstorm/ts-pub4",
            "repo_url": "https://git.example/thinkstorm/ts-pub4",
            "clone_url": "https://git.example/thinkstorm/ts-pub4.git",
            "ssh_url": "ssh://[email protected]/thinkstorm/ts-pub4.git",
            "local_path": str(isolated_db / "TS-PUB4"),
            "published": True,
        }

    monkeypatch.setattr(
        ideas_api.gitea_svc, "persist_idea_dossier_repo", fake_publish
    )

    first = asyncio.run(
        ideas_api.sync_idea_to_gitea(
            "TS-PUB4", current_user=_user("researcher")
        )
    )
    second = asyncio.run(
        ideas_api.sync_idea_to_gitea(
            "TS-PUB4", current_user=_user("researcher")
        )
    )

    assert first["gitea_repo_name"] == "thinkstorm/ts-pub4"
    assert second["gitea_repo_url"].endswith("/ts-pub4")
    assert all(call["publish_remote"] is True for call in calls)
    assert all(call["collaborator_username"] == "researcher" for call in calls)

    with database.get_db() as conn:
        idea = conn.execute(
            "SELECT gitea_repo_name, gitea_repo_url FROM ideas WHERE id = 'TS-PUB4'"
        ).fetchone()
        resources = conn.execute(
            """
            SELECT COUNT(*) AS count FROM external_resources
            WHERE idea_id = 'TS-PUB4' AND resource_type = 'GITEA_DOSSIER'
            """
        ).fetchone()["count"]

    assert idea["gitea_repo_name"] == "thinkstorm/ts-pub4"
    assert resources == 1


def test_oauth_links_existing_local_account_to_gitea(isolated_db):
    linked_user = auth.get_or_create_gitea_user(
        {"login": "researcher", "id": 9876, "is_admin": False}
    )

    with database.get_db() as conn:
        row = conn.execute(
            "SELECT gitea_id FROM users WHERE username = 'researcher'"
        ).fetchone()

    assert linked_user.username == "researcher"
    assert row["gitea_id"] == 9876


def test_local_only_dossier_persistence_never_contacts_gitea(
    isolated_db, monkeypatch
):
    artifact_root = isolated_db / "artifacts"
    monkeypatch.setattr(gitea_module, "ARTIFACTS_DIR", artifact_root)
    adapter = gitea_module.GiteaAdapter(api_token="configured-token")

    def unexpected_remote_call(*_args, **_kwargs):
        raise AssertionError("local persistence attempted a remote Gitea call")

    monkeypatch.setattr(adapter, "_ensure_repo", unexpected_remote_call)

    result = asyncio.run(
        adapter.persist_idea_dossier_repo(
            idea_id="TS-LOCAL",
            title="Local only",
            summary="Summary",
            original_text="Original",
            categories=[],
            tags=[],
            lifecycle_state="AVAILABLE",
            research_docs={"analysis.md": "Research"},
            outputs={},
            provenance_runs=[],
            publish_remote=False,
        )
    )

    assert result["published"] is False
    assert (artifact_root / "TS-LOCAL" / "README.md").exists()