389 lines
14 KiB
Python
Executable File
389 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Vendor or test upstream README/Compose snapshots.
|
|
|
|
This script intentionally uses only the Python standard library. It parses the
|
|
limited YAML shape used by docs/upstream-sources.yml instead of requiring PyYAML.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlparse
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
INVENTORY = ROOT / "docs" / "upstream-sources.yml"
|
|
COMPOSE_BLOCK_LANGS = {"", "yaml", "yml", "docker-compose", "compose"}
|
|
|
|
|
|
def parse_inline_list(value: str) -> list[str]:
|
|
value = value.strip()
|
|
if value == "[]":
|
|
return []
|
|
if not (value.startswith("[") and value.endswith("]")):
|
|
return [value.strip().strip("\"'")]
|
|
inner = value[1:-1].strip()
|
|
if not inner:
|
|
return []
|
|
return [item.strip().strip("\"'") for item in inner.split(",")]
|
|
|
|
|
|
def parse_inventory(path: Path) -> dict[str, dict]:
|
|
sources: dict[str, dict] = {}
|
|
current_role: str | None = None
|
|
current_repo: dict | None = None
|
|
in_sources = False
|
|
in_repositories = False
|
|
|
|
for raw_line in path.read_text().splitlines():
|
|
if not raw_line.strip() or raw_line.lstrip().startswith("#"):
|
|
continue
|
|
|
|
if raw_line == "sources:":
|
|
in_sources = True
|
|
continue
|
|
if not in_sources:
|
|
continue
|
|
|
|
role_match = re.match(r"^ ([A-Za-z0-9_]+):$", raw_line)
|
|
if role_match:
|
|
current_role = role_match.group(1)
|
|
sources[current_role] = {"repositories": []}
|
|
current_repo = None
|
|
in_repositories = False
|
|
continue
|
|
|
|
if current_role is None:
|
|
continue
|
|
|
|
role_key = re.match(r"^ ([A-Za-z0-9_]+):\s*(.*)$", raw_line)
|
|
if role_key:
|
|
key, value = role_key.groups()
|
|
if key == "repositories":
|
|
in_repositories = True
|
|
current_repo = None
|
|
continue
|
|
if key == "notes":
|
|
continue
|
|
sources[current_role][key] = value.strip().strip("\"'")
|
|
continue
|
|
|
|
if in_repositories:
|
|
repo_start = re.match(r"^ - purpose:\s*(.*)$", raw_line)
|
|
if repo_start:
|
|
current_repo = {"purpose": repo_start.group(1).strip().strip("\"'")}
|
|
sources[current_role]["repositories"].append(current_repo)
|
|
continue
|
|
|
|
repo_key = re.match(r"^ ([A-Za-z0-9_]+):\s*(.*)$", raw_line)
|
|
if repo_key and current_repo is not None:
|
|
key, value = repo_key.groups()
|
|
value = value.strip()
|
|
if key in {"readme_paths", "compose_paths", "compose_urls"}:
|
|
current_repo[key] = parse_inline_list(value)
|
|
else:
|
|
current_repo[key] = value.strip("\"'")
|
|
|
|
for role, config in sources.items():
|
|
if "vendor_dir" not in config:
|
|
raise ValueError(f"{role}: missing vendor_dir")
|
|
config.setdefault("local_compose_file", "")
|
|
for repo in config["repositories"]:
|
|
if "purpose" not in repo:
|
|
raise ValueError(f"{role}: repository missing purpose")
|
|
repo.setdefault("readme_paths", [])
|
|
repo.setdefault("compose_paths", [])
|
|
repo.setdefault("compose_urls", [])
|
|
needs_clone = bool(repo["readme_paths"] or repo["compose_paths"])
|
|
has_clone_source = "url" in repo or "ref" in repo
|
|
if needs_clone or has_clone_source:
|
|
for key in ("url", "ref"):
|
|
if key not in repo:
|
|
raise ValueError(f"{role}: repository missing {key}")
|
|
|
|
return sources
|
|
|
|
|
|
def run_capture(command: list[str]) -> tuple[bool, str]:
|
|
completed = subprocess.run(
|
|
command,
|
|
cwd=ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
check=False,
|
|
)
|
|
return completed.returncode == 0, completed.stdout
|
|
|
|
|
|
def copy_if_present(src_root: Path, relative_path: str, dest_root: Path, apply: bool) -> str:
|
|
src = src_root / relative_path
|
|
if not src.is_file():
|
|
return "missing"
|
|
if not apply:
|
|
return "found"
|
|
|
|
dest = dest_root / relative_path
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(src, dest)
|
|
return "copied"
|
|
|
|
|
|
def copy_local_file_if_present(relative_path: str, dest_root: Path, apply: bool) -> str:
|
|
src = ROOT / relative_path
|
|
if not src.is_file():
|
|
return "missing"
|
|
if not looks_like_compose(src.read_text(errors="replace")):
|
|
return "invalid"
|
|
if not apply:
|
|
return "found"
|
|
|
|
dest = dest_root / "local" / src.name
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(src, dest)
|
|
return "copied"
|
|
|
|
|
|
def fetch_compose_url(url: str) -> tuple[str, str, str]:
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in {"http", "https"}:
|
|
return "invalid", "", "unsupported URL scheme"
|
|
|
|
try:
|
|
request = Request(url, headers={"User-Agent": "nbops-upstream-snapshots/1.0"})
|
|
with urlopen(request, timeout=30) as response:
|
|
charset = response.headers.get_content_charset() or "utf-8"
|
|
body = response.read().decode(charset, errors="replace")
|
|
except HTTPError as exc:
|
|
return "missing", "", f"HTTP {exc.code}"
|
|
except (OSError, TimeoutError, URLError) as exc:
|
|
return "missing", "", str(exc)
|
|
|
|
if not looks_like_compose(body):
|
|
return "invalid", body, "response is not Compose-shaped YAML"
|
|
|
|
return "found", body, ""
|
|
|
|
|
|
def compose_url_filename(index: int, total: int) -> str:
|
|
if total == 1:
|
|
return "compose.from-url.yaml"
|
|
return f"compose.from-url-{index}.yaml"
|
|
|
|
|
|
def extract_fenced_blocks(markdown: str) -> list[tuple[str, str]]:
|
|
pattern = re.compile(r"^```([^\n`]*)\n(.*?)^```", re.MULTILINE | re.DOTALL)
|
|
blocks = []
|
|
for match in pattern.finditer(markdown):
|
|
lang = match.group(1).strip().split(maxsplit=1)[0].lower() if match.group(1).strip() else ""
|
|
blocks.append((lang, match.group(2)))
|
|
return blocks
|
|
|
|
|
|
def looks_like_compose(text: str) -> bool:
|
|
lines = text.splitlines()
|
|
for index, line in enumerate(lines):
|
|
if re.match(r"^services:\s*(#.*)?$", line):
|
|
return any(re.match(r"^ [A-Za-z0-9_.-]+:\s*(#.*)?$", later) for later in lines[index + 1 :])
|
|
return False
|
|
|
|
|
|
def readme_compose_candidates(readme_path: Path) -> list[dict]:
|
|
if not readme_path.is_file():
|
|
return []
|
|
|
|
candidates = []
|
|
for index, (lang, body) in enumerate(extract_fenced_blocks(readme_path.read_text()), start=1):
|
|
if lang not in COMPOSE_BLOCK_LANGS:
|
|
continue
|
|
if not looks_like_compose(body):
|
|
continue
|
|
candidates.append({"index": index, "lang": lang or "untagged", "body": body})
|
|
return candidates
|
|
|
|
|
|
def print_table(rows: list[list[str]]) -> None:
|
|
headers = ["Role", "Status", "Repos", "README", "Compose", "README parse", "Notes"]
|
|
widths = [len(header) for header in headers]
|
|
for row in rows:
|
|
for index, value in enumerate(row):
|
|
widths[index] = max(widths[index], len(str(value)))
|
|
|
|
def fmt(row: list[str]) -> str:
|
|
return " ".join(str(value).ljust(widths[index]) for index, value in enumerate(row))
|
|
|
|
print(fmt(headers))
|
|
print(" ".join("-" * width for width in widths))
|
|
for row in rows:
|
|
print(fmt(row))
|
|
|
|
|
|
def snapshot_source(vendor_dir: Path, role: str, status: str, notes: list[str], apply: bool) -> None:
|
|
if not apply:
|
|
return
|
|
text = [
|
|
"# Upstream Snapshot",
|
|
"",
|
|
f"Role: `{role}`",
|
|
"Snapshot source inventory: `docs/upstream-sources.yml`",
|
|
f"Snapshot status: `{status}`",
|
|
"",
|
|
"Fetched repositories:",
|
|
*notes,
|
|
"",
|
|
"These files are reference snapshots only. Deployments are generated from role templates.",
|
|
"",
|
|
]
|
|
vendor_dir.mkdir(parents=True, exist_ok=True)
|
|
(vendor_dir / "SOURCE.md").write_text("\n".join(text))
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Test or vendor upstream README/Compose snapshots.")
|
|
parser.add_argument("roles", nargs="*", help="Role names to process. Defaults to all roles.")
|
|
parser.add_argument("--apply", action="store_true", help="Copy fetched files into role upstream directories.")
|
|
parser.add_argument("--no-parse-readme", action="store_true", help="Do not extract Compose YAML from README blocks.")
|
|
parser.add_argument("--list", action="store_true", help="List roles in the upstream inventory and exit.")
|
|
parser.add_argument("--validate-inventory", action="store_true", help="Parse the inventory and exit without cloning.")
|
|
args = parser.parse_args()
|
|
|
|
sources = parse_inventory(INVENTORY)
|
|
|
|
if args.list:
|
|
print("\n".join(sorted(sources)))
|
|
return 0
|
|
|
|
if args.validate_inventory:
|
|
print(f"inventory ok: {len(sources)} roles")
|
|
return 0
|
|
|
|
selected = {role: config for role, config in sources.items() if not args.roles or role in args.roles}
|
|
if not selected:
|
|
print(f"no matching roles found for: {', '.join(args.roles)}", file=sys.stderr)
|
|
return 2
|
|
|
|
rows: list[list[str]] = []
|
|
failures: list[str] = []
|
|
|
|
for role, config in selected.items():
|
|
vendor_dir = ROOT / config["vendor_dir"]
|
|
if args.apply:
|
|
vendor_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
repo_count = 0
|
|
readme_found = readme_missing = 0
|
|
compose_found = compose_missing = 0
|
|
readme_candidates = readme_ambiguous = 0
|
|
notes: list[str] = []
|
|
snapshot_notes: list[str] = []
|
|
|
|
for repo in config["repositories"]:
|
|
repo_count += 1
|
|
with tempfile.TemporaryDirectory(prefix=f"nbops-upstream-{role}-") as tmp:
|
|
tmpdir = Path(tmp)
|
|
ok = False
|
|
if "url" in repo:
|
|
ok, output = run_capture(["git", "clone", "--depth", "1", "--branch", repo["ref"], repo["url"], str(tmpdir)])
|
|
if not ok:
|
|
notes.append(f"{repo['purpose']}: clone failed")
|
|
failures.append(f"{role}/{repo['purpose']}: {output.splitlines()[-1] if output.splitlines() else 'clone failed'}")
|
|
|
|
if ok:
|
|
for path in repo["readme_paths"]:
|
|
result = copy_if_present(tmpdir, path, vendor_dir / repo["purpose"], args.apply)
|
|
if result == "missing":
|
|
readme_missing += 1
|
|
else:
|
|
readme_found += 1
|
|
snapshot_notes.append(f"- {repo['purpose']} README `{path}`: {result}")
|
|
|
|
if args.no_parse_readme:
|
|
continue
|
|
candidates = readme_compose_candidates(tmpdir / path)
|
|
readme_candidates += len(candidates)
|
|
if len(candidates) > 1:
|
|
readme_ambiguous += 1
|
|
if args.apply and len(candidates) == 1:
|
|
dest = vendor_dir / repo["purpose"] / "compose.from-readme.yaml"
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_text(candidates[0]["body"])
|
|
|
|
for path in repo["compose_paths"]:
|
|
result = copy_if_present(tmpdir, path, vendor_dir / repo["purpose"], args.apply)
|
|
if result == "missing":
|
|
compose_missing += 1
|
|
else:
|
|
compose_found += 1
|
|
snapshot_notes.append(f"- {repo['purpose']} Compose `{path}`: {result}")
|
|
|
|
compose_urls = repo["compose_urls"]
|
|
for index, url in enumerate(compose_urls, start=1):
|
|
result, body, reason = fetch_compose_url(url)
|
|
if result == "found":
|
|
compose_found += 1
|
|
if args.apply:
|
|
dest = vendor_dir / repo["purpose"] / compose_url_filename(index, len(compose_urls))
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_text(body)
|
|
result = "copied"
|
|
else:
|
|
compose_missing += 1
|
|
notes.append(f"{repo['purpose']}: compose URL {result}")
|
|
suffix = f" ({reason})" if reason else ""
|
|
snapshot_notes.append(f"- {repo['purpose']} Compose URL `{url}`: {result}{suffix}")
|
|
|
|
if config["local_compose_file"]:
|
|
result = copy_local_file_if_present(config["local_compose_file"], vendor_dir, args.apply)
|
|
if result in {"missing", "invalid"}:
|
|
compose_missing += 1
|
|
notes.append(f"local compose file {result}")
|
|
else:
|
|
compose_found += 1
|
|
snapshot_notes.append(f"- local Compose file `{config['local_compose_file']}`: {result}")
|
|
|
|
if config.get("status") == "verify":
|
|
notes.append("verify source")
|
|
if readme_ambiguous:
|
|
notes.append("ambiguous README compose")
|
|
|
|
status = "error" if any("clone failed" in note for note in notes) else "partial" if readme_missing or compose_missing else "ok"
|
|
snapshot_notes.append(f"README Compose candidates: {readme_candidates}")
|
|
snapshot_notes.append(f"Ambiguous README files: {readme_ambiguous}")
|
|
snapshot_source(vendor_dir, role, config.get("status", "tracked"), snapshot_notes, args.apply)
|
|
|
|
rows.append(
|
|
[
|
|
role,
|
|
status,
|
|
str(repo_count),
|
|
f"{readme_found}/{readme_found + readme_missing}",
|
|
f"{compose_found}/{compose_found + compose_missing}",
|
|
str(readme_candidates),
|
|
"; ".join(notes),
|
|
]
|
|
)
|
|
|
|
print_table(rows)
|
|
|
|
if failures:
|
|
print("\nFailures:", file=sys.stderr)
|
|
for failure in failures:
|
|
print(f"- {failure}", file=sys.stderr)
|
|
return 1
|
|
|
|
if not args.apply:
|
|
print("\nDry run only. Re-run with --apply to copy snapshots into role upstream directories.", file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|