Decode git forge content responses for agents
All checks were successful
Container Image / build-and-push (push) Successful in 22s
All checks were successful
Container Image / build-and-push (push) Successful in 22s
This commit is contained in:
@@ -4,6 +4,7 @@ Covers:
|
||||
(a) Large JSON list response -> sentinel appended, valid JSON returned
|
||||
(b) Small response -> returned unchanged, no truncation
|
||||
"""
|
||||
import base64
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
@@ -72,7 +73,7 @@ def _make_response(json_data, status=200):
|
||||
return resp
|
||||
|
||||
|
||||
async def _call(json_data, status=200):
|
||||
async def _call(json_data, status=200, *, integration=None, path="/items"):
|
||||
mock_resp = _make_response(json_data, status)
|
||||
|
||||
mock_client = AsyncMock()
|
||||
@@ -81,13 +82,13 @@ async def _call(json_data, status=200):
|
||||
mock_client.request = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with (
|
||||
patch.object(integrations, "_find_integration", return_value=DUMMY_INTEGRATION),
|
||||
patch.object(integrations, "_find_integration", return_value=integration or DUMMY_INTEGRATION),
|
||||
patch("httpx.AsyncClient", return_value=mock_client),
|
||||
# api.example.com doesn't resolve; the SSRF guard would fail closed.
|
||||
# These tests are about truncation, so stub the guard open.
|
||||
patch("src.url_safety.check_outbound_url", return_value=(True, "ok")),
|
||||
):
|
||||
return await integrations.execute_api_call("test_integ", "GET", "/items")
|
||||
return await integrations.execute_api_call("test_integ", "GET", path)
|
||||
|
||||
|
||||
async def _call_with_integration(integration, path="/items"):
|
||||
@@ -230,6 +231,32 @@ async def test_small_json_dict_not_truncated():
|
||||
assert "_truncated" not in parsed
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_gitea_contents_response_decodes_base64_and_omits_raw_blob():
|
||||
text = 'APP_VERSION = "1.0.1"\n'
|
||||
payload = {
|
||||
"name": "constants.py",
|
||||
"path": "src/constants.py",
|
||||
"encoding": "base64",
|
||||
"content": base64.b64encode(text.encode("utf-8")).decode("ascii"),
|
||||
}
|
||||
integration = {**DUMMY_INTEGRATION, "preset": "gitea"}
|
||||
|
||||
result = await _call(
|
||||
payload,
|
||||
integration=integration,
|
||||
path="/api/v1/repos/MrSphay/odysseus/contents/src/constants.py",
|
||||
)
|
||||
|
||||
assert result.get("exit_code") == 0
|
||||
body = result["output"].split(chr(10), 1)[1]
|
||||
parsed = json.loads(body)
|
||||
assert parsed["decoded_content"] == text
|
||||
assert parsed["decoded_truncated"] is False
|
||||
assert parsed["content_omitted"] is True
|
||||
assert "content" not in parsed
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_truncation_respects_limit_including_sentinel():
|
||||
"""After list truncation the total serialized body must not exceed 12000 chars,
|
||||
|
||||
Reference in New Issue
Block a user