"""Tests del enricher split_words (offline, regex puro). Mismo patron que test_split_sentences.py: nodo text con `notes` largo, verificamos tokens, fallback a name, threshold/grouping y dedupe. """ from __future__ import annotations from conftest import ( base_ctx, list_entities, list_relations, make_node, run_enricher, ) # Texto con suficientes palabras unicas para superar threshold (50) # si dedupe=true: cuento ~85 unicas a ojo. Sin dedupe (count=true) # todas las ocurrencias dan ~140+ tokens. LONG_TEXT = ( "Las estrellas brillan suavemente sobre el horizonte mientras " "la marea retrocede dejando huellas mojadas en la arena fina. " "Caminamos lentamente conversando sobre proyectos antiguos, " "ideas frescas, libros leidos durante el invierno pasado, " "viajes pendientes hacia tierras lejanas con culturas vibrantes. " "Recordamos infancias compartidas, amigos perdidos, victorias " "modestas, fracasos instructivos. Cada palabra dibuja un mapa " "diferente del territorio interno que habitamos. Los nombres de " "ciudades antiguas resuenan: Estambul, Marrakech, Kioto, Lisboa, " "Praga, Budapest, Cuzco, Cartagena. Tambien tecnologia: servidores, " "bases datos, redes neuronales, modelos linguisticos, sistemas " "distribuidos, criptografia moderna. La conversacion fluye sin " "esfuerzo aparente entre dominios completamente distintos." ) def test_split_words_creates_word_nodes(ops_db, app_dir, registry_root): """Texto corto < threshold genera Words sueltos sin Group.""" short_text = "uno dos tres cuatro cinco seis siete ocho nueve diez." make_node(ops_db, node_id="t1", name="short", type_ref="text", notes=short_text) ctx = base_ctx(ops_db=ops_db, app_dir=app_dir, registry_root=registry_root, node_id="t1", node_name="short", node_type="text") rc, out, err = run_enricher("split_words", ctx) assert rc == 0, err assert out["grouped"] is False assert out["entities_added"] == out["words"] assert out["words"] >= 5 # filtra <3 chars: dos, tres, cuatro, ... words = list_entities(ops_db, type_ref="Word") assert len(words) == out["words"] def test_split_words_above_threshold_creates_group(ops_db, app_dir, registry_root): """Texto largo (≥50 tokens unicos) → Group + 10 sueltos + resto.""" make_node(ops_db, node_id="t1", name="largo", type_ref="text", notes=LONG_TEXT) ctx = base_ctx(ops_db=ops_db, app_dir=app_dir, registry_root=registry_root, node_id="t1", node_name="largo", node_type="text") rc, out, err = run_enricher("split_words", ctx, timeout=60) assert rc == 0, err assert out["words"] >= 50 assert out["grouped"] is True assert out["group_id"] # Group + words = entities_added. assert out["entities_added"] == out["words"] + 1 # Hay 10 Words sueltos (group_id NULL) y resto agrupados. import sqlite3 cn = sqlite3.connect(ops_db) n_loose = cn.execute( "SELECT count(*) FROM entities WHERE type_ref='Word' " "AND group_id IS NULL" ).fetchone()[0] n_grouped = cn.execute( "SELECT count(*) FROM entities WHERE type_ref='Word' " "AND group_id = ?", (out["group_id"],) ).fetchone()[0] cn.close() assert n_loose == 10 assert n_loose + n_grouped == out["words"] def test_split_words_dedupe_default_true(ops_db, app_dir, registry_root): """Por defecto dedupe=true: 'casa casa casa' produce 1 Word.""" make_node(ops_db, node_id="t1", name="dup", type_ref="text", notes="casa casa casa perro perro gato.") ctx = base_ctx(ops_db=ops_db, app_dir=app_dir, registry_root=registry_root, node_id="t1", node_name="dup", node_type="text") rc, out, err = run_enricher("split_words", ctx) assert rc == 0, err assert out["deduped"] is True # casa, perro, gato (todas ≥3 chars). assert out["words"] == 3 def test_split_words_dedupe_false(ops_db, app_dir, registry_root): """Con dedupe=false cada ocurrencia es un nodo Word.""" make_node(ops_db, node_id="t1", name="dup", type_ref="text", notes="casa casa casa perro perro gato.") ctx = base_ctx(ops_db=ops_db, app_dir=app_dir, registry_root=registry_root, node_id="t1", node_name="dup", node_type="text", params={"dedupe": False}) rc, out, err = run_enricher("split_words", ctx) assert rc == 0, err assert out["deduped"] is False # 3 casa + 2 perro + 1 gato = 6 tokens. assert out["words"] == 6 def test_split_words_min_length_filters(ops_db, app_dir, registry_root): """min_length filtra tokens cortos. Default 3.""" make_node(ops_db, node_id="t1", name="cortos", type_ref="text", notes="a el de la y o un casa perro elefante.") ctx = base_ctx(ops_db=ops_db, app_dir=app_dir, registry_root=registry_root, node_id="t1", node_name="cortos", node_type="text") rc, out, err = run_enricher("split_words", ctx) assert rc == 0, err # >=3 chars: casa, perro, elefante. (un=2, de=2, la=2 quedan fuera). assert out["words"] == 3 def test_split_words_uses_notes_priority(ops_db, app_dir, registry_root): """Lee `entities.notes` por encima de node_name.""" make_node(ops_db, node_id="t1", name="ignorame", type_ref="text", notes="estos cinco tokens deberian ganar.") ctx = base_ctx(ops_db=ops_db, app_dir=app_dir, registry_root=registry_root, node_id="t1", node_name="ignorame", node_type="text") rc, out, err = run_enricher("split_words", ctx) assert rc == 0, err # estos, cinco, tokens, deberian, ganar (todos ≥3). assert out["words"] == 5 def test_split_words_no_text_fails(ops_db, app_dir, registry_root): """Sin notes y name corto → exit 2.""" make_node(ops_db, node_id="t1", name="x", type_ref="text", metadata={}) ctx = base_ctx(ops_db=ops_db, app_dir=app_dir, registry_root=registry_root, node_id="t1", node_name="x", node_type="text") rc, out, err = run_enricher("split_words", ctx) assert rc == 2 assert out is not None assert "demasiado corto" in (out.get("error") or "") def test_split_words_max_words_truncates(ops_db, app_dir, registry_root): """max_words limita el output.""" make_node(ops_db, node_id="t1", name="largo", type_ref="text", notes=LONG_TEXT) ctx = base_ctx(ops_db=ops_db, app_dir=app_dir, registry_root=registry_root, node_id="t1", node_name="largo", node_type="text", params={"max_words": 12}) rc, out, err = run_enricher("split_words", ctx) assert rc == 0, err assert out["words"] == 12