From cbb1a72f60eb90957e69c7eeb91914ce3aed2c47 Mon Sep 17 00:00:00 2001 From: Wolfvin Date: Tue, 30 Jun 2026 15:57:21 +0000 Subject: [PATCH] =?UTF-8?q?feat(snapshot):=20export-snapshot=20+=20import-?= =?UTF-8?q?snapshot=20=E2=80=94=20share=20CodeLens=20graph=20across=20team?= =?UTF-8?q?=20(closes=20#12)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 10 +- SKILL-QUICK.md | 10 +- SKILL.md | 4 +- pyproject.toml | 2 +- scripts/commands/export_snapshot.py | 168 ++++++++++ scripts/commands/import_snapshot.py | 178 ++++++++++ scripts/graph_model.py | 2 +- scripts/snapshot_io.py | 442 +++++++++++++++++++++++++ skill.json | 2 +- tests/test_integration.py | 4 +- tests/test_snapshot.py | 487 ++++++++++++++++++++++++++++ 11 files changed, 1292 insertions(+), 17 deletions(-) create mode 100644 scripts/commands/export_snapshot.py create mode 100644 scripts/commands/import_snapshot.py create mode 100644 scripts/snapshot_io.py create mode 100644 tests/test_snapshot.py diff --git a/README.md b/README.md index d4eaaedd..7c07e279 100644 --- a/README.md +++ b/README.md @@ -2,12 +2,12 @@ > **Before an AI writes a new class/id/function, CodeLens must be checked. This is not optional.** -CodeLens is an AI-native code intelligence platform that gives AI agents **full visibility** into a codebase before they write any code. It prevents collision, overwrite of existing logic, security vulnerabilities, and dead code through 63 CLI commands, an MCP server with 61 tools (50 static + 11 dynamic), AST-based taint analysis, live CVE/OSV scanning, a plugin system with OWASP Top 10 + Compliance rule packs, a true graph data model (nodes + edges) for structural code queries, and token-efficient `--format compact` output for high-volume agent workflows (issue #17). +CodeLens is an AI-native code intelligence platform that gives AI agents **full visibility** into a codebase before they write any code. It prevents collision, overwrite of existing logic, security vulnerabilities, and dead code through 65 CLI commands, an MCP server with 63 tools (50 static + 13 dynamic), AST-based taint analysis, live CVE/OSV scanning, a plugin system with OWASP Top 10 + Compliance rule packs, a true graph data model (nodes + edges) for structural code queries, and token-efficient `--format compact` output for high-volume agent workflows (issue #17). ## Features -- **63 CLI Commands** — From basic scan/query to AST taint analysis, CVE scanning, plugin management, auto-fix, dashboards, CI/CD quality gates, and `graph-schema` for cheap graph-shape introspection -- **MCP Server (61 Tools)** — Native AI agent integration via Model Context Protocol (JSON-RPC over stdio), 50 statically-defined tools + 11 dynamically discovered, every tool accepts a `format` parameter (`json`/`markdown`/`ai`/`sarif`/`compact`) +- **65 CLI Commands** — From basic scan/query to AST taint analysis, CVE scanning, plugin management, auto-fix, dashboards, CI/CD quality gates, and `graph-schema` for cheap graph-shape introspection +- **MCP Server (63 Tools)** — Native AI agent integration via Model Context Protocol (JSON-RPC over stdio), 50 statically-defined tools + 13 dynamically discovered, every tool accepts a `format` parameter (`json`/`markdown`/`ai`/`sarif`/`compact`) - **Token-Efficient Compact Output (v8.2, issue #17)** — `--format compact` produces single-char-key JSON with abbreviated types, omitted null fields, and relative paths — ~50% smaller than `json` on real trace output. Combined with `--limit`/`--offset` pagination, 5 structural queries now cost <5k tokens (down from 30-80k) - **AST Taint Engine** — Tree-sitter based taint analysis with return-value propagation, scope hierarchy, and branch condition refinement - **Live CVE/OSV Scanning** — Real-time vulnerability data from OSV.dev API with SQLite cache, 9 ecosystems (PyPI, npm, crates.io, Go, Maven, NuGet, RubyGems, Pub, Hex) @@ -225,8 +225,8 @@ codelens/ │ ├── changelog.md # Older changelog (per-version highlights) │ └── agent-integration.md # AI agent integration guide ├── scripts/ -│ ├── codelens.py # CLI entry point (63 commands registered) -│ ├── mcp_server.py # MCP JSON-RPC server (61 tools) +│ ├── codelens.py # CLI entry point (65 commands registered) +│ ├── mcp_server.py # MCP JSON-RPC server (63 tools) │ ├── registry.py # Registry read/write/build │ ├── persistent_registry.py # SQLite persistent storage (WAL mode) │ ├── base_parser.py # Base tree-sitter parser diff --git a/SKILL-QUICK.md b/SKILL-QUICK.md index 511e7454..5da0036b 100755 --- a/SKILL-QUICK.md +++ b/SKILL-QUICK.md @@ -114,7 +114,7 @@ $CLI list --limit 5 --offset 10 --format compact # → paginated + co | "Cross-file taint" | `dataflow` | `taint` (taint is single-file, AST-deep) | | "Auto-fix issues" | `fix` | `check` (check just gates, doesn't fix) | -## All 63 Commands +## All 65 Commands ### Setup & Lifecycle (8+) `init` · `scan [--incremental] [--max-files N] [--full]` · `registry-validate` · `detect` · `watch [--debounce SECS] [--git-mode] [--interval SECS]` · `git-status` · `migrate` · `serve` · `lsp-status` (issue #33: `codelens --lsp-status` top-level flag is an alias of `codelens lsp-status` — both delegate to `hybrid_engine.get_lsp_status()` and return the identical payload) @@ -146,9 +146,9 @@ $CLI list --limit 5 --offset 10 --format compact # → paginated + co ### Tooling (1) `plugin ` -**Total: 63 commands** (auto-registered via `commands/__init__.py`; rerun `python3 scripts/sync_command_count.py --apply` after adding/removing a command) +**Total: 65 commands** (auto-registered via `commands/__init__.py`; rerun `python3 scripts/sync_command_count.py --apply` after adding/removing a command) -## MCP Server (61 Tools) +## MCP Server (63 Tools) Start the MCP server for AI agent integration: @@ -156,9 +156,9 @@ Start the MCP server for AI agent integration: python3 scripts/codelens.py serve ``` -Exposes 61 tools as `codelens_` (e.g., `codelens_query`, `codelens_taint`, `codelens_graph_schema`, `codelens_architecture`, `codelens_resolve_types`, `codelens_git_status`): +Exposes 63 tools as `codelens_` (e.g., `codelens_query`, `codelens_taint`, `codelens_graph_schema`, `codelens_architecture`, `codelens_resolve_types`, `codelens_git_status`): - 50 statically-defined tools (full JSON schemas in `mcp_server.py`) -- 11 dynamically-discovered tools (auto-discovered from `COMMAND_REGISTRY`; long-running `watch` and `serve` are excluded) +- 13 dynamically-discovered tools (auto-discovered from `COMMAND_REGISTRY`; long-running `watch` and `serve` are excluded) - Every tool accepts a `format` parameter (`json`/`markdown`/`ai`/`sarif`/`compact`). Use `format: "compact"` for token-efficient responses (~50% smaller than `json`). - `watch` and `serve` itself are excluded (long-running) diff --git a/SKILL.md b/SKILL.md index e969de2e..cb753460 100755 --- a/SKILL.md +++ b/SKILL.md @@ -1,10 +1,10 @@ --- name: codelens description: > - CodeLens — AI-Native Code Intelligence. 63 commands for AI-powered code analysis, + CodeLens — AI-Native Code Intelligence. 65 commands for AI-powered code analysis, security auditing, quality scoring, AST-based taint analysis, live CVE scanning, and pre-write safety checks. Supports 28+ languages with tree-sitter + regex - fallback parsing. MCP server exposes 61 tools for AI agent integration. + fallback parsing. MCP server exposes 63 tools for AI agent integration. For quick command reference with validated output schemas, see SKILL-QUICK.md. For version history, see CHANGELOG.md. --- diff --git a/pyproject.toml b/pyproject.toml index 72b78d70..9bfe9646 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "codelens" version = "8.2.0" -description = "Live Codebase Reference Intelligence — 63 commands for AI-powered code analysis, security auditing, and quality scoring" +description = "Live Codebase Reference Intelligence — 65 commands for AI-powered code analysis, security auditing, and quality scoring" readme = "README.md" license = {text = "MIT"} requires-python = ">=3.8" diff --git a/scripts/commands/export_snapshot.py b/scripts/commands/export_snapshot.py new file mode 100644 index 00000000..74082f86 --- /dev/null +++ b/scripts/commands/export_snapshot.py @@ -0,0 +1,168 @@ +"""Export-snapshot command — Export the CodeLens graph as a compressed archive. + +Issue #12: each developer previously had to ``codelens scan`` separately. +``export-snapshot`` packages the SQLite graph tables (``graph_nodes``, +``graph_edges``, ``symbols``, ``refs``, ``files``) as a gzip-compressed +archive that can be committed to the repo and shared with the team via +``codelens import-snapshot``. + +The snapshot contains **graph metadata only** — file paths, symbol +names/kinds/line spans, edge relationships, content hashes, timestamps. +It NEVER contains file content. + +Usage:: + + codelens export-snapshot [workspace] [--output path] [--db-path path] + +Example output (the human-readable message is also embedded in the JSON +result under the ``message`` key and printed to stderr so stdout stays +machine-parseable):: + + Snapshot exported: .codelens/snapshot.codelens.gz (1.2 MB) +""" + +import os +import sys +from typing import Any, Dict, Optional + +from commands import register_command +from utils import default_db_path, logger +from snapshot_io import ( + DEFAULT_SNAPSHOT_FILENAME, + SNAPSHOT_TABLES, + build_snapshot, + default_snapshot_path, + format_size, + write_snapshot, +) + + +def add_args(parser): + """Add export-snapshot arguments to the parser.""" + parser.add_argument("workspace", nargs="?", default=None, + help="Path to workspace root (auto-detected if omitted)") + parser.add_argument("--output", default=None, + help="Output path for the snapshot archive " + "(default: .codelens/snapshot.codelens.gz)") + parser.add_argument("--db-path", default=None, + help="Custom path for the source SQLite database file") + + +def execute(args, workspace): + """Execute the export-snapshot command.""" + output = getattr(args, "output", None) + db_path = getattr(args, "db_path", None) + return cmd_export_snapshot(workspace, output_path=output, db_path=db_path) + + +def cmd_export_snapshot( + workspace: str, + output_path: Optional[str] = None, + db_path: Optional[str] = None, +) -> Dict[str, Any]: + """Export the CodeLens graph database to a compressed snapshot archive. + + Reads the SQLite graph tables (graph_nodes, graph_edges, symbols, + refs, files) and writes them as gzip-compressed JSON to + ``output_path`` (default: ``/.codelens/snapshot.codelens.gz``). + + Args: + workspace: Path to the workspace root. + output_path: Optional explicit output path. If None, defaults to + ``/.codelens/snapshot.codelens.gz``. + db_path: Optional source SQLite db path. Defaults to + ``/.codelens/codelens.db``. + + Returns: + Dict with keys: ``status``, ``message``, ``snapshot_path``, + ``size_bytes``, ``size_human``, ``header``, ``workspace``. + On error: ``status="error"`` with an ``error`` message. + """ + workspace = os.path.abspath(workspace) + + # Resolve source db path + effective_db = db_path or default_db_path(workspace) + + # Resolve output path. If the user gave a relative path, resolve it + # against the workspace root so the default ``.codelens/...`` form + # works regardless of cwd. + if output_path: + effective_output = output_path if os.path.isabs(output_path) \ + else os.path.join(workspace, output_path) + else: + effective_output = default_snapshot_path(workspace) + + # Build the snapshot (reads the DB, raises FileNotFoundError if absent) + try: + snapshot = build_snapshot(workspace, db_path=effective_db) + except FileNotFoundError as exc: + return { + "status": "error", + "error": str(exc), + "workspace": workspace, + } + except Exception as exc: # pragma: no cover - defensive + logger.error(f"export-snapshot: build failed: {exc}", exc_info=True) + return { + "status": "error", + "error": f"Failed to build snapshot: {exc}", + "workspace": workspace, + } + + header = snapshot.get("header", {}) + + # Write gzip-compressed JSON to disk + try: + size_bytes = write_snapshot(snapshot, effective_output) + except OSError as exc: + return { + "status": "error", + "error": f"Failed to write snapshot to {effective_output}: {exc}", + "workspace": workspace, + } + + size_human = format_size(size_bytes) + + # Human-readable message — also surfaced in the JSON result so + # scripted consumers can read it. Matches the issue #12 format: + # "Snapshot exported: .codelens/snapshot.codelens.gz (1.2 MB)" + # Use a workspace-relative path in the message when possible so the + # message is portable across machines (the snapshot is meant to be + # committed to the repo and shared). + try: + rel_output = os.path.relpath(effective_output, workspace) + # If the output lives outside the workspace, relpath produces + # something with '..' — fall back to the absolute path in that case. + if rel_output.startswith(".."): + display_path = effective_output + else: + display_path = rel_output + except (ValueError, OSError): + display_path = effective_output + + message = f"Snapshot exported: {display_path} ({size_human})" + + # Print the message to stderr so stdout (JSON) stays machine-clean, + # matching the convention used by scan's status messages. + print(message, file=sys.stderr) + + return { + "status": "ok", + "message": message, + "snapshot_path": effective_output, + "display_path": display_path, + "size_bytes": size_bytes, + "size_human": size_human, + "header": header, + "workspace": workspace, + "tables": list(SNAPSHOT_TABLES), + } + + +register_command( + "export-snapshot", + "Export the CodeLens graph as a compressed snapshot archive (.codelens.gz) " + "for team sharing (issue #12)", + add_args, + execute, +) diff --git a/scripts/commands/import_snapshot.py b/scripts/commands/import_snapshot.py new file mode 100644 index 00000000..974386c7 --- /dev/null +++ b/scripts/commands/import_snapshot.py @@ -0,0 +1,178 @@ +"""Import-snapshot command — Load a CodeLens graph snapshot into the database. + +Issue #12: companion to ``export-snapshot``. Restores a previously +exported snapshot (``.codelens/snapshot.codelens.gz``) into +``.codelens/codelens.db`` so a developer can use the shared team graph +without running a full ``codelens scan``. + +Validates the snapshot header and warns if it came from a different +CodeLens version. Supports ``--merge`` to combine the snapshot with the +existing graph (deduplicating nodes/edges by their natural key) instead +of the default replace behavior. + +Usage:: + + codelens import-snapshot [workspace] [--input path] [--merge] [--db-path path] + +The snapshot contains graph metadata only (paths, symbols, edges) — +never file content. +""" + +import os +import sqlite3 +import sys +from typing import Any, Dict, List, Optional + +from commands import register_command +from utils import default_db_path, logger +from snapshot_io import ( + SNAPSHOT_TABLES, + default_snapshot_path, + load_snapshot_into_db, + read_snapshot, + validate_header, +) + + +def add_args(parser): + """Add import-snapshot arguments to the parser.""" + parser.add_argument("workspace", nargs="?", default=None, + help="Path to workspace root (auto-detected if omitted)") + parser.add_argument("--input", default=None, + help="Input path for the snapshot archive " + "(default: .codelens/snapshot.codelens.gz)") + parser.add_argument("--merge", action="store_true", default=False, + help="Merge with existing graph (deduplicate nodes/edges " + "by id) instead of replacing the existing graph") + parser.add_argument("--db-path", default=None, + help="Custom path for the target SQLite database file") + + +def execute(args, workspace): + """Execute the import-snapshot command.""" + input_path = getattr(args, "input", None) + merge = getattr(args, "merge", False) + db_path = getattr(args, "db_path", None) + return cmd_import_snapshot( + workspace, input_path=input_path, merge=merge, db_path=db_path + ) + + +def cmd_import_snapshot( + workspace: str, + input_path: Optional[str] = None, + merge: bool = False, + db_path: Optional[str] = None, +) -> Dict[str, Any]: + """Load a snapshot archive into the CodeLens SQLite database. + + Args: + workspace: Path to the workspace root. + input_path: Optional explicit snapshot path. If None, defaults to + ``/.codelens/snapshot.codelens.gz``. + merge: If True, merge with existing data (deduplicate by natural + key). If False (default), replace existing graph tables. + db_path: Optional target SQLite db path. Defaults to + ``/.codelens/codelens.db``. + + Returns: + Dict with keys: ``status``, ``message``, ``mode``, ``warnings``, + ``inserted``, ``skipped``, ``header``, ``workspace``, ``db_path``. + On error: ``status="error"`` with an ``error`` message. + """ + workspace = os.path.abspath(workspace) + effective_db = db_path or default_db_path(workspace) + + # Resolve input path (relative paths resolve against the workspace root). + if input_path: + effective_input = input_path if os.path.isabs(input_path) \ + else os.path.join(workspace, input_path) + else: + effective_input = default_snapshot_path(workspace) + + # Read + parse the snapshot + try: + snapshot = read_snapshot(effective_input) + except FileNotFoundError as exc: + return { + "status": "error", + "error": str(exc), + "workspace": workspace, + } + except ValueError as exc: + return { + "status": "error", + "error": str(exc), + "workspace": workspace, + } + + header = snapshot.get("header", {}) + warnings: List[str] = validate_header(header) + + # Print warnings to stderr so they're visible without polluting stdout. + for w in warnings: + print(f"[CodeLens] Warning: {w}", file=sys.stderr) + + # Load into the database (creates schema if absent) + try: + result = load_snapshot_into_db( + snapshot, workspace, db_path=effective_db, merge=merge + ) + except sqlite3.Error as exc: + return { + "status": "error", + "error": f"Failed to import snapshot: {exc}", + "workspace": workspace, + } + except Exception as exc: + logger.error(f"import-snapshot: load failed: {exc}", exc_info=True) + return { + "status": "error", + "error": f"Failed to import snapshot: {exc}", + "workspace": workspace, + } + + inserted = result.get("inserted", {}) + skipped = result.get("skipped", {}) + total_inserted = sum(inserted.values()) + total_skipped = sum(skipped.values()) + + mode = "merge" if merge else "replace" + # Human-readable message; also embedded in the JSON result. + if merge: + message = ( + f"Snapshot imported ({mode}): {total_inserted} row(s) inserted, " + f"{total_skipped} duplicate(s) skipped — " + f"{total_inserted + total_skipped} total in snapshot." + ) + else: + message = ( + f"Snapshot imported ({mode}): {total_inserted} row(s) loaded " + f"across {len(SNAPSHOT_TABLES)} tables." + ) + print(message, file=sys.stderr) + + return { + "status": "ok", + "message": message, + "mode": mode, + "input_path": effective_input, + "warnings": warnings, + "inserted": inserted, + "skipped": skipped, + "total_inserted": total_inserted, + "total_skipped": total_skipped, + "header": header, + "workspace": workspace, + "db_path": effective_db, + "tables": list(SNAPSHOT_TABLES), + } + + +register_command( + "import-snapshot", + "Import a CodeLens graph snapshot (.codelens.gz) into the database; " + "use --merge to deduplicate with the existing graph (issue #12)", + add_args, + execute, +) diff --git a/scripts/graph_model.py b/scripts/graph_model.py index 971eabb8..d7f7d6d8 100644 --- a/scripts/graph_model.py +++ b/scripts/graph_model.py @@ -13,7 +13,7 @@ - New tables `graph_nodes` and `graph_edges` are additive (prefixed `graph_` to avoid colliding with any existing table name). - The flat registry tables and JSON files are untouched. -- All 63 existing CLI commands continue to work unchanged. +- All 65 existing CLI commands continue to work unchanged. Schema: graph_nodes( diff --git a/scripts/snapshot_io.py b/scripts/snapshot_io.py new file mode 100644 index 00000000..b32802da --- /dev/null +++ b/scripts/snapshot_io.py @@ -0,0 +1,442 @@ +"""Snapshot I/O helpers for CodeLens graph sharing (issue #12). + +Single source of truth for the on-disk snapshot format used by both the +``export-snapshot`` and ``import-snapshot`` CLI commands. Keeping the +format definition here (instead of duplicating it across two command +modules) guarantees export and import stay in lockstep. + +Snapshot format +--------------- +A snapshot is a **gzip-compressed JSON** document with this shape:: + + { + "header": { + "format_version": 1, + "codelens_version": "8.2.0", + "scan_timestamp": 1234567890.0, + "exported_at": "2024-01-01T00:00:00+00:00", + "file_count": 31, + "node_count": 31, + "edge_count": 97, + "table_counts": {"graph_nodes": 31, "graph_edges": 97, ...}, + "tables": ["graph_nodes", "graph_edges", "symbols", "refs", "files"], + "workspace": "/abs/path/to/workspace", + "note": "Contains graph metadata only — no file content." + }, + "data": { + "graph_nodes": {"columns": [...], "rows": [[...], ...]}, + "graph_edges": {"columns": [...], "rows": [[...], ...]}, + ... + } + } + +Constraint (issue #12): a snapshot MUST NOT contain file content. Only +graph metadata is exported — file paths, symbol names/kinds/line spans, +edge relationships, content hashes, and timestamps. The ``files`` table +stores ``content_hash`` (a digest), never the bytes themselves. +""" + +from __future__ import annotations + +import gzip +import json +import os +import sqlite3 +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional, Tuple + +from utils import CODELENS_VERSION, default_db_path, logger + + +# ─── Format constants ───────────────────────────────────────── + +#: Snapshot on-disk format version. Bump on backward-incompatible +#: structural changes; ``import-snapshot`` warns on mismatch. +SNAPSHOT_FORMAT_VERSION = 1 + +#: Default snapshot filename inside ``.codelens/``. +DEFAULT_SNAPSHOT_FILENAME = "snapshot.codelens.gz" + +#: Tables exported, in deterministic order. Only graph metadata — +#: never file content. See module docstring for the constraint. +SNAPSHOT_TABLES: List[str] = [ + "graph_nodes", + "graph_edges", + "symbols", + "refs", + "files", +] + +#: Column order for each exported table. Stored in the snapshot so +#: import can map columns by name (robust to future schema additions) +#: rather than by position. +TABLE_COLUMNS: Dict[str, List[str]] = { + "graph_nodes": ["id", "node_id", "node_type", "name", "file", "line", "extra_json"], + "graph_edges": [ + "id", "source_id", "target_id", "edge_type", + "file", "line", "confidence", "extra_json", + ], + "symbols": [ + "id", "name", "kind", "file_path", "line_start", "line_end", + "language", "signature", "hash", "extra_json", + ], + "refs": [ + "id", "source_symbol", "target_symbol", "reference_type", + "source_file", "extra_json", + ], + "files": [ + "id", "file_path", "language", "last_modified", + "content_hash", "last_scanned", + ], +} + +#: Natural-key columns used to deduplicate rows when importing in +#: ``--merge`` mode (issue #12). Rows whose natural key already +#: exists in the target table are skipped. The autoincrement ``id`` +#: column is intentionally NOT part of any natural key — import lets +#: SQLite assign fresh ids to avoid sequence collisions. +NATURAL_KEY_COLUMNS: Dict[str, List[str]] = { + "graph_nodes": ["node_id"], + "graph_edges": ["source_id", "target_id", "edge_type", "file", "line"], + "symbols": ["name", "kind", "file_path", "line_start"], + "refs": ["source_symbol", "target_symbol", "reference_type", "source_file"], + "files": ["file_path"], +} + + +# ─── Path helpers ───────────────────────────────────────────── + + +def default_snapshot_path(workspace: str) -> str: + """Return the default snapshot path: ``/.codelens/snapshot.codelens.gz``.""" + return os.path.join(workspace, ".codelens", DEFAULT_SNAPSHOT_FILENAME) + + +# ─── Size formatting ────────────────────────────────────────── + + +def format_size(num_bytes: int) -> str: + """Format a byte count as a human-readable string. + + Examples: ``512`` → ``"512 B"``; ``1500`` → ``"1.5 KB"``; + ``1250000`` → ``"1.2 MB"``. Used by the export command's + ``"Snapshot exported: ... (1.2 MB)"`` message. + """ + n = float(num_bytes) + for unit in ("B", "KB", "MB", "GB", "TB"): + if n < 1024.0 or unit == "TB": + if unit == "B": + return f"{int(n)} {unit}" + return f"{n:.1f} {unit}" + n /= 1024.0 + return f"{n:.1f} TB" + + +# ─── DB helpers ─────────────────────────────────────────────── + + +def _table_exists(conn: sqlite3.Connection, table: str) -> bool: + row = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name=?", + (table,), + ).fetchone() + return row is not None + + +def _quote_ident(name: str) -> str: + """Quote a SQL identifier (table/column) defensively. + + All identifiers here are hard-coded constants, but quoting keeps + static analyzers happy and is forward-compatible if a future + schema ever uses a reserved word. + """ + return '"' + name.replace('"', '""') + '"' + + +# ─── Snapshot construction (export side) ────────────────────── + + +def build_snapshot( + workspace: str, + db_path: Optional[str] = None, +) -> Dict[str, Any]: + """Read the SQLite graph tables and return a snapshot dict. + + Args: + workspace: Absolute path to the workspace root. + db_path: Optional SQLite db path. Defaults to + ``/.codelens/codelens.db``. + + Returns: + Snapshot dict with ``header`` and ``data`` keys (see module + docstring). + + Raises: + FileNotFoundError: if the database file does not exist. + """ + workspace = os.path.abspath(workspace) + db_path = db_path or default_db_path(workspace) + if not os.path.exists(db_path): + raise FileNotFoundError( + f"CodeLens database not found at {db_path}. " + f"Run 'codelens scan' first to build the graph." + ) + + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + try: + data: Dict[str, Any] = {} + counts: Dict[str, int] = {} + for table in SNAPSHOT_TABLES: + cols = TABLE_COLUMNS[table] + if not _table_exists(conn, table): + data[table] = {"columns": cols, "rows": []} + counts[table] = 0 + continue + col_list = ", ".join(_quote_ident(c) for c in cols) + rows = conn.execute( + f"SELECT {col_list} FROM {_quote_ident(table)}" + ).fetchall() + data[table] = { + "columns": cols, + "rows": [[r[i] for i in range(len(cols))] for r in rows], + } + counts[table] = len(rows) + + # Pull the original scan timestamp from scan_metadata (if present) + # so consumers can tell how stale the graph is. + scan_ts: float = 0.0 + if _table_exists(conn, "scan_metadata"): + r = conn.execute( + "SELECT scan_timestamp FROM scan_metadata WHERE id = 1" + ).fetchone() + if r is not None: + scan_ts = float(r[0] or 0.0) + + header: Dict[str, Any] = { + "format_version": SNAPSHOT_FORMAT_VERSION, + "codelens_version": CODELENS_VERSION, + "scan_timestamp": scan_ts, + "exported_at": datetime.now(timezone.utc).isoformat(), + "file_count": counts.get("files", 0), + "node_count": counts.get("graph_nodes", 0), + "edge_count": counts.get("graph_edges", 0), + "table_counts": counts, + "tables": list(SNAPSHOT_TABLES), + "workspace": workspace, + "note": "Contains graph metadata only — no file content.", + } + return {"header": header, "data": data} + finally: + conn.close() + + +def write_snapshot(snapshot: Dict[str, Any], output_path: str) -> int: + """Write a snapshot dict to ``output_path`` as gzip-compressed JSON. + + Creates parent directories as needed. + + Returns: + The size of the written file in bytes. + """ + parent = os.path.dirname(output_path) + if parent: + os.makedirs(parent, exist_ok=True) + payload = json.dumps(snapshot, default=str, ensure_ascii=False).encode("utf-8") + with gzip.open(output_path, "wb") as f: + f.write(payload) + return os.path.getsize(output_path) + + +# ─── Snapshot loading (import side) ─────────────────────────── + + +def read_snapshot(input_path: str) -> Dict[str, Any]: + """Read a gzip-compressed JSON snapshot from ``input_path``. + + Raises: + FileNotFoundError: if the file does not exist. + ValueError: if the file is not a valid snapshot. + """ + if not os.path.exists(input_path): + raise FileNotFoundError(f"Snapshot file not found: {input_path}") + try: + with gzip.open(input_path, "rb") as f: + payload = f.read() + except OSError as exc: + raise ValueError( + f"Failed to decompress snapshot {input_path}: {exc}" + ) from exc + try: + snapshot = json.loads(payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ValueError( + f"Snapshot {input_path} is not valid JSON: {exc}" + ) from exc + if not isinstance(snapshot, dict) or "header" not in snapshot or "data" not in snapshot: + raise ValueError( + f"Snapshot {input_path} is missing required 'header'/'data' keys." + ) + return snapshot + + +# ─── Snapshot import ────────────────────────────────────────── + + +def _ensure_schema(workspace: str, db_path: str) -> None: + """Create all CodeLens SQLite tables (graph + persistent registry) if absent. + + Delegates to ``PersistentRegistry._init_schema`` (which also calls + ``graph_model.init_graph_schema``) so the import target always has + the full schema, even on a fresh workspace that has never been scanned. + Idempotent — safe to call on an already-initialized database. + """ + from persistent_registry import PersistentRegistry # local import avoids cycles + reg = PersistentRegistry(workspace, db_path=db_path) + try: + reg._connect() # triggers _init_schema on first call + finally: + reg.close() + + +def _existing_natural_keys( + conn: sqlite3.Connection, table: str +) -> set: + """Return the set of existing natural-key tuples for ``table``. + + Used by ``--merge`` mode to skip rows whose natural key is already + present. Returns an empty set if the table doesn't exist. + """ + if not _table_exists(conn, table): + return set() + key_cols = NATURAL_KEY_COLUMNS[table] + col_list = ", ".join(_quote_ident(c) for c in key_cols) + rows = conn.execute( + f"SELECT {col_list} FROM {_quote_ident(table)}" + ).fetchall() + return {tuple(r[i] for i in range(len(key_cols))) for r in rows} + + +def load_snapshot_into_db( + snapshot: Dict[str, Any], + workspace: str, + db_path: Optional[str] = None, + merge: bool = False, +) -> Dict[str, Any]: + """Load a snapshot dict into the SQLite database at ``db_path``. + + Args: + snapshot: Snapshot dict (as returned by :func:`build_snapshot` / + :func:`read_snapshot`). + workspace: Absolute path to the workspace root. + db_path: Optional SQLite db path. Defaults to + ``/.codelens/codelens.db``. + merge: If True, merge with existing data — rows whose natural key + already exists are skipped (deduplication). If False (default), + the target tables are cleared before inserting (replace). + + Returns: + Dict with per-table inserted/skipped counts and the validated + header, e.g. ``{"inserted": {...}, "skipped": {...}, "header": {...}}``. + """ + workspace = os.path.abspath(workspace) + db_path = db_path or default_db_path(workspace) + _ensure_schema(workspace, db_path) + + data = snapshot.get("data", {}) + inserted: Dict[str, int] = {} + skipped: Dict[str, int] = {} + + conn = sqlite3.connect(db_path) + try: + for table in SNAPSHOT_TABLES: + cols = TABLE_COLUMNS[table] + # Insert columns exclude the autoincrement ``id`` so SQLite + # assigns fresh ids (avoids sequence collisions in both merge + # and replace modes). + insert_cols = [c for c in cols if c != "id"] + tbl_data = data.get(table, {}) + # Honor the snapshot's column list if present (forward-compat); + # otherwise fall back to the current schema's columns. + snap_cols = tbl_data.get("columns", cols) + col_index = {c: i for i, c in enumerate(snap_cols)} + rows = tbl_data.get("rows", []) + + if not rows: + inserted[table] = 0 + skipped[table] = 0 + continue + + if not merge: + conn.execute(f"DELETE FROM {_quote_ident(table)}") + existing_keys: set = set() + else: + existing_keys = _existing_natural_keys(conn, table) + + key_cols = NATURAL_KEY_COLUMNS[table] + key_idx = [col_index[c] for c in key_cols if c in col_index] + insert_idx = [col_index[c] for c in insert_cols if c in col_index] + + placeholders = ", ".join("?" for _ in insert_cols) + col_list_sql = ", ".join(_quote_ident(c) for c in insert_cols) + insert_sql = ( + f"INSERT INTO {_quote_ident(table)} ({col_list_sql}) " + f"VALUES ({placeholders})" + ) + + to_insert: List[Tuple[Any, ...]] = [] + ins_count = 0 + skip_count = 0 + for row in rows: + if merge and key_idx: + key = tuple(row[i] for i in key_idx) + if key in existing_keys: + skip_count += 1 + continue + existing_keys.add(key) + values = tuple(row[i] for i in insert_idx) + to_insert.append(values) + + if to_insert: + conn.executemany(insert_sql, to_insert) + ins_count = len(to_insert) + inserted[table] = ins_count + skipped[table] = skip_count + + conn.commit() + except sqlite3.Error as exc: + conn.rollback() + logger.error(f"Snapshot import failed: {exc}", exc_info=True) + raise + finally: + conn.close() + + return { + "inserted": inserted, + "skipped": skipped, + "header": snapshot.get("header", {}), + } + + +def validate_header(header: Dict[str, Any]) -> List[str]: + """Validate a snapshot header; return a list of warning messages. + + An empty list means no warnings. Currently checks: + - ``format_version`` mismatch (warns on future versions we don't know) + - ``codelens_version`` mismatch (warns if snapshot came from a + different CodeLens version — per issue #12 requirement) + """ + warnings: List[str] = [] + fmt_ver = header.get("format_version") + if fmt_ver is not None and fmt_ver != SNAPSHOT_FORMAT_VERSION: + warnings.append( + f"Snapshot format version is {fmt_ver}, expected " + f"{SNAPSHOT_FORMAT_VERSION}. Import may be incomplete or inaccurate." + ) + snap_ver = header.get("codelens_version") + if snap_ver is not None and snap_ver != CODELENS_VERSION: + warnings.append( + f"Snapshot was created with CodeLens v{snap_ver}; current version is " + f"v{CODELENS_VERSION}. Graph schema may differ — verify results." + ) + return warnings diff --git a/skill.json b/skill.json index f7e1290a..8041b7c8 100755 --- a/skill.json +++ b/skill.json @@ -1,7 +1,7 @@ { "name": "codelens", "version": "8.2.0", - "description": "Live Codebase Reference Intelligence. 63 commands for AI-powered code analysis, security auditing, quality scoring, and pre-write safety checks. Supports 28+ languages with regex+AST hybrid parsing. Must activate before writing/editing/deleting any class, id, or function.", + "description": "Live Codebase Reference Intelligence. 65 commands for AI-powered code analysis, security auditing, quality scoring, and pre-write safety checks. Supports 28+ languages with regex+AST hybrid parsing. Must activate before writing/editing/deleting any class, id, or function.", "author": "codelens", "command_categories": { "setup": [ diff --git a/tests/test_integration.py b/tests/test_integration.py index 7c67414a..4303c159 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -1,5 +1,5 @@ """ -Integration smoke tests for all 63 CodeLens commands. +Integration smoke tests for all 65 CodeLens commands. Tests that every command: 1. Runs without crash (valid JSON output) @@ -300,7 +300,7 @@ def test_command_registry_has_all_commands(self): sys.path.insert(0, SCRIPT_DIR) from commands import COMMAND_REGISTRY # Regression sentinel — see docstring above for update procedure. - EXPECTED_COMMAND_COUNT = 63 + EXPECTED_COMMAND_COUNT = 65 actual = len(COMMAND_REGISTRY) assert actual == EXPECTED_COMMAND_COUNT, ( f"Command count drift detected: expected {EXPECTED_COMMAND_COUNT}, " diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py new file mode 100644 index 00000000..51bb1044 --- /dev/null +++ b/tests/test_snapshot.py @@ -0,0 +1,487 @@ +"""Round-trip tests for export-snapshot + import-snapshot (issue #12). + +Verifies that exporting a CodeLens graph snapshot and importing it into +a fresh workspace produces a database with identical graph metadata: +same node/edge/symbol/ref/file rows (modulo the autoincrement ``id`` +column, which is intentionally not preserved across export/import). + +Also covers: +- The export ``"Snapshot exported: ... (N.N MB)"`` message format. +- ``--merge`` deduplication (importing the same snapshot twice does not + duplicate rows). +- Version-mismatch validation warnings. +- The constraint that the snapshot contains metadata only (no file + content is stored in any of the exported tables). +""" + +import json +import os +import shutil +import sqlite3 +import sys +import tempfile + +import pytest + +# Add scripts directory to path (matches other test files). +SCRIPT_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts" +) +sys.path.insert(0, SCRIPT_DIR) + +from commands import COMMAND_REGISTRY # noqa: E402 +from snapshot_io import ( # noqa: E402 + DEFAULT_SNAPSHOT_FILENAME, + SNAPSHOT_TABLES, + TABLE_COLUMNS, + default_snapshot_path, + format_size, +) + + +# ─── Fixtures ───────────────────────────────────────────────── + + +@pytest.fixture +def workspace_a(): + """A temp workspace with a populated CodeLens database. + + The DB schema is initialized via PersistentRegistry (which also + creates the graph_* tables), then a small known set of rows is + inserted directly so the round-trip has deterministic data to + compare against. + """ + workspace = tempfile.mkdtemp(prefix="codelens_snap_export_") + try: + _populate_workspace(workspace) + yield workspace + finally: + shutil.rmtree(workspace, ignore_errors=True) + + +@pytest.fixture +def workspace_b(): + """A fresh, empty temp workspace (import target).""" + workspace = tempfile.mkdtemp(prefix="codelens_snap_import_") + try: + yield workspace + finally: + shutil.rmtree(workspace, ignore_errors=True) + + +def _populate_workspace(workspace: str) -> None: + """Initialize the DB schema and insert a small known graph.""" + from persistent_registry import PersistentRegistry + + # Initializing the registry creates all tables (symbols, refs, files, + # analysis_cache, scan_metadata + graph_nodes + graph_edges). + reg = PersistentRegistry(workspace) + reg._connect() + reg.close() + + db_path = os.path.join(workspace, ".codelens", "codelens.db") + conn = sqlite3.connect(db_path) + try: + # graph_nodes — (node_id, node_type, name, file, line, extra_json) + conn.executemany( + "INSERT INTO graph_nodes " + "(node_id, node_type, name, file, line, extra_json) " + "VALUES (?, ?, ?, ?, ?, ?)", + [ + ("src/app.py:10:main", "function", "main", "src/app.py", 10, None), + ("src/app.py:20:helper", "function", "helper", "src/app.py", 20, + json.dumps({"async": True})), + ("src/models.py:5:User", "class", "User", "src/models.py", 5, None), + ], + ) + # graph_edges — (source_id, target_id, edge_type, file, line, confidence, extra_json) + conn.executemany( + "INSERT INTO graph_edges " + "(source_id, target_id, edge_type, file, line, confidence, extra_json) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + [ + ("src/app.py:10:main", "src/app.py:20:helper", "CALLS", + "src/app.py", 10, 1.0, None), + ("src/app.py:10:main", "src/models.py:5:User", "USES_TYPE", + "src/app.py", 12, 0.9, json.dumps({"to_fn": "User"})), + ], + ) + # symbols — (name, kind, file_path, line_start, line_end, language, signature, hash, extra_json) + conn.executemany( + "INSERT INTO symbols " + "(name, kind, file_path, line_start, line_end, language, signature, hash, extra_json) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + [ + ("main", "function", "src/app.py", 10, 15, "python", + "def main()", "sha1:abc", None), + ("User", "class", "src/models.py", 5, 30, "python", + "class User", "sha1:def", None), + ], + ) + # refs — (source_symbol, target_symbol, reference_type, source_file, extra_json) + conn.executemany( + "INSERT INTO refs " + "(source_symbol, target_symbol, reference_type, source_file, extra_json) " + "VALUES (?, ?, ?, ?, ?)", + [ + ("src/app.py:10:main", "src/app.py:20:helper", "call", + "src/app.py", None), + ], + ) + # files — (file_path, language, last_modified, content_hash, last_scanned) + conn.executemany( + "INSERT INTO files " + "(file_path, language, last_modified, content_hash, last_scanned) " + "VALUES (?, ?, ?, ?, ?)", + [ + ("src/app.py", "python", 1700000000.0, "sha1:aaa", 1700000000.0), + ("src/models.py", "python", 1700000001.0, "sha1:bbb", 1700000001.0), + ], + ) + # scan_metadata — (id=1, workspace, scan_timestamp, total_files, version) + conn.execute( + "INSERT OR REPLACE INTO scan_metadata " + "(id, workspace, scan_timestamp, total_files, version) " + "VALUES (1, ?, ?, ?, ?)", + (workspace, 1700000000.0, 2, 1), + ) + conn.commit() + finally: + conn.close() + + +def _table_rows(db_path: str, table: str, exclude_id: bool = True): + """Return rows from ``table`` as a sorted list of tuples. + + The autoincrement ``id`` column is excluded by default so rows can be + compared across databases (import assigns fresh ids). Rows are sorted + for deterministic comparison. + """ + conn = sqlite3.connect(db_path) + try: + cols = TABLE_COLUMNS[table] + if exclude_id: + cols = [c for c in cols if c != "id"] + col_list = ", ".join('"' + c + '"' for c in cols) + rows = conn.execute(f'SELECT {col_list} FROM "{table}"').fetchall() + # Normalize: convert each row to a tuple of JSON-stringifiable values + # so json-encoded extra_json strings compare equal. + return sorted(tuple(r[i] for i in range(len(cols))) for r in rows) + finally: + conn.close() + + +# ─── Registration ───────────────────────────────────────────── + + +class TestCommandRegistration: + """Both commands must auto-register via register_command().""" + + def test_export_snapshot_registered(self): + assert "export-snapshot" in COMMAND_REGISTRY + info = COMMAND_REGISTRY["export-snapshot"] + assert info["help"] + assert callable(info["add_args"]) + assert callable(info["execute"]) + + def test_import_snapshot_registered(self): + assert "import-snapshot" in COMMAND_REGISTRY + info = COMMAND_REGISTRY["import-snapshot"] + assert info["help"] + assert callable(info["add_args"]) + assert callable(info["execute"]) + + +# ─── Round-trip ─────────────────────────────────────────────── + + +class TestRoundTrip: + """export → import → query produces the same results.""" + + def test_round_trip_preserves_all_tables(self, workspace_a, workspace_b): + """Export from A, import into B, then every table must match.""" + from commands.export_snapshot import cmd_export_snapshot + from commands.import_snapshot import cmd_import_snapshot + + snapshot_path = os.path.join(workspace_a, ".codelens", + DEFAULT_SNAPSHOT_FILENAME) + + # ── Export from workspace A ── + export_result = cmd_export_snapshot(workspace_a) + assert export_result["status"] == "ok", export_result + assert os.path.exists(snapshot_path), "snapshot file was not written" + + header = export_result["header"] + assert header["node_count"] == 3 + assert header["edge_count"] == 2 + assert header["file_count"] == 2 + assert header["format_version"] == 1 + assert header["codelens_version"] # non-empty + + # ── Import into workspace B (fresh, empty) ── + import_result = cmd_import_snapshot( + workspace_b, input_path=snapshot_path, merge=False + ) + assert import_result["status"] == "ok", import_result + # Same-version import should produce no warnings. + assert import_result["warnings"] == [] + assert import_result["mode"] == "replace" + + db_a = os.path.join(workspace_a, ".codelens", "codelens.db") + db_b = os.path.join(workspace_b, ".codelens", "codelens.db") + assert os.path.exists(db_b), "import did not create a database" + + # ── Every exported table must round-trip identically ── + for table in SNAPSHOT_TABLES: + rows_a = _table_rows(db_a, table) + rows_b = _table_rows(db_b, table) + assert rows_a == rows_b, ( + f"Table '{table}' differs after round-trip:\n" + f" source rows: {rows_a}\n" + f" imported rows: {rows_b}" + ) + + def test_export_message_format(self, workspace_a): + """Export message must match 'Snapshot exported: ()'.""" + from commands.export_snapshot import cmd_export_snapshot + + result = cmd_export_snapshot(workspace_a) + assert result["status"] == "ok" + msg = result["message"] + assert msg.startswith("Snapshot exported: "), msg + # Default display path is workspace-relative. + assert ".codelens/snapshot.codelens.gz" in msg, msg + # Size appears in parentheses with a unit suffix. + assert " (" in msg and msg.rstrip().endswith(")"), msg + assert any(unit in msg for unit in (" B", " KB", " MB", " GB")), msg + # size_human must match the parenthesized portion. + assert f"({result['size_human']})" in msg, msg + + def test_graph_schema_command_matches_after_round_trip( + self, workspace_a, workspace_b + ): + """The graph-schema command must report identical stats after import.""" + from commands.export_snapshot import cmd_export_snapshot + from commands.import_snapshot import cmd_import_snapshot + from commands.graph_schema import get_graph_schema + + snapshot_path = os.path.join(workspace_a, ".codelens", + DEFAULT_SNAPSHOT_FILENAME) + assert cmd_export_snapshot(workspace_a)["status"] == "ok" + assert cmd_import_snapshot( + workspace_b, input_path=snapshot_path + )["status"] == "ok" + + schema_a = get_graph_schema(workspace_a) + schema_b = get_graph_schema(workspace_b) + # Compare the queryable graph shape (ignore workspace path). + for key in ("nodes", "edges", "node_types", "edge_types", "indexes"): + assert schema_a[key] == schema_b[key], ( + f"graph-schema '{key}' differs: {schema_a[key]} vs {schema_b[key]}" + ) + + +# ─── Merge mode ─────────────────────────────────────────────── + + +class TestMergeMode: + """--merge deduplicates nodes/edges by their natural key.""" + + def test_import_twice_replace_doubles_then_merge_noop(self, workspace_a, workspace_b): + """Replace import reproduces source; a second merge import adds nothing.""" + from commands.export_snapshot import cmd_export_snapshot + from commands.import_snapshot import cmd_import_snapshot + + snapshot_path = os.path.join(workspace_a, ".codelens", + DEFAULT_SNAPSHOT_FILENAME) + assert cmd_export_snapshot(workspace_a)["status"] == "ok" + + # First import (replace) into empty B. + r1 = cmd_import_snapshot(workspace_b, input_path=snapshot_path, merge=False) + assert r1["status"] == "ok" + assert r1["total_inserted"] > 0 + assert r1["total_skipped"] == 0 + + db_b = os.path.join(workspace_b, ".codelens", "codelens.db") + + # Row counts after first import. + counts_after_first = { + t: len(_table_rows(db_b, t)) for t in SNAPSHOT_TABLES + } + + # Second import with --merge: every row's natural key already + # exists, so all rows must be skipped (0 inserted). + r2 = cmd_import_snapshot(workspace_b, input_path=snapshot_path, merge=True) + assert r2["status"] == "ok" + assert r2["mode"] == "merge" + assert r2["total_inserted"] == 0, ( + f"merge re-import should insert 0 rows, got {r2['total_inserted']}" + ) + assert r2["total_skipped"] > 0 + + # Row counts must be unchanged after the merge re-import. + counts_after_second = { + t: len(_table_rows(db_b, t)) for t in SNAPSHOT_TABLES + } + assert counts_after_first == counts_after_second, ( + f"merge re-import changed row counts: " + f"{counts_after_first} -> {counts_after_second}" + ) + + def test_merge_combines_disjoint_graphs(self, workspace_a, workspace_b): + """Merging a snapshot with extra nodes adds only the new ones.""" + from commands.export_snapshot import cmd_export_snapshot + from commands.import_snapshot import cmd_import_snapshot + + snapshot_path = os.path.join(workspace_a, ".codelens", + DEFAULT_SNAPSHOT_FILENAME) + assert cmd_export_snapshot(workspace_a)["status"] == "ok" + + # Seed workspace B with ONE node that is NOT in the snapshot. + from persistent_registry import PersistentRegistry + reg = PersistentRegistry(workspace_b) + reg._connect() + reg.close() + db_b = os.path.join(workspace_b, ".codelens", "codelens.db") + conn = sqlite3.connect(db_b) + try: + conn.execute( + "INSERT INTO graph_nodes " + "(node_id, node_type, name, file, line, extra_json) " + "VALUES (?, ?, ?, ?, ?, ?)", + ("src/extra.py:1:only_in_b", "function", "only_in_b", + "src/extra.py", 1, None), + ) + conn.commit() + finally: + conn.close() + + # Merge-import the snapshot — the 3 snapshot nodes are new, + # so they're inserted; the existing B-only node survives. + r = cmd_import_snapshot(workspace_b, input_path=snapshot_path, merge=True) + assert r["status"] == "ok" + assert r["mode"] == "merge" + # All 3 graph_nodes from the snapshot are new relative to B. + assert r["inserted"]["graph_nodes"] == 3 + assert r["skipped"]["graph_nodes"] == 0 + + # Final node count = 1 (pre-existing) + 3 (merged in) = 4. + final_nodes = len(_table_rows(db_b, "graph_nodes")) + assert final_nodes == 4, f"expected 4 nodes after merge, got {final_nodes}" + + +# ─── Validation ─────────────────────────────────────────────── + + +class TestValidation: + """Version-mismatch warnings and error handling.""" + + def test_version_mismatch_warns(self, workspace_a, workspace_b): + """A snapshot with a different codelens_version must warn on import.""" + from commands.export_snapshot import cmd_export_snapshot + from commands.import_snapshot import cmd_import_snapshot + from snapshot_io import read_snapshot, write_snapshot + + snapshot_path = os.path.join(workspace_a, ".codelens", + DEFAULT_SNAPSHOT_FILENAME) + assert cmd_export_snapshot(workspace_a)["status"] == "ok" + + # Tamper with the version to simulate a cross-version import. + snap = read_snapshot(snapshot_path) + snap["header"]["codelens_version"] = "0.0.0-mock" + write_snapshot(snap, snapshot_path) + + r = cmd_import_snapshot( + workspace_b, input_path=snapshot_path, merge=False + ) + assert r["status"] == "ok" + assert any("0.0.0-mock" in w for w in r["warnings"]), r["warnings"] + assert any("different" in w.lower() or "version" in w.lower() + for w in r["warnings"]), r["warnings"] + + def test_import_missing_snapshot_returns_error(self, workspace_b): + """Importing a non-existent snapshot must return a clean error.""" + from commands.import_snapshot import cmd_import_snapshot + + bogus = os.path.join(workspace_b, ".codelens", "nope.codelens.gz") + r = cmd_import_snapshot(workspace_b, input_path=bogus) + assert r["status"] == "error" + assert "not found" in r["error"].lower() or "nope" in r["error"] + + def test_export_missing_db_returns_error(self, workspace_b): + """Exporting with no database present must return a clean error.""" + from commands.export_snapshot import cmd_export_snapshot + + r = cmd_export_snapshot(workspace_b) + assert r["status"] == "error" + assert "not found" in r["error"].lower() or "scan" in r["error"].lower() + + +# ─── Constraint: no file content ────────────────────────────── + + +class TestNoFileContent: + """Issue #12 constraint: the snapshot must NOT contain file content.""" + + def test_snapshot_contains_no_content_blobs(self, workspace_a): + """No exported column may hold raw source content. + + The ``files`` table holds ``content_hash`` (a digest), not bytes. + ``symbols`` holds ``signature`` (a parsed signature string), not + source. This test asserts the snapshot's data shape stays free + of any obvious content-bearing column by checking the exported + column names against a denylist. + """ + from commands.export_snapshot import cmd_export_snapshot + from snapshot_io import read_snapshot, default_snapshot_path + + result = cmd_export_snapshot(workspace_a) + assert result["status"] == "ok" + + snap = read_snapshot(default_snapshot_path(workspace_a)) + data = snap["data"] + + # Forbidden column names — if any appear, file content is leaking + # into the snapshot and the issue #12 constraint is violated. + forbidden = {"content", "source", "body", "text", "raw", "bytes", "code"} + for table, tbl in data.items(): + cols = set(tbl.get("columns", [])) + leaked = cols & forbidden + assert not leaked, ( + f"Table '{table}' exports content-bearing columns {leaked} " + f"— issue #12 forbids file content in snapshots." + ) + + # Spot-check: the files table must export content_hash, not content. + assert "content_hash" in data["files"]["columns"] + assert "content" not in data["files"]["columns"] + # And no row value may be a multi-KB blob (sanity cap). + for table in SNAPSHOT_TABLES: + for row in data[table]["rows"]: + for val in row: + if isinstance(val, str): + assert len(val) < 8192, ( + f"Suspiciously large string value ({len(val)} chars) " + f"in table '{table}' — possible content leak." + ) + + +# ─── format_size helper ─────────────────────────────────────── + + +class TestFormatSize: + def test_bytes(self): + assert format_size(512) == "512 B" + + def test_kilobytes(self): + assert format_size(1500) == "1.5 KB" + + def test_megabytes(self): + assert format_size(1250000) == "1.2 MB" + + def test_zero(self): + assert format_size(0) == "0 B" + + +if __name__ == "__main__": + pytest.main([__file__, "-v", "--tb=short"])