1212Usage:
1313 python .github/scripts/validate_branch_existence.py
1414
15+ # Run built-in unit tests (no network required):
16+ python .github/scripts/validate_branch_existence.py --self-test
17+
1518Exit codes:
1619 0 all checks passed
1720 1 one or more issues found
2225 GITHUB_ACTIONS Set automatically in CI. Enables pr-comment.md output.
2326"""
2427
28+ import contextlib
2529import csv
30+ import io
2631import json
2732import os
2833import re
2934import sys
3035import urllib .error
36+ import urllib .parse
3137import urllib .request
3238from pathlib import Path
3339
@@ -68,7 +74,8 @@ def _api_get(path: str) -> dict | None:
6874
6975def branch_exists (branch : str ) -> bool :
7076 if branch not in _cache :
71- result = _api_get (f"repos/{ GENERATED_DIAGRAMS_REPO } /branches/{ branch } " )
77+ encoded = urllib .parse .quote (branch , safe = "" )
78+ result = _api_get (f"repos/{ GENERATED_DIAGRAMS_REPO } /branches/{ encoded } " )
7279 _cache [branch ] = result is not None
7380 return _cache [branch ]
7481
@@ -81,7 +88,14 @@ def load_versions_csv() -> list[dict]:
8188 return list (csv .DictReader (f ))
8289
8390
84- def run_checks (rows : list [dict ]) -> list [dict ]:
91+ def run_checks (rows : list [dict ], _exists_fn = None ) -> list [dict ]:
92+ """Check each versions.csv row for branch existence and staleness.
93+
94+ _exists_fn is injectable for unit tests; defaults to branch_exists.
95+ """
96+ if _exists_fn is None :
97+ _exists_fn = branch_exists
98+
8599 failures = []
86100 checked : set [str ] = set ()
87101
@@ -95,7 +109,7 @@ def run_checks(rows: list[dict]) -> list[dict]:
95109 if branch not in checked :
96110 checked .add (branch )
97111 print (f" { version :8s} → { branch } ..." , end = " " , flush = True )
98- if branch_exists (branch ):
112+ if _exists_fn (branch ):
99113 print ("OK" )
100114 else :
101115 print ("MISSING" )
@@ -114,7 +128,7 @@ def run_checks(rows: list[dict]) -> list[dict]:
114128 # e.g. v26.2 → release-26.1 when release-26.2 now exists.
115129 expected = f"release-{ version .lstrip ('v' )} "
116130 if branch != expected and expected not in checked :
117- if branch_exists (expected ):
131+ if _exists_fn (expected ):
118132 checked .add (expected )
119133 failures .append ({
120134 "type" : "branch_mismatch" ,
@@ -158,11 +172,58 @@ def format_comment(failures: list[dict]) -> str:
158172 return "\n " .join (lines )
159173
160174
175+ # ---------------------------------------------------------------------------
176+ # Self-tests (no network required)
177+ # ---------------------------------------------------------------------------
178+
179+ def _run_self_tests () -> None :
180+ """Unit tests for run_checks logic using injected exists functions."""
181+
182+ def _quiet (rows , exists_fn ):
183+ with contextlib .redirect_stdout (io .StringIO ()):
184+ return run_checks (rows , _exists_fn = exists_fn )
185+
186+ # branch_missing: listed branch does not exist
187+ rows = [{"major_version" : "v26.1" , "crdb_branch_name" : "release-26.1" }]
188+ failures = _quiet (rows , lambda b : False )
189+ assert len (failures ) == 1 , failures
190+ assert failures [0 ]["type" ] == "branch_missing" , failures
191+
192+ # all OK: branch exists and matches expected
193+ rows = [{"major_version" : "v26.1" , "crdb_branch_name" : "release-26.1" }]
194+ failures = _quiet (rows , lambda b : True )
195+ assert failures == [], failures
196+
197+ # branch_mismatch: listed branch exists but a newer canonical branch also exists
198+ rows = [{"major_version" : "v26.2" , "crdb_branch_name" : "release-26.1" }]
199+ known = {"release-26.1" , "release-26.2" }
200+ failures = _quiet (rows , lambda b : b in known )
201+ assert len (failures ) == 1 , failures
202+ assert failures [0 ]["type" ] == "branch_mismatch" , failures
203+ assert failures [0 ]["expected" ] == "release-26.2" , failures
204+
205+ # N/A entries are skipped entirely
206+ rows = [{"major_version" : "v24.1" , "crdb_branch_name" : "N/A" }]
207+ failures = _quiet (rows , lambda b : (_ for _ in ()).throw (AssertionError ("unexpected call" )))
208+ assert failures == [], failures
209+
210+ # empty branch field is skipped
211+ rows = [{"major_version" : "v25.1" , "crdb_branch_name" : "" }]
212+ failures = _quiet (rows , lambda b : (_ for _ in ()).throw (AssertionError ("unexpected call" )))
213+ assert failures == [], failures
214+
215+ print ("All self-tests passed." )
216+ sys .exit (0 )
217+
218+
161219# ---------------------------------------------------------------------------
162220# Entry point
163221# ---------------------------------------------------------------------------
164222
165223def main () -> None :
224+ if "--self-test" in sys .argv :
225+ _run_self_tests ()
226+
166227 rows = load_versions_csv ()
167228 print (f"Checking { len (rows )} versions.csv entries against cockroachdb/generated-diagrams...\n " )
168229 failures = run_checks (rows )
0 commit comments