scripts/check-brokers-lock.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Check broker SDK pins against requirements.lock when broker waves activate."""
from __future__ import annotations
import hashlib
import pathlib
import re
import sys
import tomllib
from importlib import metadata
REPO = pathlib.Path(__file__).resolve().parents[1]
BROKERS_LOCK = REPO / "brokers.lock"
REQUIREMENTS_LOCK = REPO / "requirements.lock"
UV_LOCK = REPO / "uv.lock"
ACTIVATED_WAVES = {"dhanhq", "growwapi", "kotakneoapi", "upstox-python-sdk"}
YANKED_VERSIONS = {"dhanhq": {"2.1.0"}}
PIN_RE = re.compile(r"^([a-zA-Z0-9_.-]+)==([^\s;]+)")
HASH_RE = re.compile(r"--hash=sha256:([a-f0-9]{64})")
STALE_PLACEHOLDER_RE = re.compile(r"PLACEHOLDER|hash pending|pending hash", re.IGNORECASE)
KOTAK_GIT = "https://github.com/Kotak-Neo/kotak-neo-python.git"
KOTAK_FIELDS = (
"source_commit",
"source_tree",
"release_version",
"release_tag",
"release_commit",
"release_tree",
"release_wheel_sha256",
"release_sdist_sha256",
"licence_sha256",
)
def _parse_requirements(text: str) -> dict[str, tuple[str, set[str]]]:
pins: dict[str, tuple[str, set[str]]] = {}
current: list[str] = []
blocks: list[str] = []
for line in text.splitlines():
if PIN_RE.match(line):
if current:
blocks.append("\n".join(current))
current = [line]
elif current:
current.append(line)
if current:
blocks.append("\n".join(current))
for block in blocks:
match = PIN_RE.match(block)
if match:
pins[match.group(1).lower()] = (match.group(2), set(HASH_RE.findall(block)))
return pins
def _parse_uv_git_sources(text: str) -> dict[str, tuple[str, str]]:
data = tomllib.loads(text)
sources: dict[str, tuple[str, str]] = {}
for package in data.get("package", []):
if not isinstance(package, dict):
continue
source = package.get("source")
if not isinstance(source, dict):
continue
git = source.get("git")
if isinstance(git, str):
sources[str(package.get("name", "")).lower()] = (str(package.get("version", "")), git)
return sources
def main() -> int:
data = tomllib.loads(BROKERS_LOCK.read_text(encoding="utf-8"))
pins = _parse_requirements(REQUIREMENTS_LOCK.read_text(encoding="utf-8")) if REQUIREMENTS_LOCK.exists() else {}
git_sources = _parse_uv_git_sources(UV_LOCK.read_text(encoding="utf-8")) if UV_LOCK.exists() else {}
failures: list[str] = []
for entry in data.get("broker", []):
name = entry["name"].lower()
if name == "kotakneoapi":
for field in KOTAK_FIELDS:
value = str(entry.get(field, ""))
if not value or STALE_PLACEHOLDER_RE.search(value):
failures.append(f"{name}: missing {field}")
elif field.endswith("_sha256") and not re.fullmatch(r"[a-f0-9]{64}", value):
failures.append(f"{name}: invalid {field}")
elif field.endswith(("_commit", "_tree")) and not re.fullmatch(r"[a-f0-9]{40}", value):
failures.append(f"{name}: invalid {field}")
release_version = entry.get("release_version")
if not isinstance(release_version, str) or re.fullmatch(r"\d+\.\d+\.\d+", release_version) is None:
failures.append(f"{name}: invalid release_version")
if entry.get("release_tag") != f"v{release_version}":
failures.append(f"{name}: release_tag does not match release_version")
if entry.get("sha256") != entry.get("release_wheel_sha256"):
failures.append(f"{name}: sha256 differs from release_wheel_sha256")
if entry.get("licence_source") != f"kotakneoapi-{entry.get('version')}.dist-info/licenses/LICENSE":
failures.append(f"{name}: invalid licence_source")
try:
distribution = metadata.distribution(name)
except metadata.PackageNotFoundError:
distribution = None
if distribution is not None:
licence_file = distribution.locate_file(str(entry.get("licence_source", "")))
if not licence_file.is_file():
failures.append(f"{name}: installed licence file is missing")
elif hashlib.sha256(licence_file.read_bytes()).hexdigest() != entry.get("licence_sha256"):
failures.append(f"{name}: installed licence_sha256 does not match brokers.lock")
if name in pins:
failures.append(f"{name}: Git distribution must be omitted from requirements.lock")
source = git_sources.get(name, ("", ""))[1]
expected = f"{KOTAK_GIT}?rev={entry.get('source_commit')}#{entry.get('source_commit')}"
if source != expected:
failures.append(f"{name}: uv.lock does not pin the official Git source and full commit")
if entry.get("version") in YANKED_VERSIONS.get(name, set()):
failures.append(f"{name}: version {entry['version']} is yanked")
activated = name in ACTIVATED_WAVES
if activated:
for field in ("version", "sha256", "licence", "licence_source", "sandbox_tested", "approved_by"):
value = str(entry.get(field, ""))
if not value or STALE_PLACEHOLDER_RE.search(value):
failures.append(f"{name}: activated wave has placeholder {field}")
for field, raw in entry.items():
if isinstance(raw, str) and STALE_PLACEHOLDER_RE.search(raw):
failures.append(f"{name}: activated wave has stale placeholder text in {field}")
if name in pins and name != "kotakneoapi":
version, hashes = pins[name]
if entry["version"] != version:
failures.append(f"{name}: brokers.lock {entry['version']} != requirements.lock {version}")
sha = entry.get("sha256", "")
if "PLACEHOLDER" not in sha and sha not in hashes:
failures.append(f"{name}: sha256 not present in requirements.lock")
elif name in git_sources:
version, git_source = git_sources[name]
if entry["version"] != version:
failures.append(f"{name}: brokers.lock {entry['version']} != uv.lock {version}")
source_commit = str(entry.get("source_commit", ""))
if not source_commit or source_commit not in git_source:
failures.append(f"{name}: source_commit not present in uv.lock git source")
elif activated:
failures.append(f"{name}: missing from requirements.lock and uv.lock git sources")
for failure in failures:
print(f"FAIL: {failure}", file=sys.stderr)
return 1 if failures else 0
if __name__ == "__main__":
raise SystemExit(main())