feat(backend): intellisense symbol server for the library long tail
GET /api/intellisense/symbols/{Name@Version} scans the content-addressed
library cache entry's headers (dependency-free C++ scanner) into an editor
symbol table, disk-cached per entry and served with long cache headers.
Plain OSS hosts without a lib cache 404 and the editor degrades gracefully.
This commit is contained in:
David Montero Crespo
parent
7c3b1c6e24
commit
63a91a08ef
|
|
@ -0,0 +1,87 @@
|
|||
"""Intellisense symbol server.
|
||||
|
||||
Serves editor symbol tables (classes, public methods, constants, enum values)
|
||||
for Arduino libraries present in the server-side content-addressed library
|
||||
cache (VELXIO_LIBCACHE_DIR, default /var/velxio/libcache — entries named
|
||||
`<norm_name>@<norm_version>-<sha12>`). Extraction is done by the
|
||||
dependency-free scanner in app.services.symbol_extract, which also mirrors
|
||||
the pro overlay's cache naming rules; this module never imports app.pro.
|
||||
|
||||
Scans are cached on disk under `<cache_root>/.symbols/<entry>.json`: entry
|
||||
names are content-addressed, so a cached scan can never go stale and the
|
||||
response is safe to cache aggressively client-side too. A host with no cache
|
||||
dir (plain OSS self-host) simply 404s — the frontend degrades gracefully.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from app.services.symbol_extract import find_cache_entry, parse_spec, scan_library
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
# Entry dir names are content-addressed (sha12 suffix), so the payload for a
|
||||
# given spec resolution can be cached hard by browsers/proxies.
|
||||
_CACHE_HEADERS = {"Cache-Control": "public, max-age=86400"}
|
||||
|
||||
|
||||
def _cache_root() -> Path:
|
||||
"""Resolved per-request so tests (and container reconfigs) can point the
|
||||
endpoint elsewhere via the environment."""
|
||||
return Path(os.environ.get("VELXIO_LIBCACHE_DIR", "/var/velxio/libcache"))
|
||||
|
||||
|
||||
def _write_symbol_cache(sym_file: Path, payload: dict) -> None:
|
||||
"""Atomically persist a scan result (tmp file + rename). Failures are
|
||||
logged and ignored — the response was already computed."""
|
||||
try:
|
||||
sym_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
fd, tmp = tempfile.mkstemp(dir=str(sym_file.parent), prefix=".tmp-", suffix=".json")
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||||
json.dump(payload, f)
|
||||
os.replace(tmp, sym_file)
|
||||
except BaseException:
|
||||
try:
|
||||
os.unlink(tmp)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
except OSError as e:
|
||||
logger.warning("[intellisense] symbol cache write failed for %s: %s", sym_file, e)
|
||||
|
||||
|
||||
@router.get("/symbols/{spec}")
|
||||
def get_symbols(spec: str) -> JSONResponse:
|
||||
"""Symbol table for one library. `spec` is `Name@Version` or a bare
|
||||
`Name` (which resolves to the newest cached version). Names and versions
|
||||
are normalized with the same rules the cache uses, so `Adafruit GFX
|
||||
Library@1.12.6` and `adafruitgfxlibrary@1.12.6` hit the same entry."""
|
||||
name, version = parse_spec(spec)
|
||||
root = _cache_root()
|
||||
if not root.is_dir():
|
||||
raise HTTPException(status_code=404, detail=f"library cache not available at {root}")
|
||||
entry = find_cache_entry(root, name, version)
|
||||
if entry is None:
|
||||
raise HTTPException(status_code=404, detail=f"library not found in cache: {spec}")
|
||||
|
||||
sym_file = root / ".symbols" / f"{entry.name}.json"
|
||||
if sym_file.is_file():
|
||||
try:
|
||||
cached = json.loads(sym_file.read_text(encoding="utf-8"))
|
||||
return JSONResponse(content=cached, headers=_CACHE_HEADERS)
|
||||
except (OSError, ValueError):
|
||||
pass # corrupt/unreadable cache file -> rescan below
|
||||
|
||||
payload = scan_library(entry)
|
||||
_write_symbol_cache(sym_file, payload)
|
||||
return JSONResponse(content=payload, headers=_CACHE_HEADERS)
|
||||
|
|
@ -13,7 +13,7 @@ if sys.platform == 'win32':
|
|||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.api.routes import compile, compile_chip, compile_rom, flash, libraries
|
||||
from app.api.routes import compile, compile_chip, compile_rom, flash, intellisense, libraries
|
||||
from app.core.config import settings
|
||||
from app.core.hooks import run_lifespan_startup
|
||||
|
||||
|
|
@ -94,6 +94,7 @@ app.include_router(compile.router, prefix="/api/compile", tags=["compilation"])
|
|||
app.include_router(compile_chip.router, prefix="/api/compile-chip", tags=["custom-chips"])
|
||||
app.include_router(compile_rom.router, prefix="/api/compile-rom", tags=["custom-chips"])
|
||||
app.include_router(libraries.router, prefix="/api/libraries", tags=["libraries"])
|
||||
app.include_router(intellisense.router, prefix="/api/intellisense", tags=["intellisense"])
|
||||
# Hardware flash: subprocesses arduino-cli upload to write a compiled
|
||||
# sketch to a real USB-attached board. Desktop-only in practice (the
|
||||
# web build has no access to local serial ports without WebSerial),
|
||||
|
|
|
|||
|
|
@ -0,0 +1,514 @@
|
|||
"""Dependency-free C++ header symbol scanner for the intellisense endpoint.
|
||||
|
||||
Scans an Arduino library tree (as laid out in the content-addressed library
|
||||
cache: library.properties + headers in the root and src/) and extracts a small,
|
||||
editor-friendly symbol table: classes, their PUBLIC methods, object-like
|
||||
#define constants and enum values. Regex/state-machine based on purpose — no
|
||||
tree-sitter, no libclang, no new pip dependency. Robustness beats
|
||||
completeness: any construct that cannot be parsed confidently is skipped.
|
||||
|
||||
This module also mirrors (reimplements — it must NOT import) the pro overlay's
|
||||
library-cache naming rules from pro/backend/app/pro/services/library_cache.py:
|
||||
cache entries are directories named `<norm_name>@<norm_version>-<sha12>`, where
|
||||
norm_name keeps lowercase alphanumerics only and norm_version collapses the
|
||||
untrusted library.properties version into a safe path token. The OSS image
|
||||
must resolve those entries without the overlay being importable, so the
|
||||
normalization and newest-version-first ordering are duplicated here verbatim.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Hard cap on symbols per library: keeps responses and the on-disk symbol
|
||||
# cache bounded even for pathological headers.
|
||||
MAX_SYMBOLS = 400
|
||||
|
||||
_HEADER_SUFFIXES = frozenset({".h", ".hpp"})
|
||||
# Directories never scanned for symbols (case-insensitive match on the name).
|
||||
_SKIP_DIRS = frozenset({"examples", "example", "test", "tests", "docs", "doc", "extras"})
|
||||
|
||||
_CACHE_META = ".velxio-cache-meta.json"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Library-cache naming (mirrors pro library_cache.py — do not import it)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def norm_name(name: str) -> str:
|
||||
"""Lowercase alphanumerics only — matches the pro cache's norm_name so a
|
||||
request spec, a folder name and a cache key all compare equal."""
|
||||
return "".join(ch for ch in (name or "").lower() if ch.isalnum())
|
||||
|
||||
|
||||
def norm_version(version: str) -> str:
|
||||
"""Collapse an untrusted version string into a single safe, traversal-free
|
||||
path token (mirrors the pro cache's norm_version)."""
|
||||
v = re.sub(r"[^A-Za-z0-9._-]", "-", (version or "").strip())
|
||||
v = v.lstrip(".") # never a leading dot (no '.'/'..' dir tokens)
|
||||
return v or "0"
|
||||
|
||||
|
||||
def parse_spec(spec: str) -> tuple[str, Optional[str]]:
|
||||
"""Split a `Name@Version` request spec into (name, version). A bare `Name`
|
||||
yields (name, None). Whitespace around either part is ignored."""
|
||||
name, sep, version = (spec or "").partition("@")
|
||||
return name.strip(), (version.strip() if sep else None)
|
||||
|
||||
|
||||
def _entry_version_key(entry: Path) -> tuple:
|
||||
"""Rank cache entries for the SAME name: newest version first when the
|
||||
sort is reversed (mirrors the pro cache's _entry_version_key). Parses the
|
||||
version token out of '<name>@<version>-<sha12>' into numeric release
|
||||
tokens + pre-release tag so '3.10.1' > '3.6.0' compares numerically, and
|
||||
'1.2.3' > '1.2.3-rc1'. Ties break on the sidecar's populated_at."""
|
||||
ver = entry.name.rsplit("-", 1)[0].partition("@")[2]
|
||||
tokens = re.split(r"[.\-_]", ver)
|
||||
release: list[int] = []
|
||||
for tok in tokens:
|
||||
if not tok.isdigit():
|
||||
break
|
||||
release.append(int(tok))
|
||||
pre = tuple(tokens[len(release):])
|
||||
populated = 0
|
||||
try:
|
||||
meta = json.loads((entry / _CACHE_META).read_text(encoding="utf-8"))
|
||||
populated = int(meta.get("populated_at") or 0)
|
||||
except (OSError, ValueError, TypeError):
|
||||
pass
|
||||
return (release, 0 if pre else 1, pre, populated)
|
||||
|
||||
|
||||
def find_cache_entry(cache_root: Path, name: str, version: Optional[str] = None) -> Optional[Path]:
|
||||
"""Resolve a cache entry dir under `cache_root` for a library name and
|
||||
optional version. With a version -> first `<norm>@<norm_version>-*` match
|
||||
in key order (mirrors the pro cache's lookup). Without one -> the NEWEST
|
||||
version for the bare name (mirrors lookup_by_name). Returns None when the
|
||||
root is missing, the name normalizes to empty, or nothing matches —
|
||||
callers degrade gracefully (OSS self-host may have no cache at all)."""
|
||||
n = norm_name(name)
|
||||
if not n or not cache_root.is_dir():
|
||||
return None
|
||||
if version is not None:
|
||||
prefix = f"{n}@{norm_version(version)}-"
|
||||
for d in sorted(cache_root.iterdir()):
|
||||
if d.is_dir() and not d.name.startswith(".") and d.name.startswith(prefix):
|
||||
return d
|
||||
return None
|
||||
prefix = f"{n}@"
|
||||
matches = [
|
||||
d for d in sorted(cache_root.iterdir())
|
||||
if d.is_dir() and not d.name.startswith(".") and d.name.startswith(prefix)
|
||||
]
|
||||
if not matches:
|
||||
return None
|
||||
matches.sort(key=_entry_version_key, reverse=True)
|
||||
return matches[0]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# C++ header scanning
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _strip_comments(text: str) -> str:
|
||||
"""Remove // and /* */ comments, respecting string and char literals so a
|
||||
quoted "//" never truncates a line. Newlines inside block comments are
|
||||
preserved so directive lines keep their own line."""
|
||||
out: list[str] = []
|
||||
i, n = 0, len(text)
|
||||
state = "code" # code | line | block | dq | sq
|
||||
while i < n:
|
||||
c = text[i]
|
||||
nxt = text[i + 1] if i + 1 < n else ""
|
||||
if state == "code":
|
||||
if c == "/" and nxt == "/":
|
||||
state = "line"
|
||||
i += 2
|
||||
continue
|
||||
if c == "/" and nxt == "*":
|
||||
state = "block"
|
||||
i += 2
|
||||
continue
|
||||
if c == '"':
|
||||
state = "dq"
|
||||
elif c == "'":
|
||||
state = "sq"
|
||||
out.append(c)
|
||||
i += 1
|
||||
elif state == "line":
|
||||
if c == "\n":
|
||||
out.append(c)
|
||||
state = "code"
|
||||
i += 1
|
||||
elif state == "block":
|
||||
if c == "*" and nxt == "/":
|
||||
state = "code"
|
||||
i += 2
|
||||
continue
|
||||
if c == "\n":
|
||||
out.append(c)
|
||||
i += 1
|
||||
else: # inside a string/char literal
|
||||
if c == "\\" and nxt:
|
||||
out.append(c)
|
||||
out.append(nxt)
|
||||
i += 2
|
||||
continue
|
||||
if (state == "dq" and c == '"') or (state == "sq" and c == "'"):
|
||||
state = "code"
|
||||
out.append(c)
|
||||
i += 1
|
||||
return "".join(out)
|
||||
|
||||
|
||||
# Object-like #define with an UPPERCASE name and a value. The mandatory space
|
||||
# after the name rejects function-like macros (`#define NAME(x)` has no space
|
||||
# before the paren); the mandatory value rejects bare include guards.
|
||||
_DEFINE_RE = re.compile(r"^\s*#\s*define\s+([A-Z][A-Z0-9_]*)\s+\S")
|
||||
|
||||
|
||||
def _extract_directives(text: str) -> tuple[str, list[str]]:
|
||||
"""Drop every preprocessor directive line (conservative #if/#ifdef
|
||||
handling: the guarded content is KEPT, only the directive lines go),
|
||||
collecting object-like uppercase #define names on the way. Backslash
|
||||
continuations are swallowed with their directive."""
|
||||
kept: list[str] = []
|
||||
defines: list[str] = []
|
||||
lines = text.split("\n")
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
line = lines[i]
|
||||
if line.lstrip().startswith("#"):
|
||||
m = _DEFINE_RE.match(line)
|
||||
if m:
|
||||
defines.append(m.group(1))
|
||||
kept.append("")
|
||||
while line.rstrip().endswith("\\") and i + 1 < len(lines):
|
||||
i += 1
|
||||
line = lines[i]
|
||||
kept.append("")
|
||||
i += 1
|
||||
continue
|
||||
kept.append(line)
|
||||
i += 1
|
||||
return "\n".join(kept), defines
|
||||
|
||||
|
||||
# Statement ends with a class/struct definition head: optional single ALL-CAPS
|
||||
# export macro between the keyword and the name, optional `final`, optional
|
||||
# base-clause. Anchored at the end so `typedef struct` (anonymous) never hits.
|
||||
_CLASS_RE = re.compile(
|
||||
r"\b(class|struct)\s+(?:[A-Z][A-Z0-9_]*\s+)?([A-Za-z_]\w*)\s*(?:final\s*)?(?::[^{;]*)?$"
|
||||
)
|
||||
_ENUM_RE = re.compile(
|
||||
r"\benum(?:\s+(class|struct))?(?:\s+([A-Za-z_]\w*))?\s*(?::\s*[\w:\s]+)?$"
|
||||
)
|
||||
_ACCESS_RE = re.compile(r"\b(public|private|protected)\s*:(?!:)")
|
||||
_TRAILER_RE = re.compile(
|
||||
r"^(?:(?:const|override|final|noexcept(?:\([^)]*\))?|=\s*0|=\s*default)\s*)*(?::.*)?$"
|
||||
)
|
||||
_HEAD_QUALIFIERS = ("virtual", "static", "inline", "explicit", "constexpr")
|
||||
_RET_OK_RE = re.compile(r"^[\w:<>,\s*&\[\]]+$")
|
||||
_NAME_TAIL_RE = re.compile(r"([A-Za-z_]\w*)\s*$")
|
||||
_KEYWORDS = frozenset({"if", "for", "while", "switch", "return", "sizeof", "new", "delete", "throw"})
|
||||
|
||||
|
||||
def _split_args(args: str) -> list[str]:
|
||||
"""Split an argument list on top-level commas (nested <>, () and []
|
||||
stay intact). '(void)' and '()' both yield no arguments."""
|
||||
args = args.strip()
|
||||
if not args or args == "void":
|
||||
return []
|
||||
parts: list[str] = []
|
||||
cur: list[str] = []
|
||||
depth = 0
|
||||
for ch in args:
|
||||
if ch in "<([":
|
||||
depth += 1
|
||||
elif ch in ">)]":
|
||||
depth -= 1
|
||||
if ch == "," and depth == 0:
|
||||
parts.append("".join(cur).strip())
|
||||
cur = []
|
||||
else:
|
||||
cur.append(ch)
|
||||
tail = "".join(cur).strip()
|
||||
if tail:
|
||||
parts.append(tail)
|
||||
return [p for p in parts if p]
|
||||
|
||||
|
||||
def _param_hint(arg: str) -> str:
|
||||
"""Snippet placeholder text for one argument: the declared parameter name
|
||||
when it can be found, else the whole trimmed argument. Snippet-breaking
|
||||
characters are removed."""
|
||||
decl = arg.split("=", 1)[0].rstrip()
|
||||
m = _NAME_TAIL_RE.search(re.sub(r"\[\s*\]\s*$", "", decl))
|
||||
hint = m.group(1) if m else arg
|
||||
return re.sub(r"[{}$]", "", hint).strip() or "arg"
|
||||
|
||||
|
||||
def _build_call(name: str, parts: list[str]) -> tuple[str, str]:
|
||||
"""(signature-args, insertText) for a callable with the given split args."""
|
||||
required = [p for p in parts if "=" not in p]
|
||||
if not required:
|
||||
insert = f"{name}()"
|
||||
else:
|
||||
placeholders = ", ".join(
|
||||
f"${{{k}:{_param_hint(p)}}}" for k, p in enumerate(required, 1)
|
||||
)
|
||||
insert = f"{name}({placeholders})$0"
|
||||
return ", ".join(parts), insert
|
||||
|
||||
|
||||
def _try_method(candidate: str, scope: dict, add: Callable[[dict], Optional[dict]], detail: str) -> None:
|
||||
"""Attempt to parse `candidate` (a statement inside a public class/struct
|
||||
section, stripped of access labels) as a method declaration. Emits a
|
||||
method symbol, or fills the owning class symbol's insertText/params from
|
||||
its first public constructor. Anything unconfident is skipped."""
|
||||
candidate = re.sub(r"\s+", " ", candidate).strip()
|
||||
if not candidate or "operator" in candidate or "~" in candidate or "template" in candidate:
|
||||
return
|
||||
p = candidate.find("(")
|
||||
if p <= 0:
|
||||
return
|
||||
depth = 0
|
||||
q = -1
|
||||
for k in range(p, len(candidate)):
|
||||
if candidate[k] == "(":
|
||||
depth += 1
|
||||
elif candidate[k] == ")":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
q = k
|
||||
break
|
||||
if q < 0:
|
||||
return
|
||||
head = candidate[:p].strip()
|
||||
args_raw = candidate[p + 1:q].strip()
|
||||
trailer = candidate[q + 1:].strip()
|
||||
if "delete" in trailer or not _TRAILER_RE.match(trailer):
|
||||
return
|
||||
if head.startswith("friend"):
|
||||
return
|
||||
nm = _NAME_TAIL_RE.search(head)
|
||||
if not nm:
|
||||
return
|
||||
name = nm.group(1)
|
||||
if name in _KEYWORDS:
|
||||
return
|
||||
ret = head[: nm.start()].strip()
|
||||
for qual in _HEAD_QUALIFIERS:
|
||||
ret = re.sub(rf"\b{qual}\b", " ", ret)
|
||||
ret = re.sub(r"\s+", " ", ret).strip()
|
||||
parts = _split_args(args_raw)
|
||||
if not ret:
|
||||
# No return type: a constructor when the name matches the class;
|
||||
# anything else (macro invocation, unparsable) is skipped.
|
||||
if name == scope["name"]:
|
||||
sym = scope.get("sym")
|
||||
if sym is not None and "insertText" not in sym:
|
||||
sig_args, insert = _build_call(name, parts)
|
||||
sym["signature"] = f"{name}({sig_args})"
|
||||
sym["insertText"] = insert
|
||||
if parts:
|
||||
sym["params"] = parts
|
||||
return
|
||||
if not _RET_OK_RE.match(ret):
|
||||
return
|
||||
sig_args, insert = _build_call(name, parts)
|
||||
method = {
|
||||
"name": name,
|
||||
"kind": "method",
|
||||
"owner": scope["name"],
|
||||
"signature": f"{ret} {name}({sig_args})",
|
||||
"detail": detail,
|
||||
"insertText": insert,
|
||||
}
|
||||
if parts:
|
||||
method["params"] = parts
|
||||
add(method)
|
||||
|
||||
|
||||
def _capture_braces(text: str, i: int) -> tuple[str, int]:
|
||||
"""Given `i` at a '{', return (body, index-just-past-matching-'}')."""
|
||||
depth = 0
|
||||
for k in range(i, len(text)):
|
||||
if text[k] == "{":
|
||||
depth += 1
|
||||
elif text[k] == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return text[i + 1:k], k + 1
|
||||
return text[i + 1:], len(text)
|
||||
|
||||
|
||||
def _emit_enum(scoped: bool, ename: Optional[str], body: str, add: Callable[[dict], Optional[dict]], detail: str) -> None:
|
||||
for piece in _split_args(body):
|
||||
m = re.match(r"^([A-Za-z_]\w*)", piece.strip())
|
||||
if not m:
|
||||
continue
|
||||
sym = {"name": m.group(1), "kind": "constant", "detail": detail}
|
||||
if scoped and ename:
|
||||
sym["owner"] = ename
|
||||
add(sym)
|
||||
|
||||
|
||||
def _scan_text(text: str, add: Callable[[dict], Optional[dict]], detail: str) -> None:
|
||||
"""Statement-oriented walk of comment/directive-free source: track brace
|
||||
depth and a scope stack; classify each statement when it terminates at
|
||||
'{', '}' or ';'. Multi-line declarations fall out naturally."""
|
||||
buf: list[str] = []
|
||||
stack: list[dict] = []
|
||||
i, n = 0, len(text)
|
||||
|
||||
def handle_statement(stmt: str) -> None:
|
||||
if not stack or stack[-1]["kind"] != "classlike":
|
||||
return
|
||||
scope = stack[-1]
|
||||
candidate = stmt
|
||||
for m in _ACCESS_RE.finditer(stmt):
|
||||
scope["access"] = m.group(1)
|
||||
candidate = stmt[m.end():]
|
||||
if scope["access"] != "public":
|
||||
return
|
||||
_try_method(candidate.strip(), scope, add, detail)
|
||||
|
||||
while i < n:
|
||||
c = text[i]
|
||||
if c == "{":
|
||||
stmt = "".join(buf).strip()
|
||||
buf = []
|
||||
if "template" not in stmt:
|
||||
em = _ENUM_RE.search(stmt)
|
||||
if em and re.search(r"\benum\b", stmt):
|
||||
body, j = _capture_braces(text, i)
|
||||
_emit_enum(em.group(1) is not None, em.group(2), body, add, detail)
|
||||
i = j
|
||||
continue
|
||||
cm = _CLASS_RE.search(stmt)
|
||||
if cm:
|
||||
cname = cm.group(2)
|
||||
sym = {"name": cname, "kind": "class", "detail": detail}
|
||||
stored = add(sym)
|
||||
stack.append({
|
||||
"kind": "classlike",
|
||||
"name": cname,
|
||||
"access": "public" if cm.group(1) == "struct" else "private",
|
||||
"sym": stored,
|
||||
})
|
||||
i += 1
|
||||
continue
|
||||
# Inline method definition (`ret name(args) {`), namespace,
|
||||
# function body, brace initializer, ... — try the method parse,
|
||||
# then descend as an opaque scope.
|
||||
handle_statement(stmt)
|
||||
stack.append({"kind": "other"})
|
||||
i += 1
|
||||
continue
|
||||
if c == "}":
|
||||
if stack:
|
||||
stack.pop()
|
||||
buf = []
|
||||
i += 1
|
||||
continue
|
||||
if c == ";":
|
||||
stmt = "".join(buf).strip()
|
||||
buf = []
|
||||
if stmt:
|
||||
handle_statement(stmt)
|
||||
i += 1
|
||||
continue
|
||||
buf.append(c)
|
||||
i += 1
|
||||
|
||||
|
||||
def _read_props(lib_dir: Path) -> dict[str, str]:
|
||||
props: dict[str, str] = {}
|
||||
try:
|
||||
text = (lib_dir / "library.properties").read_text(encoding="utf-8", errors="ignore")
|
||||
except OSError:
|
||||
return props
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
k, _, v = line.partition("=")
|
||||
props[k.strip().lower()] = v.strip()
|
||||
return props
|
||||
|
||||
|
||||
def _iter_headers(lib_dir: Path):
|
||||
"""Yield every *.h/*.hpp under the library tree, pruning examples/test/
|
||||
docs style directories, in a deterministic order."""
|
||||
for dirpath, dirnames, filenames in os.walk(lib_dir):
|
||||
dirnames[:] = sorted(
|
||||
d for d in dirnames
|
||||
if d.lower() not in _SKIP_DIRS and not d.startswith(".")
|
||||
)
|
||||
for fn in sorted(filenames):
|
||||
if Path(fn).suffix.lower() in _HEADER_SUFFIXES:
|
||||
yield Path(dirpath) / fn
|
||||
|
||||
|
||||
def scan_library(lib_dir: Path) -> dict:
|
||||
"""Scan one library tree into an intellisense payload:
|
||||
|
||||
{"id": <display name>, "triggers": [<header basenames>], "symbols": [...]}
|
||||
|
||||
Symbol dicts use the ApiSymbol field names exactly: name, kind, signature,
|
||||
detail, doc, insertText, owner, params — optional fields are simply
|
||||
omitted when unknown. `detail` is always the library display name."""
|
||||
props = _read_props(lib_dir)
|
||||
display = props.get("name") or lib_dir.name
|
||||
|
||||
triggers: list[str] = []
|
||||
for root in (lib_dir, lib_dir / "src"):
|
||||
if not root.is_dir():
|
||||
continue
|
||||
try:
|
||||
children = sorted(root.iterdir())
|
||||
except OSError:
|
||||
continue
|
||||
for child in children:
|
||||
if child.is_file() and child.suffix.lower() in _HEADER_SUFFIXES:
|
||||
if child.name not in triggers:
|
||||
triggers.append(child.name)
|
||||
|
||||
symbols: list[dict] = []
|
||||
seen: set[tuple[str, str, str]] = set()
|
||||
|
||||
def add(sym: dict) -> Optional[dict]:
|
||||
if len(symbols) >= MAX_SYMBOLS:
|
||||
return None
|
||||
key = (sym["name"], sym["kind"], sym.get("owner", ""))
|
||||
if key in seen:
|
||||
return None
|
||||
seen.add(key)
|
||||
symbols.append(sym)
|
||||
return sym
|
||||
|
||||
for hdr in _iter_headers(lib_dir):
|
||||
if len(symbols) >= MAX_SYMBOLS:
|
||||
break
|
||||
try:
|
||||
raw = hdr.read_text(encoding="utf-8", errors="ignore")
|
||||
except OSError:
|
||||
continue
|
||||
try:
|
||||
text = _strip_comments(raw)
|
||||
text, defines = _extract_directives(text)
|
||||
for dname in defines:
|
||||
add({"name": dname, "kind": "constant", "detail": display})
|
||||
_scan_text(text, add, display)
|
||||
except Exception:
|
||||
# A single hostile/bizarre header must never take the payload down.
|
||||
logger.warning("[symbol_extract] failed scanning %s", hdr, exc_info=True)
|
||||
continue
|
||||
|
||||
return {"id": display, "triggers": triggers, "symbols": symbols}
|
||||
|
|
@ -0,0 +1,172 @@
|
|||
"""Unit tests for the intellisense symbol scanner and its library-cache
|
||||
resolution helpers. Pure-Python — no FastAPI app, no HTTP server, no
|
||||
toolchain — so they run anywhere pytest does.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app.services.symbol_extract import (
|
||||
find_cache_entry,
|
||||
norm_name,
|
||||
norm_version,
|
||||
parse_spec,
|
||||
scan_library,
|
||||
)
|
||||
|
||||
|
||||
# ── fixture library ───────────────────────────────────────────────────────
|
||||
|
||||
_HEADER = """\
|
||||
#pragma once
|
||||
#include <stdint.h>
|
||||
|
||||
#define FAKE_MAX_PIXELS 64
|
||||
#define FAKE_NAME "fake" // string-valued define still counts
|
||||
#define FAKE_FUNC(x) ((x) * 2)
|
||||
#define lowercase_define 1
|
||||
#define FAKE_GUARD_H
|
||||
|
||||
enum FakeMode { FAKE_MODE_A, FAKE_MODE_B = 2 };
|
||||
enum class FakeColor : uint8_t { Red, Green };
|
||||
|
||||
/* block comment with a fake decl: void notReal(); */
|
||||
class FakeLib {
|
||||
public:
|
||||
FakeLib(uint8_t pin);
|
||||
~FakeLib();
|
||||
bool begin(uint8_t pin, int count = 8);
|
||||
void show();
|
||||
private:
|
||||
void internalUpdate();
|
||||
uint8_t pin_;
|
||||
};
|
||||
"""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_lib(tmp_path: Path) -> Path:
|
||||
lib = tmp_path / "fakelib@1.2.3-0123456789ab"
|
||||
src = lib / "src"
|
||||
src.mkdir(parents=True)
|
||||
(lib / "library.properties").write_text(
|
||||
"name=Fake Lib\nversion=1.2.3\n", encoding="utf-8"
|
||||
)
|
||||
(src / "FakeLib.h").write_text(_HEADER, encoding="utf-8")
|
||||
examples = lib / "examples"
|
||||
examples.mkdir()
|
||||
(examples / "Noise.h").write_text(
|
||||
"class NotMe { public: void nope(); };\n", encoding="utf-8"
|
||||
)
|
||||
return lib
|
||||
|
||||
|
||||
def _by_kind(result: dict, kind: str) -> dict[str, dict]:
|
||||
return {s["name"]: s for s in result["symbols"] if s["kind"] == kind}
|
||||
|
||||
|
||||
# ── scan_library ──────────────────────────────────────────────────────────
|
||||
|
||||
def test_scan_library_id_and_triggers(fake_lib: Path) -> None:
|
||||
result = scan_library(fake_lib)
|
||||
assert result["id"] == "Fake Lib"
|
||||
assert "FakeLib.h" in result["triggers"]
|
||||
# examples/ headers are neither triggers nor scanned for symbols
|
||||
assert "Noise.h" not in result["triggers"]
|
||||
assert all(s["name"] != "NotMe" for s in result["symbols"])
|
||||
|
||||
|
||||
def test_scan_library_class_and_public_methods(fake_lib: Path) -> None:
|
||||
result = scan_library(fake_lib)
|
||||
classes = _by_kind(result, "class")
|
||||
methods = _by_kind(result, "method")
|
||||
|
||||
assert "FakeLib" in classes
|
||||
# constructor is folded into the class symbol, not a separate method
|
||||
assert classes["FakeLib"]["insertText"] == "FakeLib(${1:pin})$0"
|
||||
assert classes["FakeLib"]["params"] == ["uint8_t pin"]
|
||||
|
||||
assert set(methods) == {"begin", "show"}
|
||||
for m in methods.values():
|
||||
assert m["owner"] == "FakeLib"
|
||||
assert m["detail"] == "Fake Lib"
|
||||
assert methods["begin"]["signature"] == "bool begin(uint8_t pin, int count = 8)"
|
||||
# only the required (non-defaulted) arg becomes a snippet placeholder
|
||||
assert methods["begin"]["insertText"] == "begin(${1:pin})$0"
|
||||
assert methods["begin"]["params"] == ["uint8_t pin", "int count = 8"]
|
||||
assert methods["show"]["insertText"] == "show()"
|
||||
|
||||
# private method and destructor never surface
|
||||
assert "internalUpdate" not in methods
|
||||
assert all("~" not in s["name"] for s in result["symbols"])
|
||||
|
||||
|
||||
def test_scan_library_constants_and_enums(fake_lib: Path) -> None:
|
||||
result = scan_library(fake_lib)
|
||||
constants = _by_kind(result, "constant")
|
||||
|
||||
assert "FAKE_MAX_PIXELS" in constants
|
||||
assert "FAKE_NAME" in constants
|
||||
# function-like macro, lowercase define, and value-less include guard: out
|
||||
assert "FAKE_FUNC" not in constants
|
||||
assert "lowercase_define" not in constants
|
||||
assert "FAKE_GUARD_H" not in constants
|
||||
|
||||
# plain enum values: no owner; enum class values: owner set
|
||||
assert "FAKE_MODE_A" in constants and "owner" not in constants["FAKE_MODE_A"]
|
||||
assert "FAKE_MODE_B" in constants
|
||||
assert constants["Red"]["owner"] == "FakeColor"
|
||||
assert constants["Green"]["owner"] == "FakeColor"
|
||||
|
||||
|
||||
# ── spec parsing / normalization (used by the route) ──────────────────────
|
||||
|
||||
def test_parse_spec() -> None:
|
||||
assert parse_spec("Adafruit GFX Library@1.12.6") == ("Adafruit GFX Library", "1.12.6")
|
||||
assert parse_spec("FastLED") == ("FastLED", None)
|
||||
assert parse_spec("") == ("", None)
|
||||
|
||||
|
||||
def test_norm_rules_mirror_library_cache() -> None:
|
||||
assert norm_name("Adafruit GFX Library") == "adafruitgfxlibrary"
|
||||
assert norm_name("DHT sensor library") == "dhtsensorlibrary"
|
||||
assert norm_name("") == ""
|
||||
assert norm_version("1.12.6") == "1.12.6"
|
||||
assert norm_version(" 1.2.3 ") == "1.2.3"
|
||||
# unsafe chars become '-', leading dots are stripped (no '.'/'..' tokens)
|
||||
assert norm_version("../evil/1") == "-evil-1"
|
||||
assert norm_version("") == "0"
|
||||
|
||||
|
||||
def test_find_cache_entry(tmp_path: Path) -> None:
|
||||
for entry in (
|
||||
"fastled@3.6.0-aaaaaaaaaaaa",
|
||||
"fastled@3.10.1-bbbbbbbbbbbb",
|
||||
"other@1.0.0-cccccccccccc",
|
||||
".symbols",
|
||||
):
|
||||
(tmp_path / entry).mkdir()
|
||||
|
||||
exact = find_cache_entry(tmp_path, "FastLED", "3.6.0")
|
||||
assert exact is not None and exact.name == "fastled@3.6.0-aaaaaaaaaaaa"
|
||||
|
||||
# bare name -> NEWEST version, compared numerically (3.10.1 > 3.6.0 even
|
||||
# though it sorts first lexicographically) — mirrors lookup_by_name
|
||||
bare = find_cache_entry(tmp_path, "FastLED")
|
||||
assert bare is not None and bare.name == "fastled@3.10.1-bbbbbbbbbbbb"
|
||||
|
||||
assert find_cache_entry(tmp_path, "FastLED", "9.9.9") is None
|
||||
assert find_cache_entry(tmp_path, "Nope") is None
|
||||
assert find_cache_entry(tmp_path / "missing", "FastLED") is None
|
||||
assert find_cache_entry(tmp_path, "!!!") is None # normalizes to empty
|
||||
|
||||
|
||||
# ── route wiring (skipped where fastapi is not installed) ─────────────────
|
||||
|
||||
def test_route_registers_symbols_path() -> None:
|
||||
pytest.importorskip("fastapi")
|
||||
from app.api.routes import intellisense
|
||||
|
||||
assert any(r.path == "/symbols/{spec}" for r in intellisense.router.routes)
|
||||
Loading…
Reference in New Issue