diff --git a/DEVELOPER_DOCS.md b/DEVELOPER_DOCS.md index 9e15ad5..ee46a94 100644 --- a/DEVELOPER_DOCS.md +++ b/DEVELOPER_DOCS.md @@ -4,6 +4,30 @@ This document provides instructions for developing, testing, configuring, and ru --- +## 🗄️ Relational Database Selection, Schema Migration & Rollback + +ContextCortex supports unified relational database selection across **SQLite** (default local embedded storage) and **PostgreSQL** (enterprise local deployment profiles). + +### Database Engine Selection +- **SQLite (Default)**: Used when `DATABASE_URL` is unset or starts with `sqlite:///`. Relational metadata and index caches are stored at `/app/data/index_cache.db` (configurable via `CACHE_DB_PATH`). +- **PostgreSQL**: Selected by setting `DATABASE_URL` to a valid PostgreSQL connection string (e.g., `postgresql+psycopg://user:password@postgres:5432/contextcortex`). + - When PostgreSQL is selected, all relational access across repositories, settings, credentials, indexing, file summaries, AST symbols, API routes, relationships, search, ADRs, and file deletion routes through the unified `PostgresConnectionAdapter`. + - **No Fallback SQLite Files**: When PostgreSQL is selected, the application operates exclusively against PostgreSQL and will never create or query an unrelated fallback SQLite file. + +### Schema Migration +1. **Automated Schema Initialization**: Schema tables and initial seed data (custom prompts, system metadata, vector store configuration) are initialized on startup via `init_db()`. +2. **Canonical SQLAlchemy Core Schema**: Table definitions are specified in `app/services/database/schema.py`. Running `init_db()` calls `metadata.create_all(bind=engine)` to idempotently create missing tables across SQLite and PostgreSQL. +3. **Column Migration Checks**: `ensure_git_repositories_columns()` and `ensure_file_summaries_columns()` verify table columns on startup (e.g. `keep_shallow`, `summary_text`) using `information_schema` on PostgreSQL or `PRAGMA table_info` on SQLite. +4. **pgvector Extension**: When initializing PostgreSQL, `init_db()` executes `CREATE EXTENSION IF NOT EXISTS vector;` to enable vector operations for pgvector storage. + +### Database Rollback Strategy +If you need to roll back or switch storage engines between PostgreSQL and SQLite: +1. **Switch Connection URL**: Set `DATABASE_URL` to the desired storage target (e.g., unset `DATABASE_URL` or set `DATABASE_URL="sqlite:////app/data/index_cache.db"` to revert to SQLite). +2. **Initialize Target Schema**: Start the application (`python main.py`) or call `init_db()`. The database engine factory initializes target schema tables idempotently without modifying or destroying data in the previously active database. +3. **Re-index Content**: Trigger re-indexing (`run_full_indexing()` or via the UI/API) to populate repository files, symbols, and summaries into the target relational database. + +--- + ## 🛠️ Prerequisites - **Python 3.11+** diff --git a/REQUIREMENTS.md b/REQUIREMENTS.md index 59d9c13..feb96c5 100644 --- a/REQUIREMENTS.md +++ b/REQUIREMENTS.md @@ -2,7 +2,7 @@ > **Note:** This document is automatically generated and verified against the live test suite by `scripts/generate_requirements.py` and `tests/backend/test_requirements_sync.py`. -**Test Verification Baseline:** **1052 Automated Tests** (699 Pytest Backend + 303 Vitest Frontend + 50 Playwright E2E). +**Test Verification Baseline:** **1057 Automated Tests** (704 Pytest Backend + 303 Vitest Frontend + 50 Playwright E2E). --- @@ -1179,6 +1179,16 @@ and leaves the prior indexed state intact without data loss._ - `test_poller_worker_cycle` - `test_poller_worker_disabled_interval` +#### `tests/test_postgres_unified_connection.py` (5 tests) +- `test_dict_row_compatibility` - _Verify DictRow supports dict lookup, index lookup, attribute access, and dict conversion._ +- `test_convert_sql_placeholders` - _Verify placeholder conversion converts ? to :p1, :p2 and appends RETURNING id for inserts._ +- `test_postgres_selection_never_creates_fallback_sqlite` - _Asserts that when PostgreSQL is selected, get_db_connection() returns the PostgreSQL adapter +and NEVER creates or queries the fallback SQLite index_cache.db file._ +- `test_postgres_full_relational_crud_and_indexing` - _Tests full relational lifecycle (repositories, settings, credentials, indexing, +file summaries, AST symbols/routes/relationships, ADRs, search and deletion) +over the PostgresConnectionAdapter using an initialized test engine._ +- `test_ensure_columns_postgres` - _Verify ensure_git_repositories_columns and ensure_file_summaries_columns run without error when on PostgreSQL._ + #### `tests/test_processor_caching.py` (6 tests) - `test_compute_text_hash` - `test_process_file_content_populates_embedding_cache` diff --git a/app/services/database/connection.py b/app/services/database/connection.py index 4158996..5318393 100644 --- a/app/services/database/connection.py +++ b/app/services/database/connection.py @@ -18,18 +18,36 @@ logger = logging.getLogger("contextcortex.db") CACHE_DB_PATH = get_default_db_path() -try: - os.makedirs(os.path.dirname(CACHE_DB_PATH), exist_ok=True) -except Exception: - pass +if not is_postgres(): + try: + os.makedirs(os.path.dirname(CACHE_DB_PATH), exist_ok=True) + except Exception: + pass +from app.services.database.postgres_compat import ( + DictRow, + TABLES_WITH_ID, + _convert_sql_placeholders, + _SingleRowResult, + PostgresResultAdapter, + PostgresCursorAdapter, + PostgresConnectionAdapter, +) -def get_db_connection() -> sqlite3.Connection: +def get_db_connection() -> Any: """ - Returns a SQLite database connection for backward compatibility with legacy modules. + Returns a database connection (PostgreSQL adapter or SQLite connection) + based on database selection. """ + if is_postgres(): + return PostgresConnectionAdapter() + from app.services.database.engine import get_current_db_path db_path = get_current_db_path() + try: + os.makedirs(os.path.dirname(db_path), exist_ok=True) + except Exception: + pass conn = sqlite3.connect(db_path, timeout=15.0) conn.row_factory = sqlite3.Row @@ -38,14 +56,12 @@ def get_db_connection() -> sqlite3.Connection: conn.execute("PRAGMA foreign_keys=ON;") return conn - def init_db(vault_path: str = "/docs", engine=None): """ Initializes database schema and seed data via SQLAlchemy Core engine. """ return engine_init_db(vault_path=vault_path, engine=engine) - def get_metadata(key: str, default: Optional[str] = None) -> Optional[str]: try: with get_connection() as conn: @@ -56,7 +72,6 @@ def get_metadata(key: str, default: Optional[str] = None) -> Optional[str]: except Exception: return default - def set_metadata(key: str, value: str): try: with get_connection() as conn: @@ -70,7 +85,6 @@ def set_metadata(key: str, value: str): except Exception as e: logger.error(f"Failed to set metadata key {key}: {e}") - def get_default_vector_storage_path() -> str: env_path = os.getenv("VECTOR_STORE_STORAGE_PATH") if env_path: @@ -79,7 +93,6 @@ def get_default_vector_storage_path() -> str: return "/app/data/vector_storage" return os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data", "vector_storage") - def _resolve_default_vector_store_config(conn: Optional[Any] = None) -> Dict[str, str]: provider = os.getenv("VECTOR_STORE") or os.getenv("VECTOR_STORE_PROVIDER") or ( "pgvector" if is_postgres() else "qdrant" @@ -127,7 +140,6 @@ def _resolve_default_vector_store_config(conn: Optional[Any] = None) -> Dict[str "collection": collection.strip(), } - def get_vector_store_db_config() -> Dict[str, str]: provider = get_metadata("vector_store_provider") mode = get_metadata("vector_store_mode") @@ -145,7 +157,6 @@ def get_vector_store_db_config() -> Dict[str, str]: "collection": (collection or default_cfg["collection"]).strip(), } - def set_vector_store_db_config( provider: Optional[str] = None, mode: Optional[str] = None, @@ -164,7 +175,6 @@ def set_vector_store_db_config( if collection is not None: set_metadata("vector_store_collection", collection.strip()) - def detect_system_resources() -> Dict[str, Any]: """Detects available CPU cores and RAM in the execution environment, respecting container quotas.""" import math @@ -200,7 +210,6 @@ def detect_system_resources() -> Dict[str, Any]: pass return {"cpus": cpus, "memory_gb": mem_gb} - def _resolve_default_embedding_config(conn: Optional[Any] = None) -> Dict[str, Any]: sys_res = detect_system_resources() default_threads = min(2, max(1, sys_res["cpus"])) @@ -234,7 +243,6 @@ def _resolve_default_embedding_config(conn: Optional[Any] = None) -> Dict[str, A "system_memory_gb": sys_res["memory_gb"], } - def get_embedding_db_config() -> Dict[str, Any]: provider = get_metadata("embedding_provider") dense_model = get_metadata("embedding_dense_model") @@ -273,7 +281,6 @@ def get_embedding_db_config() -> Dict[str, Any]: "system_memory_gb": default_cfg["system_memory_gb"], } - def set_embedding_db_config( provider: Optional[str] = None, dense_model: Optional[str] = None, @@ -308,7 +315,6 @@ def set_embedding_db_config( set_metadata("chat_model", chat_model.strip()) set_metadata("embedding_chat_model", chat_model.strip()) - def get_vision_ocr_model() -> str: """Returns stored vision OCR model or fallback to env/default.""" stored = get_metadata("openai_vision_model") or get_metadata("vision_ocr_model") or get_metadata("embedding_vision_ocr_model") @@ -316,7 +322,6 @@ def get_vision_ocr_model() -> str: return stored.strip() return (os.getenv("VISION_OCR_MODEL") or "").strip() - def get_chat_model() -> str: """Returns stored chat model or fallback to env/default.""" stored = get_metadata("openai_chat_model") or get_metadata("chat_model") or get_metadata("embedding_chat_model") @@ -324,7 +329,6 @@ def get_chat_model() -> str: return stored.strip() return (os.getenv("CHAT_MODEL") or "").strip() - def get_ai_gateway_config() -> Dict[str, Any]: """ Returns OpenAI-compatible AI Gateway configuration from database. @@ -366,7 +370,6 @@ def get_ai_gateway_config() -> Dict[str, Any]: "embedding_model": embedding_model, } - def set_ai_gateway_config( url: Optional[str] = None, api_key: Optional[str] = None, @@ -397,40 +400,57 @@ def set_ai_gateway_config( return get_ai_gateway_config() - def ensure_git_repositories_columns(conn=None): """Ensures keep_shallow column exists on git_repositories table.""" try: - raw_conn = get_db_connection() - try: - cols = [r["name"] for r in raw_conn.execute("PRAGMA table_info(git_repositories)").fetchall()] - if cols and "keep_shallow" not in cols: - raw_conn.execute("ALTER TABLE git_repositories ADD COLUMN keep_shallow INTEGER DEFAULT 0") - raw_conn.commit() - logger.info("Migrated git_repositories: added keep_shallow column") - finally: - raw_conn.close() + if is_postgres(): + from sqlalchemy import text + with get_connection() as sa_conn: + res = sa_conn.execute( + text("SELECT column_name FROM information_schema.columns WHERE table_name='git_repositories' AND column_name='keep_shallow'") + ).fetchone() + if not res: + sa_conn.execute(text("ALTER TABLE git_repositories ADD COLUMN keep_shallow INTEGER DEFAULT 0")) + sa_conn.commit() + logger.info("Migrated git_repositories: added keep_shallow column") + else: + raw_conn = get_db_connection() + try: + cols = [r["name"] for r in raw_conn.execute("PRAGMA table_info(git_repositories)").fetchall()] + if cols and "keep_shallow" not in cols: + raw_conn.execute("ALTER TABLE git_repositories ADD COLUMN keep_shallow INTEGER DEFAULT 0") + raw_conn.commit() + logger.info("Migrated git_repositories: added keep_shallow column") + finally: + raw_conn.close() except Exception as e: logger.debug(f"Migration error for git_repositories: {e}") - def ensure_file_summaries_columns(conn=None): """Ensures summary_text column exists on file_summaries table and git_repositories has keep_shallow.""" try: ensure_git_repositories_columns(conn) - raw_conn = get_db_connection() - try: - cols = [r["name"] for r in raw_conn.execute("PRAGMA table_info(file_summaries)").fetchall()] - if cols and "summary_text" not in cols: - raw_conn.execute("ALTER TABLE file_summaries ADD COLUMN summary_text TEXT") - raw_conn.commit() - finally: - raw_conn.close() + if is_postgres(): + from sqlalchemy import text + with get_connection() as sa_conn: + res = sa_conn.execute( + text("SELECT column_name FROM information_schema.columns WHERE table_name='file_summaries' AND column_name='summary_text'") + ).fetchone() + if not res: + sa_conn.execute(text("ALTER TABLE file_summaries ADD COLUMN summary_text TEXT")) + sa_conn.commit() + else: + raw_conn = get_db_connection() + try: + cols = [r["name"] for r in raw_conn.execute("PRAGMA table_info(file_summaries)").fetchall()] + if cols and "summary_text" not in cols: + raw_conn.execute("ALTER TABLE file_summaries ADD COLUMN summary_text TEXT") + raw_conn.commit() + finally: + raw_conn.close() except Exception as e: logger.debug(f"Migration error for file_summaries: {e}") - - def get_file_settings() -> Dict[str, Any]: """Retrieves file reading and large file summarization settings.""" enabled_val = get_metadata("file_summary_enabled") @@ -447,7 +467,6 @@ def get_file_settings() -> Dict[str, Any]: "summary_chat_model": (model_val or get_chat_model()).strip(), } - def set_file_settings(payload: Dict[str, Any]) -> Dict[str, Any]: """Updates file reading and summarization settings.""" if "summary_enabled" in payload: diff --git a/app/services/database/credentials.py b/app/services/database/credentials.py index cba4f0a..5575f5e 100644 --- a/app/services/database/credentials.py +++ b/app/services/database/credentials.py @@ -33,8 +33,12 @@ def save_git_host_credential(host: str, provider: str, auth_token: str, auth_use with get_db_connection() as conn: cur = conn.cursor() cur.execute( - """INSERT OR REPLACE INTO git_host_credentials (host, provider, auth_user, auth_token) - VALUES (?, ?, ?, ?)""", + """INSERT INTO git_host_credentials (host, provider, auth_user, auth_token) + VALUES (?, ?, ?, ?) + ON CONFLICT(host) DO UPDATE SET + provider = EXCLUDED.provider, + auth_user = EXCLUDED.auth_user, + auth_token = EXCLUDED.auth_token""", (clean_host, provider.lower().strip(), auth_user.strip() if auth_user else None, auth_token.strip()) ) conn.commit() diff --git a/app/services/database/embedding_cache.py b/app/services/database/embedding_cache.py index 8838bd4..060c0fc 100644 --- a/app/services/database/embedding_cache.py +++ b/app/services/database/embedding_cache.py @@ -47,8 +47,12 @@ def set_cached_embeddings_batch(items: List[Dict[str, Any]]) -> None: )) with get_db_connection() as conn: conn.executemany( - """INSERT OR REPLACE INTO embedding_cache (chunk_hash, dense_vector, sparse_indices, sparse_values, model_name) - VALUES (?, ?, ?, ?, ?)""", + """INSERT INTO embedding_cache (chunk_hash, dense_vector, sparse_indices, sparse_values, model_name) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(chunk_hash, model_name) DO UPDATE SET + dense_vector = EXCLUDED.dense_vector, + sparse_indices = EXCLUDED.sparse_indices, + sparse_values = EXCLUDED.sparse_values""", rows_to_insert ) conn.commit() diff --git a/app/services/database/postgres_compat.py b/app/services/database/postgres_compat.py new file mode 100644 index 0000000..06ad985 --- /dev/null +++ b/app/services/database/postgres_compat.py @@ -0,0 +1,241 @@ +import logging +from typing import Any, Optional, Tuple, Dict, List +from sqlalchemy import text + +logger = logging.getLogger("contextcortex.database.postgres_compat") + + +class DictRow(dict): + """ + Dictionary subclass providing compatibility with sqlite3.Row access patterns: + - Key lookup: row["col"] + - Positional lookup: row[0] + - Attribute lookup: row.col + - Dict conversion: dict(row) + """ + def __init__(self, mapping: dict, tuple_vals: Optional[tuple] = None): + super().__init__(mapping) + self._tuple_vals = tuple_vals if tuple_vals is not None else tuple(mapping.values()) + + def __getitem__(self, key): + if isinstance(key, int): + return self._tuple_vals[key] + return super().__getitem__(key) + + def __getattr__(self, name): + if name in self: + return self[name] + raise AttributeError(f"'DictRow' object has no attribute '{name}'") + + +TABLES_WITH_ID = { + "git_repositories", + "indexed_paths", + "git_host_credentials", + "ast_symbols", + "ast_relationships", + "api_routes", + "api_client_calls", + "custom_prompts", + "api_keys", +} + + +def _convert_sql_placeholders(sql: str, params: Any = None) -> tuple: + if not isinstance(sql, str): + return sql, params + + clean_sql = sql.strip() + + sql_upper = clean_sql.upper() + if sql_upper.startswith("INSERT") and "RETURNING" not in sql_upper: + for tbl in TABLES_WITH_ID: + if tbl in clean_sql: + clean_sql = clean_sql.rstrip(";") + " RETURNING id" + break + + if params is None: + return clean_sql, None + + if isinstance(params, dict): + return clean_sql, params + + if isinstance(params, (tuple, list)): + if "?" not in clean_sql: + return clean_sql, params + parts = clean_sql.split("?") + if len(parts) - 1 != len(params): + return clean_sql, params + param_dict = {} + new_parts = [] + for i, part in enumerate(parts[:-1]): + pname = f"p{i+1}" + param_dict[pname] = params[i] + new_parts.append(part) + new_parts.append(f":{pname}") + new_parts.append(parts[-1]) + return "".join(new_parts), param_dict + + return clean_sql, params + + +class _SingleRowResult: + def __init__(self, row_dict: Optional[dict]): + self._row_dict = row_dict + self._consumed = False + + def mappings(self): + return self + + def fetchone(self): + if self._consumed or self._row_dict is None: + return None + self._consumed = True + return self._row_dict + + def fetchall(self): + if self._consumed or self._row_dict is None: + return [] + self._consumed = True + return [self._row_dict] + + +class PostgresResultAdapter: + def __init__(self, sa_result: Any, lastrowid: Optional[Any] = None): + self._sa_result = sa_result + self.lastrowid = lastrowid + if hasattr(sa_result, "rowcount"): + self.rowcount = sa_result.rowcount + else: + self.rowcount = -1 + + def fetchone(self) -> Optional[DictRow]: + if self._sa_result is None: + return None + try: + row = self._sa_result.mappings().fetchone() + if row is None: + return None + return DictRow(dict(row)) + except Exception: + return None + + def fetchall(self) -> list: + if self._sa_result is None: + return [] + try: + rows = self._sa_result.mappings().fetchall() + return [DictRow(dict(r)) for r in rows] + except Exception: + return [] + + +class PostgresCursorAdapter: + def __init__(self, conn_adapter: "PostgresConnectionAdapter"): + self._conn_adapter = conn_adapter + self.lastrowid = None + self.rowcount = -1 + self._last_result = None + + def execute(self, sql: Any, params: Any = None): + res = self._conn_adapter.execute(sql, params) + self.lastrowid = res.lastrowid + self.rowcount = res.rowcount + self._last_result = res + return res + + def executemany(self, sql: str, param_list: list): + res = self._conn_adapter.executemany(sql, param_list) + self.lastrowid = res.lastrowid + self.rowcount = res.rowcount + self._last_result = res + return res + + def fetchone(self): + if self._last_result: + return self._last_result.fetchone() + return None + + def fetchall(self): + if self._last_result: + return self._last_result.fetchall() + return [] + + def close(self): + pass + + +class PostgresConnectionAdapter: + def __init__(self, sa_conn=None, engine=None): + from app.services.database.engine import get_db_engine + self._engine = engine or get_db_engine() + self._sa_conn = sa_conn or self._engine.connect() + self._closed = False + + def cursor(self): + return PostgresCursorAdapter(self) + + def execute(self, sql: Any, params: Any = None): + if isinstance(sql, str): + converted_sql, converted_params = _convert_sql_placeholders(sql, params) + is_insert = converted_sql.strip().upper().startswith("INSERT") and "RETURNING" in converted_sql.upper() + if converted_params is not None: + sa_res = self._sa_conn.execute(text(converted_sql), converted_params) + else: + sa_res = self._sa_conn.execute(text(converted_sql)) + + lastrowid = None + if is_insert: + try: + row = sa_res.mappings().fetchone() + if row: + lastrowid = row.get("id") or (list(row.values())[0] if row else None) + sa_res = _SingleRowResult(dict(row)) + except Exception: + pass + return PostgresResultAdapter(sa_res, lastrowid=lastrowid) + else: + sa_res = self._sa_conn.execute(sql, params) if params else self._sa_conn.execute(sql) + return PostgresResultAdapter(sa_res) + + def executemany(self, sql: str, param_list: list): + if not param_list: + return PostgresResultAdapter(None) + if isinstance(param_list[0], (tuple, list)): + converted_sql, _ = _convert_sql_placeholders(sql, param_list[0]) + converted_param_list = [] + for p in param_list: + _, p_dict = _convert_sql_placeholders(sql, p) + converted_param_list.append(p_dict if isinstance(p_dict, dict) else p) + else: + converted_sql = sql + converted_param_list = param_list + + sa_res = self._sa_conn.execute(text(converted_sql), converted_param_list) + return PostgresResultAdapter(sa_res) + + def commit(self): + try: + self._sa_conn.commit() + except Exception: + pass + + def rollback(self): + try: + self._sa_conn.rollback() + except Exception: + pass + + def close(self): + if not self._closed: + try: + self._sa_conn.close() + except Exception: + pass + self._closed = True + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() diff --git a/app/services/indexing/git_syncer.py b/app/services/indexing/git_syncer.py index f83875c..1df9d21 100644 --- a/app/services/indexing/git_syncer.py +++ b/app/services/indexing/git_syncer.py @@ -322,13 +322,32 @@ def sync_single_git_repo(repo_id: int): with db_service.get_db_connection() as conn: if batch_indexed_files: conn.executemany( - "INSERT OR REPLACE INTO indexed_files (filepath, repo, doc_type, language, commit_sha, mtime, hash) VALUES (?, ?, ?, ?, ?, ?, ?)", + """INSERT INTO indexed_files (filepath, repo, doc_type, language, commit_sha, mtime, hash) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(filepath) DO UPDATE SET + repo = EXCLUDED.repo, + doc_type = EXCLUDED.doc_type, + language = EXCLUDED.language, + commit_sha = EXCLUDED.commit_sha, + mtime = EXCLUDED.mtime, + hash = EXCLUDED.hash""", batch_indexed_files ) batch_indexed_files.clear() if batch_summaries: conn.executemany( - "INSERT OR REPLACE INTO file_summaries (filepath, repo, title, folder, category, tags, headings, keywords, summary_text, mtime) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + """INSERT INTO file_summaries (filepath, repo, title, folder, category, tags, headings, keywords, summary_text, mtime) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(filepath) DO UPDATE SET + repo = EXCLUDED.repo, + title = EXCLUDED.title, + folder = EXCLUDED.folder, + category = EXCLUDED.category, + tags = EXCLUDED.tags, + headings = EXCLUDED.headings, + keywords = EXCLUDED.keywords, + summary_text = EXCLUDED.summary_text, + mtime = EXCLUDED.mtime""", batch_summaries ) batch_summaries.clear() diff --git a/app/services/indexing/local_syncer.py b/app/services/indexing/local_syncer.py index 32ae5e8..cc211fd 100644 --- a/app/services/indexing/local_syncer.py +++ b/app/services/indexing/local_syncer.py @@ -154,17 +154,35 @@ def sync_local_paths(): logger.error(f"Failed to upsert points to vector store: {e}") return False - # Batch update SQLite tables + # Batch update database tables try: with db_service.get_db_connection() as conn: if files_to_update_cache: conn.executemany( - "INSERT OR REPLACE INTO indexed_files (filepath, repo, doc_type, language, commit_sha, mtime) VALUES (?, ?, ?, ?, ?, ?)", + """INSERT INTO indexed_files (filepath, repo, doc_type, language, commit_sha, mtime) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(filepath) DO UPDATE SET + repo = EXCLUDED.repo, + doc_type = EXCLUDED.doc_type, + language = EXCLUDED.language, + commit_sha = EXCLUDED.commit_sha, + mtime = EXCLUDED.mtime""", files_to_update_cache ) if all_summaries: conn.executemany( - "INSERT OR REPLACE INTO file_summaries (filepath, repo, title, folder, category, tags, headings, keywords, summary_text, mtime) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + """INSERT INTO file_summaries (filepath, repo, title, folder, category, tags, headings, keywords, summary_text, mtime) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(filepath) DO UPDATE SET + repo = EXCLUDED.repo, + title = EXCLUDED.title, + folder = EXCLUDED.folder, + category = EXCLUDED.category, + tags = EXCLUDED.tags, + headings = EXCLUDED.headings, + keywords = EXCLUDED.keywords, + summary_text = EXCLUDED.summary_text, + mtime = EXCLUDED.mtime""", all_summaries ) if all_symbols: diff --git a/pytest.ini b/pytest.ini index 00b9ce4..4bc8c42 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,5 +1,6 @@ [pytest] pythonpath = . +asyncio_default_fixture_loop_scope = function filterwarnings = error ignore:.*Field 'lifespan' has an incomplete definition.*:UserWarning diff --git a/tests/test_postgres_unified_connection.py b/tests/test_postgres_unified_connection.py new file mode 100644 index 0000000..2b97901 --- /dev/null +++ b/tests/test_postgres_unified_connection.py @@ -0,0 +1,215 @@ +""" +Tests for unified PostgreSQL relational selection and compatibility via get_db_connection(). +Asserts that when PostgreSQL is selected, the application uses the PostgreSQL connection adapter, +does not create or query an unrelated fallback SQLite database, and supports full CRUD/indexing/search/ADRs/settings/deletion. +""" + +import os +import tempfile +import pytest +from unittest.mock import patch, MagicMock +from sqlalchemy import create_engine, text + +from app.services.database.schema import metadata +import app.services.database as db_service +from app.services.database.connection import ( + get_db_connection, + is_postgres, + DictRow, + PostgresConnectionAdapter, + _convert_sql_placeholders, + ensure_git_repositories_columns, + ensure_file_summaries_columns, + get_file_settings, + set_file_settings, + get_ai_gateway_config, + set_ai_gateway_config, +) +from app.services.database.credentials import ( + save_git_host_credential, + list_git_host_credentials, + get_git_host_credential, + delete_git_host_credential, +) +from app.services.database.adrs import ( + create_adr, + get_adr, + list_adrs, + update_adr, + supersede_adr, + upsert_adr, +) +from app.services.local_storage import LocalStorageService + + +def test_dict_row_compatibility(): + """Verify DictRow supports dict lookup, index lookup, attribute access, and dict conversion.""" + data = {"id": 10, "name": "my-repo", "branch": "main"} + row = DictRow(data) + + # Key lookup + assert row["id"] == 10 + assert row["name"] == "my-repo" + + # Index lookup + assert row[0] == 10 + assert row[1] == "my-repo" + assert row[2] == "main" + + # Attribute access + assert row.name == "my-repo" + assert row.branch == "main" + + # Dict conversion + assert dict(row) == data + assert list(row.keys()) == ["id", "name", "branch"] + assert row.get("name") == "my-repo" + assert row.get("missing", "default") == "default" + + +def test_convert_sql_placeholders(): + """Verify placeholder conversion converts ? to :p1, :p2 and appends RETURNING id for inserts.""" + sql = "SELECT * FROM git_repositories WHERE name = ? AND branch = ?" + params = ("repo1", "main") + c_sql, c_params = _convert_sql_placeholders(sql, params) + assert c_sql == "SELECT * FROM git_repositories WHERE name = :p1 AND branch = :p2" + assert c_params == {"p1": "repo1", "p2": "main"} + + # Insert on table with ID should append RETURNING id + insert_sql = "INSERT INTO git_repositories (name, url) VALUES (?, ?)" + insert_params = ("repo1", "https://github.com/org/repo1") + i_sql, i_params = _convert_sql_placeholders(insert_sql, insert_params) + assert i_sql == "INSERT INTO git_repositories (name, url) VALUES (:p1, :p2) RETURNING id" + assert i_params == {"p1": "repo1", "p2": "https://github.com/org/repo1"} + + +def test_postgres_selection_never_creates_fallback_sqlite(tmp_path, monkeypatch): + """ + Asserts that when PostgreSQL is selected, get_db_connection() returns the PostgreSQL adapter + and NEVER creates or queries the fallback SQLite index_cache.db file. + """ + db_file = tmp_path / "never_created_sqlite.db" + monkeypatch.setenv("CACHE_DB_PATH", str(db_file)) + monkeypatch.setenv("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/testdb") + + # Patch is_postgres to True and get_db_engine to return a mock engine + mock_engine = MagicMock() + mock_conn = MagicMock() + mock_engine.connect.return_value = mock_conn + mock_engine.dialect.name = "postgresql" + + with patch("app.services.database.connection.is_postgres", return_value=True), \ + patch("app.services.database.engine.is_postgres", return_value=True), \ + patch("app.services.database.engine.get_db_engine", return_value=mock_engine): + + conn = get_db_connection() + assert isinstance(conn, PostgresConnectionAdapter) + + # Ensure SQLite file was NOT created on disk + assert not os.path.exists(db_file) + + +def test_postgres_full_relational_crud_and_indexing(tmp_path, monkeypatch): + """ + Tests full relational lifecycle (repositories, settings, credentials, indexing, + file summaries, AST symbols/routes/relationships, ADRs, search and deletion) + over the PostgresConnectionAdapter using an initialized test engine. + """ + # Create SQLite DB engine that accepts RETURNING id and ON CONFLICT + db_file = tmp_path / "pg_test.db" + db_url = f"sqlite:///{db_file}" + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + metadata.create_all(bind=engine) + + with patch("app.services.database.connection.is_postgres", return_value=True), \ + patch("app.services.database.engine.is_postgres", return_value=True), \ + patch("app.services.database.engine.get_db_engine", return_value=engine): + + # 1. Test repositories CRUD + with get_db_connection() as conn: + cur = conn.cursor() + cur.execute( + "INSERT INTO git_repositories (name, url, branch, provider, status) VALUES (?, ?, ?, ?, 'pending')", + ("test-repo", "https://github.com/org/test-repo.git", "main", "github") + ) + repo_id = cur.lastrowid + assert repo_id is not None + + rows = conn.execute("SELECT * FROM git_repositories WHERE id = ?", (repo_id,)).fetchall() + assert len(rows) == 1 + assert rows[0]["name"] == "test-repo" + assert rows[0]["branch"] == "main" + + conn.execute("UPDATE git_repositories SET status = 'synced' WHERE id = ?", (repo_id,)) + updated_row = conn.execute("SELECT status FROM git_repositories WHERE id = ?", (repo_id,)).fetchone() + assert updated_row["status"] == "synced" + + # 2. Test credentials + cred_id = save_git_host_credential("github.com", "github", "token_12345", auth_user="octocat") + assert cred_id is not None + creds = list_git_host_credentials() + assert len(creds) >= 1 + c = get_git_host_credential("github.com") + assert c is not None + assert c["auth_token"] == "token_12345" + assert delete_git_host_credential(cred_id) is True + + # 3. Test settings + set_file_settings({"summary_enabled": True, "read_file_max_lines": 1500}) + f_settings = get_file_settings() + assert f_settings["summary_enabled"] is True + assert f_settings["read_file_max_lines"] == 1500 + + # 4. Test ADRs + adr = create_adr("test-repo", "Use PostgreSQL Engine", status="PROPOSED", context="Scaling DB", decision="Adopt SQLAlchemy PostgreSQL") + assert adr["id"] == "ADR-001" + assert adr["title"] == "Use PostgreSQL Engine" + + listed_adrs = list_adrs("test-repo") + assert len(listed_adrs) == 1 + + up_adr = update_adr("ADR-001", "test-repo", status="ACCEPTED") + assert up_adr["status"] == "ACCEPTED" + + adr2 = create_adr("test-repo", "New ADR", status="PROPOSED") + sup_adr = supersede_adr("ADR-001", adr2["id"], "test-repo") + assert sup_adr["status"] == "SUPERSEDED" + + # 5. Test indexing & deletion via LocalStorageService + storage_dir = tmp_path / "storage" + storage_dir.mkdir(parents=True, exist_ok=True) + storage_svc = LocalStorageService(storage_root=str(storage_dir)) + + save_res = storage_svc.save_file("sample.py", "def my_func():\n return 42\n", repo="local_storage") + assert save_res["status"] == "success" + + with get_db_connection() as conn: + idx_file = conn.execute("SELECT * FROM indexed_files WHERE repo = ?", ("local_storage",)).fetchone() + assert idx_file is not None + syms = conn.execute("SELECT * FROM ast_symbols WHERE repo = ?", ("local_storage",)).fetchall() + assert len(syms) >= 1 + assert syms[0]["name"] == "my_func" + + # Test deletion + del_res = storage_svc.delete_file("sample.py", repo="local_storage") + assert del_res["deleted"] is True + + with get_db_connection() as conn: + idx_file_after = conn.execute("SELECT * FROM indexed_files WHERE repo = ?", ("local_storage",)).fetchone() + assert idx_file_after is None + syms_after = conn.execute("SELECT * FROM ast_symbols WHERE repo = ?", ("local_storage",)).fetchall() + assert len(syms_after) == 0 + + +def test_ensure_columns_postgres(): + """Verify ensure_git_repositories_columns and ensure_file_summaries_columns run without error when on PostgreSQL.""" + mock_sa_conn = MagicMock() + mock_sa_conn.execute.return_value.fetchone.return_value = ("keep_shallow",) + + with patch("app.services.database.connection.is_postgres", return_value=True), \ + patch("app.services.database.connection.get_connection") as mock_get_conn: + + mock_get_conn.return_value.__enter__.return_value = mock_sa_conn + ensure_git_repositories_columns() + ensure_file_summaries_columns() + assert mock_sa_conn.execute.called