deusdata--codebase-memory-mcp/tests/test_py_lsp_bench.c

/*
 * test_py_lsp_bench.c — parity-tracking benchmark for py_lsp.
 *
 * The fixture is a synthetic but representative ~250-line Python file
 * that exercises every parity feature py_lsp implements:
 *   - imports + dotted submodule walks (os.path)
 *   - dataclass + class-body annotated fields
 *   - inheritance with super() calls (method + __init__)
 *   - Self return type (fluent chains)
 *   - generic containers + comprehension element typing
 *   - dict / list / tuple subscript value types
 *   - narrowing (isinstance / is not None / walrus + None)
 *   - match/case class patterns
 *   - async / await
 *   - classmethod / staticmethod
 *   - @property attribute access
 *   - SQLAlchemy 2.0-style Mapped[T] annotations
 *   - Pydantic-style BaseModel with field annotations
 *   - typing.Annotated / ClassVar / Final / InitVar wrappers
 *   - generator yield with for-loop element typing
 *   - operator dunder dispatch
 *   - typing.cast / assert_type
 *   - common stdlib calls (logging, pathlib, json, functools)
 *
 * Asserts: resolution ratio >= 50% under sanitizers (production target
 * is 95%+ — see docs/BENCHMARK_PYTHON.md).
 */
#include "test_framework.h"
#include "cbm.h"
#include "lsp/py_lsp.h"
#include <stdlib.h>
#include <time.h>

static const char *bench_source =
    "import os\n"
    "import os.path\n"
    "import json\n"
    "import logging\n"
    "import functools\n"
    "from dataclasses import dataclass, field\n"
    "from pathlib import Path\n"
    "from collections import defaultdict\n"
    "from typing import (\n"
    "    Optional, Self, cast, ClassVar, Final, Annotated, Generator,\n"
    "    NamedTuple, Protocol\n"
    ")\n"
    "\n"
    "log = logging.getLogger('bench')\n"
    "\n"
    "@dataclass\n"
    "class Config:\n"
    "    name: str\n"
    "    debug: bool\n"
    "    extras: list[str]\n"
    "    LIMIT: ClassVar[int] = 100\n"
    "    def display(self) -> str:\n"
    "        return self.name\n"
    "    def with_debug(self, on: bool) -> Self:\n"
    "        self.debug = on\n"
    "        return self\n"
    "\n"
    "class Mapped:\n"
    "    pass\n"
    "\n"
    "class User:\n"
    "    id: Annotated[int, 'primary_key']\n"
    "    name: Final[str]\n"
    "    email: str\n"
    "    posts: list[str]\n"
    "    def __init__(self, id: int, name: str, email: str):\n"
    "        self.id = id\n"
    "        self.name = name\n"
    "        self.email = email\n"
    "        self.posts = []\n"
    "    def display_name(self) -> str:\n"
    "        return self.name\n"
    "    @property\n"
    "    def domain(self) -> str:\n"
    "        return self.email.split('@')[1]\n"
    "\n"
    "class BaseStore:\n"
    "    def __init__(self, root: Path):\n"
    "        self.root: Path = root\n"
    "        self.metadata: dict[str, str] = {}\n"
    "    def get(self, key: str) -> Optional[str]:\n"
    "        return self.metadata.get(key)\n"
    "    def put(self, key: str, value: str) -> None:\n"
    "        self.metadata[key] = value\n"
    "    def keys(self) -> list[str]:\n"
    "        return list(self.metadata.keys())\n"
    "\n"
    "class FileStore(BaseStore):\n"
    "    def __init__(self, root: Path):\n"
    "        super().__init__(root)\n"
    "        self.cache: dict[str, str] = {}\n"
    "    @classmethod\n"
    "    def open(cls, root: Path) -> Self:\n"
    "        return cls(root)\n"
    "    @staticmethod\n"
    "    def is_writable(p: Path) -> bool:\n"
    "        return True\n"
    "    def get(self, key: str) -> Optional[str]:\n"
    "        if key in self.cache:\n"
    "            return self.cache[key]\n"
    "        full = self.root / key\n"
    "        if full.exists():\n"
    "            text = full.read_text()\n"
    "            self.cache[key] = text\n"
    "            return text\n"
    "        return None\n"
    "    def put(self, key: str, value: str) -> None:\n"
    "        super().put(key, value)\n"
    "        self.cache[key] = value\n"
    "\n"
    "class Result:\n"
    "    def __init__(self, ok: bool, value: Optional[str]):\n"
    "        self.ok = ok\n"
    "        self.value = value\n"
    "    def display(self) -> str:\n"
    "        return self.value or 'empty'\n"
    "\n"
    "class App:\n"
    "    def __init__(self, cfg: Config, store: BaseStore):\n"
    "        self.cfg = cfg\n"
    "        self.store = store\n"
    "        self.users: dict[int, User] = {}\n"
    "        self.counts: dict[str, int] = defaultdict(int)\n"
    "        self.logger = logging.getLogger('app')\n"
    "    def display_config(self) -> str:\n"
    "        return self.cfg.display()\n"
    "    def add_user(self, u: User) -> None:\n"
    "        self.users[u.id] = u\n"
    "        self.logger.info('added %s', u.display_name())\n"
    "    def get_user(self, uid: int) -> Optional[User]:\n"
    "        return self.users.get(uid)\n"
    "    def lookup(self, key: str) -> Result:\n"
    "        result = self.store.get(key)\n"
    "        self.counts[key] += 1\n"
    "        return Result(result is not None, result)\n"
    "    def write(self, key: str, value: str) -> None:\n"
    "        self.store.put(key, value)\n"
    "        self.logger.info('wrote %s', key)\n"
    "    def display_results(self, results: list[Result]) -> list[str]:\n"
    "        return [r.display() for r in results]\n"
    "    def display_users(self) -> list[str]:\n"
    "        return [u.display_name() for u in self.users.values()]\n"
    "    def filter_active(self, keys: list[str]) -> list[str]:\n"
    "        return [k for k in keys if self.lookup(k).ok]\n"
    "    def with_debug(self) -> Self:\n"
    "        self.cfg = self.cfg.with_debug(True)\n"
    "        return self\n"
    "\n"
    "@functools.lru_cache(maxsize=128)\n"
    "def slow_lookup(key: str) -> Optional[str]:\n"
    "    return None\n"
    "\n"
    "def parse_response(payload: str) -> dict[str, str]:\n"
    "    return json.loads(payload)\n"
    "\n"
    "async def fetch(key: str) -> Optional[str]:\n"
    "    return slow_lookup(key)\n"
    "\n"
    "async def fetch_many(keys: list[str]) -> list[Optional[str]]:\n"
    "    out: list[Optional[str]] = []\n"
    "    for k in keys:\n"
    "        v = await fetch(k)\n"
    "        out.append(v)\n"
    "    return out\n"
    "\n"
    "def gen_users(n: int) -> Generator[User, None, None]:\n"
    "    for i in range(n):\n"
    "        yield User(i, 'name' + str(i), 'a@b.com')\n"
    "\n"
    "def classify(x) -> str:\n"
    "    match x:\n"
    "        case Config():\n"
    "            return x.display()\n"
    "        case User():\n"
    "            return x.display_name()\n"
    "        case Result():\n"
    "            return x.display()\n"
    "        case App():\n"
    "            return x.display_config()\n"
    "        case _:\n"
    "            return 'unknown'\n"
    "\n"
    "def maybe_get(s: Optional[BaseStore], key: str) -> Optional[str]:\n"
    "    if s is not None:\n"
    "        return s.get(key)\n"
    "    return None\n"
    "\n"
    "def first_writable(stores: list[BaseStore]) -> Optional[BaseStore]:\n"
    "    for s in stores:\n"
    "        if isinstance(s, FileStore):\n"
    "            return s\n"
    "    return None\n"
    "\n"
    "def main() -> None:\n"
    "    cfg = Config('app', True, ['x', 'y'])\n"
    "    fs = FileStore.open(Path(os.getcwd()))\n"
    "    app = App(cfg, fs)\n"
    "    app.with_debug().write('a', '1')\n"
    "    res = app.lookup('a')\n"
    "    classify(cfg)\n"
    "    classify(res)\n"
    "    classify(app)\n"
    "    payload = parse_response('{}')\n"
    "    log.info('payload=%s', payload)\n"
    "    for u in gen_users(3):\n"
    "        app.add_user(u)\n"
    "    summary = app.display_users()\n"
    "    log.info('users=%d', len(summary))\n";

static double elapsed_ms(struct timespec t0, struct timespec t1) {
    double s = (double)(t1.tv_sec - t0.tv_sec);
    double ns = (double)(t1.tv_nsec - t0.tv_nsec);
    return s * 1000.0 + ns / 1000000.0;
}

TEST(pylsp_bench_resolution_ratio) {
    /* Perf benchmark: time-budgeted. Under ASan+UBSan the budget is scaled up
     * (see the sanitizer-aware budget below) and the result is freed before
     * asserting so a budget miss doesn't leak. */
    int slen = (int)strlen(bench_source);

    struct timespec t0;
    struct timespec t1;
    clock_gettime(CLOCK_MONOTONIC, &t0);
    CBMFileResult *r = cbm_extract_file(bench_source, slen, CBM_LANG_PYTHON,
                                        "test", "bench.py", 0, NULL, NULL);
    clock_gettime(CLOCK_MONOTONIC, &t1);
    ASSERT_NOT_NULL(r);

    double ms = elapsed_ms(t0, t1);
    int calls = r->calls.count;
    int resolved = r->resolved_calls.count;
    int loc = 0;
    for (const char *p = bench_source; *p; p++) {
        if (*p == '\n') loc++;
    }
    double ratio = calls > 0 ? (double)resolved / (double)calls : 0.0;

    printf("    bench: %d lines, %d calls, %d resolved (%.0f%%), %.2f ms\n",
           loc, calls, resolved, ratio * 100.0, ms);

    /* Free the result BEFORE asserting so a budget miss doesn't leak. */
    cbm_free_result(r);

    ASSERT_GTE(calls, 1);
    ASSERT_GTE(resolved, 1);

    /* Floor at 50% under sanitizers; production target is 95% per
     * docs/BENCHMARK_PYTHON.md. */
    if (calls >= 30) {
        ASSERT_GTE(resolved * 2, calls);
    }

    /* Liveness backstop ONLY — not a perf gate. The perf content of this test
     * is the resolution-ratio assertion above; wall-clock on a shared CI
     * runner is not a property of the code (the 1500 ms sanitized budget was
     * exceeded twice on loaded ubuntu-24.04-arm runners by the SAME sha that
     * passed hours earlier). Absolute timings belong to
     * scripts/benchmark-*.sh, which record rather than gate. This bound
     * exists solely so a catastrophic hang fails the suite instead of the
     * 900 s shard wall clock. */
    ASSERT(ms < 30000.0);
    PASS();
}

SUITE(py_lsp_bench) {
    RUN_TEST(pylsp_bench_resolution_ratio);
}