117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from aios_chatgpt_bridge_v2 import ( # noqa: E402
|
|
OPERATION_REQUIRED_FIELDS,
|
|
operation_contracts,
|
|
render_openapi_contract_fragment,
|
|
)
|
|
from aios_chatgpt_bridge_v2_server import GovernedRuntime # noqa: E402
|
|
from aios_gitops_job_controller import JobState # noqa: E402
|
|
from gitea_tag_publication import validate_gitea_api_url # noqa: E402
|
|
from release_version import validate_release_version # noqa: E402
|
|
|
|
|
|
def test_canonical_release_version_is_shared_and_accepts_v2r2c1():
|
|
assert validate_release_version("V2R2C1") == "V2R2C1"
|
|
assert validate_release_version("V2R2C1-repair") == "V2R2C1-repair"
|
|
with pytest.raises(ValueError):
|
|
validate_release_version("v2r2c1")
|
|
|
|
|
|
def test_gitea_url_validation_is_startup_safe():
|
|
assert validate_gitea_api_url("https://g.cnnetmkt.top/git/api/v1") == (
|
|
"https://g.cnnetmkt.top/git/api/v1"
|
|
)
|
|
with pytest.raises(ValueError, match="HTTP"):
|
|
validate_gitea_api_url("g.cnnetmkt.top/git/api/v1")
|
|
|
|
|
|
def test_commit_files_runtime_and_published_schema_have_same_required_fields():
|
|
runtime = operation_contracts()["commitFilesGoverned"]["required_fields"]
|
|
assert runtime == list(OPERATION_REQUIRED_FIELDS["commitFilesGoverned"])
|
|
fragment = render_openapi_contract_fragment()
|
|
published = (ROOT / "docs/gitops/openapi-chatgpt-actions.yaml").read_text()
|
|
assert fragment in published
|
|
for field in runtime:
|
|
assert f" - {field}" in fragment
|
|
assert "required: [path, content_base64]" in fragment
|
|
|
|
|
|
class _Store:
|
|
def __init__(self, row):
|
|
self.row = row
|
|
self.receipt_calls = 0
|
|
|
|
def read(self, _job_id):
|
|
return self.row
|
|
|
|
def _receipt(self, row, duplicate=False):
|
|
self.receipt_calls += 1
|
|
return type("Receipt", (), {"as_dict": lambda _self: {
|
|
"job_id": row["job_id"],
|
|
"state": row["state"],
|
|
"duplicate": duplicate,
|
|
}})()
|
|
|
|
|
|
class _Controller:
|
|
def __init__(self, row):
|
|
self.store = _Store(row)
|
|
|
|
|
|
def _runtime_for(row):
|
|
runtime = object.__new__(GovernedRuntime)
|
|
controller = _Controller(row)
|
|
runtime._controller_for_job = lambda _job_id: controller
|
|
runtime._executor_from_row = lambda *_args: (_ for _ in ()).throw(
|
|
AssertionError("terminal reconcile must not reconstruct an executor")
|
|
)
|
|
return runtime, controller
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"state",
|
|
[
|
|
JobState.COMPLETED.value,
|
|
JobState.FAILED.value,
|
|
JobState.CANCELLED.value,
|
|
JobState.MANUAL_OWNER_REVIEW_REQUIRED.value,
|
|
],
|
|
)
|
|
def test_terminal_reconcile_is_idempotent_without_durable_binding(state):
|
|
row = {
|
|
"job_id": "job-terminal",
|
|
"state": state,
|
|
"request_json": json.dumps({}),
|
|
}
|
|
runtime, controller = _runtime_for(row)
|
|
result = GovernedRuntime.job(runtime, {"job_id": row["job_id"]}, "reconcile")
|
|
assert result["duplicate"] is True
|
|
assert controller.store.receipt_calls == 1
|
|
|
|
|
|
def test_active_reconcile_still_uses_executor():
|
|
row = {
|
|
"job_id": "job-active",
|
|
"state": JobState.RECONCILIATION_REQUIRED.value,
|
|
"request_json": json.dumps({}),
|
|
}
|
|
runtime, controller = _runtime_for(row)
|
|
class Executor:
|
|
def reconcile(self, job_id):
|
|
return {"job_id": job_id, "state": "COMPLETED"}
|
|
runtime._executor_from_row = lambda *_args: Executor()
|
|
assert GovernedRuntime.job(runtime, {"job_id": row["job_id"]}, "reconcile") == {
|
|
"job_id": "job-active",
|
|
"state": "COMPLETED",
|
|
}
|