Compare commits
No commits in common. "main" and "feat/opax-domain-decouple" have entirely different histories.
main
...
feat/opax-
|
|
@ -56,23 +56,6 @@ 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)
|
## CI1e — ADK-integrasjon (neste steg)
|
||||||
|
|
||||||
Når `opax-mcp` er live, legges det til som ADK-tool i `agents/core-logic/agent.py`:
|
Når `opax-mcp` er live, legges det til som ADK-tool i `agents/core-logic/agent.py`:
|
||||||
|
|
|
||||||
|
|
@ -18,9 +18,6 @@ CLOUD_RUN_SERVICE = os.environ.get("CLOUD_RUN_SERVICE", "osvauco-agent")
|
||||||
MCP_SECRET = os.environ.get("MCP_SECRET", "")
|
MCP_SECRET = os.environ.get("MCP_SECRET", "")
|
||||||
BUILD_TRIGGER_ID = os.environ.get("CLOUD_BUILD_TRIGGER_ID", "38423976-91ff-4ff4-859e-1f262344c609")
|
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="")):
|
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)
|
# Aksepter både X-MCP-Key og api-key (Perplexity MCP connector bruker api-key)
|
||||||
|
|
@ -73,107 +70,6 @@ def _parse_ts(ts):
|
||||||
return None
|
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):
|
class PushStaticRequest(BaseModel):
|
||||||
file_path: str
|
file_path: str
|
||||||
content: str
|
content: str
|
||||||
|
|
|
||||||
|
|
@ -1,202 +0,0 @@
|
||||||
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,97 +34,6 @@ to the existing Cloud Run service.
|
||||||
|
|
||||||
## MCP Deployment Behavior
|
## 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:** `e9b51245a4684f36f0c3bfcea48ce8f910a61650`
|
|
||||||
- **Short commit:** `e9b5124`
|
|
||||||
- **Cloud Build ID:** `1b1fe463-aaf9-427c-b247-d7dab94e46ed`
|
|
||||||
- **Build result:** `SUCCESS`
|
|
||||||
- **Immutable image:** `us-central1-docker.pkg.dev/propane-will-491900-m5/osvauco-repo/opax-mcp@sha256:7dfc5b640c82e152a29641de1ba7c0d6f7435d33bc79fef5598d4a1a805cfd98`
|
|
||||||
- **Cloud Run service:** `opax-mcp`
|
|
||||||
- **Region:** `us-central1`
|
|
||||||
- **Candidate revision:** `opax-mcp-00220-vob`
|
|
||||||
- **Candidate tag:** `candidate-e9b5124`
|
|
||||||
- **Promotion result:** `opax-mcp-00220-vob` now receives 100% traffic.
|
|
||||||
- **Primary service URL:** `https://opax-mcp-zjbqp3prqq-uc.a.run.app`
|
|
||||||
- **Production health result:** `{"status":"ok","service":"opax-mcp","version":"3.6.0"}`
|
|
||||||
- **Deployment validation:** Build succeeded; candidate revision became Ready=True; candidate `/health` passed; promotion succeeded; primary `/health` passed.
|
|
||||||
- **Deferred verification:** Live authenticated MCP `tools/list` verification of `patch_gitea_file` was not completed because the endpoint requires the application-level `MCP_SECRET`. No secret was retrieved, exposed, or used.
|
|
||||||
- **Status:** Promoted successfully on 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
|
## Required OPAX-MCP Runtime Contract
|
||||||
|
|
||||||
- `opax-mcp` uses Direct VPC egress to reach the internal Emma/Ollama runtime.
|
- `opax-mcp` uses Direct VPC egress to reach the internal Emma/Ollama runtime.
|
||||||
|
|
|
||||||
|
|
@ -6,158 +6,6 @@ import re
|
||||||
import base64
|
import base64
|
||||||
import json
|
import json
|
||||||
import binascii
|
import binascii
|
||||||
from typing import List, Literal, Callable, Awaitable, Any, Dict
|
|
||||||
from pydantic import BaseModel, Field, ValidationError
|
|
||||||
|
|
||||||
|
|
||||||
# --- Gitea Change Operation Models & Logic ---
|
|
||||||
|
|
||||||
MAX_GITEA_PATCH_OPERATIONS = 50
|
|
||||||
MAX_GITEA_PATCH_INPUT_BYTES = 131_072
|
|
||||||
|
|
||||||
|
|
||||||
class ExactReplacementOperation(BaseModel, extra="forbid"):
|
|
||||||
"""A single, deterministic replacement operation."""
|
|
||||||
type: Literal["replace_once"]
|
|
||||||
find: str = Field(min_length=1)
|
|
||||||
replace: str
|
|
||||||
|
|
||||||
|
|
||||||
def apply_gitea_patch_operations(
|
|
||||||
original_content: str,
|
|
||||||
operations: List[ExactReplacementOperation],
|
|
||||||
) -> str:
|
|
||||||
"""
|
|
||||||
Applies a sequence of deterministic patch operations to a string.
|
|
||||||
Ensures that each `find` pattern matches exactly once in the content
|
|
||||||
as it evolves through the sequence of operations.
|
|
||||||
"""
|
|
||||||
if not isinstance(original_content, str):
|
|
||||||
raise TypeError("original_content must be a string.")
|
|
||||||
|
|
||||||
if not operations:
|
|
||||||
raise ValueError("Change operations cannot be empty.")
|
|
||||||
|
|
||||||
if len(operations) > MAX_GITEA_PATCH_OPERATIONS:
|
|
||||||
raise ValueError(
|
|
||||||
f"Too many patch operations: {len(operations)} > "
|
|
||||||
f"{MAX_GITEA_PATCH_OPERATIONS}."
|
|
||||||
)
|
|
||||||
|
|
||||||
total_patch_bytes = sum(
|
|
||||||
len(operation.find.encode("utf-8"))
|
|
||||||
+ len(operation.replace.encode("utf-8"))
|
|
||||||
for operation in operations
|
|
||||||
)
|
|
||||||
if total_patch_bytes > MAX_GITEA_PATCH_INPUT_BYTES:
|
|
||||||
raise ValueError(
|
|
||||||
f"Patch input exceeds {MAX_GITEA_PATCH_INPUT_BYTES} bytes."
|
|
||||||
)
|
|
||||||
|
|
||||||
content = original_content
|
|
||||||
for i, op in enumerate(operations):
|
|
||||||
if op.type != "replace_once":
|
|
||||||
raise ValueError(f"Unsupported operation type at index {i}: {op.type}")
|
|
||||||
|
|
||||||
if not op.find:
|
|
||||||
raise ValueError(f"Operation at index {i} has an empty 'find' value.")
|
|
||||||
|
|
||||||
match_count = content.count(op.find)
|
|
||||||
if match_count == 0:
|
|
||||||
raise ValueError(
|
|
||||||
f"Operation at index {i} failed: The 'find' string was not found."
|
|
||||||
)
|
|
||||||
if match_count > 1:
|
|
||||||
raise ValueError(
|
|
||||||
f"Operation at index {i} failed: The 'find' string matched {match_count} times (expected 1)."
|
|
||||||
)
|
|
||||||
|
|
||||||
content = content.replace(op.find, op.replace, 1)
|
|
||||||
|
|
||||||
if content == original_content:
|
|
||||||
raise ValueError("The applied operations resulted in no net change to the file content.")
|
|
||||||
|
|
||||||
return content
|
|
||||||
|
|
||||||
async def orchestrate_gitea_patch(
|
|
||||||
params: Dict[str, Any],
|
|
||||||
read_file_func: Callable[
|
|
||||||
[Dict[str, Any], str],
|
|
||||||
Awaitable[Dict[str, Any]],
|
|
||||||
],
|
|
||||||
write_file_func: Callable[
|
|
||||||
[str, Dict[str, Any]],
|
|
||||||
Awaitable[Dict[str, Any]],
|
|
||||||
],
|
|
||||||
) -> Dict[str, Any]:
|
|
||||||
"""
|
|
||||||
Orchestrates a safe, server-side patch workflow against Gitea.
|
|
||||||
This pure helper contains the core logic and is dependency-injected for testability.
|
|
||||||
"""
|
|
||||||
if not isinstance(params, dict):
|
|
||||||
raise ValueError("patch_gitea_file input must be an object.")
|
|
||||||
|
|
||||||
allowed_keys = {"repo", "branch", "path", "commit_message", "operations"}
|
|
||||||
unsupported_keys = set(params) - allowed_keys
|
|
||||||
if unsupported_keys:
|
|
||||||
raise ValueError(
|
|
||||||
f"Unsupported patch_gitea_file input fields: {sorted(unsupported_keys)}"
|
|
||||||
)
|
|
||||||
|
|
||||||
# 1. Extract and validate all inputs before any network access.
|
|
||||||
repo = params.get("repo")
|
|
||||||
branch = params.get("branch")
|
|
||||||
path = params.get("path")
|
|
||||||
commit_message = params.get("commit_message")
|
|
||||||
operations_data = params.get("operations")
|
|
||||||
|
|
||||||
validate_repo_for_write(repo)
|
|
||||||
validate_branch_for_write(branch)
|
|
||||||
validate_path_for_write(path)
|
|
||||||
|
|
||||||
if not isinstance(commit_message, str) or not commit_message.strip():
|
|
||||||
raise ValueError("commit_message must be a non-empty string.")
|
|
||||||
|
|
||||||
if not isinstance(operations_data, list) or not operations_data:
|
|
||||||
raise ValueError("operations must be a non-empty list.")
|
|
||||||
|
|
||||||
try:
|
|
||||||
parsed_operations = [ExactReplacementOperation(**op) for op in operations_data]
|
|
||||||
except (TypeError, ValidationError) as e:
|
|
||||||
raise ValueError(f"Invalid operation object provided: {e}") from e
|
|
||||||
|
|
||||||
# 2. Read current file state from Gitea.
|
|
||||||
file_state = await read_file_func(
|
|
||||||
{"repo": repo, "path": path, "ref": branch},
|
|
||||||
repo,
|
|
||||||
)
|
|
||||||
original_content = file_state["content"]
|
|
||||||
file_sha = file_state["file_sha"]
|
|
||||||
|
|
||||||
# 3. Apply the patch operations to the fetched content.
|
|
||||||
try:
|
|
||||||
patched_content = apply_gitea_patch_operations(
|
|
||||||
original_content,
|
|
||||||
parsed_operations,
|
|
||||||
)
|
|
||||||
except ValueError as e:
|
|
||||||
raise ValueError(f"Failed to apply patch: {e}") from e
|
|
||||||
|
|
||||||
# 4. Write the update to Gitea.
|
|
||||||
encoded_patched_content = base64.b64encode(patched_content.encode("utf-8")).decode("ascii")
|
|
||||||
put_body = {
|
|
||||||
"branch": branch,
|
|
||||||
"message": commit_message,
|
|
||||||
"content": encoded_patched_content,
|
|
||||||
"sha": file_sha,
|
|
||||||
}
|
|
||||||
|
|
||||||
return await write_file_func(
|
|
||||||
f"/repos/{repo}/contents/{path}",
|
|
||||||
put_body,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
@ -170,10 +18,6 @@ _BRANCH_RE = re.compile(
|
||||||
r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$"
|
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(
|
MAX_SOURCE_ARCHIVE_BYTES = int(
|
||||||
os.environ.get("OPAX_MAX_SOURCE_ARCHIVE_BYTES", "104857600")
|
os.environ.get("OPAX_MAX_SOURCE_ARCHIVE_BYTES", "104857600")
|
||||||
)
|
)
|
||||||
|
|
@ -344,34 +188,6 @@ def validate_path_for_write(path: str) -> str:
|
||||||
return path
|
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(
|
async def resolve_branch_to_commit_sha(
|
||||||
branch_name: str,
|
branch_name: str,
|
||||||
repo_id: str,
|
repo_id: str,
|
||||||
|
|
@ -526,39 +342,10 @@ 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:
|
if caller_repo is not None and caller_repo != validated_server_repo:
|
||||||
raise ValueError("Repository file request is not allowed.")
|
raise ValueError("Repository file request is not allowed.")
|
||||||
|
|
||||||
|
ref = _validate_commit_sha(p.get("ref"))
|
||||||
path = _validate_safe_path(p.get("path"))
|
path = _validate_safe_path(p.get("path"))
|
||||||
|
|
||||||
requested_ref = p.get("ref")
|
url = f"{gitea_url}/api/v1/repos/{validated_server_repo}/contents/{quote(path, safe='')}?ref={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:
|
try:
|
||||||
async with httpx.AsyncClient(timeout=15) as c:
|
async with httpx.AsyncClient(timeout=15) as c:
|
||||||
|
|
@ -611,19 +398,7 @@ async def handle_get_file_content(p: dict, server_repo_id: str) -> dict:
|
||||||
if '\0' in text_content:
|
if '\0' in text_content:
|
||||||
raise ValueError("Repository file content is not readable text.")
|
raise ValueError("Repository file content is not readable text.")
|
||||||
|
|
||||||
file_sha = data.get("sha")
|
return {"path": path, "content": text_content, "encoding": "utf-8"}
|
||||||
if not isinstance(file_sha, str) or not file_sha:
|
|
||||||
raise ValueError("Repository file is unavailable: upstream response missing file SHA.")
|
|
||||||
|
|
||||||
return {
|
|
||||||
"repo": validated_server_repo,
|
|
||||||
"path": path,
|
|
||||||
"requested_ref": requested_ref,
|
|
||||||
"resolved_commit_sha": resolved_commit_sha,
|
|
||||||
"file_sha": _validate_commit_sha(file_sha),
|
|
||||||
"content": text_content,
|
|
||||||
"encoding": "utf-8",
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
async def handle_list_repo_files(p: dict, default_repo: str) -> dict:
|
async def handle_list_repo_files(p: dict, default_repo: str) -> dict:
|
||||||
|
|
|
||||||
|
|
@ -70,7 +70,6 @@ from gitea_handler import (
|
||||||
validate_repo_for_write,
|
validate_repo_for_write,
|
||||||
validate_branch_for_write,
|
validate_branch_for_write,
|
||||||
validate_path_for_write,
|
validate_path_for_write,
|
||||||
orchestrate_gitea_patch,
|
|
||||||
)
|
)
|
||||||
from capability_bridge import build_capability_system_context
|
from capability_bridge import build_capability_system_context
|
||||||
from deployment_policy import get_deployment_target
|
from deployment_policy import get_deployment_target
|
||||||
|
|
@ -101,21 +100,13 @@ class EncryptedPayloadV1(BaseModel):
|
||||||
ciphertext: str
|
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):
|
class GiteaChangePlan(BaseModel):
|
||||||
"""Immutable admin-approved one-file Gitea change plan."""
|
"""Immutable admin-approved one-file Gitea change plan."""
|
||||||
|
|
||||||
plan_id: str = Field(
|
plan_id: str = Field(
|
||||||
default_factory=lambda: f"gitea-change-{uuid.uuid4().hex}"
|
default_factory=lambda: f"gitea-change-{uuid.uuid4().hex}"
|
||||||
)
|
)
|
||||||
status: str = GITEA_CHANGE_PLAN_STATUS_PENDING
|
status: str = "PENDING"
|
||||||
|
|
||||||
created_at: datetime = Field(default_factory=datetime.utcnow)
|
created_at: datetime = Field(default_factory=datetime.utcnow)
|
||||||
expires_at: datetime = Field(
|
expires_at: datetime = Field(
|
||||||
|
|
@ -143,38 +134,6 @@ class GiteaChangePlan(BaseModel):
|
||||||
apply_error_message: Optional[str] = None
|
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:
|
async def create_gitea_change_plan(plan: GiteaChangePlan) -> dict:
|
||||||
"""Persist a pending one-file Gitea change plan in Firestore."""
|
"""Persist a pending one-file Gitea change plan in Firestore."""
|
||||||
from google.cloud import firestore
|
from google.cloud import firestore
|
||||||
|
|
@ -271,42 +230,6 @@ async def get_gitea_change_plan(
|
||||||
return GiteaChangePlan(**document.to_dict())
|
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:
|
def _validate_admin_gitea_path(path: str) -> None:
|
||||||
"""Validate a repository-relative admin file path."""
|
"""Validate a repository-relative admin file path."""
|
||||||
if not isinstance(path, str) or not path or path.isspace():
|
if not isinstance(path, str) or not path or path.isspace():
|
||||||
|
|
@ -1355,15 +1278,6 @@ async def list_repo_files(p: dict) -> dict:
|
||||||
raise ValueError("Unsupported list_repo_files input field.")
|
raise ValueError("Unsupported list_repo_files input field.")
|
||||||
return await handle_list_repo_files({"path": p.get("path")}, GITEA_REPO)
|
return await handle_list_repo_files({"path": p.get("path")}, GITEA_REPO)
|
||||||
async def create_issue(p): return await _gitea_post(f"/repos/{p.get('repo', GITEA_REPO)}/issues", {"title": p.get("title"), "body": p.get("body", "")})
|
async def create_issue(p): return await _gitea_post(f"/repos/{p.get('repo', GITEA_REPO)}/issues", {"title": p.get("title"), "body": p.get("body", "")})
|
||||||
async def patch_gitea_file(p: dict) -> dict:
|
|
||||||
"""Applies a patch to a file in Gitea via the injected orchestration helper."""
|
|
||||||
return await orchestrate_gitea_patch(
|
|
||||||
p,
|
|
||||||
handle_get_file_content,
|
|
||||||
_gitea_put,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
async def push_file(p):
|
async def push_file(p):
|
||||||
# ... (beholdt uendret)
|
# ... (beholdt uendret)
|
||||||
repo, path = p.get("repo", GITEA_REPO), p.get("path")
|
repo, path = p.get("repo", GITEA_REPO), p.get("path")
|
||||||
|
|
@ -1740,41 +1654,6 @@ TOOLS = {
|
||||||
"required": ["repo", "branch", "path", "new_content", "base_sha", "commit_message"]
|
"required": ["repo", "branch", "path", "new_content", "base_sha", "commit_message"]
|
||||||
},
|
},
|
||||||
),
|
),
|
||||||
"patch_gitea_file": (
|
|
||||||
patch_gitea_file,
|
|
||||||
"Applies ordered exact-once replacements to an existing text file and commits the result. Each find value must match exactly once.",
|
|
||||||
{
|
|
||||||
"type": "object",
|
|
||||||
"additionalProperties": False,
|
|
||||||
"properties": {
|
|
||||||
"repo": {"type": "string"},
|
|
||||||
"branch": {"type": "string"},
|
|
||||||
"path": {"type": "string"},
|
|
||||||
"commit_message": {"type": "string"},
|
|
||||||
"operations": {
|
|
||||||
"type": "array",
|
|
||||||
"minItems": 1,
|
|
||||||
"items": {
|
|
||||||
"type": "object",
|
|
||||||
"additionalProperties": False,
|
|
||||||
"properties": {
|
|
||||||
"type": {
|
|
||||||
"type": "string",
|
|
||||||
"enum": ["replace_once"],
|
|
||||||
},
|
|
||||||
"find": {
|
|
||||||
"type": "string",
|
|
||||||
"minLength": 1,
|
|
||||||
},
|
|
||||||
"replace": {"type": "string"},
|
|
||||||
},
|
|
||||||
"required": ["type", "find", "replace"],
|
|
||||||
},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"required": ["repo", "branch", "path", "commit_message", "operations"],
|
|
||||||
},
|
|
||||||
),
|
|
||||||
# Google Workspace
|
# Google Workspace
|
||||||
"create_email_alias": (create_email_alias, "Opprett et nytt e-postalias", {"type":"object","properties":{"user_key":{"type":"string"},"alias":{"type":"string"}},"required":["user_key","alias"]}),
|
"create_email_alias": (create_email_alias, "Opprett et nytt e-postalias", {"type":"object","properties":{"user_key":{"type":"string"},"alias":{"type":"string"}},"required":["user_key","alias"]}),
|
||||||
"list_user_aliases": (list_user_aliases, "List en brukers e-postaliaser", {"type":"object","properties":{"user_key":{"type":"string"}},"required":["user_key"]}),
|
"list_user_aliases": (list_user_aliases, "List en brukers e-postaliaser", {"type":"object","properties":{"user_key":{"type":"string"}},"required":["user_key"]}),
|
||||||
|
|
@ -1865,4 +1744,4 @@ async def mcp_handler(request: Request):
|
||||||
|
|
||||||
@app.get("/health")
|
@app.get("/health")
|
||||||
async def health():
|
async def health():
|
||||||
return {"status": "ok", "service": "opax-mcp", "version": "3.6.0"}
|
return {"status": "ok", "service": "opax-mcp", "version": "3.6.0", "ollama": OLLAMA_BASE_URL} # BUMPED
|
||||||
|
|
|
||||||
|
|
@ -3,73 +3,12 @@ from pathlib import Path
|
||||||
import sys
|
import sys
|
||||||
from unittest.mock import AsyncMock, MagicMock, patch
|
from unittest.mock import AsyncMock, MagicMock, patch
|
||||||
import os
|
import os
|
||||||
import base64
|
|
||||||
import json
|
|
||||||
import httpx
|
import httpx
|
||||||
|
|
||||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||||
sys.path.insert(0, str(REPO_ROOT / "opax-mcp"))
|
sys.path.insert(0, str(REPO_ROOT / "opax-mcp"))
|
||||||
|
|
||||||
from gitea_handler import (
|
from gitea_handler import handle_list_repo_files
|
||||||
handle_list_repo_files,
|
|
||||||
_validate_read_branch_ref,
|
|
||||||
handle_get_file_content,
|
|
||||||
orchestrate_gitea_patch,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
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):
|
class TestGiteaHandler(unittest.TestCase):
|
||||||
|
|
||||||
|
|
@ -401,444 +340,5 @@ class TestGiteaHandler(unittest.TestCase):
|
||||||
import asyncio
|
import asyncio
|
||||||
asyncio.run(run_test())
|
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()
|
|
||||||
|
|
||||||
|
|
||||||
class TestOrchestrateGiteaPatch(unittest.IsolatedAsyncioTestCase):
|
|
||||||
def setUp(self):
|
|
||||||
self.base_params = {
|
|
||||||
"repo": "chris/OSVauco",
|
|
||||||
"branch": "agent/test-patch",
|
|
||||||
"path": "path/to/file.txt",
|
|
||||||
"commit_message": "Patch file",
|
|
||||||
"operations": [{"type": "replace_once", "find": "before", "replace": "after"}],
|
|
||||||
}
|
|
||||||
self.read_func = AsyncMock()
|
|
||||||
self.write_func = AsyncMock()
|
|
||||||
|
|
||||||
async def test_success_flow(self):
|
|
||||||
"""Tests the ideal success path, verifying exact collaborator calls."""
|
|
||||||
self.read_func.return_value = {"content": "before", "file_sha": "f" * 40}
|
|
||||||
self.write_func.return_value = {"commit": {"sha": "c" * 40}}
|
|
||||||
|
|
||||||
result = await orchestrate_gitea_patch(
|
|
||||||
self.base_params, self.read_func, self.write_func
|
|
||||||
)
|
|
||||||
self.assertEqual(result, {"commit": {"sha": "c" * 40}})
|
|
||||||
|
|
||||||
self.read_func.assert_awaited_once_with(
|
|
||||||
{"repo": "chris/OSVauco", "path": "path/to/file.txt", "ref": "agent/test-patch"},
|
|
||||||
"chris/OSVauco",
|
|
||||||
)
|
|
||||||
|
|
||||||
expected_body = {
|
|
||||||
"branch": "agent/test-patch",
|
|
||||||
"message": "Patch file",
|
|
||||||
"content": base64.b64encode(b"after").decode("ascii"),
|
|
||||||
"sha": "f" * 40,
|
|
||||||
}
|
|
||||||
self.write_func.assert_awaited_once_with(
|
|
||||||
"/repos/chris/OSVauco/contents/path/to/file.txt",
|
|
||||||
expected_body,
|
|
||||||
)
|
|
||||||
|
|
||||||
async def test_non_dict_params_rejected_before_io(self):
|
|
||||||
"""Tests that non-dict input is rejected with a precise error."""
|
|
||||||
with self.assertRaisesRegex(
|
|
||||||
ValueError, r"^patch_gitea_file input must be an object\.$"
|
|
||||||
):
|
|
||||||
await orchestrate_gitea_patch("not a dict", self.read_func, self.write_func)
|
|
||||||
self.read_func.assert_not_awaited()
|
|
||||||
self.write_func.assert_not_awaited()
|
|
||||||
|
|
||||||
async def test_unsupported_top_level_field_rejected_before_io(self):
|
|
||||||
"""Tests that unsupported top-level fields are rejected with a precise error."""
|
|
||||||
params = {**self.base_params, "extra": "field"}
|
|
||||||
with self.assertRaisesRegex(
|
|
||||||
ValueError, r"^Unsupported patch_gitea_file input fields: \['extra'\]$"
|
|
||||||
):
|
|
||||||
await orchestrate_gitea_patch(params, self.read_func, self.write_func)
|
|
||||||
self.read_func.assert_not_awaited()
|
|
||||||
self.write_func.assert_not_awaited()
|
|
||||||
|
|
||||||
async def test_field_and_operation_validation_failures(self):
|
|
||||||
"""Tests that invalid required fields and operations fail before any I/O."""
|
|
||||||
test_cases = {
|
|
||||||
"bad_repo": ("repo", "other/repo"),
|
|
||||||
"protected_branch": ("branch", "main"),
|
|
||||||
"invalid_path": ("path", "../secrets"),
|
|
||||||
"empty_commit": ("commit_message", " "),
|
|
||||||
"empty_ops": ("operations", []),
|
|
||||||
"non_list_ops": ("operations", "not-a-list"),
|
|
||||||
"malformed_ops_obj": ("operations", [{"find": "a"}]),
|
|
||||||
"unknown_operation_field": (
|
|
||||||
"operations",
|
|
||||||
[{"type": "replace_once", "find": "a", "replace": "b", "extra": True}],
|
|
||||||
),
|
|
||||||
}
|
|
||||||
for name, (key, value) in test_cases.items():
|
|
||||||
with self.subTest(name=name):
|
|
||||||
bad_params = self.base_params.copy()
|
|
||||||
bad_params[key] = value
|
|
||||||
with self.assertRaises(ValueError):
|
|
||||||
await orchestrate_gitea_patch(bad_params, self.read_func, self.write_func)
|
|
||||||
self.read_func.assert_not_awaited()
|
|
||||||
self.write_func.assert_not_awaited()
|
|
||||||
self.read_func.reset_mock(return_value=True, side_effect=True)
|
|
||||||
self.write_func.reset_mock(return_value=True, side_effect=True)
|
|
||||||
|
|
||||||
async def test_patch_logic_failures_prevent_write(self):
|
|
||||||
"""Tests that patch application failures (no-match, multi-match) prevent writes."""
|
|
||||||
test_cases = {
|
|
||||||
"no_match": "content does not match",
|
|
||||||
"multiple_matches": "before before",
|
|
||||||
}
|
|
||||||
for name, content in test_cases.items():
|
|
||||||
with self.subTest(name=name):
|
|
||||||
self.read_func.return_value = {"content": content, "file_sha": "f" * 40}
|
|
||||||
with self.assertRaisesRegex(ValueError, "Failed to apply patch"):
|
|
||||||
await orchestrate_gitea_patch(
|
|
||||||
self.base_params, self.read_func, self.write_func
|
|
||||||
)
|
|
||||||
self.read_func.assert_awaited_once()
|
|
||||||
self.write_func.assert_not_awaited()
|
|
||||||
self.read_func.reset_mock(return_value=True, side_effect=True)
|
|
||||||
self.write_func.reset_mock(return_value=True, side_effect=True)
|
|
||||||
|
|
||||||
async def test_read_failure_prevents_write(self):
|
|
||||||
"""Tests that a failure in the injected read function prevents writes."""
|
|
||||||
self.read_func.side_effect = ValueError("Upstream read failed")
|
|
||||||
with self.assertRaisesRegex(ValueError, "Upstream read failed"):
|
|
||||||
await orchestrate_gitea_patch(
|
|
||||||
self.base_params, self.read_func, self.write_func
|
|
||||||
)
|
|
||||||
self.read_func.assert_awaited_once()
|
|
||||||
self.write_func.assert_not_awaited()
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|
|
||||||
|
|
@ -297,19 +297,11 @@ class TestGetFileHardening(unittest.IsolatedAsyncioTestCase):
|
||||||
)
|
)
|
||||||
client.stream.assert_not_called()
|
client.stream.assert_not_called()
|
||||||
|
|
||||||
@patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
|
async def test_get_file_ref_validation(self):
|
||||||
async def test_get_file_ref_validation(self, mock_resolver):
|
with self.assertRaisesRegex(ValueError, "Invalid commit SHA"):
|
||||||
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")
|
await gitea_handler.handle_get_file_content({"path": "README.md", "ref": "main"}, "chris/OSVauco")
|
||||||
mock_resolver.assert_awaited_once()
|
with self.assertRaisesRegex(ValueError, "Invalid commit SHA"):
|
||||||
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")
|
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):
|
async def test_gitea_repo_validation(self):
|
||||||
with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"):
|
with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"):
|
||||||
|
|
@ -317,24 +309,17 @@ class TestGetFileHardening(unittest.IsolatedAsyncioTestCase):
|
||||||
with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"):
|
with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"):
|
||||||
await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "")
|
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")
|
@patch("gitea_handler.httpx.AsyncClient")
|
||||||
async def test_get_file_valid_paths(self, mock_client):
|
async def test_get_file_path_validation(self, mock_client):
|
||||||
client = mock_client.return_value.__aenter__.return_value
|
client = mock_client.return_value.__aenter__.return_value
|
||||||
configure_stream(client, json.dumps({'content': 'YQ==', 'size': 1}).encode("utf-8"))
|
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"]:
|
for path in ["README.md", "docs/ARCHITECTURE.md", ".gemini/GEMINI.md"]:
|
||||||
with self.subTest(path=path):
|
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")
|
@patch("gitea_handler.httpx.AsyncClient")
|
||||||
async def test_get_file_size_limit(self, mock_client):
|
async def test_get_file_size_limit(self, mock_client):
|
||||||
|
|
|
||||||
|
|
@ -18,26 +18,6 @@ KMS_KEY_NAME = "projects/p/locations/l/keyRings/k/cryptoKeys/k"
|
||||||
|
|
||||||
class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
|
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):
|
def setUp(self):
|
||||||
"""Set up valid parameters for tests."""
|
"""Set up valid parameters for tests."""
|
||||||
self.valid_params = {
|
self.valid_params = {
|
||||||
|
|
@ -364,252 +344,5 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
|
||||||
self.assertNotEqual(base_hash, server._calculate_approval_subject_hash(modified_plan_cipher))
|
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__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue
Block a user