egutierrez
b620cc38c2
Privacidad (decisión del usuario: al móvil solo datos de agenda): - _compose_agenda_vcard compone el vCard desde el contacto (fn/tels/emails) + las direcciones (ADR) y aliases (NICKNAME) de la persona enlazada por note_path, SIN pasar nunca el dict osint a build_vcard → el vCard jamás lleva X-OSINT-* (DNI/sexo/fecha-nac quedan solo en DuckDB+Obsidian). Usado en upsert_contact y en el push masivo push_all_dav (que antes leía solo contacts y perdía las direcciones). Sync inverso DAVx5→DuckDB (last-write-wins por etag): - Tras cada push se captura el etag nuevo del recurso (dav_list_resources) y se persiste en contacts.etag, para no confundir el push propio con una edición del móvil. - POST /api/sync/dav-pull: pull incremental — compara etags, descarga SOLO los recursos cambiados/nuevos (dav_get_resource + parse_vcard + upsert), borra los que el móvil quitó, re-enlaza. Distinto del ingest_dav (DELETE+INSERT ciego): respeta la verdad de la DB salvo donde el etag prueba un cambio externo. 20 tests verdes (18 + 2 nuevos: vCard sin OSINT con direcciones; pull incremental por etag).
703 lines
24 KiB
Python
703 lines
24 KiB
Python
"""Tests del service osint_db: migraciones, ingest del vault, API y render.
|
|
|
|
Todo corre contra un vault temporal y una base DuckDB temporal, SIN red: el
|
|
ingest DAV no se ejercita aquí (requiere Xandikos + pass). El enlace
|
|
contacto→ficha sí se prueba insertando un contacto a mano y relanzando el
|
|
ingest del vault, que re-enlaza.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from server.config import Config # noqa: E402
|
|
from server.db import apply_migrations, write_conn # noqa: E402
|
|
from server.main import create_app # noqa: E402
|
|
|
|
PERSONA_MD = """---
|
|
tipo: persona
|
|
nombre: "Ana García Pérez"
|
|
slug: ana-garcia-perez
|
|
aliases: ["Anita"]
|
|
sexo: mujer
|
|
fecha_nacimiento: 1990-04-12
|
|
dni: 12345678Z
|
|
telefono: "+34 600 111 222"
|
|
email: ana@example.com
|
|
direccion: null
|
|
pais: españa
|
|
relaciones: []
|
|
contexto: familia
|
|
fuente: "test fixture"
|
|
tags: [persona, osint]
|
|
---
|
|
|
|
## Notas
|
|
Ficha de prueba.
|
|
"""
|
|
|
|
PERSONA2_MD = """---
|
|
tipo: persona
|
|
nombre: "Luis Pérez"
|
|
slug: luis-perez
|
|
aliases: []
|
|
sexo: hombre
|
|
fecha_nacimiento: null
|
|
dni: null
|
|
telefono: null
|
|
email: null
|
|
direccion: null
|
|
pais: null
|
|
relaciones: []
|
|
contexto: movil
|
|
fuente: "Xandikos UID abc-123"
|
|
tags: [persona, osint, movil]
|
|
---
|
|
|
|
## Notas
|
|
"""
|
|
|
|
ORG_MD = """---
|
|
tipo: organizacion
|
|
nombre: "Acme S.L."
|
|
slug: acme-sl
|
|
tags: [organizacion, osint]
|
|
---
|
|
|
|
## Notas
|
|
"""
|
|
|
|
DOC_MD = """---
|
|
tipo: documento
|
|
doc_tipo: dni
|
|
---
|
|
|
|
Sub-nota de documento (NO debe contar como ficha de persona).
|
|
"""
|
|
|
|
|
|
@pytest.fixture()
|
|
def cfg(tmp_path):
|
|
"""Vault temporal con fichas de fixture + base DuckDB temporal migrada."""
|
|
vault = tmp_path / "vault"
|
|
(vault / "personas" / "ana-garcia-perez").mkdir(parents=True)
|
|
(vault / "organizaciones").mkdir()
|
|
(vault / "personas" / "ana-garcia-perez.md").write_text(
|
|
PERSONA_MD, encoding="utf-8"
|
|
)
|
|
(vault / "personas" / "luis-perez.md").write_text(PERSONA2_MD, encoding="utf-8")
|
|
(vault / "personas" / "_plantilla.md").write_text(
|
|
"---\ntipo: plantilla\n---\n", encoding="utf-8"
|
|
)
|
|
(vault / "personas" / "ana-garcia-perez" / "dni.md").write_text(
|
|
DOC_MD, encoding="utf-8"
|
|
)
|
|
(vault / "organizaciones" / "acme-sl.md").write_text(ORG_MD, encoding="utf-8")
|
|
|
|
config = Config(
|
|
vault_dir=str(vault),
|
|
db_path=str(tmp_path / "data" / "osint.duckdb"),
|
|
port=0,
|
|
)
|
|
apply_migrations(config.db_path)
|
|
return config
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(cfg):
|
|
return TestClient(create_app(cfg))
|
|
|
|
|
|
def test_migrations_son_idempotentes(cfg):
|
|
"""La segunda pasada de migraciones no aplica nada (tabla _migrations)."""
|
|
assert apply_migrations(cfg.db_path) == []
|
|
|
|
|
|
def test_health(client, cfg):
|
|
r = client.get("/api/health").json()
|
|
assert r["status"] == "ok"
|
|
assert r["db_path"] == cfg.db_path
|
|
assert r["tables"] >= 8
|
|
|
|
|
|
def test_ingest_vault_cuenta_entidades(client):
|
|
r = client.post("/api/ingest/vault").json()
|
|
assert r["status"] == "ok"
|
|
# 5 notas: 2 personas + plantilla + sub-nota documento + organización.
|
|
assert r["notes"] == 5
|
|
# Solo las fichas de nivel-1 sin prefijo _ cuentan como persona.
|
|
assert r["persons"] == 2
|
|
assert r["organizations"] == 1
|
|
assert r["domains"] == 0
|
|
assert sorted(r["derived_rebuilt"]) == [
|
|
"derived.contact_link_quality",
|
|
"derived.event_monthly",
|
|
"derived.person_stats",
|
|
]
|
|
|
|
|
|
def test_ingest_vault_extrae_dav_uid_de_fuente(client):
|
|
client.post("/api/ingest/vault")
|
|
r = client.post(
|
|
"/api/query",
|
|
json={"sql": "SELECT dav_uid FROM persons WHERE slug = 'luis-perez'"},
|
|
).json()
|
|
assert r["status"] == "ok"
|
|
assert r["rows"][0]["dav_uid"] == "abc-123"
|
|
|
|
|
|
def test_api_query_ok_y_error_siempre_http_200(client):
|
|
client.post("/api/ingest/vault")
|
|
ok = client.post(
|
|
"/api/query",
|
|
json={"sql": "SELECT slug, nombre FROM persons ORDER BY slug", "max_rows": 10},
|
|
)
|
|
assert ok.status_code == 200
|
|
body = ok.json()
|
|
assert body["status"] == "ok"
|
|
assert body["columns"] == ["slug", "nombre"]
|
|
assert body["row_count"] == 2
|
|
assert body["truncated"] is False
|
|
assert body["rows"][0]["slug"] == "ana-garcia-perez"
|
|
|
|
err = client.post("/api/query", json={"sql": "SELECT * FROM tabla_que_no_existe"})
|
|
assert err.status_code == 200
|
|
assert err.json()["status"] == "error"
|
|
assert err.json()["error"]
|
|
|
|
|
|
def test_api_query_es_solo_lectura(client):
|
|
client.post("/api/ingest/vault")
|
|
r = client.post(
|
|
"/api/query", json={"sql": "DELETE FROM persons"}
|
|
).json()
|
|
assert r["status"] == "error"
|
|
|
|
|
|
def test_catalogo_de_queries_con_nombre(client):
|
|
r = client.get("/api/queries").json()
|
|
assert r["status"] == "ok"
|
|
names = {q["name"] for q in r["queries"]}
|
|
assert {
|
|
"personas_por_contexto",
|
|
"personas_recientes",
|
|
"eventos_proximos",
|
|
"contactos_sin_nota",
|
|
"stats_personas",
|
|
} <= names
|
|
assert all(q["sql"] and q["description"] for q in r["queries"])
|
|
|
|
|
|
def test_query_named_ok_y_desconocida(client):
|
|
client.post("/api/ingest/vault")
|
|
r = client.post(
|
|
"/api/query/named", json={"name": "personas_por_contexto"}
|
|
).json()
|
|
assert r["status"] == "ok"
|
|
contextos = {row["contexto"]: row["personas"] for row in r["rows"]}
|
|
assert contextos == {"familia": 1, "movil": 1}
|
|
|
|
bad = client.post("/api/query/named", json={"name": "no_existe"}).json()
|
|
assert bad["status"] == "error"
|
|
|
|
|
|
def test_tables_inventario(client):
|
|
client.post("/api/ingest/vault")
|
|
r = client.get("/api/tables").json()
|
|
assert r["status"] == "ok"
|
|
by_name = {(t["schema"], t["name"]): t for t in r["tables"]}
|
|
persons = by_name[("main", "persons")]
|
|
assert persons["kind"] == "master"
|
|
assert persons["row_count"] == 2
|
|
assert {"name": "note_path", "type": "VARCHAR"} in persons["columns"]
|
|
stats = by_name[("derived", "person_stats")]
|
|
assert stats["kind"] == "derived"
|
|
assert ("main", "_migrations") not in by_name
|
|
|
|
|
|
def test_derivadas_sin_note_path(client):
|
|
"""Regla dura: ninguna tabla del schema derived referencia notas."""
|
|
client.post("/api/ingest/vault")
|
|
r = client.post(
|
|
"/api/query",
|
|
json={
|
|
"sql": (
|
|
"SELECT table_name, column_name FROM information_schema.columns "
|
|
"WHERE table_schema = 'derived' AND column_name LIKE '%note%'"
|
|
)
|
|
},
|
|
).json()
|
|
assert r["status"] == "ok"
|
|
assert r["rows"] == []
|
|
# Y las tres derivadas existen de verdad.
|
|
t = client.post(
|
|
"/api/query",
|
|
json={
|
|
"sql": (
|
|
"SELECT table_name FROM information_schema.tables "
|
|
"WHERE table_schema = 'derived' ORDER BY table_name"
|
|
)
|
|
},
|
|
).json()
|
|
assert [row["table_name"] for row in t["rows"]] == [
|
|
"contact_link_quality",
|
|
"event_monthly",
|
|
"person_stats",
|
|
]
|
|
|
|
|
|
def test_link_contacts_por_telefono(client, cfg):
|
|
"""Un contacto con teléfono que casa con una ficha queda enlazado al re-ingestar."""
|
|
client.post("/api/ingest/vault")
|
|
now = datetime.now(tz=timezone.utc)
|
|
with write_conn(cfg.db_path) as conn:
|
|
conn.execute(
|
|
"INSERT INTO contacts VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
[
|
|
"uid-movil-1",
|
|
"/enmanuel/contacts/addressbook/",
|
|
"etag1",
|
|
"Ana G.",
|
|
'["600111222"]',
|
|
"[]",
|
|
"BEGIN:VCARD...",
|
|
None,
|
|
now,
|
|
],
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO contacts VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
[
|
|
"uid-movil-2",
|
|
"/enmanuel/contacts/addressbook/",
|
|
"etag2",
|
|
"Desconocido",
|
|
'["699999999"]',
|
|
"[]",
|
|
"BEGIN:VCARD...",
|
|
None,
|
|
now,
|
|
],
|
|
)
|
|
# El ingest del vault re-enlaza contacts y reconstruye derivadas.
|
|
client.post("/api/ingest/vault")
|
|
r = client.post(
|
|
"/api/query",
|
|
json={"sql": "SELECT uid, note_path FROM contacts ORDER BY uid"},
|
|
).json()
|
|
rows = {row["uid"]: row["note_path"] for row in r["rows"]}
|
|
assert rows["uid-movil-1"] == os.path.join("personas", "ana-garcia-perez.md")
|
|
assert rows["uid-movil-2"] is None
|
|
|
|
q = client.post("/api/query/named", json={"name": "contactos_sin_nota"}).json()
|
|
assert [row["uid"] for row in q["rows"]] == ["uid-movil-2"]
|
|
|
|
quality = client.post(
|
|
"/api/query/named", json={"name": "calidad_enlace_contactos"}
|
|
).json()
|
|
assert quality["rows"] == [{"total": 2, "linked": 1, "unlinked": 1}]
|
|
|
|
|
|
def test_render_note_crea_bloque_sentinel_y_es_idempotente(client, cfg):
|
|
client.post("/api/ingest/vault")
|
|
body = {
|
|
"note_path": "tableros/personas.md",
|
|
"block_id": "personas",
|
|
"query": "personas_por_contexto",
|
|
"title": "Personas por contexto",
|
|
}
|
|
r = client.post("/api/render/note", json=body).json()
|
|
assert r["status"] == "ok"
|
|
assert r["note_path"] == "tableros/personas.md"
|
|
assert r["rows_rendered"] == 2
|
|
|
|
note_file = os.path.join(cfg.vault_dir, "tableros", "personas.md")
|
|
content = open(note_file, encoding="utf-8").read()
|
|
assert "<!-- osintdb:begin id=personas -->" in content
|
|
assert "<!-- osintdb:end id=personas -->" in content
|
|
assert "### Personas por contexto" in content
|
|
assert "| contexto | personas |" in content
|
|
assert "| familia | 1 |" in content
|
|
|
|
# Idempotente: un segundo render no duplica el bloque ni la tabla.
|
|
r2 = client.post("/api/render/note", json=body).json()
|
|
assert r2["status"] == "ok"
|
|
content2 = open(note_file, encoding="utf-8").read()
|
|
assert content2.count("<!-- osintdb:begin id=personas -->") == 1
|
|
assert content2.count("| familia | 1 |") == 1
|
|
|
|
|
|
def test_render_note_valida_inputs(client):
|
|
client.post("/api/ingest/vault")
|
|
# Ni sql ni query.
|
|
r = client.post(
|
|
"/api/render/note", json={"note_path": "t.md", "block_id": "x"}
|
|
).json()
|
|
assert r["status"] == "error"
|
|
# Query con nombre desconocida.
|
|
r = client.post(
|
|
"/api/render/note",
|
|
json={"note_path": "t.md", "block_id": "x", "query": "nope"},
|
|
).json()
|
|
assert r["status"] == "error"
|
|
# Path traversal fuera del vault.
|
|
r = client.post(
|
|
"/api/render/note",
|
|
json={"note_path": "../fuera.md", "block_id": "x", "query": "stats_personas"},
|
|
).json()
|
|
assert r["status"] == "error"
|
|
assert "fuera del vault" in r["error"]
|
|
|
|
|
|
# --- F1: migraciones multi-valor + addressbooks ----------------------------
|
|
|
|
|
|
def test_migracion_multivalue_y_addressbooks(client, cfg):
|
|
"""002 añade columnas multi-valor a persons; 003 crea+seed addressbooks."""
|
|
r = client.get("/api/tables").json()
|
|
by_name = {(t["schema"], t["name"]): t for t in r["tables"]}
|
|
persons_cols = {c["name"] for c in by_name[("main", "persons")]["columns"]}
|
|
assert {"telefonos", "emails", "direcciones", "extra_fm"} <= persons_cols
|
|
# Las singulares siguen existiendo (compat).
|
|
assert {"telefono", "email", "direccion"} <= persons_cols
|
|
# La libreta por defecto quedó sembrada.
|
|
assert ("main", "addressbooks") in by_name
|
|
ab = client.post(
|
|
"/api/query", json={"sql": "SELECT slug, collection_path FROM addressbooks"}
|
|
).json()
|
|
rows = {row["slug"]: row["collection_path"] for row in ab["rows"]}
|
|
assert rows["addressbook"] == "/enmanuel/contacts/addressbook/"
|
|
|
|
|
|
# --- F2: ingest selectivo (la DB es dueña de los campos OWNED) --------------
|
|
|
|
|
|
def test_ingest_vault_no_pisa_campo_owned(client, cfg):
|
|
"""Un valor escrito por la API persiste tras re-ingestar el vault.
|
|
|
|
Simula la escritura de un teléfono por la futura API con un UPDATE directo
|
|
a la DB; el re-ingest del vault NO debe pisarlo con el frontmatter viejo de
|
|
la nota (que tenía '+34 600 111 222').
|
|
"""
|
|
client.post("/api/ingest/vault")
|
|
# La API escribe un teléfono nuevo (multi-valor) en la DB.
|
|
with write_conn(cfg.db_path) as conn:
|
|
conn.execute(
|
|
"UPDATE persons SET telefonos = ?, telefono = ? WHERE slug = ?",
|
|
['["+34 999 888 777"]', "+34 999 888 777", "ana-garcia-perez"],
|
|
)
|
|
# Re-ingest del vault: la ficha de Ana YA existe -> solo refresca
|
|
# note_path + extra_fm, NO los campos OWNED.
|
|
r = client.post("/api/ingest/vault").json()
|
|
assert r["status"] == "ok"
|
|
assert r["persons"] == 2
|
|
assert r["persons_updated"] == 2 # ambas fichas ya existían
|
|
assert r["persons_inserted"] == 0
|
|
|
|
q = client.post(
|
|
"/api/query",
|
|
json={
|
|
"sql": "SELECT telefono, telefonos FROM persons WHERE slug = 'ana-garcia-perez'"
|
|
},
|
|
).json()
|
|
assert q["rows"][0]["telefono"] == "+34 999 888 777" # el valor de la API, NO el del FM
|
|
assert "+34 999 888 777" in q["rows"][0]["telefonos"]
|
|
|
|
|
|
def test_ingest_vault_bootstrapea_ficha_nueva(client, cfg):
|
|
"""Una ficha cuyo slug no está en la DB se inserta completa desde el FM."""
|
|
# Primer ingest: solo las dos fichas del fixture.
|
|
client.post("/api/ingest/vault")
|
|
# Añadimos una ficha nueva al vault con teléfono singular en el FM.
|
|
nueva = (
|
|
"---\n"
|
|
"tipo: persona\n"
|
|
'nombre: "Marta Ruiz"\n'
|
|
"slug: marta-ruiz\n"
|
|
'telefono: "+34 611 000 111"\n'
|
|
"contexto: trabajo\n"
|
|
"campo_libre: valor_raro\n"
|
|
"tags: [persona]\n"
|
|
"---\n\n## Notas\n"
|
|
)
|
|
with open(
|
|
os.path.join(cfg.vault_dir, "personas", "marta-ruiz.md"), "w", encoding="utf-8"
|
|
) as fh:
|
|
fh.write(nueva)
|
|
|
|
r = client.post("/api/ingest/vault").json()
|
|
assert r["status"] == "ok"
|
|
assert r["persons"] == 3
|
|
assert r["persons_inserted"] == 1 # marta-ruiz nueva
|
|
|
|
q = client.post(
|
|
"/api/query",
|
|
json={
|
|
"sql": "SELECT telefono, telefonos, contexto, extra_fm "
|
|
"FROM persons WHERE slug = 'marta-ruiz'"
|
|
},
|
|
).json()
|
|
row = q["rows"][0]
|
|
assert row["telefono"] == "+34 611 000 111" # singular derivado de la lista
|
|
assert "+34 611 000 111" in row["telefonos"] # lista poblada desde el singular
|
|
assert row["contexto"] == "trabajo"
|
|
# extra_fm captura el frontmatter no-owned (campo_libre), no los OWNED.
|
|
extra = json.loads(row["extra_fm"])
|
|
assert extra.get("campo_libre") == "valor_raro"
|
|
assert "telefono" not in extra and "contexto" not in extra
|
|
|
|
|
|
# --- F3: endpoints de escritura estructurada (persons, sin red) ------------
|
|
|
|
|
|
def test_api_person_crud_y_materializa(client, cfg):
|
|
"""POST /api/person con 2 teléfonos -> fila en DB + nota .md materializada."""
|
|
client.post("/api/ingest/vault")
|
|
body = {
|
|
"slug": "nuevo-contacto",
|
|
"nombre": "Nuevo Contacto",
|
|
"telefonos": ["+34 600 000 001", "+34 600 000 002"],
|
|
"emails": ["nc@example.com"],
|
|
"contexto": "trabajo",
|
|
"tags": ["persona"],
|
|
}
|
|
r = client.post("/api/person", json=body).json()
|
|
assert r["status"] == "ok"
|
|
assert r["inserted"] == 1
|
|
assert r["materialized"] is True
|
|
|
|
# Fila en DB: telefonos como lista, singular = primer elemento.
|
|
q = client.post(
|
|
"/api/query",
|
|
json={
|
|
"sql": "SELECT telefono, telefonos, emails, email FROM persons "
|
|
"WHERE slug = 'nuevo-contacto'"
|
|
},
|
|
).json()
|
|
row = q["rows"][0]
|
|
assert json.loads(row["telefonos"]) == ["+34 600 000 001", "+34 600 000 002"]
|
|
assert row["telefono"] == "+34 600 000 001"
|
|
assert row["email"] == "nc@example.com"
|
|
|
|
# Nota .md materializada con la lista telefonos.
|
|
note_file = os.path.join(cfg.vault_dir, "personas", "nuevo-contacto.md")
|
|
assert os.path.exists(note_file)
|
|
content = open(note_file, encoding="utf-8").read()
|
|
assert "telefonos:" in content
|
|
assert "+34 600 000 001" in content
|
|
assert "+34 600 000 002" in content
|
|
|
|
# PUT actualiza (un solo teléfono ahora).
|
|
r = client.put(
|
|
"/api/person/nuevo-contacto",
|
|
json={"slug": "nuevo-contacto", "nombre": "NC", "telefonos": ["+34 600 000 009"]},
|
|
).json()
|
|
assert r["status"] == "ok"
|
|
assert r["updated"] == 1
|
|
q = client.post(
|
|
"/api/query",
|
|
json={"sql": "SELECT telefono FROM persons WHERE slug = 'nuevo-contacto'"},
|
|
).json()
|
|
assert q["rows"][0]["telefono"] == "+34 600 000 009"
|
|
|
|
# DELETE quita la fila de la DB.
|
|
r = client.request("DELETE", "/api/person/nuevo-contacto").json()
|
|
assert r["status"] == "ok"
|
|
q = client.post(
|
|
"/api/query",
|
|
json={"sql": "SELECT COUNT(*) AS n FROM persons WHERE slug = 'nuevo-contacto'"},
|
|
).json()
|
|
assert q["rows"][0]["n"] == 0
|
|
|
|
|
|
def test_api_person_render_preserva_prosa(client, cfg):
|
|
"""POST /api/person/{slug}/render reescribe el frontmatter SIN tocar la prosa."""
|
|
client.post("/api/ingest/vault")
|
|
# Ana ya tiene cuerpo "## Notas\nFicha de prueba." en el fixture.
|
|
note_file = os.path.join(cfg.vault_dir, "personas", "ana-garcia-perez.md")
|
|
before = open(note_file, encoding="utf-8").read()
|
|
assert "Ficha de prueba." in before
|
|
|
|
# Cambiamos el teléfono por la API y re-materializamos.
|
|
client.put(
|
|
"/api/person/ana-garcia-perez",
|
|
json={
|
|
"slug": "ana-garcia-perez",
|
|
"nombre": "Ana García Pérez",
|
|
"telefonos": ["+34 622 333 444"],
|
|
},
|
|
)
|
|
r = client.post("/api/person/ana-garcia-perez/render").json()
|
|
assert r["status"] == "ok"
|
|
|
|
after = open(note_file, encoding="utf-8").read()
|
|
assert "Ficha de prueba." in after # prosa preservada
|
|
assert "+34 622 333 444" in after # frontmatter actualizado
|
|
|
|
|
|
# --- F4: push de agenda SIN OSINT + sync inverso por etag ------------------
|
|
|
|
|
|
def test_compose_agenda_vcard_sin_osint_con_direcciones(client, cfg, monkeypatch):
|
|
"""upsert_contact compone un vCard de agenda con ADR de la persona y SIN X-OSINT.
|
|
|
|
Crea una persona con DNI/sexo/dirección y un contacto enlazado por teléfono;
|
|
captura el vCard que se manda a Xandikos y verifica que (a) NO lleva ninguna
|
|
línea X-OSINT-*, (b) SÍ lleva la dirección de la persona como ADR.
|
|
"""
|
|
from server import writes
|
|
|
|
client.post("/api/ingest/vault")
|
|
|
|
# Persona con campos OSINT (dni/sexo/fecha_nac) + direccion + alias.
|
|
client.post(
|
|
"/api/person",
|
|
json={
|
|
"slug": "marca-osint",
|
|
"nombre": "Marca Osint",
|
|
"dni": "99999999R",
|
|
"sexo": "mujer",
|
|
"fecha_nacimiento": "1985-01-01",
|
|
"pais": "españa",
|
|
"contexto": "investigacion",
|
|
"telefonos": ["+34 655 100 200"],
|
|
"direcciones": ["Calle Falsa 123, Madrid"],
|
|
"aliases": ["La Marca"],
|
|
},
|
|
)
|
|
# El contacto se enlaza a la ficha por teléfono al re-ingestar el vault.
|
|
with write_conn(cfg.db_path) as conn:
|
|
conn.execute(
|
|
"INSERT INTO contacts VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
[
|
|
"uid-marca",
|
|
"/enmanuel/contacts/addressbook/",
|
|
None,
|
|
"Marca Osint",
|
|
'["+34 655 100 200"]',
|
|
"[]",
|
|
"BEGIN:VCARD...",
|
|
None,
|
|
datetime.now(tz=timezone.utc),
|
|
],
|
|
)
|
|
client.post("/api/ingest/vault") # enlaza uid-marca -> personas/marca-osint.md
|
|
|
|
# Capturamos el vCard que upsert_contact empuja a Xandikos.
|
|
pushed_vcards: list = []
|
|
|
|
def fake_put(base, user, pwd, collection, uid, vcard, **kw):
|
|
pushed_vcards.append(vcard)
|
|
return {"status": "ok", "http_status": 201, "url": "x"}
|
|
|
|
monkeypatch.setattr(writes, "pass_get_secret", lambda *_: {"status": "ok", "value": "pw"})
|
|
monkeypatch.setattr(writes, "carddav_put_vcard", fake_put)
|
|
monkeypatch.setattr(
|
|
writes,
|
|
"dav_list_resources",
|
|
lambda *a, **k: {
|
|
"status": "ok",
|
|
"http_status": 207,
|
|
"resources": [{"href": "/enmanuel/contacts/addressbook/uid-marca.vcf", "etag": '"newtag"'}],
|
|
},
|
|
)
|
|
|
|
r = client.put(
|
|
"/api/contact/uid-marca",
|
|
json={"uid": "uid-marca", "fn": "Marca Osint", "tels": ["+34 655 100 200"]},
|
|
).json()
|
|
assert r["status"] == "ok"
|
|
assert r["pushed"] is True
|
|
assert r["etag"] == '"newtag"' # etag del recurso capturado tras el push
|
|
|
|
assert pushed_vcards, "no se empujó ningún vCard"
|
|
vcard = pushed_vcards[-1]
|
|
# (a) Privacidad: ninguna línea X-OSINT-* (DNI/sexo/fecha-nac/pais/contexto).
|
|
assert "X-OSINT-" not in vcard
|
|
assert "99999999R" not in vcard # el DNI no se filtra al móvil
|
|
# (b) Direcciones y alias de la persona enlazada SÍ viajan (agenda).
|
|
assert "ADR;TYPE=HOME:;;Calle Falsa 123\\, Madrid;;;;" in vcard
|
|
assert "NICKNAME:La Marca" in vcard
|
|
|
|
# El etag nuevo quedó persistido en la DB (sync inverso fiable).
|
|
q = client.post(
|
|
"/api/query", json={"sql": "SELECT etag, raw FROM contacts WHERE uid = 'uid-marca'"}
|
|
).json()
|
|
assert q["rows"][0]["etag"] == '"newtag"'
|
|
assert "X-OSINT-" not in q["rows"][0]["raw"] # el raw guardado tampoco lleva OSINT
|
|
|
|
|
|
def test_pull_dav_incremental_por_etag(client, cfg, monkeypatch):
|
|
"""pull_dav actualiza solo el etag cambiado, no toca el igual y borra el ausente."""
|
|
from server import writes
|
|
|
|
client.post("/api/ingest/vault")
|
|
|
|
coll = "/enmanuel/contacts/addressbook/"
|
|
now = datetime.now(tz=timezone.utc)
|
|
# Tres contactos en la DB con etags conocidos.
|
|
with write_conn(cfg.db_path) as conn:
|
|
for uid, etag, fn in [
|
|
("c-same", '"e-same"', "Sin Cambios"),
|
|
("c-changed", '"e-old"', "Antes"),
|
|
("c-gone", '"e-gone"', "Se Borra"),
|
|
]:
|
|
conn.execute(
|
|
"INSERT INTO contacts VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
[uid, coll, etag, fn, "[]", "[]", "BEGIN:VCARD...", None, now],
|
|
)
|
|
|
|
# Remoto: c-same igual, c-changed con etag nuevo, c-gone ausente.
|
|
def fake_list(base, user, pwd, collection, **kw):
|
|
return {
|
|
"status": "ok",
|
|
"http_status": 207,
|
|
"resources": [
|
|
{"href": coll + "c-same.vcf", "etag": '"e-same"'},
|
|
{"href": coll + "c-changed.vcf", "etag": '"e-new"'},
|
|
],
|
|
}
|
|
|
|
def fake_get(base, user, pwd, href, **kw):
|
|
# Solo debería pedirse el recurso cambiado (c-same tiene etag igual).
|
|
assert "c-changed" in href, f"GET inesperado de {href}"
|
|
return {
|
|
"status": "ok",
|
|
"http_status": 200,
|
|
"text": (
|
|
"BEGIN:VCARD\r\nVERSION:3.0\r\nUID:c-changed\r\n"
|
|
"FN:Despues\r\nTEL;TYPE=CELL:+34 600 000 000\r\nEND:VCARD\r\n"
|
|
),
|
|
"url": href,
|
|
}
|
|
|
|
monkeypatch.setattr(writes, "pass_get_secret", lambda *_: {"status": "ok", "value": "pw"})
|
|
monkeypatch.setattr(writes, "dav_list_resources", fake_list)
|
|
monkeypatch.setattr(writes, "dav_get_resource", fake_get)
|
|
|
|
r = client.post("/api/sync/dav-pull").json()
|
|
assert r["status"] == "ok"
|
|
assert r["pulled"] == 1 # solo c-changed se descargó
|
|
assert r["updated"] == 1 # y era una actualización (ya existía)
|
|
assert r["unchanged"] == 1 # c-same no se tocó
|
|
assert r["deleted"] == 1 # c-gone se borró (ausente en el PROPFIND)
|
|
|
|
rows = client.post(
|
|
"/api/query",
|
|
json={"sql": "SELECT uid, etag, fn FROM contacts ORDER BY uid"},
|
|
).json()["rows"]
|
|
by_uid = {row["uid"]: row for row in rows}
|
|
assert set(by_uid) == {"c-changed", "c-same"} # c-gone desapareció
|
|
assert by_uid["c-same"]["fn"] == "Sin Cambios" # intacto
|
|
assert by_uid["c-same"]["etag"] == '"e-same"'
|
|
assert by_uid["c-changed"]["fn"] == "Despues" # FN del vCard nuevo
|
|
assert by_uid["c-changed"]["etag"] == '"e-new"' # etag remoto persistido
|