graph_explorer/tests/test_schema_migration_group_id.py

"""Tests de la migracion idempotente de operations.db -> + group_id.

El binario C++ implementa `project_migrate_schema` en project_manager.cpp:
detecta si `entities.group_id` existe via PRAGMA table_info; si no,
ejecuta `ALTER TABLE entities ADD COLUMN group_id TEXT`. Es idempotente
— al volver a abrir una BD ya migrada NO debe fallar ni duplicar.

Como la logica es pequena y deterministica (PRAGMA + ALTER), aqui la
replicamos en Python para testear el contrato sin depender del
binario. Si el contrato cambia, este mirror tiene que actualizarse
junto con project_manager.cpp.
"""
from __future__ import annotations

import sqlite3
from pathlib import Path

import pytest


def _table_has_column(conn: sqlite3.Connection, table: str, column: str) -> bool:
    cur = conn.execute(f"PRAGMA table_info({table})")
    return any(row[1] == column for row in cur.fetchall())


def _migrate_group_id(db_path: Path) -> bool:
    """Mirror Python de project_migrate_schema (issue 0035a + 0035e).

    Devuelve True si la migracion se completo (con o sin trabajo), False
    si la BD no se pudo abrir.
    """
    cn = sqlite3.connect(db_path)
    try:
        if not _table_has_column(cn, "entities", "group_id"):
            cn.execute("ALTER TABLE entities ADD COLUMN group_id TEXT")
            cn.commit()
        return True
    finally:
        cn.close()


# Schema "viejo" — sin la columna group_id. Reproduce el estado de una
# operations.db previa al issue 0035a (pre-2026-05-03).
LEGACY_SCHEMA = """
CREATE TABLE entities (
    id          TEXT PRIMARY KEY,
    name        TEXT NOT NULL,
    type_ref    TEXT NOT NULL,
    status      TEXT NOT NULL DEFAULT 'active',
    description TEXT NOT NULL DEFAULT '',
    domain      TEXT NOT NULL DEFAULT '',
    tags        TEXT NOT NULL DEFAULT '[]',
    source      TEXT NOT NULL,
    metadata    TEXT NOT NULL DEFAULT '{}',
    notes       TEXT NOT NULL DEFAULT '',
    created_at  TEXT NOT NULL,
    updated_at  TEXT NOT NULL
);
CREATE TABLE relations (
    id          TEXT PRIMARY KEY,
    name        TEXT NOT NULL,
    from_entity TEXT NOT NULL DEFAULT '',
    to_entity   TEXT NOT NULL,
    created_at  TEXT NOT NULL,
    updated_at  TEXT NOT NULL
);
"""


@pytest.fixture
def legacy_db(tmp_path):
    db = tmp_path / "operations.db"
    cn = sqlite3.connect(db)
    cn.executescript(LEGACY_SCHEMA)
    # Datos previos para verificar que sobreviven la migracion.
    cn.execute(
        "INSERT INTO entities(id, name, type_ref, status, source, "
        "                     metadata, created_at, updated_at) "
        "VALUES ('e_pre', 'pre-existing', 'Url', 'active', 'manual', "
        "        '{}', '2025-01-01', '2025-01-01')"
    )
    cn.commit()
    cn.close()
    return db


def test_schema_migration_idempotent(legacy_db):
    """1a apertura migra; 2a apertura no rompe ni duplica la columna."""
    # Estado inicial: sin group_id.
    cn = sqlite3.connect(legacy_db)
    assert not _table_has_column(cn, "entities", "group_id"), \
        "fixture ya tenia group_id (schema legacy roto)"
    cn.close()

    # 1a migracion: anade la columna.
    assert _migrate_group_id(legacy_db) is True
    cn = sqlite3.connect(legacy_db)
    assert _table_has_column(cn, "entities", "group_id")
    # Datos previos sobreviven y la columna nueva es NULL por defecto.
    row = cn.execute(
        "SELECT id, name, group_id FROM entities WHERE id = 'e_pre'"
    ).fetchone()
    assert row == ("e_pre", "pre-existing", None)
    cn.close()

    # 2a migracion: idempotente, no debe fallar ni duplicar.
    assert _migrate_group_id(legacy_db) is True
    cn = sqlite3.connect(legacy_db)
    # Una sola columna group_id (no duplicada).
    cur = cn.execute("PRAGMA table_info(entities)")
    cols = [row[1] for row in cur.fetchall()]
    assert cols.count("group_id") == 1, cols
    # Y los datos siguen intactos.
    cnt = cn.execute("SELECT COUNT(*) FROM entities").fetchone()[0]
    assert cnt == 1
    cn.close()


def test_schema_migration_already_migrated_db_is_noop(tmp_path):
    """BD ya creada con la columna group_id desde el inicio: noop."""
    db = tmp_path / "operations.db"
    cn = sqlite3.connect(db)
    cn.executescript(LEGACY_SCHEMA)
    cn.execute("ALTER TABLE entities ADD COLUMN group_id TEXT")
    cn.commit()
    cn.close()

    # Migrar no debe fallar y la columna sigue siendo unica.
    assert _migrate_group_id(db) is True
    cn = sqlite3.connect(db)
    cur = cn.execute("PRAGMA table_info(entities)")
    cols = [row[1] for row in cur.fetchall()]
    assert cols.count("group_id") == 1
    cn.close()


def test_schema_migration_preserves_existing_group_id_values(tmp_path):
    """Si una BD ya tiene valores en group_id, la migracion los respeta."""
    db = tmp_path / "operations.db"
    cn = sqlite3.connect(db)
    cn.executescript(LEGACY_SCHEMA)
    cn.execute("ALTER TABLE entities ADD COLUMN group_id TEXT")
    cn.execute(
        "INSERT INTO entities(id, name, type_ref, status, source, "
        "                     metadata, group_id, "
        "                     created_at, updated_at) "
        "VALUES ('child', 'c', 'Url', 'active', 'manual', '{}', "
        "        'parent_grp', '2026-05-04', '2026-05-04')"
    )
    cn.commit()
    cn.close()

    _migrate_group_id(db)

    cn = sqlite3.connect(db)
    val = cn.execute(
        "SELECT group_id FROM entities WHERE id = 'child'"
    ).fetchone()[0]
    assert val == "parent_grp"
    cn.close()