challenger

check-db-naming.py (5743B)

---

 #!/usr/bin/env python3
 """Enforce the database layer naming convention.
 
 Rules checked:
 
   1. every public function's name starts with the name of the file it lives in
   2. the file's own name uses a prefix from the closed set, or is exempt
   3. iterate_ iff the function takes a caller-supplied callback; get_ iff not
   4. a .h exists beside every .c, and every .sql that is not schema or
      boilerplate is named after the .c that runs it
   5. every prepared statement is named after the function preparing it
   6. no prepared statement name is used in two files
   7. no numeric suffixes and no repeated library prefix
 
 Run from the top of the source tree; exits non-zero on any violation.
 """
 import re
 import sys
 import pathlib
 
 # ---------------------------------------------------------------- CONFIG --
 # (impl dir, header dir, symbol prefix, .c basename prefix)
 LAYERS = [
     ("src/challengerdb", "src/include/challenger-database", "CHALLENGERDB_",
      ""),
 ]
 
 # Prefixes that describe what the caller gets.
 PREFIXES = ("get_", "get_count_", "get_exists_", "iterate_",
             "insert_", "update_", "update_to_", "delete_", "do_")
 
 # Not CRUD: DDL and maintenance.  Keep this list short and justified -- every
 # entry is a rule the checker stops enforcing.  Connection handling
 # (CHALLENGERDB_connect/_connect_admin/_disconnect) lives in pg.c, which is
 # skipped below, and challenger has no explicit transaction control.
 EXEMPT = {
     "create_tables", "drop_tables", "gc", "preflight",
 }
 
 # Files that are not part of the API surface at all: the connection handling,
 # the shared PREPARE macro, the copy-me template, and the dbinit binary.
 SKIP = {"pg", "pg_helper", "template", "challenger-dbinit"}
 
 # SQL that belongs to the schema rather than to one API function.
 SQL_NOT_A_FUNCTION = {"versioning", "drop", "commit", "procedures-prelude"}
 SQL_SCHEMA_RE = re.compile(r"^challenger-\d+$")
 # ------------------------------------------------------------ END CONFIG --
 
 PREP = re.compile(r'PREPARE\s*\(\s*\w+\s*,\s*"([A-Za-z0-9_]+)"', re.S)
 
 errors = []
 
 
 def check_layer(impl_dir, hdr_dir, prefix, fpfx):
     impl, hdr = pathlib.Path(impl_dir), pathlib.Path(hdr_dir)
     if not impl.is_dir():
         return
     decl = re.compile(r"^" + re.escape(prefix) + r"(\w+)\s*\((.*?)\)\s*;",
                       re.S | re.M)
     stmt_owner = {}
     stems = set()
 
     for c in sorted(impl.glob("*.c")):
         if c.stem in SKIP or c.stem.startswith("test_") \
            or c.stem.startswith("plugin_"):
             continue
         if fpfx and not c.stem.startswith(fpfx):
             continue
         stem = c.stem[len(fpfx):] if fpfx else c.stem
         stems.add(stem)
 
         # 2. prefix from the closed set, or exempt
         if stem not in EXEMPT and not stem.startswith(PREFIXES):
             errors.append(f"{c}: '{stem}' uses no approved prefix")
 
         # 7. numeric suffix
         if re.search(r"\d$", stem):
             errors.append(f"{c}: '{stem}' ends in a digit")
 
         # 4. matching header
         h = hdr / (stem + ".h")
         if stem not in EXEMPT and not h.exists():
             errors.append(f"{c}: no matching header {h}")
 
         text = c.read_text(errors="replace")
 
         # 5. prepared statements named after the function
         for s in dict.fromkeys(PREP.findall(text)):
             if s != stem and not s.startswith(stem + "_"):
                 errors.append(
                     f"{c}: prepared statement '{s}' is not named after "
                     f"'{stem}'")
             # 6. cross-file uniqueness
             if s in stmt_owner and stmt_owner[s] != str(c):
                 errors.append(
                     f"{c}: prepared statement '{s}' also prepared in "
                     f"{stmt_owner[s]}")
             stmt_owner[s] = str(c)
 
         if not h.exists():
             continue
         htext = re.sub(r"/\*.*?\*/", "", h.read_text(errors="replace"),
                        flags=re.S)
         for name, params in decl.findall(htext):
             # 1. declared names belong to their file.  Exempt entries are the
             # documented multi-function modules, so this does not apply there.
             if (stem not in EXEMPT
                     and name != stem and not name.startswith(stem + "_")):
                 errors.append(
                     f"{h}: declares '{name}', which does not belong to "
                     f"'{stem}'")
             # 3. iterate_ iff callback
             has_cb = bool(re.search(r"\w*(Callback|Iterator)\s+\w+", params))
             if name.startswith("iterate_") and not has_cb:
                 errors.append(
                     f"{h}: '{name}' is iterate_ but takes no callback")
             if name.startswith("get_") and has_cb:
                 errors.append(f"{h}: '{name}' takes a callback; use iterate_")
 
     # 4. an amalgamated .sql is named after the .c that runs it, so that a
     # renamed function cannot leave its stored procedure behind.
     for s in sorted(impl.glob("*.sql")):
         if s.stem in SQL_NOT_A_FUNCTION or SQL_SCHEMA_RE.match(s.stem):
             continue
         if s.stem not in stems:
             errors.append(f"{s}: no {impl}/{s.stem}.c runs this SQL")
 
     # 7. repeated library prefix anywhere in the tree
     head = prefix.split("_")[0]
     for f in list(impl.glob("*.[ch]")) + list(hdr.glob("*.h")):
         if "unc-backup" in f.name:
             continue
         if re.search(r"\b" + re.escape(head) + r"_" + re.escape(prefix),
                      f.read_text(errors="replace")):
             errors.append(f"{f}: repeated '{prefix}' prefix")
 
 
 for a in LAYERS:
     check_layer(*a)
 
 for e in errors:
     print("ERROR:", e)
 print(f"\ndb-naming: {len(errors)} violation(s)")
 sys.exit(1 if errors else 0)