py/tests/test_korp_wire.py
"""**The korp stubs really are korp's wire, not a fork of it.**
`korp/proto/korp.proto` is the SINGLE SOURCE of the demo's data plane. This
package ships the *generated* stubs (`korp_pb2` / `korp_pb2_grpc`) so a thin
install needs no protoc — but generated stubs are a photograph, and a photograph
goes stale silently.
It already did. Until the commit that added this file, this package also carried a
hand-maintained *fourth copy* of `korp.proto` next to the stubs, and that copy was
**three RPCs behind** the server: `AdhocSql`, `ListGraphs` and `GraphSchema` simply
did not exist on the client side. The hand copy is now gone (regenerate from the
source of truth — see `scripts/regen_protos.py`), and this is the guard that keeps
it gone.
Why a guard and not just a sync: protobuf decodes by NUMBER, not by name. A field
whose number is reused for a different field does not error — it decodes happily
into the WRONG SLOT. That failure is invisible to every other test in this repo,
because every other test asks the client what it thinks it received.
The check reads the checked-in *descriptors* (what actually goes on the wire) and
compares them against the upstream `.proto` TEXT, parsed here by hand. No protoc,
no grpc_tools, no build step — so this can never be the reason the suite is
skipped. When the sibling `korp` checkout is missing the test FAILS, loudly, with
the env var to set: a silent skip is how the drift got three RPCs deep.
"""
from __future__ import annotations
import os
import re
from pathlib import Path
import pytest
from google.protobuf.descriptor import FieldDescriptor
import facett_demo
HERE = Path(__file__).resolve().parent
# The package UNDER TEST — the installed one in the container (`cargo ops test-demo`
# installs the wheel), the source tree only when the source is on sys.path.
PKG_PROTO_DIR = Path(facett_demo.__file__).resolve().parent / "proto"
# The source of truth lives in the sibling `korp` checkout. `KORP_PROTO` overrides
# it for a checkout laid out differently.
# Built by WALKING the parents rather than indexing them. `HERE.parents[2]` raises
# `IndexError: 2` when the package is mounted at `/app` — which is every container
# run — and it raised it at IMPORT time, so this module could not be collected on
# the one road that installs the real wheel. `cargo xtask test-demo` ran a single
# file and read as green. `KORP_PROTO` (the container passes it) still wins.
_KORP_REL = Path("korp") / "proto" / "korp.proto"
_KORP_CANDIDATES = [parent / _KORP_REL for parent in HERE.parents]
def _locate_korp_proto() -> Path:
env = os.environ.get("KORP_PROTO")
if env:
p = Path(env)
if not p.is_file():
pytest.fail(f"KORP_PROTO={env!r} is not a file")
return p
for cand in _KORP_CANDIDATES:
if cand.is_file():
return cand
pytest.fail(
"korp's source-of-truth proto was not found — this guard must not be skipped.\n"
"Looked at:\n " + "\n ".join(str(c) for c in _KORP_CANDIDATES) + "\n"
"Check out the sibling `korp` repo, or point KORP_PROTO at its proto/korp.proto."
)
raise AssertionError("unreachable") # pragma: no cover
# ── a tiny hand-rolled .proto parser (no protoc, no dependency) ────────────────
_SCALARS = {
"double", "float", "int32", "int64", "uint32", "uint64", "sint32", "sint64",
"fixed32", "fixed64", "sfixed32", "sfixed64", "bool", "string", "bytes",
}
def _strip_comments(text: str) -> str:
text = re.sub(r"/\*.*?\*/", "", text, flags=re.S)
return re.sub(r"//[^\n]*", "", text)
def _blocks(text: str, keyword: str) -> dict[str, str]:
"""`{name: body}` for every top-level `<keyword> <Name> { ... }`, brace-balanced."""
out: dict[str, str] = {}
for m in re.finditer(rf"\b{keyword}\s+(\w+)\s*\{{", text):
depth, i = 1, m.end()
while i < len(text) and depth:
if text[i] == "{":
depth += 1
elif text[i] == "}":
depth -= 1
i += 1
out[m.group(1)] = text[m.end() : i - 1]
return out
def _fields(body: str) -> dict[int, tuple[str, str, bool]]:
"""`{number: (name, type, repeated)}` for one message body."""
out: dict[int, tuple[str, str, bool]] = {}
for fm in re.finditer(r"(?:(repeated|optional)\s+)?([\w.]+)\s+(\w+)\s*=\s*(\d+)\s*;", body):
label, ftype, fname, num = fm.groups()
n = int(num)
assert n not in out, f"field number {n} declared twice in the same message"
out[n] = (fname, ftype.rsplit(".", 1)[-1], label == "repeated")
return out
def _rpcs(body: str) -> dict[str, tuple[str, str, bool, bool]]:
"""`{rpc: (request, response, client_streaming, server_streaming)}`."""
out: dict[str, tuple[str, str, bool, bool]] = {}
pat = r"rpc\s+(\w+)\s*\(\s*(stream\s+)?([\w.]+)\s*\)\s*returns\s*\(\s*(stream\s+)?([\w.]+)\s*\)"
for name, cs, req, ss, resp in re.findall(pat, body):
out[name] = (req.rsplit(".", 1)[-1], resp.rsplit(".", 1)[-1], bool(cs), bool(ss))
return out
# ── the checked-in generated descriptors (what really goes on the wire) ────────
_TYPE_NAME = {
FieldDescriptor.TYPE_DOUBLE: "double",
FieldDescriptor.TYPE_FLOAT: "float",
FieldDescriptor.TYPE_INT64: "int64",
FieldDescriptor.TYPE_UINT64: "uint64",
FieldDescriptor.TYPE_INT32: "int32",
FieldDescriptor.TYPE_FIXED64: "fixed64",
FieldDescriptor.TYPE_FIXED32: "fixed32",
FieldDescriptor.TYPE_BOOL: "bool",
FieldDescriptor.TYPE_STRING: "string",
FieldDescriptor.TYPE_BYTES: "bytes",
FieldDescriptor.TYPE_UINT32: "uint32",
FieldDescriptor.TYPE_SFIXED32: "sfixed32",
FieldDescriptor.TYPE_SFIXED64: "sfixed64",
FieldDescriptor.TYPE_SINT32: "sint32",
FieldDescriptor.TYPE_SINT64: "sint64",
}
def _is_repeated(f: FieldDescriptor) -> bool:
# protobuf >= 5.27 exposes `is_repeated`; older runtimes only have `label`.
rep = getattr(f, "is_repeated", None)
return bool(rep) if rep is not None else f.label == 3 # 3 == LABEL_REPEATED
def _descriptor_type(f: FieldDescriptor) -> str:
if f.type in (FieldDescriptor.TYPE_MESSAGE, FieldDescriptor.TYPE_GROUP):
return f.message_type.name
if f.type == FieldDescriptor.TYPE_ENUM:
return f.enum_type.name
return _TYPE_NAME[f.type]
@pytest.fixture(scope="module")
def wire() -> tuple[dict, dict, dict, dict]:
"""`(gen_messages, gen_services, src_messages, src_services)`."""
from facett_demo.proto import korp_pb2
fd = korp_pb2.DESCRIPTOR
gen_msgs = {
name: {
f.number: (f.name, _descriptor_type(f), _is_repeated(f))
for f in md.fields
}
for name, md in fd.message_types_by_name.items()
}
gen_svcs = {
sname: {
m.name: (m.input_type.name, m.output_type.name, m.client_streaming, m.server_streaming)
for m in sd.methods
}
for sname, sd in fd.services_by_name.items()
}
text = _strip_comments(_locate_korp_proto().read_text())
src_msgs = {n: _fields(b) for n, b in _blocks(text, "message").items()}
src_svcs = {n: _rpcs(b) for n, b in _blocks(text, "service").items()}
assert src_msgs, "parsed zero messages out of korp.proto — the parser is broken"
assert src_svcs, "parsed zero services out of korp.proto — the parser is broken"
return gen_msgs, gen_svcs, src_msgs, src_svcs
# ── the guard ─────────────────────────────────────────────────────────────────
def test_package_is_korp(wire):
from facett_demo.proto import korp_pb2
assert korp_pb2.DESCRIPTOR.package == "korp"
assert re.search(r"^package\s+korp;", _locate_korp_proto().read_text(), re.M)
def test_no_hand_copy_of_the_wire_survives():
"""The fourth copy is not allowed back.
A hand-maintained copy of another repo's `.proto` is a drift generator by
construction: it is only ever as fresh as the last person who remembered.
Regenerate from `korp/proto/korp.proto` instead (`scripts/regen_protos.py`).
"""
stray = PKG_PROTO_DIR / "korp.proto"
assert not stray.exists(), (
f"a hand copy of korp's wire reappeared at {stray}.\n"
"Delete it and regenerate the stubs from korp/proto/korp.proto:\n"
" python scripts/regen_protos.py"
)
def test_no_field_number_means_two_different_things(wire):
"""The nastiest drift: one NUMBER, two different fields.
Both sides decode without error and the value lands in the wrong slot. Nothing
else in this repo can see it — every other test asks the client what it thinks
it got.
"""
gen_msgs, _, src_msgs, _ = wire
collisions = []
for name, gen in gen_msgs.items():
src = src_msgs.get(name)
if src is None:
continue
for num, (gname, gtype, grep) in gen.items():
if num not in src:
continue
sname, stype, srep = src[num]
if (gname, gtype, grep) != (sname, stype, srep):
collisions.append(
f"{name}.{num}: stubs say {'repeated ' if grep else ''}{gtype} {gname}"
f" / korp says {'repeated ' if srep else ''}{stype} {sname}"
)
assert not collisions, (
"FIELD NUMBER REUSED FOR A DIFFERENT FIELD — this decodes silently into the "
"wrong slot:\n " + "\n ".join(collisions)
)
def test_every_generated_message_exists_upstream(wire):
gen_msgs, _, src_msgs, _ = wire
orphans = sorted(set(gen_msgs) - set(src_msgs))
assert not orphans, (
f"the stubs carry messages korp does not define: {orphans}\n"
"They are a fork, not a copy. Regenerate: python scripts/regen_protos.py"
)
def test_no_message_lost_a_field(wire):
gen_msgs, _, src_msgs, _ = wire
missing = []
for name, src in src_msgs.items():
gen = gen_msgs.get(name)
if gen is None:
missing.append(f"{name}: MESSAGE MISSING from the stubs")
continue
for num, (fname, ftype, rep) in sorted(src.items()):
if num not in gen:
missing.append(f"{name}.{fname} = {num} ({'repeated ' if rep else ''}{ftype})")
assert not missing, (
"the checked-in stubs are STALE — korp declares fields they do not have:\n "
+ "\n ".join(missing)
+ "\nRegenerate: python scripts/regen_protos.py"
)
def test_service_surface_matches_exactly(wire):
"""RPC-for-RPC, including which direction streams.
The three RPCs the deleted hand copy was missing (`AdhocSql`, `ListGraphs`,
`GraphSchema`) are exactly what this catches.
"""
_, gen_svcs, _, src_svcs = wire
assert set(gen_svcs) == set(src_svcs), (
f"service sets differ — stubs: {sorted(gen_svcs)}, korp: {sorted(src_svcs)}"
)
for sname, src in src_svcs.items():
gen = gen_svcs[sname]
behind = sorted(set(src) - set(gen))
ahead = sorted(set(gen) - set(src))
assert not behind, (
f"service {sname} is BEHIND korp — the stubs cannot call: {behind}\n"
"Regenerate: python scripts/regen_protos.py"
)
assert not ahead, (
f"service {sname} claims RPCs korp does not serve: {ahead}\n"
"Every call to these fails UNIMPLEMENTED at runtime."
)
for rpc, sig in sorted(src.items()):
assert gen[rpc] == sig, (
f"{sname}.{rpc} signature drifted — stubs: {gen[rpc]}, korp: {sig}\n"
"(request, response, client_streaming, server_streaming)"
)
def test_the_grpc_stub_exposes_every_rpc(wire):
"""The pb2 descriptor and the pb2_grpc stub are generated separately.
A regeneration that updates one and not the other leaves a client that can
describe an RPC it cannot call.
"""
from facett_demo.proto import korp_pb2_grpc
_, _, _, src_svcs = wire
for sname, src in src_svcs.items():
stub_cls = getattr(korp_pb2_grpc, f"{sname}Stub", None)
assert stub_cls is not None, f"korp_pb2_grpc has no {sname}Stub"
src_lines = Path(korp_pb2_grpc.__file__).read_text()
for rpc in sorted(src):
assert f"'/{'korp'}.{sname}/{rpc}'" in src_lines, (
f"{sname}Stub cannot call {rpc} — the grpc stub is stale.\n"
"Regenerate: python scripts/regen_protos.py"
)