mirror of
https://github.com/Sea-Haven-Industries/seahaven-org-baseline.git
synced 2026-09-30 08:03:19 +00:00
225 lines
7.3 KiB
Python
Executable file
225 lines
7.3 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Flag HCP workspaces whose VCS file triggers omit packaged source paths.
|
|
|
|
Lists workspaces in seahaven-mgmt, seahaven-prod, and seahaven-dev with
|
|
file-triggers-enabled=true. Fails when trigger-prefixes and trigger-patterns
|
|
are both empty and the repo working directory references source trees outside
|
|
itself (Lambda src, functions, lambda, lambdas). PLAT-183.
|
|
|
|
Token: TFE_TOKEN, TF_TOKEN_app_terraform_io, or
|
|
~/.terraform.d/credentials.tfrc.json (app.terraform.io).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ORG = "seahaven"
|
|
PROJECTS = ("seahaven-mgmt", "seahaven-prod", "seahaven-dev")
|
|
API = "https://app.terraform.io/api/v2"
|
|
OUTSIDE_SOURCE = re.compile(
|
|
r"(?:\.\./(?:src|functions|lambda|lambdas)\b)"
|
|
r"|(?:\$\{(?:ROOT|REPO|FUNCS)\}/(?:src|functions|lambda|lambdas)\b)"
|
|
)
|
|
SCAN_SUFFIXES = {".tf", ".sh"}
|
|
|
|
|
|
def load_token() -> str:
|
|
for key in ("TFE_TOKEN", "TF_TOKEN_app_terraform_io"):
|
|
value = os.environ.get(key)
|
|
if value:
|
|
return value
|
|
creds = Path.home() / ".terraform.d" / "credentials.tfrc.json"
|
|
if creds.is_file():
|
|
data = json.loads(creds.read_text())
|
|
token = data.get("credentials", {}).get("app.terraform.io", {}).get("token")
|
|
if token:
|
|
return token
|
|
raise SystemExit(
|
|
"no HCP token: set TFE_TOKEN or configure ~/.terraform.d/credentials.tfrc.json"
|
|
)
|
|
|
|
|
|
def api_get(token: str, path: str) -> dict[str, Any]:
|
|
req = urllib.request.Request(
|
|
API + path,
|
|
headers={
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/vnd.api+json",
|
|
},
|
|
)
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.load(resp)
|
|
|
|
|
|
def paginate(token: str, path: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
|
items: list[dict[str, Any]] = []
|
|
included: list[dict[str, Any]] = []
|
|
while path:
|
|
payload = api_get(token, path)
|
|
items.extend(payload.get("data") or [])
|
|
included.extend(payload.get("included") or [])
|
|
nxt = (payload.get("links") or {}).get("next")
|
|
if not nxt:
|
|
break
|
|
if nxt.startswith(API):
|
|
path = nxt[len(API) :]
|
|
elif "/api/v2" in nxt:
|
|
path = nxt.split("/api/v2", 1)[1]
|
|
else:
|
|
path = nxt
|
|
return items, included
|
|
|
|
|
|
def working_dir_reads_outside(repo_path: Path, working_directory: str) -> list[str]:
|
|
"""Return relative files under the working directory that reference source trees."""
|
|
wd = working_directory.strip().strip("/")
|
|
root = repo_path / wd if wd else repo_path
|
|
if not root.is_dir():
|
|
return []
|
|
hits: list[str] = []
|
|
for path in root.rglob("*"):
|
|
if not path.is_file() or path.suffix not in SCAN_SUFFIXES:
|
|
continue
|
|
if "build" in path.parts:
|
|
continue
|
|
text = path.read_text(errors="replace")
|
|
if OUTSIDE_SOURCE.search(text):
|
|
hits.append(str(path.relative_to(repo_path)))
|
|
return hits
|
|
|
|
|
|
def local_repo_path(repo_root: Path, identifier: str | None) -> Path | None:
|
|
if not identifier or "/" not in identifier:
|
|
return None
|
|
name = identifier.split("/", 1)[1]
|
|
candidate = repo_root / name
|
|
return candidate if candidate.is_dir() else None
|
|
|
|
|
|
def classify_workspace(
|
|
attrs: dict[str, Any],
|
|
project: str,
|
|
repo_root: Path,
|
|
) -> dict[str, Any]:
|
|
vcs = attrs.get("vcs-repo") or {}
|
|
identifier = vcs.get("identifier") if isinstance(vcs, dict) else None
|
|
prefixes = attrs.get("trigger-prefixes") or []
|
|
patterns = attrs.get("trigger-patterns") or []
|
|
wd = attrs.get("working-directory") or ""
|
|
file_triggers = bool(attrs.get("file-triggers-enabled"))
|
|
local = local_repo_path(repo_root, identifier)
|
|
outside: list[str] = []
|
|
inspect = "skipped"
|
|
if local is not None:
|
|
outside = working_dir_reads_outside(local, wd)
|
|
inspect = "ok"
|
|
elif identifier:
|
|
inspect = "missing-clone"
|
|
|
|
empty_triggers = not prefixes and not patterns
|
|
defect = bool(file_triggers and empty_triggers and outside)
|
|
return {
|
|
"name": attrs.get("name"),
|
|
"project": project,
|
|
"file_triggers": file_triggers,
|
|
"working_directory": wd,
|
|
"trigger_prefixes": prefixes,
|
|
"trigger_patterns": patterns,
|
|
"vcs": identifier,
|
|
"inspect": inspect,
|
|
"outside_refs": outside,
|
|
"defect": defect,
|
|
}
|
|
|
|
|
|
def check(
|
|
token: str,
|
|
repo_root: Path,
|
|
org: str = ORG,
|
|
projects: tuple[str, ...] = PROJECTS,
|
|
) -> list[dict[str, Any]]:
|
|
workspaces, included = paginate(
|
|
token, f"/organizations/{org}/workspaces?page%5Bsize%5D=100&include=project"
|
|
)
|
|
project_names = {
|
|
item["id"]: item["attributes"]["name"]
|
|
for item in included
|
|
if item.get("type") == "projects"
|
|
}
|
|
rows: list[dict[str, Any]] = []
|
|
for workspace in workspaces:
|
|
attrs = workspace["attributes"]
|
|
if not attrs.get("file-triggers-enabled"):
|
|
continue
|
|
project_id = (
|
|
(((workspace.get("relationships") or {}).get("project") or {}).get("data") or {}).get(
|
|
"id"
|
|
)
|
|
)
|
|
project = project_names.get(project_id, "")
|
|
if project not in projects:
|
|
continue
|
|
rows.append(classify_workspace(attrs, project, repo_root))
|
|
return rows
|
|
|
|
|
|
def format_report(rows: list[dict[str, Any]]) -> str:
|
|
lines = [
|
|
"workspace project prefixes/patterns outside-refs",
|
|
"-" * 96,
|
|
]
|
|
for row in sorted(rows, key=lambda item: (item["project"], item["name"] or "")):
|
|
prefixes = row["trigger_prefixes"]
|
|
patterns = row["trigger_patterns"]
|
|
trigger = ",".join(prefixes or patterns) or "(empty)"
|
|
mark = "FAIL" if row["defect"] else "ok"
|
|
refs = ",".join(row["outside_refs"][:3]) or row["inspect"]
|
|
lines.append(
|
|
f"{mark:4} {row['name']:36} {row['project']:18} {trigger:18} {refs}"
|
|
)
|
|
defects = [row for row in rows if row["defect"]]
|
|
lines.append("")
|
|
lines.append(f"file-triggered workspaces: {len(rows)} defects: {len(defects)}")
|
|
if defects:
|
|
lines.append(
|
|
"empty trigger-prefixes and trigger-patterns while the working "
|
|
"directory reads source trees outside itself:"
|
|
)
|
|
for row in defects:
|
|
lines.append(f" {row['name']}: {', '.join(row['outside_refs'])}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--repo-root",
|
|
type=Path,
|
|
default=None,
|
|
help="Directory of GitHub clones named after the repo (default: parent of this repo)",
|
|
)
|
|
parser.add_argument("--org", default=ORG)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def default_repo_root() -> Path:
|
|
return Path(__file__).resolve().parents[1].parent
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
repo_root = (args.repo_root or default_repo_root()).resolve()
|
|
rows = check(load_token(), repo_root, org=args.org)
|
|
print(format_report(rows))
|
|
return 1 if any(row["defect"] for row in rows) else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|