FabledGrimoire/tests/test_forge_webhook.py

"""Forge push webhook — signature gate, payload parsing, and the
end-to-end drift flag on real Postgres.

The webhook is the seam that makes verification scale: a push names exactly
which files moved, so only the records that point at them get flagged. The
properties pinned here:

  - No secret configured → the endpoint does not exist (404); a bad
    signature → 401. Both BEFORE any payload parsing.
  - Matching is O(bindings + snippets-in-project + changed files) and goes
    through repo bindings — recorded location repos are free-form names and
    cannot address a forge.
  - A replayed delivery (same head commit) flags nothing new; re-verifying
    clears the flag by construction.
"""
import hashlib
import hmac

import pytest
import pytest_asyncio

from grimoire.routes.webhooks import delivered_signature, push_facts, signature_ok
from grimoire.services.snippets import _path_touches
from tests.helpers import ensure_user

SECRET = "wh-secret"
HEAD = "e" * 40


# --- unit: the signature gate ------------------------------------------------

def _sign(body: bytes, secret: str = SECRET) -> str:
    return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()


def test_signature_gate():
    body = b'{"x": 1}'
    assert signature_ok(SECRET, body, _sign(body)) is True
    assert signature_ok(SECRET, body, _sign(body).upper()) is True  # hex case
    assert signature_ok(SECRET, body, _sign(body, "wrong")) is False
    assert signature_ok(SECRET, body, "") is False
    assert signature_ok("", body, _sign(body)) is False


def test_delivered_signature_reads_both_forges_headers():
    """GitHub signs the same HMAC but ships it as
    X-Hub-Signature-256: sha256=<hex> — the whole webhook payload mapping is
    this header, so pin it."""
    hexsig = _sign(b"{}")
    assert delivered_signature({"X-Gitea-Signature": hexsig}) == hexsig
    assert delivered_signature({"X-Hub-Signature-256": f"sha256={hexsig}"}) == hexsig
    # Gitea's header wins when both appear; absence reads as empty (→ 401).
    assert delivered_signature({}) == ""
    # The stripped GitHub form still passes the gate end to end.
    assert signature_ok(
        SECRET, b'{"x": 1}',
        delivered_signature({"X-Hub-Signature-256": "sha256=" + _sign(b'{"x": 1}')}),
    )


def test_push_facts_reads_a_github_shaped_payload():
    """GitHub's push payload carries the same fields push_facts consumes —
    asserted against a real-shaped sample so a rename on either side of the
    mapping breaks a test instead of silently flagging nothing."""
    payload = {
        "ref": "refs/heads/main",
        "after": HEAD,
        "repository": {
            "full_name": "alice/widget",
            "clone_url": "https://github.com/alice/widget.git",
            "html_url": "https://github.com/alice/widget",
        },
        "commits": [
            {"id": "a" * 40, "added": [], "modified": ["src/x.py"], "removed": []},
        ],
        "head_commit": {"id": HEAD},
    }
    raw, changed, removed, head = push_facts(payload)
    assert raw == "https://github.com/alice/widget.git"
    assert changed == ["src/x.py"]
    assert removed == []
    assert head == HEAD


# --- unit: payload parsing ---------------------------------------------------

def test_push_facts_collects_and_dedups_paths():
    payload = {
        "after": HEAD,
        "repository": {"clone_url": "https://git.example.com/alice/widget.git"},
        "commits": [
            {"added": ["a.py"], "modified": ["b.py"], "removed": []},
            {"added": [], "modified": ["b.py", "c/d.py"], "removed": ["gone.py"]},
        ],
    }
    raw, changed, removed, head = push_facts(payload)
    assert raw.endswith("alice/widget.git")
    assert changed == ["a.py", "b.py", "c/d.py"]
    assert removed == ["gone.py"]
    assert head == HEAD


def test_push_facts_tolerates_an_empty_payload():
    assert push_facts({}) == ("", [], [], "")


def test_path_touches_uses_the_location_semantics():
    assert _path_touches("src/x.py", "src/x.py")
    assert _path_touches("src", "src/lib/x.py")       # recorded dir, file below
    assert not _path_touches("src/x.py", "src/x_test.py")
    assert not _path_touches("src/lib", "src/library/x.py")  # no prefix bleed
    assert not _path_touches("", "src/x.py")


def test_route_is_registered_and_unauthenticated_by_design():
    from grimoire.app import create_app
    from grimoire.routes import webhooks as wh

    assert callable(wh.forge_push)
    rules = {r.rule for r in create_app().url_map.iter_rules()}
    assert "/api/webhooks/forge" in rules


# --- integration: the flag lands and clears on real Postgres -----------------


@pytest_asyncio.fixture
async def seeded(_dispose_engine):
    """User + project + binding + two verified snippets + one unverified."""
    from grimoire.models import async_session
    from grimoire.models.project import Project
    from grimoire.services import snippets as svc
    from grimoire.services.repo_bindings import set_binding

    async with async_session() as s:
        user = await ensure_user(s, "webhook_itest")
        project = Project(user_id=user.id, title="Widget")
        s.add(project)
        await s.flush()
        uid, pid = user.id, project.id
        await s.commit()

    await set_binding(uid, "https://git.example.com/alice/widget.git", pid)

    async def _snippet(name, path, code, verify=True):
        note = await svc.create_snippet(
            uid, name=name, code=code, language="python",
            repo="Widget", path=path, symbol=name, project_id=pid,
        )
        if verify:
            await svc.record_verification(uid, note.id, status="ok")
        return note.id

    return {
        "uid": uid,
        "hit": await _snippet("wh_hit", "src/x.py", "def wh_hit():\n    return 1\n"),
        "miss": await _snippet("wh_miss", "src/other.py", "def wh_miss():\n    return 2\n"),
        "gone": await _snippet("wh_gone", "src/gone.py", "def wh_gone():\n    return 3\n"),
        "unchecked": await _snippet(
            "wh_unchecked", "src/x.py", "def wh_unchecked():\n    return 4\n",
            verify=False,
        ),
    }


@pytest.mark.integration
async def test_push_flags_matched_verdicts_and_replay_is_quiet(seeded):
    from grimoire.services import snippets as svc

    flagged = await svc.invalidate_for_push(
        "git.example.com/alice/widget",
        changed=["src/x.py"], removed=["src/gone.py"], commit_sha=HEAD,
    )
    # wh_hit (modified) + wh_gone (removed). wh_miss untouched; wh_unchecked
    # carries no verdict and is skipped by design.
    assert flagged == 2

    uid = seeded["uid"]
    hit = svc.snippet_to_dict(await svc.get_snippet(uid, seeded["hit"]))
    assert hit["verification"]["needs_attention"] is True
    assert hit["verification"]["invalidated_by"]["commit_sha"] == HEAD
    assert hit["verification"]["invalidated_by"]["removed"] is False

    gone = svc.snippet_to_dict(await svc.get_snippet(uid, seeded["gone"]))
    assert gone["verification"]["invalidated_by"]["removed"] is True

    miss = svc.snippet_to_dict(await svc.get_snippet(uid, seeded["miss"]))
    assert miss["verification"]["needs_attention"] is False

    unchecked = svc.snippet_to_dict(await svc.get_snippet(uid, seeded["unchecked"]))
    assert unchecked["verification"]["status"] == "unverified"

    # The attention listing — the operator's single entry point — now shows
    # exactly the flagged pair, through the SQL dialect.
    items, total = await svc.list_snippets(uid, verification="attention")
    ids = {i["id"] for i in items}
    assert {seeded["hit"], seeded["gone"]} <= ids
    assert seeded["miss"] not in ids

    # Replayed delivery: same head commit flags nothing new.
    again = await svc.invalidate_for_push(
        "git.example.com/alice/widget",
        changed=["src/x.py"], removed=["src/gone.py"], commit_sha=HEAD,
    )
    assert again == 0


@pytest.mark.integration
async def test_reverifying_clears_the_flag(seeded):
    from grimoire.services import snippets as svc

    await svc.invalidate_for_push(
        "git.example.com/alice/widget", changed=["src/x.py"], removed=[],
        commit_sha=HEAD,
    )
    uid = seeded["uid"]
    await svc.record_verification(
        uid, seeded["hit"], status="ok", detail="rechecked after push",
        commit_sha=HEAD,
    )
    hit = svc.snippet_to_dict(await svc.get_snippet(uid, seeded["hit"]))
    assert hit["verification"]["needs_attention"] is False
    assert hit["verification"]["invalidated_by"] is None
    items, _ = await svc.list_snippets(uid, verification="attention")
    assert seeded["hit"] not in {i["id"] for i in items}


@pytest.mark.integration
async def test_unbound_repo_flags_nothing(seeded):
    from grimoire.services import snippets as svc

    flagged = await svc.invalidate_for_push(
        "github.com/somebody/else", changed=["src/x.py"], removed=[],
        commit_sha=HEAD,
    )
    assert flagged == 0