Files
awoooi/ops/runner/test_guard_gitea_runner_pressure.py
Your Name ba92daa92d
Some checks failed
CD Pipeline / workflow-shape (push) Has been cancelled
CD Pipeline / cancel-stale-cd (push) Has been cancelled
CD Pipeline / tests (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
fix(ci): allow controlled cd push trigger
2026-06-28 19:49:22 +08:00

118 lines
3.1 KiB
Python

from __future__ import annotations
import importlib.util
import sys
from pathlib import Path
def load_guard_module():
module_path = Path(__file__).with_name("guard-gitea-runner-pressure.py")
spec = importlib.util.spec_from_file_location("guard_gitea_runner_pressure", module_path)
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
def write_workflow(root: Path, body: str) -> None:
workflow_dir = root / ".gitea" / "workflows"
workflow_dir.mkdir(parents=True)
(workflow_dir / "cd.yaml").write_text(body, encoding="utf-8")
def write_controlled_service(root: Path) -> None:
service_dir = root / "ops" / "runner"
service_dir.mkdir(parents=True)
(service_dir / "awoooi-cd-lane-drain.service").write_text(
"\n".join(
(
"Environment=AWOOOI_CD_LANE_CONTROLLED=1",
"CPUQuota=300%",
"MemoryMax=10G",
"TasksMax=1024",
"NoNewPrivileges=true",
)
),
encoding="utf-8",
)
def test_blocks_auto_push_to_incident_runner(tmp_path: Path) -> None:
guard = load_guard_module()
write_workflow(
tmp_path,
"""
on:
push:
branches:
- main
jobs:
deploy:
runs-on: awoooi-host
""",
)
_, violations = guard.inspect(tmp_path, tmp_path / ".gitea" / "workflows")
assert [item.reason for item in violations] == [
"auto_branch_event_targets_110_incident_runner"
]
def test_allows_controlled_cd_push_contract(tmp_path: Path) -> None:
guard = load_guard_module()
write_workflow(
tmp_path,
"""
on:
push:
branches:
- main
workflow_dispatch:
env:
HOST_WEB_BUILD_PRESSURE_WARN_ONLY: "1"
DOCKER_BUILD_LOCK_WARN_ONLY: "1"
jobs:
deploy:
if: ${{ github.event_name != 'push' || (!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, 'cancel-stale-cd')) }}
runs-on: awoooi-host
steps:
- run: bash scripts/ci/wait-host-web-build-pressure.sh
""",
)
write_controlled_service(tmp_path)
_, violations = guard.inspect(tmp_path, tmp_path / ".gitea" / "workflows")
assert violations == []
def test_blocks_controlled_cd_push_without_service_contract(tmp_path: Path) -> None:
guard = load_guard_module()
write_workflow(
tmp_path,
"""
on:
push:
branches:
- main
workflow_dispatch:
env:
HOST_WEB_BUILD_PRESSURE_WARN_ONLY: "1"
DOCKER_BUILD_LOCK_WARN_ONLY: "1"
jobs:
deploy:
if: ${{ github.event_name != 'push' || (!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, 'cancel-stale-cd')) }}
runs-on: awoooi-host
steps:
- run: bash scripts/ci/wait-host-web-build-pressure.sh
""",
)
_, violations = guard.inspect(tmp_path, tmp_path / ".gitea" / "workflows")
assert [item.reason for item in violations] == [
"auto_branch_event_targets_110_incident_runner"
]