"""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.org_contacts",
        "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 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",
        "org_contacts",
        "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 (uid, collection, etag, fn, tels, emails, raw, note_path, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
            [
                "uid-movil-1",
                "/enmanuel/contacts/addressbook/",
                "etag1",
                "Ana G.",
                '["600111222"]',
                "[]",
                "BEGIN:VCARD...",
                None,
                now,
            ],
        )
        conn.execute(
            "INSERT INTO contacts (uid, collection, etag, fn, tels, emails, raw, note_path, updated_at) 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 (uid, collection, etag, fn, tels, emails, raw, note_path, updated_at) 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 (uid, collection, etag, fn, tels, emails, raw, note_path, updated_at) 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


# --- F5: push masivo por disco (1 rsync + 1 commit + 1 PROPFIND) ------------


def test_write_agenda_vcards_to_dir_nombres_y_sin_osint(client, cfg, tmp_path):
    """_write_agenda_vcards_to_dir escribe un .vcf por contacto, sin OSINT.

    Parte LOCAL del push masivo por disco, testeable sin SSH: genera los .vcf en
    un tmpdir con el nombre de recurso EXACTO del push HTTP (_safe_resource(uid)
    + '.vcf') y compone el vCard de agenda (con ADR de la persona enlazada, SIN
    ninguna línea X-OSINT-*).
    """
    from server import writes

    client.post("/api/ingest/vault")
    # Persona con campos OSINT + dirección, contacto enlazado por teléfono.
    client.post(
        "/api/person",
        json={
            "slug": "disco-osint",
            "nombre": "Disco Osint",
            "dni": "11111111H",
            "sexo": "hombre",
            "telefonos": ["+34 655 900 900"],
            "direcciones": ["Av. Disco 7, Málaga"],
        },
    )
    now = datetime.now(tz=timezone.utc)
    with write_conn(cfg.db_path) as conn:
        conn.execute(
            "INSERT INTO contacts (uid, collection, etag, fn, tels, emails, raw, note_path, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
            [
                "uid con espacios/raro",  # fuerza saneo del nombre del recurso
                "/enmanuel/contacts/addressbook/",
                None,
                "Disco Osint",
                '["+34 655 900 900"]',
                "[]",
                "BEGIN:VCARD...",
                None,
                now,
            ],
        )
    client.post("/api/ingest/vault")  # enlaza el contacto a la ficha

    rows = client.post(
        "/api/query",
        json={
            "sql": "SELECT uid, collection, fn, tels, emails, note_path "
            "FROM contacts WHERE uid = 'uid con espacios/raro'"
        },
    ).json()["rows"]

    out_dir = str(tmp_path / "vcards_out")
    os.makedirs(out_dir)
    res = writes._write_agenda_vcards_to_dir(cfg, rows, out_dir)

    assert res["written"] == 1
    # Nombre de recurso = saneo del uid + .vcf (idéntico al push HTTP).
    expected_name = writes._safe_resource("uid con espacios/raro") + ".vcf"
    assert expected_name == "uid_con_espacios_raro.vcf"
    assert list(res["by_resource"]) == [expected_name]
    assert res["by_resource"][expected_name] == "uid con espacios/raro"

    written_path = os.path.join(out_dir, expected_name)
    assert os.path.exists(written_path)
    content = open(written_path, encoding="utf-8").read()
    # (a) Privacidad: sin OSINT.
    assert "X-OSINT-" not in content
    assert "11111111H" not in content
    # (b) Agenda: la dirección de la persona enlazada SÍ viaja como ADR.
    assert "ADR;TYPE=HOME:;;Av. Disco 7\\, Málaga;;;;" in content


def test_push_all_dav_bulk_flujo_mockeado(client, cfg, monkeypatch):
    """push_all_dav_bulk: genera .vcf, rsync+commit (mock) y persiste etags por uid.

    Mockea rsync (subprocess), el commit remoto (HEAD before/after) y el PROPFIND
    final, verificando: written = nº de contactos, committed True (HEAD cambió) y
    que contacts.etag queda poblado con el etag del PROPFIND casado por uid.
    """
    from server import writes

    client.post("/api/ingest/vault")
    coll = "/enmanuel/contacts/addressbook/"
    now = datetime.now(tz=timezone.utc)
    with write_conn(cfg.db_path) as conn:
        for uid, fn in [("c-a", "Contacto A"), ("c-b", "Contacto B")]:
            conn.execute(
                "INSERT INTO contacts (uid, collection, etag, fn, tels, emails, raw, note_path, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
                [uid, coll, None, fn, "[]", "[]", "BEGIN:VCARD...", None, now],
            )

    # rsync + ssh (commit / rev-parse) mockeados a nivel de helper.
    monkeypatch.setattr(
        writes, "_rsync_vcards", lambda *a, **k: {"status": "ok", "stdout": "", "stderr": ""}
    )
    monkeypatch.setattr(
        writes,
        "_git_commit_remote",
        lambda *a, **k: {
            "status": "ok",
            "head_before": "aaaa111",
            "head_after": "bbbb222",
            "committed": True,
        },
    )
    monkeypatch.setattr(
        writes, "pass_get_secret", lambda *_: {"status": "ok", "value": "pw"}
    )

    # PROPFIND: devuelve el etag de cada recurso por su nombre saneado.
    def fake_list(base, user, pwd, collection, **kw):
        return {
            "status": "ok",
            "http_status": 207,
            "resources": [
                {"href": coll + "c-a.vcf", "etag": '"etag-a"'},
                {"href": coll + "c-b.vcf", "etag": '"etag-b"'},
            ],
        }

    monkeypatch.setattr(writes, "dav_list_resources", fake_list)

    r = writes.push_all_dav_bulk(cfg)
    assert r["status"] == "ok"
    assert r["written"] == 2
    assert r["rsynced"] is True
    assert r["committed"] is True
    assert r["head_before"] == "aaaa111"
    assert r["head_after"] == "bbbb222"
    assert r["etags_updated"] == 2
    assert isinstance(r["elapsed_s"], float)

    # Los etags del PROPFIND quedaron persistidos por uid (sync inverso fiable).
    rows = client.post(
        "/api/query",
        json={"sql": "SELECT uid, etag FROM contacts ORDER BY uid"},
    ).json()["rows"]
    by_uid = {row["uid"]: row["etag"] for row in rows}
    assert by_uid["c-a"] == '"etag-a"'
    assert by_uid["c-b"] == '"etag-b"'