stackchain-dashboard/tests/test_issue_attachments.py

import asyncio
import base64

import httpx
import pytest

from src import gitea_proxy, main


PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"mobile screenshot"


@pytest.fixture(autouse=True)
def clear_attachment_idempotency():
    main._authored_action_operations.clear()
    main._idempotency_ledger.clear()
    yield
    main._authored_action_operations.clear()
    main._idempotency_ledger.clear()


@pytest.mark.anyio
async def test_attachment_endpoint_uploads_valid_screenshot_to_assigned_issue(monkeypatch):
    calls = []

    async def upload(repository, number, filename, content_type, content):
        calls.append((repository, number, filename, content_type, content))
        return {
            "name": "checkout.png",
            "url": "https://forge.example/attachments/checkout.png",
            "size": len(content),
        }

    monkeypatch.setattr(
        main.gitea_proxy, "upload_assigned_issue_attachment", upload, raising=False
    )
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json={
                "filename": "checkout.png",
                "content_type": "image/png",
                "data": base64.b64encode(PNG_BYTES).decode("ascii"),
            },
        )

    assert response.status_code == 201
    assert response.headers["cache-control"] == "no-store"
    assert response.json() == {
        "name": "checkout.png",
        "url": "https://forge.example/attachments/checkout.png",
        "size": len(PNG_BYTES),
        "markdown": "![checkout.png](<https://forge.example/attachments/checkout.png>)",
    }
    assert calls == [
        ("stackchain/api", 17, "checkout.png", "image/png", PNG_BYTES)
    ]


@pytest.mark.anyio
async def test_attachment_endpoint_accepts_binary_multipart_without_base64_expansion(monkeypatch):
    calls = []

    async def upload(repository, number, filename, content_type, content):
        calls.append((repository, number, filename, content_type, content))
        return {
            "name": filename,
            "url": "https://forge.example/attachments/checkout.png",
            "size": len(content),
        }

    monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            files={"file": ("checkout.png", PNG_BYTES, "image/png")},
        )

    assert response.status_code == 201
    assert calls == [
        ("stackchain/api", 17, "checkout.png", "image/png", PNG_BYTES)
    ]


@pytest.mark.anyio
async def test_attachment_endpoint_replays_confirmed_upload_for_same_key(monkeypatch):
    calls = []

    async def upload(repository, number, filename, content_type, content):
        calls.append((repository, number, filename, content_type, content))
        return {
            "name": filename,
            "url": "https://forge.example/attachments/checkout.png",
            "size": len(content),
        }

    monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
    payload = {
        "filename": "checkout.png",
        "content_type": "image/png",
        "data": base64.b64encode(PNG_BYTES).decode("ascii"),
    }
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        first = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json=payload,
            headers={"Idempotency-Key": "attachment-retry-471"},
        )
        replay = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json=payload,
            headers={"Idempotency-Key": "attachment-retry-471"},
        )

    assert first.status_code == replay.status_code == 201
    assert replay.json() == first.json()
    assert len(calls) == 1


@pytest.mark.anyio
async def test_attachment_endpoint_coalesces_concurrent_upload_retries(monkeypatch):
    calls = 0

    async def upload(_repository, _number, filename, _content_type, content):
        nonlocal calls
        calls += 1
        await asyncio.sleep(0.05)
        return {
            "name": filename,
            "url": "https://forge.example/attachments/checkout.png",
            "size": len(content),
        }

    monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
    payload = {
        "filename": "checkout.png",
        "content_type": "image/png",
        "data": base64.b64encode(PNG_BYTES).decode("ascii"),
    }
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        first, retry = await asyncio.gather(*[
            client.post(
                "/api/v1/repos/stackchain/api/issues/17/attachments",
                json=payload,
                headers={"Idempotency-Key": "attachment-concurrent-471"},
            )
            for _ in range(2)
        ])

    assert first.status_code == retry.status_code == 201
    assert first.json() == retry.json()
    assert calls == 1


@pytest.mark.anyio
async def test_attachment_endpoint_rejects_changed_upload_for_used_key(monkeypatch):
    calls = 0

    async def upload(_repository, _number, filename, _content_type, content):
        nonlocal calls
        calls += 1
        return {
            "name": filename,
            "url": "https://forge.example/attachments/checkout.png",
            "size": len(content),
        }

    monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        accepted = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json={
                "filename": "checkout.png",
                "content_type": "image/png",
                "data": base64.b64encode(PNG_BYTES).decode("ascii"),
            },
            headers={"Idempotency-Key": "attachment-conflict-471"},
        )
        conflict = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json={
                "filename": "checkout.png",
                "content_type": "image/png",
                "data": base64.b64encode(PNG_BYTES + b" changed").decode("ascii"),
            },
            headers={"Idempotency-Key": "attachment-conflict-471"},
        )

    assert accepted.status_code == 201
    assert conflict.status_code == 409
    assert calls == 1


@pytest.mark.anyio
async def test_attachment_endpoint_admits_a_normal_phone_screenshot(monkeypatch):
    screenshot = b"\x89PNG\r\n\x1a\n" + (b"x" * (100 * 1024))

    async def upload(_repository, _number, filename, _content_type, content):
        return {
            "name": filename,
            "url": "https://forge.example/attachments/screen.png",
            "size": len(content),
        }

    monkeypatch.setattr(
        main.gitea_proxy, "upload_assigned_issue_attachment", upload, raising=False
    )
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json={
                "filename": "screen.png",
                "content_type": "image/png",
                "data": base64.b64encode(screenshot).decode("ascii"),
            },
        )

    assert response.status_code == 201
    assert response.json()["size"] == len(screenshot)


@pytest.mark.anyio
async def test_attachment_endpoint_rejects_spoofed_image_before_upstream(monkeypatch):
    called = False

    async def upload(*_args):
        nonlocal called
        called = True

    monkeypatch.setattr(
        main.gitea_proxy, "upload_assigned_issue_attachment", upload, raising=False
    )
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json={
                "filename": "not-really.png",
                "content_type": "image/png",
                "data": base64.b64encode(b"<script>alert(1)</script>").decode("ascii"),
            },
        )

    assert response.status_code == 422
    assert response.headers["cache-control"] == "no-store"
    assert called is False


@pytest.mark.anyio
async def test_attachment_endpoint_rejects_image_over_two_megabytes_before_upstream(monkeypatch):
    called = False

    async def upload(*_args):
        nonlocal called
        called = True
        return {"name": "large.png", "url": "https://forge.example/a", "size": 1}

    monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
    oversized = b"\x89PNG\r\n\x1a\n" + b"x" * (2 * 1024 * 1024 - 7)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json={
                "filename": "large.png",
                "content_type": "image/png",
                "data": base64.b64encode(oversized).decode("ascii"),
            },
        )

    assert len(oversized) == 2 * 1024 * 1024 + 1
    assert response.status_code == 422
    assert called is False


@pytest.mark.anyio
async def test_gitea_attachment_revalidates_assignment_and_sends_multipart():
    requests = []

    async def handler(request):
        requests.append(request)
        if request.url.path == "/api/v1/user":
            return httpx.Response(200, json={"login": "timmy"})
        if request.method == "GET":
            return httpx.Response(200, json={
                "state": "open", "pull_request": None,
                "assignees": [{"login": "timmy"}],
            })
        return httpx.Response(201, json={
            "id": 9,
            "name": "checkout.png",
            "size": len(PNG_BYTES),
            "browser_download_url": "https://forge.example/attachments/checkout.png",
        })

    gitea_proxy.start_client(transport=httpx.MockTransport(handler))
    try:
        result = await gitea_proxy.upload_assigned_issue_attachment(
            "stackchain/api", 17, "checkout.png", "image/png", PNG_BYTES
        )
    finally:
        await gitea_proxy.stop_client()

    assert [(request.method, request.url.path) for request in requests] == [
        ("GET", "/api/v1/user"),
        ("GET", "/api/v1/repos/stackchain/api/issues/17"),
        ("POST", "/api/v1/repos/stackchain/api/issues/17/assets"),
    ]
    upload = requests[-1]
    assert upload.url.params["name"] == "checkout.png"
    assert "multipart/form-data" in upload.headers["content-type"]
    assert b'name="attachment"; filename="checkout.png"' in upload.content
    assert PNG_BYTES in upload.content
    assert result == {
        "name": "checkout.png",
        "url": "https://forge.example/attachments/checkout.png",
        "size": len(PNG_BYTES),
    }


@pytest.mark.anyio
async def test_attachment_markdown_escapes_untrusted_confirmed_filename(monkeypatch):
    async def upload(*_args):
        return {
            "name": "screen](not-an-image).png",
            "url": "https://forge.example/attachments/a.png",
            "size": len(PNG_BYTES),
        }

    monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json={
                "filename": "screen.png",
                "content_type": "image/png",
                "data": base64.b64encode(PNG_BYTES).decode("ascii"),
            },
        )

    assert response.status_code == 201
    assert response.json()["markdown"] == (
        "![screen\\](not-an-image).png](<https://forge.example/attachments/a.png>)"
    )


@pytest.mark.anyio
async def test_attachment_endpoint_hides_an_issue_that_is_no_longer_assigned(monkeypatch):
    async def upload(*_args):
        raise gitea_proxy.IssueNotAvailableError("not assigned")

    monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json={
                "filename": "screen.png",
                "content_type": "image/png",
                "data": base64.b64encode(PNG_BYTES).decode("ascii"),
            },
        )

    assert response.status_code == 404
    assert response.headers["cache-control"] == "no-store"


@pytest.mark.anyio
async def test_attachment_endpoint_rejects_unsafe_or_mismatched_filename(monkeypatch):
    called = False

    async def upload(*_args):
        nonlocal called
        called = True

    monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        traversal = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json={
                "filename": "../payload.png",
                "content_type": "image/png",
                "data": base64.b64encode(PNG_BYTES).decode("ascii"),
            },
        )
        mismatch = await client.post(
            "/api/v1/repos/stackchain/api/issues/17/attachments",
            json={
                "filename": "payload.html",
                "content_type": "image/png",
                "data": base64.b64encode(PNG_BYTES).decode("ascii"),
            },
        )

    assert [traversal.status_code, mismatch.status_code] == [422, 422]
    assert called is False