#!/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())