305 lines
10 KiB
Python
305 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import re
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from aios_chatgpt_bridge_v2_server import GovernedRuntime
|
|
from aios_continuation_executor import ContinuationExecutor
|
|
from aios_continuation_lease import WriterLeaseRequestV2
|
|
from aios_gitops_job_controller import JobState
|
|
|
|
|
|
TASK = "continuation-normalization-test"
|
|
JOB = "job-continuation-normalization-test"
|
|
REPO = "aios/aios"
|
|
BASE = "a" * 40
|
|
BRANCH = "agent/continuation-normalization-test"
|
|
|
|
|
|
class Authority:
|
|
def __init__(self, head: str | None) -> None:
|
|
self.head = head
|
|
|
|
def read_ref(self, ref: str) -> str | None:
|
|
assert ref == "refs/heads/" + BRANCH
|
|
return self.head
|
|
|
|
def commit_exists(self, oid: str) -> bool:
|
|
return oid == BASE
|
|
|
|
def branch_exists(self, branch: str) -> bool:
|
|
assert branch == BRANCH
|
|
return self.head is not None
|
|
|
|
def branch_head(self, branch: str) -> str:
|
|
assert branch == BRANCH and self.head is not None
|
|
return self.head
|
|
|
|
def is_ancestor(self, older: str, newer: str) -> bool:
|
|
return older == BASE and newer == BASE
|
|
|
|
|
|
def lease(**overrides: object) -> WriterLeaseRequestV2:
|
|
values = {
|
|
"task_id": TASK,
|
|
"job_id": JOB,
|
|
"repository": REPO,
|
|
"branch_mode": "CREATE_NEW",
|
|
"base_oid": BASE,
|
|
"expected_branch_oid": BASE,
|
|
"target_branch": BRANCH,
|
|
"generation": 1,
|
|
"lease_seconds": 120,
|
|
}
|
|
values.update(overrides)
|
|
return WriterLeaseRequestV2(**values)
|
|
|
|
|
|
def runtime(*, head: str | None = BASE, **binding_overrides: object) -> GovernedRuntime:
|
|
value = GovernedRuntime.__new__(GovernedRuntime)
|
|
value.authority = Authority(head)
|
|
binding = {
|
|
"task_id": TASK,
|
|
"job_id": JOB,
|
|
"repository": REPO,
|
|
"base_oid": BASE,
|
|
"expected_branch_oid": BASE,
|
|
"target_branch": BRANCH,
|
|
"generation": 1,
|
|
"branch_mode": "CREATE_NEW",
|
|
}
|
|
binding.update(binding_overrides)
|
|
value.bindings = {TASK: binding}
|
|
return value
|
|
|
|
|
|
def test_same_lineage_create_new_normalizes_to_continue_existing() -> None:
|
|
normalized = runtime()._normalize_runtime_lease(lease())
|
|
assert normalized.branch_mode == "CONTINUE_EXISTING"
|
|
assert normalized.task_id == TASK
|
|
assert normalized.job_id == JOB
|
|
assert normalized.repository == REPO
|
|
assert normalized.target_branch == BRANCH
|
|
assert normalized.expected_branch_oid == BASE
|
|
assert normalized.generation == 1
|
|
|
|
|
|
def test_unprovisioned_create_new_stays_create_new() -> None:
|
|
assert runtime(head=None)._normalize_runtime_lease(lease()).branch_mode == "CREATE_NEW"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value"),
|
|
(
|
|
("repository", "other/repo"),
|
|
("base_oid", "b" * 40),
|
|
("expected_branch_oid", "b" * 40),
|
|
("target_branch", "agent/foreign"),
|
|
("generation", 2),
|
|
),
|
|
)
|
|
def test_binding_mismatch_fails_closed(field: str, value: object) -> None:
|
|
with pytest.raises(RuntimeError, match="LEASE_BINDING_CONFLICT"):
|
|
runtime(**{field: value})._normalize_runtime_lease(lease())
|
|
|
|
|
|
def test_drifted_existing_head_fails_closed() -> None:
|
|
with pytest.raises(RuntimeError, match="EXPECTED_BRANCH_OID_CONFLICT"):
|
|
runtime(head="b" * 40)._normalize_runtime_lease(lease())
|
|
|
|
|
|
def test_openapi_contract_parses_and_requires_exact_commit_lineage() -> None:
|
|
document = yaml.safe_load(
|
|
Path("docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8")
|
|
)
|
|
schema = document["components"]["schemas"]["BridgeCommitResponse"]
|
|
assert {
|
|
"job_id",
|
|
"commit_oid",
|
|
"commit_visible",
|
|
"branch_mode",
|
|
"target_branch",
|
|
"generation",
|
|
} <= set(schema["required"])
|
|
|
|
|
|
def test_changed_production_files_pass_secret_scan() -> None:
|
|
paths = [
|
|
Path("active-runtime/app/aios_authority_git_executor.py"),
|
|
Path("active-runtime/app/aios_chatgpt_bridge_v2_server.py"),
|
|
Path("active-runtime/app/aios_continuation_executor.py"),
|
|
Path("active-runtime/app/aios_continuation_lease.py"),
|
|
Path("active-runtime/app/authorized_task_writer_provisioning.py"),
|
|
Path("active-runtime/app/aios_chatgpt_bridge_v2.py"),
|
|
Path("docs/gitops/openapi-chatgpt-actions.yaml"),
|
|
]
|
|
expression = re.compile(
|
|
r"(?i)(BEGIN (RSA|OPENSSH|EC) PRIVATE KEY|"
|
|
r"authorization:\\s*bearer\\s+[A-Za-z0-9._-]{12,}|"
|
|
r"api[_-]?key\\s*[:=]\\s*['\\\"][^'\\\"]{12,})"
|
|
)
|
|
hits = []
|
|
for path in paths:
|
|
for number, line in enumerate(
|
|
path.read_text(encoding="utf-8", errors="replace").splitlines(), 1
|
|
):
|
|
if expression.search(line):
|
|
hits.append(f"{path}:{number}")
|
|
assert hits == []
|
|
|
|
|
|
def test_remote_verified_job_is_not_completed_when_head_cas_callback_fails() -> None:
|
|
class Store:
|
|
def read(self, _job_id):
|
|
return {
|
|
"branch": BRANCH,
|
|
"request_json": json.dumps(
|
|
{"binding": {"expected_branch_oid": BASE}}
|
|
),
|
|
}
|
|
|
|
class Controller:
|
|
def __init__(self):
|
|
self.store = Store()
|
|
self.states = []
|
|
|
|
def advance(self, _job_id, state, **_kwargs):
|
|
self.states.append(state)
|
|
return SimpleNamespace(as_dict=lambda: {"state": state.value})
|
|
|
|
executor = ContinuationExecutor.__new__(ContinuationExecutor)
|
|
executor.controller = Controller()
|
|
executor._observe_remote = lambda _branch, oid: (oid, oid, True)
|
|
|
|
def fail_cas(_oid):
|
|
raise RuntimeError("LEASE_CAS_CONFLICT")
|
|
|
|
with pytest.raises(RuntimeError, match="LEASE_CAS_CONFLICT"):
|
|
executor._finish_remote_verification(
|
|
JOB,
|
|
"b" * 40,
|
|
"b" * 40,
|
|
before_complete=fail_cas,
|
|
)
|
|
|
|
assert executor.controller.states == [JobState.REMOTE_VERIFIED]
|
|
|
|
|
|
@pytest.mark.parametrize("branch_mode", ("CREATE_NEW", "CONTINUE_EXISTING"))
|
|
def test_executor_preflight_accepts_the_exact_mode_dependent_normalized_shape(
|
|
branch_mode: str,
|
|
) -> None:
|
|
from datetime import datetime, timedelta, timezone
|
|
from aios_continuation_executor import (
|
|
ContinuationPermissions,
|
|
ContinuationRuntimeBinding,
|
|
)
|
|
|
|
request = lease(branch_mode=branch_mode)
|
|
now = datetime.now(timezone.utc)
|
|
executor = ContinuationExecutor.__new__(ContinuationExecutor)
|
|
executor.authority = Authority(None if branch_mode == "CREATE_NEW" else BASE)
|
|
executor.now = lambda: now
|
|
executor._max_generation = lambda _branch, *, exclude_job_id=None: 0
|
|
executor.workspace = SimpleNamespace(
|
|
is_detached=lambda: True,
|
|
head_oid=lambda: BASE,
|
|
)
|
|
executor.remote = SimpleNamespace(repository_identity=lambda: REPO)
|
|
binding = ContinuationRuntimeBinding(
|
|
request=request,
|
|
job_id=JOB,
|
|
workspace_path=Path("/tmp/continuation-normalization-test"),
|
|
lease_not_before=now - timedelta(seconds=1),
|
|
lease_expires_at=now + timedelta(seconds=60),
|
|
original_receipt_valid=True,
|
|
original_grant_consumed=True,
|
|
permissions=ContinuationPermissions(),
|
|
)
|
|
|
|
executor._validate_preflight(binding, job_id=JOB)
|
|
|
|
|
|
def test_create_new_executor_resumes_executing_and_uses_absence_bound_writers() -> None:
|
|
from datetime import datetime, timedelta, timezone
|
|
from aios_continuation_executor import (
|
|
ContinuationCommit,
|
|
ContinuationPermissions,
|
|
ContinuationRuntimeBinding,
|
|
)
|
|
|
|
new_oid = "b" * 40
|
|
|
|
class Store:
|
|
def read(self, _job_id: str) -> dict[str, object]:
|
|
return {"state": "EXECUTING", "revision": 4}
|
|
|
|
class Controller:
|
|
def __init__(self) -> None:
|
|
self.store = Store()
|
|
self.advanced: list[str] = []
|
|
|
|
def advance(self, _job_id: str, state: JobState, **_kwargs: object) -> object:
|
|
self.advanced.append(state.value)
|
|
return SimpleNamespace(as_dict=lambda: {"state": state.value})
|
|
|
|
class Workspace:
|
|
def commit_files(self, _files: object, _message: str) -> str:
|
|
return new_oid
|
|
|
|
class CreateAuthority:
|
|
def __init__(self) -> None:
|
|
self.created: list[tuple[str, str, str]] = []
|
|
|
|
def is_ancestor(self, older: str, newer: str) -> bool:
|
|
return older == BASE and newer == new_oid
|
|
|
|
def create_task_branch_cas(self, branch: str, base: str, new: str) -> str:
|
|
self.created.append((branch, base, new))
|
|
return new
|
|
|
|
def compare_and_swap_branch(self, *_args: object) -> str:
|
|
raise AssertionError("existing-branch CAS must not serve CREATE_NEW")
|
|
|
|
class CreateRemote:
|
|
def __init__(self) -> None:
|
|
self.created: list[tuple[str, str]] = []
|
|
|
|
def push_new_exact(self, branch: str, new: str) -> None:
|
|
self.created.append((branch, new))
|
|
|
|
def push_exact(self, *_args: object) -> None:
|
|
raise AssertionError("existing-branch push must not serve CREATE_NEW")
|
|
|
|
class Credential:
|
|
def project(self) -> None: pass
|
|
def revoke(self) -> None: pass
|
|
|
|
controller=Controller(); authority=CreateAuthority(); remote=CreateRemote()
|
|
executor=ContinuationExecutor(controller,authority,Workspace(),remote,Credential())
|
|
executor._validate_preflight=lambda *_args,**_kwargs: None
|
|
executor._finish_remote_verification=lambda *_args,**_kwargs: {
|
|
"job_id":JOB,"state":"COMPLETED","receipt":{"authority_oid":new_oid,"commit_visible":True}
|
|
}
|
|
now=datetime.now(timezone.utc)
|
|
binding=ContinuationRuntimeBinding(
|
|
request=lease(branch_mode="CREATE_NEW"),job_id=JOB,
|
|
workspace_path=Path("/tmp/create-new-executor-resume"),
|
|
lease_not_before=now-timedelta(seconds=1),lease_expires_at=now+timedelta(seconds=60),
|
|
original_receipt_valid=True,original_grant_consumed=True,
|
|
permissions=ContinuationPermissions(),
|
|
)
|
|
result=executor.execute(
|
|
binding,ContinuationCommit("create-new-test",{"docs/test.txt":b"ok"},"test")
|
|
)
|
|
|
|
assert result["state"]=="COMPLETED"
|
|
assert authority.created==[(BRANCH,BASE,new_oid)]
|
|
assert remote.created==[(BRANCH,new_oid)]
|
|
assert controller.advanced==["LOCAL_VERIFIED","REMOTE_SYNCING"]
|