coulomb-social/tests/test_content_sync.py

import hashlib
import hmac
import json
from unittest.mock import patch

import pytest
from django.urls import reverse

from coulomb_social.apps.members.models import Member
from coulomb_social.apps.spaces.content import (
    clear_content_cache,
    clear_content_cache_for_repo,
    forgejo_edit_url,
)
from coulomb_social.apps.spaces.forgejo import FetchedFile
from coulomb_social.apps.spaces.models import Space


@pytest.mark.django_db
def test_forgejo_edit_url():
    space = Space(
        tenant_id="t",
        slug="demo",
        title="D",
        forgejo_owner="coulomb",
        forgejo_repo="coulomb-social",
        default_branch="main",
        content_root="docs/space-fixtures/demo/pages",
    )
    url = forgejo_edit_url(space, "index")
    assert url.endswith(
        "/coulomb/coulomb-social/_edit/main/docs/space-fixtures/demo/pages/index.md"
    )


@pytest.mark.django_db
def test_refresh_clears_cache_and_refetches(client, settings):
    clear_content_cache()
    settings.DEBUG = True
    settings.OIDC_ENABLED = False
    client.post(
        reverse("identity:dev_login"),
        {
            "subject": "writer",
            "issuer": "https://local.dev/issuer",
            "name": "Writer",
            "tenant": "tenant:coulomb",
        },
    )
    member = Member.objects.get(subject="writer")
    Space.objects.create(
        tenant_id="tenant:coulomb",
        slug="demo",
        title="Demo",
        created_by=member,
        forgejo_owner="coulomb",
        forgejo_repo="coulomb-social",
        content_root="docs/space-fixtures/demo/pages",
    )
    calls = {"n": 0}

    def fake_fetch(**kwargs):
        calls["n"] += 1
        return FetchedFile(
            path=kwargs["path"],
            text=f"# Version {calls['n']}\n\nbody\n",
            source="forgejo-raw",
        )

    with patch(
        "coulomb_social.apps.spaces.content.fetch_raw_file", side_effect=fake_fetch
    ):
        r1 = client.get(reverse("spaces:detail", kwargs={"slug": "demo"}))
        assert r1.status_code == 200
        assert b"Version 1" in r1.content
        # cached — still v1
        r2 = client.get(reverse("spaces:detail", kwargs={"slug": "demo"}))
        assert b"Version 1" in r2.content
        assert calls["n"] == 1
        # refresh
        r3 = client.post(reverse("spaces:refresh", kwargs={"slug": "demo"}))
        assert r3.status_code == 302
        r4 = client.get(reverse("spaces:detail", kwargs={"slug": "demo"}))
        assert b"Version 2" in r4.content
        assert calls["n"] == 2


@pytest.mark.django_db
def test_webhook_hmac_invalidates_repo_cache(client, settings):
    clear_content_cache()
    settings.FORGEJO_WEBHOOK_SECRET = "test-webhook-secret-at-least-16"
    Space.objects.create(
        tenant_id="tenant:coulomb",
        slug="demo",
        title="Demo",
        forgejo_owner="coulomb",
        forgejo_repo="coulomb-social",
        content_root="pages",
    )
    # prime cache
    from coulomb_social.apps.spaces import content as content_mod

    key = (
        "coulomb",
        "coulomb-social",
        "main",
        "pages/index.md",
        settings.FORGEJO_BASE_URL,
    )
    content_mod._FETCH_CACHE[key] = FetchedFile(
        path="pages/index.md", text="# old", source="forgejo-raw"
    )

    body = json.dumps(
        {"repository": {"full_name": "coulomb/coulomb-social", "name": "coulomb-social"}}
    ).encode()
    sig = hmac.new(
        settings.FORGEJO_WEBHOOK_SECRET.encode(), body, hashlib.sha256
    ).hexdigest()
    r = client.post(
        reverse("spaces:forgejo_webhook"),
        data=body,
        content_type="application/json",
        headers={"X-Gitea-Signature": sig},
    )
    assert r.status_code == 200
    data = r.json()
    assert data["ok"] is True
    assert data["cache_entries_cleared"] >= 1
    assert data["spaces_matched"] == 1
    assert key not in content_mod._FETCH_CACHE


@pytest.mark.django_db
def test_webhook_rejects_bad_secret(client, settings):
    settings.FORGEJO_WEBHOOK_SECRET = "test-webhook-secret-at-least-16"
    r = client.post(
        reverse("spaces:forgejo_webhook"),
        data=b"{}",
        content_type="application/json",
        headers={"X-Coulomb-Webhook-Secret": "wrong"},
    )
    assert r.status_code == 401