91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
import base64
|
|
from pathlib import Path
|
|
|
|
from aios_authority_git_executor import AuthorityGitExecutor
|
|
from aios_chatgpt_bridge_v2_server import BridgeHandler, RuntimeConfig
|
|
from aios_continuation_executor import GiteaContinuationRemote, ServerSideCredentialProjection
|
|
from aios_continuation_lease import WriterLeaseRequestV2
|
|
|
|
|
|
class Probe:
|
|
def __init__(self, present=False):
|
|
self.present = present
|
|
|
|
def branch_exists(self, _branch):
|
|
return self.present
|
|
|
|
def branch_head(self, _branch):
|
|
return "a" * 40
|
|
|
|
def is_ancestor(self, _older, _newer):
|
|
return True
|
|
|
|
def commit_exists(self, _oid):
|
|
return True
|
|
|
|
|
|
def test_create_new_lease_is_exact_base_and_absent():
|
|
request = WriterLeaseRequestV2(
|
|
task_id="TASK-P0-BRIDGE-V21-DURABLE-RECONCILIATION-20260809",
|
|
repository="aios/aios",
|
|
branch_mode="CREATE_NEW",
|
|
base_oid="a" * 40,
|
|
expected_branch_oid="a" * 40,
|
|
target_branch="agent/p0-bridge-v21-durable-reconciliation-20260809",
|
|
generation=1,
|
|
)
|
|
request.validate(Probe())
|
|
|
|
|
|
def test_create_new_lease_rejects_existing_target():
|
|
request = WriterLeaseRequestV2(
|
|
task_id="task",
|
|
repository="aios/aios",
|
|
branch_mode="CREATE_NEW",
|
|
base_oid="a" * 40,
|
|
expected_branch_oid="a" * 40,
|
|
target_branch="agent/existing-target",
|
|
generation=1,
|
|
)
|
|
try:
|
|
request.validate(Probe(True))
|
|
except Exception as exc:
|
|
assert str(exc) == "TARGET_BRANCH_ALREADY_EXISTS"
|
|
else:
|
|
raise AssertionError("existing CREATE_NEW target must fail")
|
|
|
|
|
|
def test_slash_branch_uses_paginated_collection():
|
|
credential = ServerSideCredentialProjection(Path("/tmp/key"), Path("/tmp/token"), active=True)
|
|
calls = []
|
|
|
|
class FakeRemote(GiteaContinuationRemote):
|
|
def _api(self, path):
|
|
calls.append(path)
|
|
if "page=1" in path:
|
|
return [{"name": f"agent/other-{i}", "commit": {"id": "b" * 40}} for i in range(50)]
|
|
return [{"name": "agent/slash/branch", "commit": {"id": "c" * 40}}]
|
|
|
|
remote = FakeRemote(Path("/tmp"), credential)
|
|
assert remote.read_api_branch("agent/slash/branch") == "c" * 40
|
|
assert calls[0].endswith("branches?page=1&limit=50")
|
|
assert calls[1].endswith("branches?page=2&limit=50")
|
|
|
|
|
|
def test_http_route_contract_is_complete():
|
|
expected = {
|
|
"/v2/runtime", "/v2/capabilities", "/v2/governed-task",
|
|
"/v2/writer-lease", "/v2/commit-file", "/v2/commit-files",
|
|
"/v2/job", "/v2/job/resume", "/v2/job/reconcile",
|
|
"/v2/tag/approval-request",
|
|
"/v2/tag/publish",
|
|
"/v2/tag/status",
|
|
"/v2/release/artifact",
|
|
"/v2/release/execute",
|
|
"/v2/release/consistency",
|
|
"/v2/release/acceptance",
|
|
}
|
|
from aios_chatgpt_bridge_v2_server import ROUTES
|
|
|
|
assert set(ROUTES) == expected
|