stackchain-dashboard/tests/test_update_reply_attachments.py

import io
import json
import subprocess
from pathlib import Path

import httpx
import pytest
from PIL import Image

from src import gitea_proxy, main
from tests.dashboard_bundle import dashboard


ROOT = Path(__file__).parents[1]
OUTBOX = ROOT / "frontend" / "authored-outbox.js"
SYNC = ROOT / "frontend" / "background-issue-sync.js"
_png = io.BytesIO()
Image.new("RGB", (2, 2), "red").save(_png, format="PNG")
PNG = _png.getvalue()


def run_node(script: str):
    return json.loads(subprocess.run(
        ["node", "-e", script], check=True, capture_output=True, text=True
    ).stdout)


@pytest.fixture(autouse=True)
def clear_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_notification_attachment_endpoint_resolves_exact_pull_server_side(monkeypatch):
    calls = []

    async def upload(thread_id, filename, content_type, content):
        calls.append((thread_id, filename, content_type, content))
        return {"name": filename, "url": "http://127.0.0.1:3000/a/proof.webp", "size": len(content)}

    monkeypatch.setattr(main.gitea_proxy, "upload_notification_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/notifications/527/attachments",
            files={"file": ("proof.png", PNG, "image/png")},
            headers={"Idempotency-Key": "reply-527:attachment"},
        )

    assert response.status_code == 201
    assert response.json()["markdown"] == "![proof.png](<http://127.0.0.1:3000/a/proof.webp>)"
    assert calls == [(527, "proof.png", "image/png", PNG)]
    assert main.request_body_limit("POST", "/api/v1/notifications/527/attachments") == 2 * 1024 * 1024 + 64 * 1024


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

    async def handler(request):
        requests.append(request)
        if request.method == "GET":
            return httpx.Response(200, json={
                "repository": {"full_name": "stackchain/web"},
                "subject": {
                    "type": "Pull",
                    "url": "http://127.0.0.1:3000/api/v1/repos/stackchain/web/pulls/31",
                },
            })
        return httpx.Response(201, json={
            "name": "proof.png", "size": len(PNG),
            "browser_download_url": "http://127.0.0.1:3000/a/proof.png",
        })

    gitea_proxy.start_client(transport=httpx.MockTransport(handler))
    try:
        result = await gitea_proxy.upload_notification_attachment(527, "proof.png", "image/png", PNG)
    finally:
        await gitea_proxy.stop_client()

    assert [(request.method, request.url.path) for request in requests] == [
        ("GET", "/api/v1/notifications/threads/527"),
        ("POST", "/api/v1/repos/stackchain/web/issues/31/assets"),
    ]
    assert result["url"] == "http://127.0.0.1:3000/a/proof.png"


@pytest.mark.anyio
async def test_update_reply_composer_offers_mobile_safe_removable_screenshot_preview():
    html = await dashboard()

    assert 'id="update-reply-attachment"' in html
    assert 'accept="image/png,image/jpeg,image/webp"' in html
    assert 'id="update-reply-attachment-preview"' in html
    assert 'id="remove-update-reply-attachment"' in html
    assert "const updateReplyAttachmentController = issueAttachment.mount({" in html
    assert "await updateReplyAttachmentController.serialize()" in html
    assert ".update-reply .issue-attachment-preview { width:100%; min-width:0; }" in html
    assert "@media (max-width:390px)" in html


def test_authored_outbox_durably_keeps_account_bound_update_screenshot_out_of_localstorage():
    script = f"""
const createOutbox=require({json.dumps(str(OUTBOX))});
const values=new Map();const mirrored=[];
const storage={{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}};
const outbox=createOutbox({{storage,getOwnerLogin:()=>'timmy',backgroundSync:{{
  reconcile:async items=>mirrored.push(items),requestSync:async()=>{{}},
}}}});
(async()=>{{const result=await outbox.enqueueDurably({{
  kind:'update-reply-read',notificationId:527,body:'',operationId:'reply-image',
  attachment:{{filename:'phone.png',contentType:'image/png',blob:new Blob(['private-bytes'],{{type:'image/png'}})}},
}});process.stdout.write(JSON.stringify({{
  result,local:outbox.list()[0],raw:values.get('stackchain.authored-outbox.v1'),
  durable:{{ownerLogin:mirrored[0][0].ownerLogin,text:await mirrored[0][0].attachment.blob.text()}},
}}));}})();
"""
    output = run_node(script)

    assert output["local"]["attachment"] == {
        "filename": "phone.png", "contentType": "image/png", "stored": True
    }
    assert output["durable"] == {"ownerLogin": "timmy", "text": "private-bytes"}
    assert "private-bytes" not in output["raw"]


def test_update_screenshot_is_not_admitted_without_indexeddb_durability():
    script = f"""
const createOutbox=require({json.dumps(str(OUTBOX))});
const values=new Map();const storage={{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}};
const outbox=createOutbox({{storage,getOwnerLogin:()=>'timmy'}});
(async()=>{{let error='';try{{await outbox.enqueueDurably({{
  kind:'update-reply',notificationId:527,body:'proof',operationId:'no-db',
  attachment:{{filename:'phone.png',contentType:'image/png',data:'private-bytes'}},
}});}}catch(caught){{error=caught.message;}}
process.stdout.write(JSON.stringify({{error,items:outbox.list(),raw:values.get('stackchain.authored-outbox.v1')}}));}})();
"""
    output = run_node(script)

    assert output["error"] == "Screenshot delivery needs IndexedDB. Your reply and screenshot are still here; retry."
    assert output["items"] == []
    assert "private-bytes" not in output["raw"]


def test_background_update_reply_screenshot_checkpoints_upload_then_reply_then_read():
    script = f"""
const createSync=require({json.dumps(str(SYNC))});
let item={{id:'reply-image',operationId:'reply-image',ownerLogin:'timmy',status:'queued',
  kind:'update-reply-read',notificationId:527,body:'',replyConfirmed:false,
  attachment:{{filename:'phone.webp',contentType:'image/webp',blob:new Blob(['pixels'],{{type:'image/webp'}})}}}};
const calls=[];let replyAttempts=0;
const store={{claimNext:async()=>item?{{...item}}:null,update:async(_id,fn)=>{{item=fn(item);}},
  complete:async()=>{{item=null;}},release:async()=>{{item={{...item,status:'queued'}};}},fail:async()=>{{}},countBlocked:async()=>0}};
const fetchJson=async(url,options={{}})=>{{if(url==='api/v1/background-identity')return{{login:'timmy'}};
  calls.push({{url,key:options.headers?.['Idempotency-Key'],body:options.body instanceof FormData?'multipart':options.body?JSON.parse(options.body):null}});
  if(url.endsWith('/attachments'))return{{markdown:'![phone.webp](http://127.0.0.1:3000/phone.webp)'}};
  if(url.endsWith('/reply') && replyAttempts++===0){{const error=new Error('offline');error.status=503;throw error;}}
  return url.endsWith('/reply')?{{id:8}}:{{status:'read'}};
}};
(async()=>{{const sync=createSync({{store,fetchJson}});try{{await sync.flush();}}catch(_error){{}}
  const checkpoint={{attachmentMarkdown:item.attachmentMarkdown,replyConfirmed:item.replyConfirmed}};
  const result=await sync.flush();process.stdout.write(JSON.stringify({{calls,checkpoint,result}}));}})();
"""
    output = run_node(script)

    assert output["checkpoint"]["attachmentMarkdown"].startswith("![phone.webp]")
    assert output["checkpoint"]["replyConfirmed"] is False
    assert [call["url"] for call in output["calls"]] == [
        "api/v1/notifications/527/attachments",
        "api/v1/notifications/527/reply",
        "api/v1/notifications/527/reply",
        "api/v1/notifications/527/read",
    ]
    assert [call["key"] for call in output["calls"][:3]] == [
        "reply-image:attachment", "reply-image:reply", "reply-image:reply"
    ]
    assert output["calls"][2]["body"] == {"body": "![phone.webp](http://127.0.0.1:3000/phone.webp)"}
    assert output["result"]["confirmed"] == [{"status": "read"}]