183 lines
8.1 KiB
Python
183 lines
8.1 KiB
Python
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": "https://forge.example/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"] == ""
|
|
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": "https://forge.example/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"] == "https://forge.example/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:''}};
|
|
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": ""}
|
|
assert output["result"]["confirmed"] == [{"status": "read"}]
|