MIL-001: Stabilise the metadata sync #26

Merged
Tirsvad merged 2 commits from mil-001-stabilise-metadata-sync into main 2026-10-03 14:52:26 +02:00
10 changed files with 555 additions and 220 deletions
@@ -23,10 +23,12 @@ jobs:
python3 - <<'PY' python3 - <<'PY'
import json import json
import os import os
import sys
import urllib.error import urllib.error
import urllib.parse import urllib.parse
import urllib.request import urllib.request
def request_json(url, method="GET", headers=None, body=None): def request_json(url, method="GET", headers=None, body=None):
request = urllib.request.Request( request = urllib.request.Request(
url, url,
@@ -43,7 +45,9 @@ jobs:
f"API request failed with HTTP {error.code} ({error.reason})" f"API request failed with HTTP {error.code} ({error.reason})"
) from None ) from None
raw_credentials = os.environ.get("GITHUB_CREDENTIALS", "").strip()
def parse_github_token(raw_credentials):
raw_credentials = (raw_credentials or "").strip()
if not raw_credentials: if not raw_credentials:
raise RuntimeError("CREDENTIALS_FOR_GITHUB is missing or empty.") raise RuntimeError("CREDENTIALS_FOR_GITHUB is missing or empty.")
@@ -59,54 +63,47 @@ jobs:
"CREDENTIALS_FOR_GITHUB must be a JSON object." "CREDENTIALS_FOR_GITHUB must be a JSON object."
) )
github_token = credentials.get("GITHUB_PAT") token = credentials.get("GITHUB_PAT")
if not isinstance(github_token, str) or not github_token.strip(): if not isinstance(token, str) or not token.strip():
raise RuntimeError( raise RuntimeError(
"CREDENTIALS_FOR_GITHUB must contain a non-empty GITHUB_PAT." "CREDENTIALS_FOR_GITHUB must contain a non-empty GITHUB_PAT."
) )
return token.strip()
raw_gitea_credentials = os.environ.get("GITEA_CREDENTIALS", "").strip()
if not raw_gitea_credentials: def parse_gitea_token(raw_credentials):
raw_credentials = (raw_credentials or "").strip()
if not raw_credentials:
raise RuntimeError("TOKEN_FOR_GITEA is missing or empty.") raise RuntimeError("TOKEN_FOR_GITEA is missing or empty.")
try: try:
gitea_credentials = json.loads(raw_gitea_credentials) credentials = json.loads(raw_credentials)
except json.JSONDecodeError: except json.JSONDecodeError:
gitea_token = raw_gitea_credentials token = raw_credentials
else: else:
if isinstance(gitea_credentials, dict): if isinstance(credentials, dict):
gitea_token = gitea_credentials.get("GITEA_TOKEN") token = credentials.get("GITEA_TOKEN")
elif isinstance(gitea_credentials, str): elif isinstance(credentials, str):
gitea_token = gitea_credentials token = credentials
else: else:
gitea_token = None token = None
if not isinstance(gitea_token, str) or not gitea_token.strip(): if not isinstance(token, str) or not token.strip():
raise RuntimeError( raise RuntimeError(
"TOKEN_FOR_GITEA must contain a non-empty GITEA_TOKEN." "TOKEN_FOR_GITEA must contain a non-empty GITEA_TOKEN."
) )
source_owner, separator, source_repo = os.environ[ return token.strip()
"SOURCE_REPOSITORY"
].partition("/")
if not separator or not source_owner or not source_repo: def split_repository(full_name):
owner, separator, repo = (full_name or "").partition("/")
if not separator or not owner or not repo:
raise RuntimeError("Could not determine the Gitea source repository.") raise RuntimeError("Could not determine the Gitea source repository.")
return owner, repo
gitea_api_url = os.environ["GITEA_API_URL"].rstrip("/")
source_url = f"{gitea_api_url}/repos/{source_owner}/{source_repo}"
gitea_headers = {
"Authorization": f"token {gitea_token.strip()}",
"Accept": "application/json",
}
source = request_json(source_url, headers=gitea_headers)
mirrors = request_json(
f"{source_url}/push_mirrors",
headers=gitea_headers,
)
if not isinstance(mirrors, list):
print("Gitea returned an invalid push mirror list.")
exit 0
github_targets = [] def find_github_targets(mirrors):
targets = []
for mirror in mirrors: for mirror in mirrors:
remote_address = mirror.get("remote_address", "") remote_address = mirror.get("remote_address", "")
if remote_address.startswith("git@github.com:"): if remote_address.startswith("git@github.com:"):
@@ -124,22 +121,49 @@ jobs:
"Could not determine the GitHub owner and repository " "Could not determine the GitHub owner and repository "
"from a configured push mirror." "from a configured push mirror."
) )
github_targets.append(tuple(path_parts)) targets.append(tuple(path_parts))
if len(github_targets) != 1: if len(targets) != 1:
raise RuntimeError( raise RuntimeError(
"Expected exactly one GitHub push mirror for this repository; " "Expected exactly one GitHub push mirror for this repository; "
f"found {len(github_targets)}." f"found {len(targets)}."
)
return targets[0]
def main():
github_token = parse_github_token(os.environ.get("GITHUB_CREDENTIALS"))
gitea_token = parse_gitea_token(os.environ.get("GITEA_CREDENTIALS"))
source_owner, source_repo = split_repository(
os.environ.get("SOURCE_REPOSITORY")
) )
github_owner, github_repo = github_targets[0] gitea_api_url = os.environ["GITEA_API_URL"].rstrip("/")
source_url = f"{gitea_api_url}/repos/{source_owner}/{source_repo}"
gitea_headers = {
"Authorization": f"token {gitea_token}",
"Accept": "application/json",
}
source = request_json(source_url, headers=gitea_headers)
mirrors = request_json(
f"{source_url}/push_mirrors",
headers=gitea_headers,
)
if not isinstance(mirrors, list):
print(
"WARNING: Gitea returned an invalid push mirror list; "
"GitHub metadata was not synced."
)
return 0
github_owner, github_repo = find_github_targets(mirrors)
github_api_url = ( github_api_url = (
"https://api.github.com/repos/" "https://api.github.com/repos/"
f"{urllib.parse.quote(github_owner, safe='')}/" f"{urllib.parse.quote(github_owner, safe='')}/"
f"{urllib.parse.quote(github_repo, safe='')}" f"{urllib.parse.quote(github_repo, safe='')}"
) )
github_headers = { github_headers = {
"Authorization": f"Bearer {github_token.strip()}", "Authorization": f"Bearer {github_token}",
"Accept": "application/vnd.github+json", "Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28", "X-GitHub-Api-Version": "2022-11-28",
"Content-Type": "application/json", "Content-Type": "application/json",
@@ -159,4 +183,13 @@ jobs:
) )
print(f"Synced description and topics to {github_owner}/{github_repo}.") print(f"Synced description and topics to {github_owner}/{github_repo}.")
return 0
if __name__ == "__main__":
try:
sys.exit(main())
except RuntimeError as error:
print(f"ERROR: {error}", file=sys.stderr)
sys.exit(1)
PY PY
+24
View File
@@ -0,0 +1,24 @@
name: Validate workflows
on:
pull_request:
push:
branches: [ main ]
workflow_dispatch:
jobs:
validate:
permissions:
contents: read
runs-on: ubuntu-latest
steps:
- name: Check out the repository
uses: actions/checkout@v4
- name: Create a virtual environment and install test dependencies
run: |
python3 -m venv .venv
.venv/bin/python -m pip install --quiet -r tests/requirements.txt
- name: Validate workflow files and run the offline tests
run: .venv/bin/python -m pytest -q tests
+5
View File
@@ -9,3 +9,8 @@
# .claude/skills/ <- same copy, for the standalone Claude Code CLI # .claude/skills/ <- same copy, for the standalone Claude Code CLI
/.agents/ /.agents/
/.claude/ /.claude/
# Python virtual environment and caches
.venv/
__pycache__/
.pytest_cache/
+14 -3
View File
@@ -43,7 +43,18 @@ Workflows run automatically on push to `main` and on the daily schedule. To run
## 🧪 Tests ## 🧪 Tests
There are no automated tests yet (planned in MIL-001). `tests/test_.ps1` is a manual probe, not a test. Verify changes by running the workflow manually and checking the description and topics on the GitHub mirror. The workflow files and the Python embedded in them are validated by offline tests that need no network access and no secrets:
```bash
python3 -m venv .venv
source .venv/bin/activate # Windows PowerShell: .venv\Scripts\Activate.ps1
python -m pip install -r tests/requirements.txt
python -m pytest -q tests
```
The same checks run on every pull request and push to `main` through [`validate-workflows.yml`](.gitea/workflows/validate-workflows.yml). They parse every workflow as YAML, compile the embedded Python, require explicit job permissions, and test credential parsing, push mirror address parsing and the sync flow against a fake API.
After a change to the sync workflow, also trigger **Sync GitHub mirror metadata** manually in Gitea and compare the description and topics on the GitHub mirror.
## 🟢 Capabilities ## 🟢 Capabilities
@@ -51,7 +62,7 @@ Only the first row is implemented. The rest is planned and tracked in the [Proje
| Capability | Status | | Capability | Status |
| --- | --- | | --- | --- |
| Synchronize repository metadata (Gitea to GitHub mirror) | Implemented in `.gitea/scoped_workflows/`; the script there currently has a syntax error (`exit 0`) that MIL-001 fixes | | Synchronize repository metadata (Gitea to GitHub mirror) | Implemented in `.gitea/scoped_workflows/`, with offline tests and a validation workflow |
| Configure and validate workflow credentials | Partial: secrets are checked for format only | | Configure and validate workflow credentials | Partial: secrets are checked for format only |
| Onboard a repository, manage runners, diagnose failures | Planned (MIL-002) | | Onboard a repository, manage runners, diagnose failures | Planned (MIL-002) |
| Shared quality checks | Planned (MIL-003) | | Shared quality checks | Planned (MIL-003) |
@@ -99,7 +110,7 @@ Workflow -> Gitea: GET /repos/{owner}/{repo}/push_mirrors
Gitea --> Workflow: Push mirror response Gitea --> Workflow: Push mirror response
alt mirrors is not a list alt mirrors is not a list
Workflow --> Trigger: Workflow fails\n"Gitea returned an invalid push mirror list." Workflow --> Trigger: Log a warning and succeed\n(GitHub metadata not synced)
else mirrors is a list else mirrors is a list
Workflow -> Workflow: Find GitHub owner and repo from remote_address Workflow -> Workflow: Find GitHub owner and repo from remote_address
@@ -9,8 +9,8 @@
## Version History ## Version History
| Date | Status | Author | Reviewer | Change | Commit | | Date | Status | Author | Reviewer | Change | Commit |
| --- | --- | --- | --- | --- | --- | | --- | --- | --- | --- | --- | --- |
| 2026-10-03 | Rejected | Jens Tirsvad Nielsen | S01 | Initial version | [4de5438] |
| 2026-10-03 | Accepted | Jens Tirsvad Nielsen | S01 | Owner is S05 (security authority); S01 remains approving reviewer | [4de5438] | | 2026-10-03 | Accepted | Jens Tirsvad Nielsen | S01 | Owner is S05 (security authority); S01 remains approving reviewer | [4de5438] |
| 2026-10-03 | Proposed | Jens Tirsvad Nielsen | S01 | Recorded decisions for tasks 1 and 3 | [cbb9688] |
--- ---
@@ -62,9 +62,9 @@ successful manual run on the Gitea instance.
| # | Task | Summary | Needs its own Use Case/User Story? | Reference | | # | Task | Summary | Needs its own Use Case/User Story? | Reference |
| --- | --- | --- | --- | --- | | --- | --- | --- | --- | --- |
| 1 | Decide behavior for an invalid push mirror response | The committed workflow fails with an error when Gitea returns a non-list push mirror response. The uncommitted scoped copy instead prints a message and tries to exit successfully, and the README diagram showed that. Choose fail or skip, record the decision, and align code and README. | No | | | 1 | Decide behavior for an invalid push mirror response | The committed workflow fails with an error when Gitea returns a non-list push mirror response. The uncommitted scoped copy instead prints a message and tries to exit successfully, and the README diagram showed that. Choose fail or skip, record the decision, and align code and README. Decision: skip with a visible warning and a successful run, following the maintainer's edit to the scoped copy; zero, several or malformed GitHub mirrors still fail. | No | |
| 2 | Fix the Python syntax error in the scoped workflow | `exit 0` is not valid Python, so the whole embedded script fails to compile in the scoped copy. Replace it according to the decision in task 1 (`sys.exit(0)` also needs `import sys`). | No | | | 2 | Fix the Python syntax error in the scoped workflow | `exit 0` is not valid Python, so the whole embedded script fails to compile in the scoped copy. Replace it according to the decision in task 1 (`sys.exit(0)` also needs `import sys`). | No | |
| 3 | Reduce the workflow to one canonical source | The same workflow exists in `src/` and `.gitea/scoped_workflows/` (and a legacy copy that is no longer documented), and the scoped copy has diverged. Choose the canonical location, and generate or remove the others. | No | | | 3 | Reduce the workflow to one canonical source | The same workflow exists in `src/` and `.gitea/scoped_workflows/` (and a legacy copy that is no longer documented), and the scoped copy has diverged. Choose the canonical location, and generate or remove the others. Decision: `.gitea/scoped_workflows/` is canonical because the script must be embedded in the YAML to work as a scoped workflow; the `src/` copy is removed. | No | |
| 4 | Add static validation of workflow files | Parse each workflow YAML and compile its embedded Python in an automated check that runs on pull requests, so defects such as task 2 fail before merge. | No | | | 4 | Add static validation of workflow files | Parse each workflow YAML and compile its embedded Python in an automated check that runs on pull requests, so defects such as task 2 fail before merge. | No | |
| 5 | Add offline unit tests for the sync logic | Move credential and mirror URL parsing into testable code and test bare token, JSON token, SSH and HTTPS mirror addresses, and zero or several mirrors. Replace `tests/test_.ps1`, which probes a hard-coded real repository and reads `.env`. | No | | | 5 | Add offline unit tests for the sync logic | Move credential and mirror URL parsing into testable code and test bare token, JSON token, SSH and HTTPS mirror addresses, and zero or several mirrors. Replace `tests/test_.ps1`, which probes a hard-coded real repository and reads `.env`. | No | |
| 6 | Keep the README accurate | Maintain the capability table (implemented, partial, planned), and align the sequence diagram with the final behavior from task 1. | No | | | 6 | Keep the README accurate | Maintain the capability table (implemented, partial, planned), and align the sequence diagram with the final behavior from task 1. | No | |
@@ -74,3 +74,4 @@ successful manual run on the Gitea instance.
[BC-001]: ../business-case.md [BC-001]: ../business-case.md
[4de5438]: https://git.tirsystem.com/TirSystem/github-action/commit/4de5438a88b4bbf18b165f52f797d1a52af89f6d [4de5438]: https://git.tirsystem.com/TirSystem/github-action/commit/4de5438a88b4bbf18b165f52f797d1a52af89f6d
[cbb9688]: https://git.tirsystem.com/TirSystem/github-action/commit/cbb9688ed4217ee953d5843b846e2e79f22aaf45
+2
View File
@@ -0,0 +1,2 @@
pytest>=8,<10
PyYAML>=6,<7
-52
View File
@@ -1,52 +0,0 @@
# Tests Gitea API access and checks for GitHub mirrors on a repository
$tokenLine = Get-Content .env |
Where-Object { $_ -match '^\s*GITEA_TOKEN\s*=' } |
Select-Object -First 1
if (-not $tokenLine) { throw "GITEA_TOKEN was not found in .env" }
$env:GITEA_TOKEN = ($tokenLine -replace '^\s*GITEA_TOKEN\s*=\s*', '').Trim().Trim('"').Trim("'")
$api = "https://git.tirsystem.com/api/v1"
$repo = "Tirsvad-Udemy-100_days_of_code/001-band_name_generator"
$headers = @(
"Authorization: token $env:GITEA_TOKEN",
"Accept: application/json"
)
foreach ($path in @("/repos/$repo", "/repos/$repo/push_mirrors")) {
$raw = (& curl.exe --silent --show-error --max-time 20 `
--write-out "`n__HTTP_STATUS__%{http_code}" `
--header $headers[0] --header $headers[1] "$api$path" | Out-String)
$marker = "__HTTP_STATUS__"
$index = $raw.LastIndexOf($marker)
if ($index -lt 0) {
Write-Output "No HTTP response for $path. curl exit code: $LASTEXITCODE"
continue
}
$body = $raw.Substring(0, $index).TrimEnd("`r", "`n")
$status = $raw.Substring($index + $marker.Length).Trim()
Write-Output "$path -> HTTP $status"
if ($status -eq "200" -and $path.EndsWith("/push_mirrors")) {
$body | ConvertFrom-Json | ForEach-Object {
$match = [regex]::Match(
[string]$_.remote_address,
'github\.com[:/](?<target>[^?#]+)'
)
if ($match.Success) {
Write-Output ("GitHub mirror: github.com/{0}" -f ($match.Groups["target"].Value -replace '\.git$', ''))
} else {
Write-Output "Mirror found; GitHub target could not be identified."
}
}
} elseif ($status -ne "200") {
try {
$errorBody = $body | ConvertFrom-Json
if ($errorBody.message) { Write-Output $errorBody.message }
} catch {}
}
}
Remove-Item Env:\GITEA_TOKEN
+213
View File
@@ -0,0 +1,213 @@
"""Offline tests for the Python embedded in sync-github-metadata.yml."""
import io
import json
import urllib.error
import pytest
from workflow_source import load_sync_module
@pytest.fixture()
def sync():
return load_sync_module()
class TestGithubToken:
def test_returns_stripped_token_from_json_object(self, sync):
raw = json.dumps({"GITHUB_PAT": " ghp_abc "})
assert sync["parse_github_token"](raw) == "ghp_abc"
@pytest.mark.parametrize("raw", [None, "", " "])
def test_rejects_missing_or_empty_value(self, sync, raw):
with pytest.raises(RuntimeError, match="missing or empty"):
sync["parse_github_token"](raw)
def test_rejects_invalid_json(self, sync):
with pytest.raises(RuntimeError, match="valid JSON"):
sync["parse_github_token"]("ghp_plain_token")
def test_rejects_json_that_is_not_an_object(self, sync):
with pytest.raises(RuntimeError, match="JSON object"):
sync["parse_github_token"]('["x"]')
@pytest.mark.parametrize("value", [{}, {"GITHUB_PAT": ""}, {"GITHUB_PAT": 5}])
def test_rejects_missing_or_invalid_pat(self, sync, value):
with pytest.raises(RuntimeError, match="GITHUB_PAT"):
sync["parse_github_token"](json.dumps(value))
class TestGiteaToken:
def test_accepts_bare_token(self, sync):
assert sync["parse_gitea_token"]("abc123") == "abc123"
def test_accepts_json_object(self, sync):
assert sync["parse_gitea_token"]('{"GITEA_TOKEN": " t "}') == "t"
def test_accepts_json_string(self, sync):
assert sync["parse_gitea_token"]('"quoted"') == "quoted"
@pytest.mark.parametrize("raw", [None, "", " "])
def test_rejects_missing_or_empty_value(self, sync, raw):
with pytest.raises(RuntimeError, match="missing or empty"):
sync["parse_gitea_token"](raw)
@pytest.mark.parametrize("raw", ["{}", '{"GITEA_TOKEN": ""}', "12", "[]"])
def test_rejects_value_without_token(self, sync, raw):
with pytest.raises(RuntimeError, match="GITEA_TOKEN"):
sync["parse_gitea_token"](raw)
class TestSplitRepository:
def test_splits_owner_and_repo(self, sync):
assert sync["split_repository"]("TirSystem/github-action") == (
"TirSystem",
"github-action",
)
@pytest.mark.parametrize("value", [None, "", "owner", "/repo", "owner/"])
def test_rejects_incomplete_names(self, sync, value):
with pytest.raises(RuntimeError, match="source repository"):
sync["split_repository"](value)
class TestFindGithubTargets:
@pytest.mark.parametrize(
"address",
[
"git@github.com:Org/repo.git",
"https://github.com/Org/repo.git",
"https://user:token@github.com/Org/repo",
"ssh://git@github.com/Org/repo.git",
],
)
def test_extracts_owner_and_repo(self, sync, address):
mirrors = [{"remote_address": address}]
assert sync["find_github_targets"](mirrors) == ("Org", "repo")
def test_ignores_non_github_mirrors(self, sync):
mirrors = [
{"remote_address": "https://gitlab.com/Org/other.git"},
{"remote_address": "https://github.com/Org/repo.git"},
]
assert sync["find_github_targets"](mirrors) == ("Org", "repo")
@pytest.mark.parametrize(
"mirrors",
[
[],
[{"remote_address": "https://gitlab.com/Org/other.git"}],
],
)
def test_fails_when_no_github_mirror_exists(self, sync, mirrors):
with pytest.raises(RuntimeError, match="found 0"):
sync["find_github_targets"](mirrors)
def test_fails_when_several_github_mirrors_exist(self, sync):
mirrors = [
{"remote_address": "https://github.com/Org/one.git"},
{"remote_address": "https://github.com/Org/two.git"},
]
with pytest.raises(RuntimeError, match="found 2"):
sync["find_github_targets"](mirrors)
def test_fails_when_path_has_wrong_shape(self, sync):
mirrors = [{"remote_address": "https://github.com/only-owner"}]
with pytest.raises(RuntimeError, match="owner and repository"):
sync["find_github_targets"](mirrors)
class TestRequestJson:
def test_maps_http_error_without_leaking_the_response_body(self, sync, monkeypatch):
def fake_urlopen(request, timeout):
raise urllib.error.HTTPError(
request.full_url, 403, "Forbidden", {}, io.BytesIO(b"secret detail")
)
monkeypatch.setattr(sync["urllib"].request, "urlopen", fake_urlopen)
with pytest.raises(RuntimeError) as caught:
sync["request_json"]("https://example.test/x")
assert str(caught.value) == "API request failed with HTTP 403 (Forbidden)"
class FakeApi:
"""Records calls and answers like the Gitea and GitHub REST APIs."""
def __init__(self, mirrors, source=None):
self.mirrors = mirrors
self.source = source or {"description": "Desc", "topics": ["a", "b"]}
self.calls = []
def __call__(self, url, method="GET", headers=None, body=None):
self.calls.append((method, url, headers, body))
if url.endswith("/push_mirrors"):
return self.mirrors
if url.startswith("https://git.example.test"):
return self.source
return None
@pytest.fixture()
def environment(monkeypatch):
monkeypatch.setenv("GITEA_API_URL", "https://git.example.test/api/v1/")
monkeypatch.setenv("GITEA_CREDENTIALS", "gitea-token")
monkeypatch.setenv("SOURCE_REPOSITORY", "TirSystem/repo")
monkeypatch.setenv("GITHUB_CREDENTIALS", json.dumps({"GITHUB_PAT": "gh-token"}))
class TestMain:
def test_syncs_description_and_topics_to_the_single_github_mirror(
self, sync, environment, capsys
):
api = FakeApi([{"remote_address": "git@github.com:Org/repo.git"}])
sync["request_json"] = api
assert sync["main"]() == 0
requests = [(call[0], call[1]) for call in api.calls]
assert requests == [
("GET", "https://git.example.test/api/v1/repos/TirSystem/repo"),
("GET", "https://git.example.test/api/v1/repos/TirSystem/repo/push_mirrors"),
("PATCH", "https://api.github.com/repos/Org/repo"),
("PUT", "https://api.github.com/repos/Org/repo/topics"),
]
assert api.calls[0][2]["Authorization"] == "token gitea-token"
assert api.calls[2][2]["Authorization"] == "Bearer gh-token"
assert api.calls[2][3] == {"description": "Desc"}
assert api.calls[3][3] == {"names": ["a", "b"]}
assert "Synced description and topics to Org/repo." in capsys.readouterr().out
def test_sends_empty_values_when_gitea_has_none(self, sync, environment):
api = FakeApi(
[{"remote_address": "git@github.com:Org/repo.git"}],
source={"description": None, "topics": None},
)
sync["request_json"] = api
sync["main"]()
assert api.calls[2][3] == {"description": ""}
assert api.calls[3][3] == {"names": []}
def test_skips_with_a_warning_when_mirror_list_is_invalid(
self, sync, environment, capsys
):
api = FakeApi({"message": "unexpected"})
sync["request_json"] = api
assert sync["main"]() == 0
assert all(call[0] == "GET" for call in api.calls)
assert "WARNING" in capsys.readouterr().out
def test_fails_before_any_request_when_credentials_are_missing(
self, sync, environment, monkeypatch
):
monkeypatch.setenv("GITHUB_CREDENTIALS", "")
api = FakeApi([])
sync["request_json"] = api
with pytest.raises(RuntimeError, match="CREDENTIALS_FOR_GITHUB"):
sync["main"]()
assert api.calls == []
+54
View File
@@ -0,0 +1,54 @@
"""Static validation of every workflow file in this repository."""
import ast
import pytest
import yaml
from workflow_source import REPO_ROOT, SYNC_WORKFLOW, embedded_python, workflow_files
WORKFLOWS = workflow_files()
def test_workflow_files_exist():
assert WORKFLOWS, "no workflow files found"
@pytest.mark.parametrize("path", WORKFLOWS, ids=lambda p: p.name)
class TestEachWorkflow:
def test_is_valid_yaml_with_name_and_jobs(self, path):
document = yaml.safe_load(path.read_text(encoding="utf-8"))
assert isinstance(document, dict)
assert document.get("name"), "a workflow needs a name"
assert document.get("jobs"), "a workflow needs at least one job"
def test_embedded_python_compiles(self, path):
for source in embedded_python(path):
ast.parse(source, filename=str(path))
def test_every_job_declares_permissions_and_runner(self, path):
document = yaml.safe_load(path.read_text(encoding="utf-8"))
for name, job in document["jobs"].items():
assert "runs-on" in job, f"job {name} has no runs-on"
assert "permissions" in job, f"job {name} has no explicit permissions"
def test_sync_workflow_embeds_exactly_one_script():
assert len(embedded_python(SYNC_WORKFLOW)) == 1
def test_each_workflow_name_has_one_source():
names = [path.name for path in WORKFLOWS]
assert len(names) == len(set(names)), "a workflow exists in several directories"
def test_no_workflow_copy_outside_the_workflow_directories():
stray = [
path.relative_to(REPO_ROOT)
for pattern in ("*.yml", "*.yaml")
for folder in (REPO_ROOT, REPO_ROOT / "src")
if folder.is_dir()
for path in folder.glob(pattern)
if path.name == SYNC_WORKFLOW.name
]
assert not stray, f"duplicate workflow copies: {stray}"
+44
View File
@@ -0,0 +1,44 @@
"""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