Files
awoooi/ops/runner/test_read_public_gitea_actions_queue.py
2026-06-29 09:05:28 +08:00

120 lines
4.0 KiB
Python

#!/usr/bin/env python3
from __future__ import annotations
import importlib.util
import json
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
SCRIPT = ROOT / "ops/runner/read-public-gitea-actions-queue.py"
def _load_module():
spec = importlib.util.spec_from_file_location(
"read_public_gitea_actions_queue",
SCRIPT,
)
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 _actions_html() -> str:
return """
<span data-tooltip-content="No matching online runner with label: awoooi-non110-ubuntu">
<span><b>ai-technology-watch.yaml #3857</b>:</span>Scheduled</div>
<span data-tooltip-content="Canceled">
<span><b>ai-technology-watch.yaml #3856</b>:</span>Scheduled</div>
<span data-tooltip-content="Waiting">
<span><b>e2e-health.yaml #3855</b>:</span>Scheduled</div>
"""
def test_parse_visible_runs_extracts_no_matching_runner_label() -> None:
module = _load_module()
runs = module.parse_visible_runs(_actions_html())
assert runs[0]["run_id"] == "3857"
assert runs[0]["workflow"] == "ai-technology-watch.yaml"
assert runs[0]["kind"] == "Scheduled"
assert runs[0]["no_matching_runner_label"] == "awoooi-non110-ubuntu"
assert runs[1]["no_matching_runner_label"] == ""
def test_build_readback_sanitizes_actions_api_internal_url() -> None:
module = _load_module()
payload = module.build_readback(
actions_html=_actions_html(),
actions_list_http_status=401,
actions_list_payload={
"message": "token is required",
"url": "http://192.168.0.110:3001/api/swagger",
},
cd_jobs_http_status=200,
cd_jobs_payload={"jobs": [], "total_count": 0},
)
text = json.dumps(payload, sort_keys=True)
assert payload["schema_version"] == module.SCHEMA_VERSION
assert payload["status"] == "blocked_no_matching_online_runner"
assert payload["readback"]["actions_list_without_token_message"] == (
"token is required"
)
assert payload["readback"]["latest_visible_no_matching_runner_run_id"] == "3857"
assert payload["readback"]["latest_visible_no_matching_runner_label"] == (
"awoooi-non110-ubuntu"
)
assert payload["rollups"]["actions_list_requires_token"] is True
assert payload["operation_boundaries"]["github_api_used"] is False
assert "192.168.0." not in text
def test_cli_json_uses_fixture_files_without_network(tmp_path: Path) -> None:
html_path = tmp_path / "actions.html"
list_path = tmp_path / "actions-list.json"
jobs_path = tmp_path / "jobs.json"
html_path.write_text(_actions_html(), encoding="utf-8")
list_path.write_text(
json.dumps(
{
"message": "token is required",
"url": "http://192.168.0.110:3001/api/swagger",
}
),
encoding="utf-8",
)
jobs_path.write_text(json.dumps({"jobs": [], "total_count": 0}), encoding="utf-8")
result = subprocess.run(
[
sys.executable,
str(SCRIPT),
"--actions-html-file",
str(html_path),
"--actions-list-json-file",
str(list_path),
"--actions-list-http-status",
"401",
"--cd-run-jobs-json-file",
str(jobs_path),
"--cd-run-jobs-http-status",
"200",
"--json",
],
check=False,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
assert result.returncode == 0, result.stdout + result.stderr
payload = json.loads(result.stdout)
assert payload["readback"]["actions_page_visible_run_count"] == 3
assert payload["readback"]["cd_run_jobs_total_count"] == 0
assert payload["readback"]["latest_visible_no_matching_runner_label"] == (
"awoooi-non110-ubuntu"
)
assert "192.168.0." not in result.stdout