test(0035e): cobertura del visual heredado, threshold override y migracion idempotente
- test_group_visual_inheritance.py (4 tests): homogeneo->Url heredado, heterogeneo->generico Group, vacio->generico, subgrupos anidados ignorados. - test_manifest_threshold_override.py (4 tests): override 100 con 80 unicas no agrupa; override bajo (20) si agrupa cuando se supera; threshold=0 cae al default 50; mirror Python del parser de manifest C++ confirma el campo se extrae como int. - test_schema_migration_group_id.py (3 tests): mirror Python de project_migrate_schema, verifica idempotencia (1a y 2a apertura no duplican columna), no-op sobre BD ya migrada, datos previos sobreviven la migracion.
This commit is contained in:
@@ -0,0 +1,160 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user