Files
GaCy-GO/tools/audit_protocol_handlers.py
T

71 lines
3.6 KiB
Python

"""Compare command IDs and message field numbers with a local PearlSR checkout.
This is a source audit, not a client compatibility test. It does not verify
field types, default values, assets, or gameplay behavior.
"""
import argparse
import pathlib
import re
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("upstream", type=pathlib.Path)
args = parser.parse_args()
root = pathlib.Path(__file__).resolve().parents[1]
upstream = args.upstream / "protocol/src"
command_source = (upstream / "root.zig").read_text(encoding="utf-8")
commands = {n: int(v) for n, v in re.findall(r"\bCmd(\w+)\s*=\s*(\d+)", command_source)}
go_commands = {}
command_conflicts = []
for filename in ("cmdid.go", "upstream_cmdids.go"):
source = (root / "protocol/proto" / filename).read_text(encoding="utf-8")
for name, value in re.findall(r'^\s*"(\w+)":\s*(\d+),', source, re.M):
number = int(value)
if name in go_commands and go_commands[name] != number:
command_conflicts.append((name, go_commands[name], number))
go_commands[name] = number
command_differences = [(n, go_commands.get(n), v) for n, v in commands.items() if go_commands.get(n) != v]
command_differences.extend(command_conflicts)
zig_messages = {}
source = (upstream / "protocol.pb.zig").read_text(encoding="utf-8")
for name, body in re.findall(r"^pub const (\w+) = struct \{\n(.*?)^};", source, re.M | re.S):
descriptor = re.search(r"^ pub const _desc_table = .\{\n(.*?)^ };", body, re.M | re.S)
if descriptor is None:
continue
numbers = {int(v) for v in re.findall(r"^ \.\w+ = fd\((\d+),", descriptor[1], re.M)}
for union in re.findall(r"pub const _union_desc = .\{(.*?)\n };", body, re.S):
numbers.update(int(v) for v in re.findall(r"\.\w+ = fd\((\d+),", union))
zig_messages[name] = numbers
go_messages, wrappers = {}, {}
for path in (root / "protocol/proto").glob("*.pb.go"):
for name, body in re.findall(r"^type (\w+) struct \{\n(.*?)^}", path.read_text(encoding="utf-8"), re.M | re.S):
numbers = {int(v) for v in re.findall(r'protobuf:"[^,]+,(\d+),', body)}
go_messages[name] = numbers
if ',oneof"' in body and "_" in name:
wrappers.setdefault(name.split("_", 1)[0], set()).update(numbers)
for name, numbers in wrappers.items():
go_messages[name].update(numbers)
shared = go_messages.keys() & zig_messages.keys()
field_differences = [(n, sorted(go_messages[n]), sorted(zig_messages[n])) for n in sorted(shared)
if go_messages[n] != zig_messages[n]]
current_pairs = {}
for name, number in commands.items():
for suffix in ("CsReq", "CSReq", "Req"):
if name.endswith(suffix):
stem = name[:-len(suffix)]
candidates = [commands[n] for n in (stem + "ScRsp", stem + "CSScRsp", stem + "Rsp") if n in commands]
if len(candidates) == 1:
current_pairs[number] = candidates[0]
break
source = (root / "protocol/proto/cmdpairs.go").read_text(encoding="utf-8")
pairs = {int(a): int(b) for a, b in re.findall(r"^\s*(\d+):\s*(\d+),", source, re.M)}
pair_differences = [(a, b, current_pairs.get(a)) for a, b in pairs.items() if current_pairs.get(a) != b]
print(f"Command IDs: {len(commands)} compared, {len(command_differences)} differences")
print(f"Existing request/response pairs: {len(pairs)} compared, {len(pair_differences)} differences")
print(f"Shared message field-number sets: {len(shared)} compared, {len(field_differences)} differences")
for difference in command_differences + pair_differences + field_differences:
print(difference)
raise SystemExit(bool(command_differences or pair_differences or field_differences))