Compare commits
7 Commits
feat/opax-
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 73b03dff0e | |||
| f0039088c2 | |||
| c3fde1e38c | |||
| 1122fb5b3a | |||
| 39edd6ef83 | |||
| 56d2768473 | |||
| 9e76526f45 |
|
|
@ -56,6 +56,23 @@ gcloud builds submit \
|
|||
.
|
||||
```
|
||||
|
||||
### Observed Staged Rollout Behavior (2026-09-23)
|
||||
|
||||
During the 2026-09-23 rollout, the `opax-mcp` service traffic configuration was
|
||||
explicitly pinned to a single production revision. Observations from that rollout
|
||||
include:
|
||||
|
||||
- Candidate revisions created outside the tagged staged flow were observed to
|
||||
retire before a later promotion decision.
|
||||
- A candidate deployed with both `--no-traffic` and a unique `--tag` remained
|
||||
active and addressable for a health check.
|
||||
- Promotion required a separate, explicit `gcloud run services update-traffic`
|
||||
command.
|
||||
|
||||
These are observations from a specific rollout, not general Cloud Run
|
||||
guarantees. The staged workflow observed to succeed is documented in
|
||||
`docs/runbooks/opax-live-deploy.md`.
|
||||
|
||||
## CI1e — ADK-integrasjon (neste steg)
|
||||
|
||||
Når `opax-mcp` er live, legges det til som ADK-tool i `agents/core-logic/agent.py`:
|
||||
|
|
|
|||
|
|
@ -18,6 +18,9 @@ CLOUD_RUN_SERVICE = os.environ.get("CLOUD_RUN_SERVICE", "osvauco-agent")
|
|||
MCP_SECRET = os.environ.get("MCP_SECRET", "")
|
||||
BUILD_TRIGGER_ID = os.environ.get("CLOUD_BUILD_TRIGGER_ID", "38423976-91ff-4ff4-859e-1f262344c609")
|
||||
|
||||
_OPAX_MCP_SERVICE = "opax-mcp"
|
||||
_OPAX_MCP_REGION = "us-central1"
|
||||
|
||||
|
||||
def verify_token(request: Request, x_mcp_key: str = Header(default="")):
|
||||
# Aksepter både X-MCP-Key og api-key (Perplexity MCP connector bruker api-key)
|
||||
|
|
@ -70,6 +73,107 @@ def _parse_ts(ts):
|
|||
return None
|
||||
|
||||
|
||||
def _sanitize_opax_deployment_status(service_data: dict) -> dict:
|
||||
"""
|
||||
Takes a decoded Cloud Run v2 service JSON object and returns a sanitized
|
||||
dict conforming to the minimal v1 output contract.
|
||||
"""
|
||||
def _validate_generation(val):
|
||||
if isinstance(val, bool):
|
||||
return None
|
||||
if isinstance(val, int) and val >= 0:
|
||||
return val
|
||||
if isinstance(val, str) and val.isascii() and val.isdecimal():
|
||||
return int(val)
|
||||
return None
|
||||
|
||||
def _validate_update_time(val):
|
||||
from datetime import datetime, timezone
|
||||
if (
|
||||
not isinstance(val, str)
|
||||
or not val.endswith("Z")
|
||||
or val.endswith("ZZ")
|
||||
or len(val) < 20
|
||||
or val[10] != "T"
|
||||
):
|
||||
return None
|
||||
try:
|
||||
normalized_val = val[:-1] + "+00:00"
|
||||
dt = datetime.fromisoformat(normalized_val)
|
||||
if dt.tzinfo is None:
|
||||
return None # Reject timezone-naive
|
||||
return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
return {
|
||||
"service": _OPAX_MCP_SERVICE,
|
||||
"region": _OPAX_MCP_REGION,
|
||||
"generation": _validate_generation(service_data.get("generation")),
|
||||
"observed_generation": _validate_generation(service_data.get("observedGeneration")),
|
||||
"last_update_time": _validate_update_time(service_data.get("updateTime")),
|
||||
"reason_code": None,
|
||||
}
|
||||
|
||||
|
||||
def _fetch_opax_deployment_service_data() -> dict:
|
||||
"""
|
||||
Fetches the raw Cloud Run v2 service object for OPAX-MCP, returning
|
||||
only a minimal subset of fields or a sanitized failure reason.
|
||||
"""
|
||||
import json
|
||||
import socket
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
try:
|
||||
token = _get_access_token()
|
||||
if not token:
|
||||
return {"reason_code": "ADC_TOKEN_UNAVAILABLE"}
|
||||
|
||||
endpoint_url = (
|
||||
f"https://run.googleapis.com/v2/projects/{PROJECT_ID}/"
|
||||
f"locations/{_OPAX_MCP_REGION}/services/{_OPAX_MCP_SERVICE}"
|
||||
)
|
||||
req = urllib.request.Request(
|
||||
endpoint_url, headers={"Authorization": f"Bearer {token}"}, method="GET"
|
||||
)
|
||||
|
||||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||||
status = getattr(resp, "status", None)
|
||||
if not isinstance(status, int):
|
||||
return {"reason_code": "UNEXPECTED_LOCAL_FAILURE"}
|
||||
if not (200 <= status < 300):
|
||||
return {"reason_code": "UNEXPECTED_NON_2XX_STATUS"}
|
||||
|
||||
body_bytes = resp.read()
|
||||
|
||||
body = json.loads(body_bytes)
|
||||
if not isinstance(body, dict):
|
||||
return {"reason_code": "INVALID_RESPONSE_SHAPE"}
|
||||
|
||||
return {
|
||||
"generation": body.get("generation"),
|
||||
"observedGeneration": body.get("observedGeneration"),
|
||||
"updateTime": body.get("updateTime"),
|
||||
}
|
||||
|
||||
except json.JSONDecodeError:
|
||||
return {"reason_code": "INVALID_JSON_RESPONSE"}
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code in (401, 403):
|
||||
return {"reason_code": "CLOUD_RUN_UNAUTHORIZED"}
|
||||
if e.code == 404:
|
||||
return {"reason_code": "CLOUD_RUN_NOT_FOUND"}
|
||||
return {"reason_code": "UNEXPECTED_NON_2XX_STATUS"}
|
||||
except socket.timeout:
|
||||
return {"reason_code": "REQUEST_TIMEOUT"}
|
||||
except urllib.error.URLError:
|
||||
return {"reason_code": "NETWORK_FAILURE"}
|
||||
except Exception:
|
||||
return {"reason_code": "UNEXPECTED_LOCAL_FAILURE"}
|
||||
|
||||
|
||||
class PushStaticRequest(BaseModel):
|
||||
file_path: str
|
||||
content: str
|
||||
|
|
|
|||
202
agents/mcp_server/test_server.py
Normal file
202
agents/mcp_server/test_server.py
Normal file
|
|
@ -0,0 +1,202 @@
|
|||
import unittest
|
||||
import json
|
||||
import socket
|
||||
import urllib.error
|
||||
from unittest.mock import patch, MagicMock
|
||||
from agents.mcp_server.server import (
|
||||
_fetch_opax_deployment_service_data,
|
||||
_sanitize_opax_deployment_status,
|
||||
)
|
||||
|
||||
class TestSanitizeOpaxDeploymentStatus(unittest.TestCase):
|
||||
|
||||
def test_valid_inputs(self):
|
||||
"""Tests valid integer/string generations and a valid UTC timestamp."""
|
||||
service_data = {
|
||||
"generation": 10,
|
||||
"observedGeneration": "9",
|
||||
"updateTime": "2023-10-27T10:00:00Z"
|
||||
}
|
||||
result = _sanitize_opax_deployment_status(service_data)
|
||||
self.assertEqual(result["generation"], 10)
|
||||
self.assertEqual(result["observed_generation"], 9)
|
||||
self.assertEqual(result["last_update_time"], "2023-10-27T10:00:00Z")
|
||||
self.assertIsNone(result["reason_code"])
|
||||
|
||||
def test_zero_values(self):
|
||||
"""Tests handling of zero for generation values."""
|
||||
service_data = {"generation": 0, "observedGeneration": "0"}
|
||||
result = _sanitize_opax_deployment_status(service_data)
|
||||
self.assertEqual(result["generation"], 0)
|
||||
self.assertEqual(result["observed_generation"], 0)
|
||||
|
||||
def test_invalid_generation_values(self):
|
||||
"""Tests various invalid generation inputs."""
|
||||
invalid_inputs = [
|
||||
True, False, -1, 1.5, "-1", " 1 ", "abc", "١", None, {}, []
|
||||
]
|
||||
for val in invalid_inputs:
|
||||
with self.subTest(val=val):
|
||||
res1 = _sanitize_opax_deployment_status({"generation": val})
|
||||
res2 = _sanitize_opax_deployment_status({"observedGeneration": val})
|
||||
self.assertIsNone(res1["generation"])
|
||||
self.assertIsNone(res2["observed_generation"])
|
||||
|
||||
def test_fractional_second_timestamp_is_truncated(self):
|
||||
"""Tests that fractional seconds are correctly truncated."""
|
||||
service_data = {"updateTime": "2023-10-27T10:00:00.123456Z"}
|
||||
result = _sanitize_opax_deployment_status(service_data)
|
||||
self.assertEqual(result["last_update_time"], "2023-10-27T10:00:00Z")
|
||||
|
||||
def test_invalid_and_non_z_timestamps(self):
|
||||
"""Tests invalid, tz-naive, missing, and non-Z timestamps."""
|
||||
invalid_timestamps = [
|
||||
"2023-10-27T10:00:00", # Missing Z (naive)
|
||||
"2023-10-27T10:00:00ZZ", # Multiple Z
|
||||
"2023-10-27T11:00:00+01:00", # Valid but not Z-suffix
|
||||
"2023-10-27 10:00:00Z", # Malformed
|
||||
"not-a-timestamp", # Invalid
|
||||
1698399600, # Not a string
|
||||
None, # Missing
|
||||
]
|
||||
for ts in invalid_timestamps:
|
||||
with self.subTest(ts=ts):
|
||||
result = _sanitize_opax_deployment_status({"updateTime": ts})
|
||||
self.assertIsNone(result["last_update_time"])
|
||||
|
||||
def test_exact_key_set_and_fixed_values(self):
|
||||
"""Tests for exact key set and fixed service/region/reason_code."""
|
||||
result = _sanitize_opax_deployment_status({})
|
||||
expected_keys = {
|
||||
"service", "region", "generation", "observed_generation",
|
||||
"last_update_time", "reason_code"
|
||||
}
|
||||
self.assertEqual(set(result.keys()), expected_keys)
|
||||
self.assertEqual(result["service"], "opax-mcp")
|
||||
self.assertEqual(result["region"], "us-central1")
|
||||
self.assertIsNone(result["reason_code"])
|
||||
|
||||
|
||||
class TestFetchOpaxDeploymentServiceData(unittest.TestCase):
|
||||
|
||||
@patch('urllib.request.urlopen')
|
||||
@patch('agents.mcp_server.server._get_access_token')
|
||||
def test_success_filters_fields(self, mock_get_token, mock_urlopen):
|
||||
mock_get_token.return_value = "fake-token"
|
||||
mock_response = MagicMock()
|
||||
mock_response.status = 200
|
||||
mock_response.read.return_value = json.dumps({
|
||||
"generation": 5, "observedGeneration": 5, "updateTime": "T",
|
||||
"uri": "forbidden"
|
||||
}).encode()
|
||||
mock_urlopen.return_value.__enter__.return_value = mock_response
|
||||
|
||||
result = _fetch_opax_deployment_service_data()
|
||||
self.assertEqual(result, {"generation": 5, "observedGeneration": 5, "updateTime": "T"})
|
||||
self.assertEqual(set(result.keys()), {"generation", "observedGeneration", "updateTime"})
|
||||
|
||||
@patch('urllib.request.urlopen')
|
||||
@patch('agents.mcp_server.server._get_access_token')
|
||||
def test_success_missing_fields(self, mock_get_token, mock_urlopen):
|
||||
mock_get_token.return_value = "fake-token"
|
||||
mock_response = MagicMock()
|
||||
mock_response.status = 200
|
||||
mock_response.read.return_value = json.dumps({"generation": 1}).encode()
|
||||
mock_urlopen.return_value.__enter__.return_value = mock_response
|
||||
|
||||
result = _fetch_opax_deployment_service_data()
|
||||
self.assertEqual(result, {"generation": 1, "observedGeneration": None, "updateTime": None})
|
||||
|
||||
@patch('urllib.request.urlopen')
|
||||
@patch('agents.mcp_server.server._get_access_token')
|
||||
def test_falsy_token_skips_call(self, mock_get_token, mock_urlopen):
|
||||
for token in [None, ""]:
|
||||
with self.subTest(token=token):
|
||||
mock_get_token.return_value = token
|
||||
result = _fetch_opax_deployment_service_data()
|
||||
self.assertEqual(result, {"reason_code": "ADC_TOKEN_UNAVAILABLE"})
|
||||
mock_urlopen.assert_not_called()
|
||||
|
||||
@patch('urllib.request.urlopen')
|
||||
@patch('agents.mcp_server.server._get_access_token')
|
||||
def test_get_token_exception(self, mock_get_token, mock_urlopen):
|
||||
mock_get_token.side_effect = Exception("local failure")
|
||||
result = _fetch_opax_deployment_service_data()
|
||||
self.assertEqual(result, {"reason_code": "UNEXPECTED_LOCAL_FAILURE"})
|
||||
mock_urlopen.assert_not_called()
|
||||
|
||||
@patch('urllib.request.urlopen')
|
||||
@patch('agents.mcp_server.server._get_access_token')
|
||||
def test_absent_response_status(self, mock_get_token, mock_urlopen):
|
||||
mock_get_token.return_value = "fake-token"
|
||||
class MockResp:
|
||||
def __init__(self):
|
||||
self.read_called = False
|
||||
def read(self):
|
||||
self.read_called = True
|
||||
return b""
|
||||
def __enter__(self):
|
||||
return self
|
||||
def __exit__(self, *args):
|
||||
return False
|
||||
|
||||
mock_response = MockResp()
|
||||
mock_urlopen.return_value = mock_response
|
||||
result = _fetch_opax_deployment_service_data()
|
||||
self.assertEqual(result, {"reason_code": "UNEXPECTED_LOCAL_FAILURE"})
|
||||
self.assertFalse(mock_response.read_called)
|
||||
|
||||
@patch('urllib.request.urlopen')
|
||||
@patch('agents.mcp_server.server._get_access_token')
|
||||
def test_non_2xx_status_skips_read(self, mock_get_token, mock_urlopen):
|
||||
mock_get_token.return_value = "fake-token"
|
||||
mock_response = MagicMock()
|
||||
mock_response.status = 503
|
||||
mock_urlopen.return_value.__enter__.return_value = mock_response
|
||||
|
||||
result = _fetch_opax_deployment_service_data()
|
||||
self.assertEqual(result, {"reason_code": "UNEXPECTED_NON_2XX_STATUS"})
|
||||
mock_response.read.assert_not_called()
|
||||
|
||||
@patch('urllib.request.urlopen')
|
||||
@patch('agents.mcp_server.server._get_access_token')
|
||||
def test_http_and_network_errors(self, mock_get_token, mock_urlopen):
|
||||
mock_get_token.return_value = "fake-token"
|
||||
errors_to_reasons = [
|
||||
(urllib.error.HTTPError(None, 401, "", {}, None), "CLOUD_RUN_UNAUTHORIZED"),
|
||||
(urllib.error.HTTPError(None, 403, "", {}, None), "CLOUD_RUN_UNAUTHORIZED"),
|
||||
(urllib.error.HTTPError(None, 404, "", {}, None), "CLOUD_RUN_NOT_FOUND"),
|
||||
(urllib.error.HTTPError(None, 500, "", {}, None), "UNEXPECTED_NON_2XX_STATUS"),
|
||||
(socket.timeout(), "REQUEST_TIMEOUT"),
|
||||
(urllib.error.URLError("DNS failure"), "NETWORK_FAILURE"),
|
||||
]
|
||||
for error, reason in errors_to_reasons:
|
||||
with self.subTest(error=error.__class__.__name__, code=getattr(error, 'code', 'N/A')):
|
||||
mock_urlopen.side_effect = error
|
||||
result = _fetch_opax_deployment_service_data()
|
||||
self.assertEqual(result, {"reason_code": reason})
|
||||
mock_urlopen.side_effect = None
|
||||
|
||||
@patch('urllib.request.urlopen')
|
||||
@patch('agents.mcp_server.server._get_access_token')
|
||||
def test_invalid_json_body(self, mock_get_token, mock_urlopen):
|
||||
mock_get_token.return_value = "fake-token"
|
||||
mock_response = MagicMock()
|
||||
mock_response.status = 200
|
||||
mock_urlopen.return_value.__enter__.return_value = mock_response
|
||||
|
||||
# Malformed JSON
|
||||
with self.subTest(case="malformed"):
|
||||
mock_response.read.return_value = b'{"key":'
|
||||
result = _fetch_opax_deployment_service_data()
|
||||
self.assertEqual(result, {"reason_code": "INVALID_JSON_RESPONSE"})
|
||||
|
||||
# Valid JSON, but not a dictionary object
|
||||
with self.subTest(case="non-object"):
|
||||
mock_response.read.return_value = b'[1, 2, 3]'
|
||||
result = _fetch_opax_deployment_service_data()
|
||||
self.assertEqual(result, {"reason_code": "INVALID_RESPONSE_SHAPE"})
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
@ -34,6 +34,79 @@ to the existing Cloud Run service.
|
|||
|
||||
## MCP Deployment Behavior
|
||||
|
||||
### OPAX-MCP Staged Deployment Workflow (Observed 2026-09-23)
|
||||
|
||||
During the 2026-09-23 rollout, the service traffic configuration was explicitly
|
||||
pinned to a revision. Untagged zero-traffic candidate revisions were observed to
|
||||
retire before a later promotion decision. A candidate created with
|
||||
`--tag candidate-f003908 --no-traffic` remained active, addressable, was
|
||||
health-checked, and was explicitly promoted. These are observations from this
|
||||
rollout, not general Cloud Run guarantees.
|
||||
|
||||
#### Stage A: Create and verify a tagged candidate
|
||||
|
||||
1. Build and push an immutable image without deploying Cloud Run. The
|
||||
`cloudbuild.build-only.yaml` configuration was used for this purpose in the
|
||||
observed rollout.
|
||||
|
||||
2. Create a tagged, no-traffic candidate revision:
|
||||
|
||||
```bash
|
||||
gcloud run deploy opax-mcp \
|
||||
--image <immutable-image-digest> \
|
||||
--region <region> \
|
||||
--tag candidate-<short-commit> \
|
||||
--no-traffic
|
||||
```
|
||||
|
||||
3. Verify the candidate uses the expected image digest, is `Ready` and `Active`,
|
||||
has a revision tag/direct URL, and has 0% traffic while the current
|
||||
production revision retains 100%.
|
||||
|
||||
4. Health-check only the tagged candidate URL:
|
||||
|
||||
```bash
|
||||
curl --silent --show-error --fail \
|
||||
--max-time 10 \
|
||||
--connect-timeout 5 \
|
||||
--max-redirs 0 \
|
||||
--request GET \
|
||||
https://<candidate-tag-url>/health
|
||||
```
|
||||
|
||||
An empty result from a narrow Cloud Logging query means no data was returned
|
||||
for that query; it does not prove the absence of application errors.
|
||||
|
||||
#### Stage B: Promote traffic
|
||||
|
||||
Promotion is a separate, explicit traffic action after Stage A verification:
|
||||
|
||||
```bash
|
||||
gcloud run services update-traffic opax-mcp \
|
||||
--to-revisions=<candidate-revision>=100 \
|
||||
--region <region>
|
||||
```
|
||||
|
||||
#### Rollback
|
||||
|
||||
Rollback is also an explicit traffic action:
|
||||
|
||||
```bash
|
||||
gcloud run services update-traffic opax-mcp \
|
||||
--to-revisions=<last-known-good-revision>=100 \
|
||||
--region <region>
|
||||
```
|
||||
|
||||
#### Observed rollout record — 2026-09-23
|
||||
|
||||
- **Source commit:** `f0039088c2505911c73e6a7374aa105415378277`
|
||||
- **Cloud Build ID:** `5b934520-8f8c-4a8e-89f3-f2f8377c2879`
|
||||
- **Immutable image:** `us-central1-docker.pkg.dev/propane-will-491900-m5/osvauco-repo/opax-mcp@sha256:1378a7b8cd81f7f82b52dafb29c0c9ee2b2d37800b27937fbdd353618f9319bf`
|
||||
- **Candidate tag/revision:** `candidate-f003908` / `opax-mcp-00218-duk`
|
||||
- **Rollback revision:** `opax-mcp-00214-nar`
|
||||
- **Health result:** HTTP 200 with `{"status":"ok","service":"opax-mcp","version":"3.6.0"}`
|
||||
- **Status:** Observed promoted successfully on 2026-09-23.
|
||||
|
||||
## Required OPAX-MCP Runtime Contract
|
||||
|
||||
- `opax-mcp` uses Direct VPC egress to reach the internal Emma/Ollama runtime.
|
||||
|
|
|
|||
|
|
@ -18,6 +18,10 @@ _BRANCH_RE = re.compile(
|
|||
r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$"
|
||||
)
|
||||
|
||||
_READ_BRANCH_REF_RE = re.compile(
|
||||
r"^[A-Za-z0-9][A-Za-z0-9._-]*(?:/[A-Za-z0-9][A-Za-z0-9._-]*)*$"
|
||||
)
|
||||
|
||||
MAX_SOURCE_ARCHIVE_BYTES = int(
|
||||
os.environ.get("OPAX_MAX_SOURCE_ARCHIVE_BYTES", "104857600")
|
||||
)
|
||||
|
|
@ -188,6 +192,34 @@ def validate_path_for_write(path: str) -> str:
|
|||
return path
|
||||
|
||||
|
||||
def _validate_read_branch_ref(ref: str) -> str:
|
||||
"""Validate a safe branch name used only for read resolution."""
|
||||
if not isinstance(ref, str) or not ref:
|
||||
raise ValueError("Invalid git reference.")
|
||||
|
||||
if any(char.isspace() or ord(char) < 32 or ord(char) == 127 for char in ref):
|
||||
raise ValueError("Invalid git reference.")
|
||||
|
||||
if (
|
||||
ref.upper() == "HEAD"
|
||||
or ref.startswith("refs/")
|
||||
or ref.startswith("/")
|
||||
or ref.endswith("/")
|
||||
or "//" in ref
|
||||
or "\\" in ref
|
||||
):
|
||||
raise ValueError("Invalid git reference.")
|
||||
|
||||
components = ref.split("/")
|
||||
if any(component in (".", "..") for component in components):
|
||||
raise ValueError("Invalid git reference.")
|
||||
|
||||
if not _READ_BRANCH_REF_RE.fullmatch(ref):
|
||||
raise ValueError("Invalid git reference.")
|
||||
|
||||
return ref
|
||||
|
||||
|
||||
async def resolve_branch_to_commit_sha(
|
||||
branch_name: str,
|
||||
repo_id: str,
|
||||
|
|
@ -342,10 +374,39 @@ async def handle_get_file_content(p: dict, server_repo_id: str) -> dict:
|
|||
if caller_repo is not None and caller_repo != validated_server_repo:
|
||||
raise ValueError("Repository file request is not allowed.")
|
||||
|
||||
ref = _validate_commit_sha(p.get("ref"))
|
||||
path = _validate_safe_path(p.get("path"))
|
||||
|
||||
url = f"{gitea_url}/api/v1/repos/{validated_server_repo}/contents/{quote(path, safe='')}?ref={ref}"
|
||||
requested_ref = p.get("ref")
|
||||
if not isinstance(requested_ref, str) or not requested_ref:
|
||||
raise ValueError("Invalid git reference.")
|
||||
|
||||
if _SHA_RE.fullmatch(requested_ref.lower()):
|
||||
resolved_commit_sha = _validate_commit_sha(requested_ref.lower())
|
||||
else:
|
||||
validated_branch_ref = _validate_read_branch_ref(requested_ref)
|
||||
try:
|
||||
resolved_commit_sha = await resolve_branch_to_commit_sha(
|
||||
branch_name=validated_branch_ref,
|
||||
repo_id=validated_server_repo,
|
||||
gitea_url=gitea_url,
|
||||
)
|
||||
except httpx.HTTPStatusError as exc:
|
||||
if exc.response.status_code == 404:
|
||||
raise ValueError(
|
||||
"Unknown or inaccessible branch reference."
|
||||
) from exc
|
||||
logger.warning(
|
||||
"Gitea branch resolution failed",
|
||||
extra={"status_code": exc.response.status_code},
|
||||
)
|
||||
raise ValueError("Repository file is unavailable.") from exc
|
||||
except Exception:
|
||||
logger.error("Gitea branch resolution failed unexpectedly")
|
||||
raise ValueError("Repository file is unavailable.")
|
||||
|
||||
resolved_commit_sha = _validate_commit_sha(resolved_commit_sha)
|
||||
|
||||
url = f"{gitea_url}/api/v1/repos/{validated_server_repo}/contents/{quote(path, safe='')}?ref={resolved_commit_sha}"
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=15) as c:
|
||||
|
|
@ -398,7 +459,14 @@ async def handle_get_file_content(p: dict, server_repo_id: str) -> dict:
|
|||
if '\0' in text_content:
|
||||
raise ValueError("Repository file content is not readable text.")
|
||||
|
||||
return {"path": path, "content": text_content, "encoding": "utf-8"}
|
||||
return {
|
||||
"repo": validated_server_repo,
|
||||
"path": path,
|
||||
"requested_ref": requested_ref,
|
||||
"resolved_commit_sha": resolved_commit_sha,
|
||||
"content": text_content,
|
||||
"encoding": "utf-8",
|
||||
}
|
||||
|
||||
|
||||
async def handle_list_repo_files(p: dict, default_repo: str) -> dict:
|
||||
|
|
|
|||
|
|
@ -100,13 +100,21 @@ class EncryptedPayloadV1(BaseModel):
|
|||
ciphertext: str
|
||||
|
||||
|
||||
GITEA_CHANGE_PLAN_STATUS_PENDING = "PENDING"
|
||||
GITEA_CHANGE_PLAN_STATUS_APPROVED = "APPROVED"
|
||||
GITEA_CHANGE_PLAN_STATUS_APPLYING = "APPLYING"
|
||||
GITEA_CHANGE_PLAN_STATUS_APPLIED = "APPLIED"
|
||||
GITEA_CHANGE_PLAN_STATUS_REJECTED = "REJECTED"
|
||||
GITEA_CHANGE_PLAN_STATUS_EXPIRED = "EXPIRED"
|
||||
|
||||
|
||||
class GiteaChangePlan(BaseModel):
|
||||
"""Immutable admin-approved one-file Gitea change plan."""
|
||||
|
||||
plan_id: str = Field(
|
||||
default_factory=lambda: f"gitea-change-{uuid.uuid4().hex}"
|
||||
)
|
||||
status: str = "PENDING"
|
||||
status: str = GITEA_CHANGE_PLAN_STATUS_PENDING
|
||||
|
||||
created_at: datetime = Field(default_factory=datetime.utcnow)
|
||||
expires_at: datetime = Field(
|
||||
|
|
@ -134,6 +142,38 @@ class GiteaChangePlan(BaseModel):
|
|||
apply_error_message: Optional[str] = None
|
||||
|
||||
|
||||
def is_gitea_change_plan_expired(plan: GiteaChangePlan) -> bool:
|
||||
"""Checks if a Gitea change plan has expired."""
|
||||
now_utc = datetime.now(timezone.utc)
|
||||
expires_at = plan.expires_at
|
||||
|
||||
if expires_at.tzinfo is None:
|
||||
expires_at = expires_at.replace(tzinfo=timezone.utc)
|
||||
|
||||
return now_utc >= expires_at
|
||||
|
||||
|
||||
_ALLOWED_GITEA_PLAN_TRANSITIONS = {
|
||||
GITEA_CHANGE_PLAN_STATUS_PENDING: {
|
||||
GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
GITEA_CHANGE_PLAN_STATUS_REJECTED,
|
||||
GITEA_CHANGE_PLAN_STATUS_EXPIRED,
|
||||
},
|
||||
GITEA_CHANGE_PLAN_STATUS_APPROVED: {
|
||||
GITEA_CHANGE_PLAN_STATUS_APPLYING,
|
||||
GITEA_CHANGE_PLAN_STATUS_EXPIRED,
|
||||
},
|
||||
GITEA_CHANGE_PLAN_STATUS_APPLYING: {GITEA_CHANGE_PLAN_STATUS_APPLIED},
|
||||
}
|
||||
|
||||
def is_valid_gitea_change_plan_status_transition(
|
||||
current_status: str,
|
||||
next_status: str,
|
||||
) -> bool:
|
||||
"""Checks if a Gitea change plan status transition is allowed."""
|
||||
return next_status in _ALLOWED_GITEA_PLAN_TRANSITIONS.get(current_status, set())
|
||||
|
||||
|
||||
async def create_gitea_change_plan(plan: GiteaChangePlan) -> dict:
|
||||
"""Persist a pending one-file Gitea change plan in Firestore."""
|
||||
from google.cloud import firestore
|
||||
|
|
@ -230,6 +270,42 @@ async def get_gitea_change_plan(
|
|||
return GiteaChangePlan(**document.to_dict())
|
||||
|
||||
|
||||
async def transition_gitea_change_plan_status(
|
||||
plan_id: str,
|
||||
expected_status: str,
|
||||
next_status: str,
|
||||
updates: Optional[dict] = None,
|
||||
) -> GiteaChangePlan:
|
||||
if not is_valid_gitea_change_plan_status_transition(expected_status, next_status):
|
||||
raise ValueError("Invalid Gitea change plan status transition")
|
||||
if updates and "status" in updates:
|
||||
raise ValueError("Gitea change plan updates cannot include status")
|
||||
|
||||
from google.cloud import firestore
|
||||
|
||||
db = firestore.AsyncClient(project=GOOGLE_CLOUD_PROJECT)
|
||||
plan_ref = db.collection("gitea_change_plans").document(plan_id)
|
||||
transaction = db.transaction()
|
||||
|
||||
@firestore.async_transactional
|
||||
async def transactional_update(transaction):
|
||||
snapshot = await plan_ref.get(transaction=transaction)
|
||||
if not snapshot.exists:
|
||||
raise ValueError("Gitea change plan not found")
|
||||
if snapshot.get("status") != expected_status:
|
||||
raise ValueError("Gitea change plan status changed")
|
||||
|
||||
update_data = dict(updates or {})
|
||||
update_data["status"] = next_status
|
||||
transaction.update(plan_ref, update_data)
|
||||
|
||||
await transactional_update(transaction)
|
||||
updated_plan = await get_gitea_change_plan(plan_id)
|
||||
if updated_plan is None:
|
||||
raise ValueError("Gitea change plan not found")
|
||||
return updated_plan
|
||||
|
||||
|
||||
def _validate_admin_gitea_path(path: str) -> None:
|
||||
"""Validate a repository-relative admin file path."""
|
||||
if not isinstance(path, str) or not path or path.isspace():
|
||||
|
|
@ -1744,4 +1820,4 @@ async def mcp_handler(request: Request):
|
|||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
return {"status": "ok", "service": "opax-mcp", "version": "3.6.0", "ollama": OLLAMA_BASE_URL} # BUMPED
|
||||
return {"status": "ok", "service": "opax-mcp", "version": "3.6.0"}
|
||||
|
|
|
|||
|
|
@ -3,12 +3,68 @@ from pathlib import Path
|
|||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
import os
|
||||
import base64
|
||||
import json
|
||||
import httpx
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(REPO_ROOT / "opax-mcp"))
|
||||
|
||||
from gitea_handler import handle_list_repo_files
|
||||
from gitea_handler import handle_list_repo_files, _validate_read_branch_ref, handle_get_file_content
|
||||
|
||||
|
||||
class TestValidateReadBranchRef(unittest.TestCase):
|
||||
def test_accepted_refs(self):
|
||||
accepted = [
|
||||
"main",
|
||||
"feat/opax-domain-decouple",
|
||||
"docs/mcp-diagnostics",
|
||||
"fix/example",
|
||||
"chore/example",
|
||||
"release/1.2.3",
|
||||
]
|
||||
for ref in accepted:
|
||||
with self.subTest(ref=ref):
|
||||
self.assertEqual(_validate_read_branch_ref(ref), ref)
|
||||
|
||||
def test_rejected_refs(self):
|
||||
rejected = [
|
||||
"",
|
||||
" main",
|
||||
"main ",
|
||||
"feature branch",
|
||||
"main\tbranch",
|
||||
"main\nbranch",
|
||||
"main\x00branch",
|
||||
"main\x7fbranch",
|
||||
"HEAD",
|
||||
"head",
|
||||
"refs/heads/main",
|
||||
"/main",
|
||||
"main/",
|
||||
"main//next",
|
||||
"main/./next",
|
||||
"main/../next",
|
||||
"main\\next",
|
||||
"main?x=1",
|
||||
"main#fragment",
|
||||
"https://example.invalid",
|
||||
"user@host",
|
||||
"feature*",
|
||||
"feature~1",
|
||||
"feature^",
|
||||
"feature{a}",
|
||||
"feature;cmd",
|
||||
"feature|cmd",
|
||||
"feature%2Fbranch",
|
||||
"feature'quote",
|
||||
'feature"quote',
|
||||
"feature[abc]",
|
||||
]
|
||||
for ref in rejected:
|
||||
with self.subTest(ref=ref):
|
||||
with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"):
|
||||
_validate_read_branch_ref(ref)
|
||||
|
||||
class TestGiteaHandler(unittest.TestCase):
|
||||
|
||||
|
|
@ -340,5 +396,332 @@ class TestGiteaHandler(unittest.TestCase):
|
|||
import asyncio
|
||||
asyncio.run(run_test())
|
||||
|
||||
|
||||
class TestHandleGetFileContent(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.original_gitea_url = os.environ.get("GITEA_URL")
|
||||
self.original_gitea_token = os.environ.get("GITEA_TOKEN")
|
||||
os.environ["GITEA_URL"] = "https://gitea.example.com"
|
||||
os.environ["GITEA_TOKEN"] = "test-token"
|
||||
self.server_repo_id = "chris/OSVauco"
|
||||
|
||||
def tearDown(self):
|
||||
if self.original_gitea_url is None:
|
||||
os.environ.pop("GITEA_URL", None)
|
||||
else:
|
||||
os.environ["GITEA_URL"] = self.original_gitea_url
|
||||
if self.original_gitea_token is None:
|
||||
os.environ.pop("GITEA_TOKEN", None)
|
||||
else:
|
||||
os.environ["GITEA_TOKEN"] = self.original_gitea_token
|
||||
|
||||
def _mock_contents_client(self, text):
|
||||
body = json.dumps({
|
||||
"content": base64.b64encode(text.encode("utf-8")).decode("ascii"),
|
||||
"encoding": "base64",
|
||||
}).encode("utf-8")
|
||||
|
||||
async def chunks():
|
||||
yield body
|
||||
|
||||
response = MagicMock()
|
||||
response.headers = {"content-length": str(len(body))}
|
||||
response.aiter_bytes = chunks
|
||||
stream = MagicMock()
|
||||
stream.__aenter__.return_value = response
|
||||
client = MagicMock()
|
||||
client.stream.return_value = stream
|
||||
async_client = MagicMock()
|
||||
async_client.__aenter__.return_value = client
|
||||
return async_client, client
|
||||
|
||||
@patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
|
||||
@patch("httpx.AsyncClient")
|
||||
def test_direct_lowercase_sha_success(
|
||||
self, mock_async_client_constructor, mock_resolver
|
||||
):
|
||||
import asyncio
|
||||
|
||||
sha = "a" * 40
|
||||
mock_async_client, client = self._mock_contents_client("file content")
|
||||
mock_async_client_constructor.return_value = mock_async_client
|
||||
|
||||
async def run_test():
|
||||
return await handle_get_file_content(
|
||||
{
|
||||
"repo": self.server_repo_id,
|
||||
"path": "README.md",
|
||||
"ref": sha,
|
||||
},
|
||||
self.server_repo_id,
|
||||
)
|
||||
|
||||
result = asyncio.run(run_test())
|
||||
|
||||
mock_resolver.assert_not_called()
|
||||
client.stream.assert_called_once()
|
||||
request_url = client.stream.call_args.args[1]
|
||||
self.assertIn(f"ref={sha}", request_url)
|
||||
self.assertEqual(result["requested_ref"], sha)
|
||||
self.assertEqual(result["resolved_commit_sha"], sha)
|
||||
self.assertEqual(result["content"], "file content")
|
||||
self.assertEqual(result["encoding"], "utf-8")
|
||||
|
||||
@patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
|
||||
@patch("httpx.AsyncClient")
|
||||
def test_direct_uppercase_sha_normalizes_for_request(
|
||||
self, mock_async_client_constructor, mock_resolver
|
||||
):
|
||||
import asyncio
|
||||
|
||||
requested_sha = "A" * 40
|
||||
resolved_sha = requested_sha.lower()
|
||||
mock_async_client, client = self._mock_contents_client("file content")
|
||||
mock_async_client_constructor.return_value = mock_async_client
|
||||
|
||||
async def run_test():
|
||||
return await handle_get_file_content(
|
||||
{
|
||||
"repo": self.server_repo_id,
|
||||
"path": "README.md",
|
||||
"ref": requested_sha,
|
||||
},
|
||||
self.server_repo_id,
|
||||
)
|
||||
|
||||
result = asyncio.run(run_test())
|
||||
|
||||
mock_resolver.assert_not_called()
|
||||
request_url = client.stream.call_args.args[1]
|
||||
self.assertIn(f"ref={resolved_sha}", request_url)
|
||||
self.assertNotIn(f"ref={requested_sha}", request_url)
|
||||
self.assertEqual(result["requested_ref"], requested_sha)
|
||||
self.assertEqual(result["resolved_commit_sha"], resolved_sha)
|
||||
|
||||
@patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
|
||||
@patch("httpx.AsyncClient")
|
||||
def test_branch_resolves_to_sha_before_contents_request(
|
||||
self, mock_async_client_constructor, mock_resolver
|
||||
):
|
||||
import asyncio
|
||||
|
||||
resolved_sha = "b" * 40
|
||||
mock_resolver.return_value = resolved_sha
|
||||
mock_async_client, client = self._mock_contents_client("file content")
|
||||
mock_async_client_constructor.return_value = mock_async_client
|
||||
|
||||
async def run_test():
|
||||
return await handle_get_file_content(
|
||||
{
|
||||
"repo": self.server_repo_id,
|
||||
"path": "README.md",
|
||||
"ref": "main",
|
||||
},
|
||||
self.server_repo_id,
|
||||
)
|
||||
|
||||
result = asyncio.run(run_test())
|
||||
|
||||
mock_resolver.assert_awaited_once_with(
|
||||
branch_name="main",
|
||||
repo_id=self.server_repo_id,
|
||||
gitea_url="https://gitea.example.com",
|
||||
)
|
||||
request_url = client.stream.call_args.args[1]
|
||||
self.assertIn(f"ref={resolved_sha}", request_url)
|
||||
self.assertNotIn("ref=main", request_url)
|
||||
self.assertEqual(result["repo"], self.server_repo_id)
|
||||
self.assertEqual(result["path"], "README.md")
|
||||
self.assertEqual(result["requested_ref"], "main")
|
||||
self.assertEqual(result["resolved_commit_sha"], resolved_sha)
|
||||
self.assertEqual(result["content"], "file content")
|
||||
self.assertEqual(result["encoding"], "utf-8")
|
||||
|
||||
@patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
|
||||
@patch("httpx.AsyncClient")
|
||||
def test_unresolved_branch_raises_sanitized_error(
|
||||
self, mock_async_client_constructor, mock_resolver
|
||||
):
|
||||
import asyncio
|
||||
|
||||
mock_resolver.side_effect = httpx.HTTPStatusError(
|
||||
"Not Found",
|
||||
request=MagicMock(),
|
||||
response=MagicMock(status_code=404),
|
||||
)
|
||||
mock_async_client, client = self._mock_contents_client("file content")
|
||||
mock_async_client_constructor.return_value = mock_async_client
|
||||
|
||||
async def run_test():
|
||||
with self.assertRaisesRegex(
|
||||
ValueError, r"^Unknown or inaccessible branch reference\.$"
|
||||
):
|
||||
await handle_get_file_content(
|
||||
{
|
||||
"repo": self.server_repo_id,
|
||||
"path": "README.md",
|
||||
"ref": "missing-branch",
|
||||
},
|
||||
self.server_repo_id,
|
||||
)
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
client.stream.assert_not_called()
|
||||
|
||||
@patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
|
||||
@patch("httpx.AsyncClient")
|
||||
def test_resolver_non_404_http_error_is_sanitized(
|
||||
self, mock_async_client_constructor, mock_resolver
|
||||
):
|
||||
import asyncio
|
||||
|
||||
mock_resolver.side_effect = httpx.HTTPStatusError(
|
||||
"Server Error",
|
||||
request=MagicMock(),
|
||||
response=MagicMock(status_code=500),
|
||||
)
|
||||
mock_async_client, client = self._mock_contents_client("file content")
|
||||
mock_async_client_constructor.return_value = mock_async_client
|
||||
|
||||
async def run_test():
|
||||
with self.assertRaisesRegex(
|
||||
ValueError, r"^Repository file is unavailable\.$"
|
||||
):
|
||||
await handle_get_file_content(
|
||||
{
|
||||
"repo": self.server_repo_id,
|
||||
"path": "README.md",
|
||||
"ref": "main",
|
||||
},
|
||||
self.server_repo_id,
|
||||
)
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
client.stream.assert_not_called()
|
||||
|
||||
@patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
|
||||
@patch("httpx.AsyncClient")
|
||||
def test_resolver_unexpected_exception_is_sanitized(
|
||||
self, mock_async_client_constructor, mock_resolver
|
||||
):
|
||||
import asyncio
|
||||
|
||||
mock_resolver.side_effect = RuntimeError(
|
||||
"upstream internal details must not reach the caller"
|
||||
)
|
||||
mock_async_client, client = self._mock_contents_client("file content")
|
||||
mock_async_client_constructor.return_value = mock_async_client
|
||||
|
||||
async def run_test():
|
||||
with self.assertRaisesRegex(
|
||||
ValueError, r"^Repository file is unavailable\.$"
|
||||
):
|
||||
await handle_get_file_content(
|
||||
{
|
||||
"repo": self.server_repo_id,
|
||||
"path": "README.md",
|
||||
"ref": "main",
|
||||
},
|
||||
self.server_repo_id,
|
||||
)
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
client.stream.assert_not_called()
|
||||
|
||||
@patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
|
||||
@patch("httpx.AsyncClient")
|
||||
def test_resolver_returns_malformed_sha_is_rejected(
|
||||
self, mock_async_client_constructor, mock_resolver
|
||||
):
|
||||
import asyncio
|
||||
|
||||
mock_resolver.return_value = "not-a-40-character-commit-sha"
|
||||
mock_async_client, client = self._mock_contents_client("file content")
|
||||
mock_async_client_constructor.return_value = mock_async_client
|
||||
|
||||
async def run_test():
|
||||
with self.assertRaisesRegex(ValueError, r"^Invalid commit SHA$"):
|
||||
await handle_get_file_content(
|
||||
{
|
||||
"repo": self.server_repo_id,
|
||||
"path": "README.md",
|
||||
"ref": "main",
|
||||
},
|
||||
self.server_repo_id,
|
||||
)
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
client.stream.assert_not_called()
|
||||
|
||||
@patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
|
||||
@patch("httpx.AsyncClient")
|
||||
def test_invalid_branch_ref_is_rejected(
|
||||
self, mock_async_client_constructor, mock_resolver
|
||||
):
|
||||
import asyncio
|
||||
|
||||
mock_async_client, client = self._mock_contents_client("file content")
|
||||
mock_async_client_constructor.return_value = mock_async_client
|
||||
|
||||
async def run_test():
|
||||
with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"):
|
||||
await handle_get_file_content(
|
||||
{
|
||||
"repo": self.server_repo_id,
|
||||
"path": "README.md",
|
||||
"ref": "main/../secret",
|
||||
},
|
||||
self.server_repo_id,
|
||||
)
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
mock_resolver.assert_not_called()
|
||||
client.stream.assert_not_called()
|
||||
|
||||
@patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
|
||||
@patch("httpx.AsyncClient")
|
||||
def test_contents_api_404_is_sanitized(
|
||||
self, mock_async_client_constructor, mock_resolver
|
||||
):
|
||||
import asyncio
|
||||
|
||||
resolved_sha = "c" * 40
|
||||
mock_resolver.return_value = resolved_sha
|
||||
mock_async_client, client = self._mock_contents_client("file content")
|
||||
|
||||
response_mock = client.stream.return_value.__aenter__.return_value
|
||||
response_mock.raise_for_status.side_effect = httpx.HTTPStatusError(
|
||||
"Not Found", request=MagicMock(), response=MagicMock(status_code=404)
|
||||
)
|
||||
mock_async_client_constructor.return_value = mock_async_client
|
||||
|
||||
async def run_test():
|
||||
with self.assertRaisesRegex(
|
||||
ValueError, r"^Repository file is unavailable\.$"
|
||||
):
|
||||
await handle_get_file_content(
|
||||
{
|
||||
"repo": self.server_repo_id,
|
||||
"path": "README.md",
|
||||
"ref": "main",
|
||||
},
|
||||
self.server_repo_id,
|
||||
)
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
mock_resolver.assert_awaited_once_with(
|
||||
branch_name="main",
|
||||
repo_id=self.server_repo_id,
|
||||
gitea_url="https://gitea.example.com",
|
||||
)
|
||||
client.stream.assert_called_once()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
|
|
@ -297,11 +297,19 @@ class TestGetFileHardening(unittest.IsolatedAsyncioTestCase):
|
|||
)
|
||||
client.stream.assert_not_called()
|
||||
|
||||
async def test_get_file_ref_validation(self):
|
||||
with self.assertRaisesRegex(ValueError, "Invalid commit SHA"):
|
||||
@patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
|
||||
async def test_get_file_ref_validation(self, mock_resolver):
|
||||
mock_resolver.side_effect = httpx.ConnectError("Network error")
|
||||
with self.assertRaisesRegex(ValueError, r"^Repository file is unavailable\.$"):
|
||||
await gitea_handler.handle_get_file_content({"path": "README.md", "ref": "main"}, "chris/OSVauco")
|
||||
with self.assertRaisesRegex(ValueError, "Invalid commit SHA"):
|
||||
mock_resolver.assert_awaited_once()
|
||||
self.assertEqual(mock_resolver.call_args.kwargs['branch_name'], "main")
|
||||
self.assertEqual(mock_resolver.call_args.kwargs['repo_id'], "chris/OSVauco")
|
||||
|
||||
with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"):
|
||||
await gitea_handler.handle_get_file_content({"path": "README.md", "ref": None}, "chris/OSVauco")
|
||||
# Verify the resolver was not called a second time for the invalid ref
|
||||
self.assertEqual(mock_resolver.await_count, 1)
|
||||
|
||||
async def test_gitea_repo_validation(self):
|
||||
with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"):
|
||||
|
|
@ -309,17 +317,24 @@ class TestGetFileHardening(unittest.IsolatedAsyncioTestCase):
|
|||
with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"):
|
||||
await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "")
|
||||
|
||||
async def test_get_file_path_not_allowed(self):
|
||||
for path in ["../secrets.txt", "/etc/passwd"]:
|
||||
with self.subTest(path=path):
|
||||
with self.assertRaisesRegex(ValueError, r"^PATH_NOT_ALLOWED$"):
|
||||
await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco")
|
||||
|
||||
async def test_get_file_secret_path_denied(self):
|
||||
with self.assertRaisesRegex(ValueError, r"^SECRET_PATH_DENIED$"):
|
||||
await gitea_handler.handle_get_file_content({"path": "docs/.env", "ref": self.valid_sha}, "chris/OSVauco")
|
||||
|
||||
@patch("gitea_handler.httpx.AsyncClient")
|
||||
async def test_get_file_path_validation(self, mock_client):
|
||||
async def test_get_file_valid_paths(self, mock_client):
|
||||
client = mock_client.return_value.__aenter__.return_value
|
||||
configure_stream(client, json.dumps({'content': 'YQ==', 'size': 1}).encode("utf-8"))
|
||||
for path in ["../secrets.txt", "/etc/passwd", "src/main.py", "docs/.env", "file.json"]:
|
||||
with self.subTest(path=path):
|
||||
with self.assertRaisesRegex(ValueError, "Repository file request is not allowed."):
|
||||
await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco")
|
||||
for path in ["README.md", "docs/ARCHITECTURE.md", ".gemini/GEMINI.md"]:
|
||||
with self.subTest(path=path):
|
||||
await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco")
|
||||
await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco")
|
||||
self.assertEqual(client.stream.call_count, 3)
|
||||
|
||||
@patch("gitea_handler.httpx.AsyncClient")
|
||||
async def test_get_file_size_limit(self, mock_client):
|
||||
|
|
|
|||
|
|
@ -18,6 +18,26 @@ KMS_KEY_NAME = "projects/p/locations/l/keyRings/k/cryptoKeys/k"
|
|||
|
||||
class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
|
||||
|
||||
@patch('server.handle_get_file_content', new_callable=AsyncMock)
|
||||
async def test_get_file_wrapper_enforces_server_repo(self, mock_handler):
|
||||
sentinel_response = {
|
||||
"requested_ref": "main",
|
||||
"resolved_commit_sha": "a" * 40,
|
||||
"content": "file content"
|
||||
}
|
||||
mock_handler.return_value = sentinel_response
|
||||
|
||||
params = {
|
||||
"path": "README.md",
|
||||
"ref": "main",
|
||||
"repo": "untrusted/repo",
|
||||
}
|
||||
result = await server.get_file(params)
|
||||
|
||||
mock_handler.assert_awaited_once_with(params, server.GITEA_REPO)
|
||||
self.assertIs(result, sentinel_response)
|
||||
|
||||
|
||||
def setUp(self):
|
||||
"""Set up valid parameters for tests."""
|
||||
self.valid_params = {
|
||||
|
|
@ -39,7 +59,7 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
|
|||
params = self.valid_params | {"branch": "main"}
|
||||
with self.assertRaisesRegex(ValueError, "Direct writes to protected branch 'main' are not allowed."):
|
||||
await server.propose_gitea_change(params)
|
||||
|
||||
|
||||
mock_resolve_sha.assert_not_awaited()
|
||||
mock_get_details.assert_not_awaited()
|
||||
mock_create_plan.assert_not_awaited()
|
||||
|
|
@ -85,10 +105,10 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
|
|||
@patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock)
|
||||
async def test_rejects_base_sha_mismatch(self, mock_resolve_sha, mock_get_details, mock_create_plan):
|
||||
mock_resolve_sha.return_value = "c" * 40 # Mismatched SHA
|
||||
|
||||
|
||||
with self.assertRaisesRegex(ValueError, "Branch head does not match the supplied base_sha"):
|
||||
await server.propose_gitea_change(self.valid_params)
|
||||
|
||||
|
||||
mock_resolve_sha.assert_awaited_once()
|
||||
mock_get_details.assert_not_awaited()
|
||||
mock_create_plan.assert_not_awaited()
|
||||
|
|
@ -99,10 +119,10 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
|
|||
async def test_rejects_no_op_diff(self, mock_resolve_sha, mock_get_details, mock_create_plan):
|
||||
mock_resolve_sha.return_value = self.valid_params["base_sha"]
|
||||
mock_get_details.return_value = (self.valid_params["new_content"], "b" * 40)
|
||||
|
||||
|
||||
with self.assertRaisesRegex(ValueError, "Proposed content produces no file change."):
|
||||
await server.propose_gitea_change(self.valid_params)
|
||||
|
||||
|
||||
mock_resolve_sha.assert_awaited_once()
|
||||
mock_get_details.assert_awaited_once()
|
||||
mock_create_plan.assert_not_awaited()
|
||||
|
|
@ -118,7 +138,7 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
|
|||
mock_kms_constructor.return_value.__aexit__ = AsyncMock(return_value=False)
|
||||
|
||||
mock_resolve_sha.return_value = self.valid_params["base_sha"]
|
||||
|
||||
|
||||
with patch.dict(os.environ, {"GITEA_PLAN_KMS_KEY_NAME": KMS_KEY_NAME}):
|
||||
result = await server.propose_gitea_change(self.valid_params)
|
||||
|
||||
|
|
@ -127,13 +147,13 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
|
|||
self.assertEqual(result["branch"], self.valid_params["branch"])
|
||||
self.assertEqual(result["path"], self.valid_params["path"])
|
||||
self.assertIn("approval_subject_hash", result)
|
||||
|
||||
|
||||
expected_hash = hashlib.sha256(self.valid_params["new_content"].encode("utf-8")).hexdigest()
|
||||
self.assertEqual(result["content_hash"], expected_hash)
|
||||
|
||||
|
||||
mock_kms_client.encrypt.assert_awaited_once()
|
||||
mock_create_plan.assert_awaited_once()
|
||||
|
||||
|
||||
# Verify the object passed to create_gitea_change_plan is the real Pydantic model
|
||||
call_args = mock_create_plan.call_args[0][0]
|
||||
self.assertIsInstance(call_args, server.GiteaChangePlan)
|
||||
|
|
@ -344,5 +364,252 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
|
|||
self.assertNotEqual(base_hash, server._calculate_approval_subject_hash(modified_plan_cipher))
|
||||
|
||||
|
||||
class TestIsGiteaChangePlanExpired(unittest.TestCase):
|
||||
def _create_test_plan(self, expires_at):
|
||||
return server.GiteaChangePlan(
|
||||
repo="o/r",
|
||||
branch="b",
|
||||
path="p",
|
||||
base_sha="a" * 40,
|
||||
content_hash="c" * 64,
|
||||
commit_message="m",
|
||||
unified_diff="d",
|
||||
expires_at=expires_at,
|
||||
)
|
||||
|
||||
def test_future_aware_is_not_expired(self):
|
||||
from datetime import datetime, timedelta, timezone
|
||||
future_time = datetime.now(timezone.utc) + timedelta(days=1)
|
||||
plan = self._create_test_plan(future_time)
|
||||
self.assertFalse(server.is_gitea_change_plan_expired(plan))
|
||||
|
||||
def test_past_aware_is_expired(self):
|
||||
from datetime import datetime, timedelta, timezone
|
||||
past_time = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
plan = self._create_test_plan(past_time)
|
||||
self.assertTrue(server.is_gitea_change_plan_expired(plan))
|
||||
|
||||
def test_past_naive_is_expired(self):
|
||||
from datetime import datetime, timedelta
|
||||
past_time_naive = datetime.utcnow() - timedelta(days=1)
|
||||
plan = self._create_test_plan(past_time_naive)
|
||||
self.assertTrue(server.is_gitea_change_plan_expired(plan))
|
||||
|
||||
def test_default_status_is_pending_constant(self):
|
||||
from datetime import datetime, timedelta, timezone
|
||||
future_time = datetime.now(timezone.utc) + timedelta(days=1)
|
||||
plan = self._create_test_plan(future_time)
|
||||
self.assertEqual(
|
||||
plan.status,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_PENDING,
|
||||
)
|
||||
self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_PENDING, "PENDING")
|
||||
self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_APPROVED, "APPROVED")
|
||||
self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_APPLYING, "APPLYING")
|
||||
self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_APPLIED, "APPLIED")
|
||||
self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_REJECTED, "REJECTED")
|
||||
self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_EXPIRED, "EXPIRED")
|
||||
|
||||
|
||||
class TestGiteaChangePlanStatusTransitions(unittest.TestCase):
|
||||
def test_allowed_transitions(self):
|
||||
self.assertTrue(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_PENDING,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
)
|
||||
)
|
||||
self.assertTrue(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_PENDING,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_REJECTED,
|
||||
)
|
||||
)
|
||||
self.assertTrue(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_PENDING,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_EXPIRED,
|
||||
)
|
||||
)
|
||||
self.assertTrue(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPLYING,
|
||||
)
|
||||
)
|
||||
self.assertTrue(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_EXPIRED,
|
||||
)
|
||||
)
|
||||
self.assertTrue(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPLYING,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPLIED,
|
||||
)
|
||||
)
|
||||
|
||||
def test_disallowed_transitions(self):
|
||||
self.assertFalse(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_PENDING,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_PENDING,
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPLIED,
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPLYING,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPLIED,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPLYING,
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_REJECTED,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_EXPIRED,
|
||||
server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
"UNKNOWN", server.GITEA_CHANGE_PLAN_STATUS_PENDING
|
||||
)
|
||||
)
|
||||
self.assertFalse(
|
||||
server.is_valid_gitea_change_plan_status_transition(
|
||||
server.GITEA_CHANGE_PLAN_STATUS_PENDING, "UNKNOWN"
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class TestTransitionGiteaChangePlanStatus(unittest.IsolatedAsyncioTestCase):
|
||||
@patch("google.cloud.firestore.AsyncClient")
|
||||
async def test_rejects_invalid_transition_before_firestore(self, mock_db_client):
|
||||
with self.assertRaisesRegex(
|
||||
ValueError, r"^Invalid Gitea change plan status transition$"
|
||||
):
|
||||
await server.transition_gitea_change_plan_status(
|
||||
plan_id="plan-123",
|
||||
expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING,
|
||||
next_status=server.GITEA_CHANGE_PLAN_STATUS_APPLIED,
|
||||
)
|
||||
mock_db_client.assert_not_called()
|
||||
|
||||
@patch("google.cloud.firestore.AsyncClient")
|
||||
async def test_rejects_status_in_updates_before_firestore(self, mock_db_client):
|
||||
with self.assertRaisesRegex(
|
||||
ValueError, r"^Gitea change plan updates cannot include status$"
|
||||
):
|
||||
await server.transition_gitea_change_plan_status(
|
||||
plan_id="plan-123",
|
||||
expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING,
|
||||
next_status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
updates={"status": server.GITEA_CHANGE_PLAN_STATUS_REJECTED},
|
||||
)
|
||||
mock_db_client.assert_not_called()
|
||||
|
||||
@patch("server.get_gitea_change_plan", new_callable=AsyncMock)
|
||||
@patch("google.cloud.firestore.async_transactional")
|
||||
@patch("google.cloud.firestore.AsyncClient")
|
||||
async def test_transitions_matching_status_and_returns_refetched_plan(
|
||||
self, mock_db_client, mock_transactional, mock_get_plan
|
||||
):
|
||||
def transactional_side_effect(callback):
|
||||
async def wrapped(transaction):
|
||||
return await callback(transaction)
|
||||
return wrapped
|
||||
mock_transactional.side_effect = transactional_side_effect
|
||||
|
||||
mock_transaction = MagicMock()
|
||||
mock_transaction.update = MagicMock()
|
||||
mock_db_client.return_value.transaction.return_value = mock_transaction
|
||||
|
||||
snapshot = MagicMock()
|
||||
snapshot.exists = True
|
||||
snapshot.get.return_value = server.GITEA_CHANGE_PLAN_STATUS_PENDING
|
||||
|
||||
mock_plan_ref = MagicMock()
|
||||
mock_plan_ref.get = AsyncMock(return_value=snapshot)
|
||||
mock_db_client.return_value.collection.return_value.document.return_value = mock_plan_ref
|
||||
|
||||
final_plan = server.GiteaChangePlan(
|
||||
status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
repo="o/r", branch="b", path="p", base_sha="a"*40,
|
||||
content_hash="c"*64, commit_message="m", unified_diff="d",
|
||||
approved_by="admin@example.com"
|
||||
)
|
||||
mock_get_plan.return_value = final_plan
|
||||
|
||||
updates = {"approved_by": "admin@example.com"}
|
||||
result = await server.transition_gitea_change_plan_status(
|
||||
plan_id="plan-123",
|
||||
expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING,
|
||||
next_status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
updates=updates,
|
||||
)
|
||||
|
||||
mock_transaction.update.assert_called_once_with(
|
||||
mock_plan_ref,
|
||||
{
|
||||
"approved_by": "admin@example.com",
|
||||
"status": server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
},
|
||||
)
|
||||
self.assertEqual(updates, {"approved_by": "admin@example.com"})
|
||||
mock_get_plan.assert_awaited_once_with("plan-123")
|
||||
self.assertIs(result, final_plan)
|
||||
|
||||
@patch("server.get_gitea_change_plan", new_callable=AsyncMock)
|
||||
@patch("google.cloud.firestore.async_transactional")
|
||||
@patch("google.cloud.firestore.AsyncClient")
|
||||
async def test_rejects_stale_status_inside_transaction(
|
||||
self, mock_db_client, mock_transactional, mock_get_plan
|
||||
):
|
||||
def transactional_side_effect(callback):
|
||||
async def wrapped(transaction):
|
||||
return await callback(transaction)
|
||||
return wrapped
|
||||
mock_transactional.side_effect = transactional_side_effect
|
||||
|
||||
mock_transaction = MagicMock()
|
||||
mock_transaction.update = MagicMock()
|
||||
mock_db_client.return_value.transaction.return_value = mock_transaction
|
||||
|
||||
snapshot = MagicMock()
|
||||
snapshot.exists = True
|
||||
snapshot.get.return_value = server.GITEA_CHANGE_PLAN_STATUS_APPROVED
|
||||
|
||||
mock_plan_ref = MagicMock()
|
||||
mock_plan_ref.get = AsyncMock(return_value=snapshot)
|
||||
mock_db_client.return_value.collection.return_value.document.return_value = mock_plan_ref
|
||||
|
||||
with self.assertRaisesRegex(ValueError, r"^Gitea change plan status changed$"):
|
||||
await server.transition_gitea_change_plan_status(
|
||||
plan_id="plan-123",
|
||||
expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING,
|
||||
next_status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
|
||||
)
|
||||
|
||||
mock_transaction.update.assert_not_called()
|
||||
mock_get_plan.assert_not_awaited()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user