Restructure the script embedded in the scoped sync workflow into testable functions, fix the invalid `exit 0` (now a successful run with a visible warning when Gitea returns an invalid push mirror list) and report errors as `ERROR: <message>` with exit code 1. Zero, several or malformed GitHub mirrors still fail. Add pytest-based offline tests (workflow YAML, embedded Python, credential and mirror parsing, sync flow against a fake API) that run in a virtual environment, and a validation workflow that runs them on pull requests and pushes to main. Remove the manual tests/test_.ps1 probe and the duplicate src/ copy of the workflow. Update the README and record the decisions for tasks 1 and 3 in MIL-001. Task: MIL-001#1 Task: MIL-001#2 Task: MIL-001#3 Task: MIL-001#4 Task: MIL-001#5 Task: MIL-001#6 Refs #1 Refs #2 Refs #3 Refs #4 Refs #5 Refs #6 Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
"""Helpers that read workflow files and the Python embedded in them."""
|
|
|
|
import re
|
|
import textwrap
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
WORKFLOW_DIRS = (
|
|
REPO_ROOT / ".gitea" / "scoped_workflows",
|
|
REPO_ROOT / ".gitea" / "workflows",
|
|
)
|
|
SYNC_WORKFLOW = WORKFLOW_DIRS[0] / "sync-github-metadata.yml"
|
|
|
|
_HEREDOC = re.compile(
|
|
r"^[ \t]*python3 - <<'PY'\r?\n(?P<code>.*?)^[ \t]*PY[ \t]*\r?$",
|
|
re.DOTALL | re.MULTILINE,
|
|
)
|
|
|
|
|
|
def workflow_files() -> list[Path]:
|
|
"""Return every workflow file in the known workflow directories."""
|
|
files: list[Path] = []
|
|
for directory in WORKFLOW_DIRS:
|
|
if directory.is_dir():
|
|
files += sorted(directory.glob("*.yml")) + sorted(directory.glob("*.yaml"))
|
|
return files
|
|
|
|
|
|
def embedded_python(path: Path) -> list[str]:
|
|
"""Return the source of each `python3 - <<'PY'` heredoc in a workflow."""
|
|
text = path.read_text(encoding="utf-8")
|
|
return [
|
|
textwrap.dedent(match.group("code").replace("\r\n", "\n"))
|
|
for match in _HEREDOC.finditer(text)
|
|
]
|
|
|
|
|
|
def load_sync_module(name: str = "embedded_sync") -> dict[str, Any]:
|
|
"""Execute the embedded sync script without running `main`."""
|
|
(source,) = embedded_python(SYNC_WORKFLOW)
|
|
namespace: dict[str, Any] = {"__name__": name}
|
|
exec(compile(source, str(SYNC_WORKFLOW), "exec"), namespace)
|
|
return namespace
|