egutierrez
9cbea2d036
- ContactIn + frontmatter + vCard multi-valor: emite N TEL, N EMAIL, N ADR; _vcard_to_json parsea ADR -> direcciones[] (y sigue leyendo X-OSINT-DIRECCION legacy). Los singulares telefono/email/direccion se mantienen por compat (= primer elemento de cada lista). - Libretas de contactos (addressbooks): endpoints GET/POST /api/addressbooks; en ContactsView un selector de libreta + boton 'Nueva libreta' (replica del patron de crear calendario) + filtro por libreta en la lista. - Frontend ContactsView: TagsInput para telefonos/emails/direcciones, cargando TODOS los valores al editar (antes solo el primero). - Feature flag OSINT_DB_BACKEND (dev/feature_flags.json, default off): con ON, osint_web lee/escribe contra el service osint_db (DuckDB = fuente de verdad) via server/osintdb_client.py; con OFF, el comportamiento historico (vault .md + vCard Xandikos) queda intacto byte a byte. Verificado: 52 tests backend (40 + 12 nuevos), tsc --noEmit limpio. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1102 lines
42 KiB
Python
1102 lines
42 KiB
Python
"""Tests del backend osint_web.
|
|
|
|
Cubren el contrato del DoD del issue 0172 sin depender de la red (Xandikos):
|
|
- Path traversal en /api/attachment (seguridad obligatoria).
|
|
- Vault inexistente -> error claro al arrancar, no 500.
|
|
- Grafo / tablas filtradas por tipo / ficha con attachments sobre un vault
|
|
mínimo sintético construido en un tmpdir.
|
|
- Endpoints DAV: degradación clara (no crash) cuando Xandikos no responde, y
|
|
parseo vCard/iCalendar a JSON sin red.
|
|
|
|
Se usa el ``TestClient`` de Starlette/FastAPI sobre un vault temporal, así los
|
|
tests son herméticos y deterministas (no tocan el vault real con PII).
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
# El backend orquesta funciones del registry: hay que poder importarlas.
|
|
_HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.join(_HERE, "..", "server"))
|
|
|
|
import main as srv # noqa: E402
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures: vault sintético mínimo
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _write(path: str, content: str) -> None:
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write(content)
|
|
|
|
|
|
@pytest.fixture()
|
|
def vault(tmp_path):
|
|
"""Construye un vault de Obsidian mínimo con personas, una org y un attachment."""
|
|
root = tmp_path / "osint"
|
|
# Una persona con foto embebida (por path) y un wikilink a una org y a una
|
|
# persona inexistente.
|
|
_write(
|
|
str(root / "personas" / "ana-gomez.md"),
|
|
"---\n"
|
|
"tipo: persona\n"
|
|
"nombre: Ana Gómez\n"
|
|
"dni: 12345678A\n"
|
|
"tags: [objetivo]\n"
|
|
"---\n\n"
|
|
"## Relaciones\n"
|
|
"- [[acme-sl]]\n"
|
|
"- [[Persona-Inexistente]]\n\n"
|
|
"## Documentos\n"
|
|
"![[attachments/personas/ana-gomez/ana-foto.jpg]]\n",
|
|
)
|
|
_write(
|
|
str(root / "organizaciones" / "acme-sl.md"),
|
|
"---\ntipo: organizacion\nnombre: Acme SL\ncif: B12345678\n---\n\nOrg de prueba.\n",
|
|
)
|
|
# El attachment embebido (basta un archivo cualquiera).
|
|
_write(str(root / "attachments" / "personas" / "ana-gomez" / "ana-foto.jpg"), "FAKEJPEGDATA")
|
|
# Un archivo secreto FUERA del vault, para el test de path traversal.
|
|
_write(str(tmp_path / "secret.txt"), "TOP SECRET")
|
|
return str(root)
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(vault):
|
|
app = srv.create_app(vault)
|
|
return TestClient(app)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Golden: grafo carga el vault
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_graph_loads_vault(client):
|
|
resp = client.get("/api/graph")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
# 2 notas reales (ana, acme) + 1 nodo fantasma (Persona-Inexistente).
|
|
ids = {n["id"] for n in data["nodes"]}
|
|
assert "ana-gomez" in ids
|
|
assert "acme-sl" in ids
|
|
assert data["total_edges"] >= 1
|
|
# Conteos por tipo presentes para la leyenda.
|
|
assert data["counts"].get("persona") == 1
|
|
assert data["counts"].get("organizacion") == 1
|
|
|
|
|
|
def test_node_card_with_attachments(client):
|
|
resp = client.get("/api/node/ana-gomez")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["frontmatter"]["nombre"] == "Ana Gómez"
|
|
assert data["body"].strip() != ""
|
|
# La galería de attachments resuelve la foto embebida (por path).
|
|
foto = next(a for a in data["attachments"] if a["name"].endswith("ana-foto.jpg"))
|
|
assert foto["kind"] == "image"
|
|
assert foto["path"] # path relativo al vault, no vacío
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Edge: tabla filtrada por tipo
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_nodes_filtered_by_tipo(client):
|
|
resp = client.get("/api/nodes", params={"tipo": "organizacion"})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["count"] == 1
|
|
assert all(r["tipo"] == "organizacion" for r in data["rows"])
|
|
assert data["rows"][0]["id"] == "acme-sl"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Edge: wikilink dangling -> nodo fantasma, sin crash
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_dangling_wikilink_is_phantom(client):
|
|
data = client.get("/api/graph").json()
|
|
phantom_ids = {n["id"] for n in data["nodes"] if n.get("dangling")}
|
|
assert "persona-inexistente" in phantom_ids
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Edge: nombre con mayúsculas/acentos -> slug estable
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_slugify_accents():
|
|
assert srv.slugify_obsidian_name("María del Mar") == "maria-del-mar"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Error: path traversal en attachment (SEGURIDAD obligatoria)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_attachment_path_traversal_blocked(client):
|
|
resp = client.get("/api/attachment", params={"path": "../../etc/passwd"})
|
|
assert resp.status_code in (403, 404)
|
|
assert "root:" not in resp.text
|
|
|
|
resp2 = client.get("/api/attachment", params={"path": "../secret.txt"})
|
|
assert resp2.status_code in (403, 404)
|
|
assert "TOP SECRET" not in resp2.text
|
|
|
|
|
|
def test_attachment_legit_served(client):
|
|
rel = os.path.join("attachments", "personas", "ana-gomez", "ana-foto.jpg")
|
|
resp = client.get("/api/attachment", params={"path": rel})
|
|
assert resp.status_code == 200
|
|
assert resp.content == b"FAKEJPEGDATA"
|
|
|
|
|
|
def test_attachment_nonexistent_inside_vault_404(client):
|
|
resp = client.get("/api/attachment", params={"path": "attachments/no-existe.png"})
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Error: vault inexistente -> error claro al arrancar, no 500
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_vault_inexistent_raises_clear_error():
|
|
with pytest.raises(FileNotFoundError):
|
|
srv.create_app("/no/existe/vault/osint")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Búsqueda
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_search_finds_node(client):
|
|
resp = client.get("/api/search", params={"q": "Ana"})
|
|
assert resp.status_code == 200
|
|
ids = {r["id"] for r in resp.json()["results"]}
|
|
assert "ana-gomez" in ids
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DAV: parseo a JSON (sin red) + degradación clara
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_vcard_to_json():
|
|
vcard = (
|
|
"BEGIN:VCARD\r\n"
|
|
"VERSION:3.0\r\n"
|
|
"UID:abc-123\r\n"
|
|
"FN:Juan Pérez\r\n"
|
|
"NICKNAME:Juanito\r\n"
|
|
"ORG:Acme;Ventas\r\n"
|
|
"TEL;TYPE=CELL:+34600111222\r\n"
|
|
"EMAIL;TYPE=HOME:juan@example.com\r\n"
|
|
"NOTE:Contacto de prueba\r\n"
|
|
"END:VCARD\r\n"
|
|
)
|
|
out = srv._vcard_to_json(vcard)
|
|
assert out["uid"] == "abc-123"
|
|
assert out["fn"] == "Juan Pérez"
|
|
assert out["nickname"] == "Juanito"
|
|
assert out["org"] == "Acme Ventas"
|
|
assert out["phones"][0]["value"] == "+34600111222"
|
|
assert out["emails"][0]["value"] == "juan@example.com"
|
|
|
|
|
|
def test_vevent_to_json_and_range():
|
|
vcal = (
|
|
"BEGIN:VCALENDAR\r\n"
|
|
"BEGIN:VEVENT\r\n"
|
|
"UID:evt-1\r\n"
|
|
"SUMMARY:Reunión OSINT\r\n"
|
|
"DTSTART:20260615T090000Z\r\n"
|
|
"DTEND:20260615T100000Z\r\n"
|
|
"LOCATION:Oficina\r\n"
|
|
"END:VEVENT\r\n"
|
|
"END:VCALENDAR\r\n"
|
|
)
|
|
events = srv._vcalendar_to_events(vcal)
|
|
assert len(events) == 1
|
|
evt = events[0]
|
|
assert evt["summary"] == "Reunión OSINT"
|
|
# Contrato nuevo: dtstart en ISO con offset (UTC -> +00:00) + tz original.
|
|
assert evt["dtstart"] == "2026-06-15T09:00:00+00:00"
|
|
assert evt["tz"] == "UTC"
|
|
assert evt["all_day"] is False
|
|
# dtstart_ical conserva el prefijo crudo para el filtro de rango.
|
|
assert evt["dtstart_ical"] == "20260615"
|
|
assert srv._event_in_range(evt, "20260601", "20260630") is True
|
|
assert srv._event_in_range(evt, "20260101", "20260131") is False
|
|
|
|
|
|
def test_dav_endpoints_degrade_without_network(client, monkeypatch):
|
|
"""Sin Xandikos accesible los endpoints DAV devuelven 503 claro, no crash.
|
|
|
|
Y los endpoints del vault siguen funcionando offline (no se ven afectados).
|
|
"""
|
|
monkeypatch.setattr(
|
|
srv,
|
|
"dav_get_collection",
|
|
lambda *a, **k: {"status": "error", "error": "sin red"},
|
|
)
|
|
monkeypatch.setattr(
|
|
srv, "dav_collection_ctag", lambda *a, **k: {"status": "error", "error": "sin red"}
|
|
)
|
|
# Evita leer pass en el test (cachea una password ficticia).
|
|
client.app.state.vault._xandikos_password = "x"
|
|
|
|
r1 = client.get("/api/contacts")
|
|
assert r1.status_code == 503
|
|
assert r1.json()["status"] == "error"
|
|
|
|
r2 = client.get("/api/calendar")
|
|
assert r2.status_code == 503
|
|
assert r2.json()["status"] == "error"
|
|
|
|
# El fallo DAV NO contamina los endpoints del vault (offline-OK).
|
|
assert client.get("/api/graph").status_code == 200
|
|
assert client.get("/api/health").status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DAV: campos osint / alias / nota / itemN. + caché + invalidación
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_vcard_to_json_alias_nota_osint_y_item_prefix():
|
|
"""Parsea alias (NICKNAME), nota (NOTE), osint (X-OSINT-*) y prefijo itemN."""
|
|
vcard = (
|
|
"BEGIN:VCARD\r\n"
|
|
"VERSION:3.0\r\n"
|
|
"UID:maria-001\r\n"
|
|
"FN:María del Mar Pérez\r\n"
|
|
"NICKNAME:Marimar\r\n"
|
|
"item1.TEL;TYPE=CELL:+34 600 111 222\r\n"
|
|
"item2.EMAIL;TYPE=INTERNET:maria@example.com\r\n"
|
|
"NOTE:Objetivo principal.\r\n"
|
|
"X-OSINT-DNI:12345678Z\r\n"
|
|
"X-OSINT-PAIS:España\r\n"
|
|
"X-OSINT-SEXO:F\r\n"
|
|
"END:VCARD\r\n"
|
|
)
|
|
out = srv._vcard_to_json(vcard)
|
|
assert out["uid"] == "maria-001"
|
|
assert out["nombre"] == "María del Mar Pérez"
|
|
assert out["alias"] == "Marimar"
|
|
assert out["nota"] == "Objetivo principal."
|
|
# El prefijo itemN. se elimina: TEL/EMAIL se reconocen.
|
|
assert out["telefonos"] == ["+34 600 111 222"]
|
|
assert out["correos"] == ["maria@example.com"]
|
|
# Bloque osint derivado de X-OSINT-*.
|
|
assert out["osint"] == {"dni": "12345678Z", "pais": "España", "sexo": "F"}
|
|
|
|
|
|
def test_vcard_to_json_nombre_desde_N_sin_fn():
|
|
vcard = (
|
|
"BEGIN:VCARD\r\nVERSION:3.0\r\nUID:juan-002\r\n"
|
|
"N:García;Juan;;;\r\nTEL:+34 611 222 333\r\nEND:VCARD\r\n"
|
|
)
|
|
out = srv._vcard_to_json(vcard)
|
|
assert out["nombre"] == "Juan García"
|
|
assert out["osint"] == {}
|
|
|
|
|
|
# Fixture DAV mockeado: dos contactos (uno con osint) y dos eventos.
|
|
_VCF_BODY = (
|
|
"BEGIN:VCARD\r\nVERSION:3.0\r\nUID:maria-001\r\nFN:María Pérez\r\n"
|
|
"NICKNAME:Mari\r\nX-OSINT-DNI:12345678Z\r\nX-OSINT-PAIS:España\r\n"
|
|
"item1.TEL;TYPE=CELL:+34600111222\r\nEND:VCARD\r\n"
|
|
)
|
|
_VCF_BODY_2 = (
|
|
"BEGIN:VCARD\r\nVERSION:3.0\r\nUID:juan-002\r\nFN:Juan García\r\n"
|
|
"EMAIL:juan@example.com\r\nEND:VCARD\r\n"
|
|
)
|
|
_ICS_BODY = (
|
|
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nUID:evt-001\r\n"
|
|
"SUMMARY:Reunión\r\nDTSTART:20260611T090000Z\r\nDTEND:20260611T100000Z\r\n"
|
|
"LOCATION:Madrid\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n"
|
|
)
|
|
_ICS_BODY_2 = (
|
|
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nUID:evt-002\r\n"
|
|
"SUMMARY:Vigilancia\r\nDTSTART:20260620T200000Z\r\nEND:VEVENT\r\n"
|
|
"END:VCALENDAR\r\n"
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def fake_dav(monkeypatch, tmp_path):
|
|
"""Parchea las funciones del registry DAV con fixtures en memoria (sin red).
|
|
|
|
Mockea ``dav_get_collection`` (UN REPORT que trae todos los recursos con su
|
|
contenido inline) y ``dav_collection_ctag`` (token de versión para la caché
|
|
en disco). Redirige la caché en disco a un tmpdir para no escribir en
|
|
``server/.cache``. Devuelve un dict con ``{"reports": int, "ctag": str}``:
|
|
``reports`` cuenta las descargas reales (REPORT) para verificar el cacheo, y
|
|
``ctag`` es mutable para simular un cambio en la colección.
|
|
"""
|
|
state = {"reports": 0, "ctag": "ctag-v1"}
|
|
contacts_res = [
|
|
{"href": "/enmanuel/contacts/addressbook/maria-001.vcf", "etag": '"a"', "data": _VCF_BODY},
|
|
{"href": "/enmanuel/contacts/addressbook/juan-002.vcf", "etag": '"b"', "data": _VCF_BODY_2},
|
|
]
|
|
calendar_res = [
|
|
{"href": "/enmanuel/calendars/calendar/evt-001.ics", "etag": '"c"', "data": _ICS_BODY},
|
|
{"href": "/enmanuel/calendars/calendar/evt-002.ics", "etag": '"d"', "data": _ICS_BODY_2},
|
|
]
|
|
|
|
def _get_collection(base, user, pw, collection, content_type="vcard", **kw):
|
|
state["reports"] += 1
|
|
res = contacts_res if "contacts" in collection else calendar_res
|
|
return {"status": "ok", "http_status": 207, "resources": res}
|
|
|
|
def _ctag(base, user, pw, collection, **kw):
|
|
return {"status": "ok", "http_status": 207, "ctag": state["ctag"]}
|
|
|
|
monkeypatch.setattr(srv, "dav_get_collection", _get_collection)
|
|
monkeypatch.setattr(srv, "dav_collection_ctag", _ctag)
|
|
monkeypatch.setattr(srv, "pass_get_secret", lambda *a, **k: {"status": "ok", "value": "x"})
|
|
# Caché en disco aislada por test (no toca server/.cache).
|
|
cache_dir = tmp_path / "dav_cache"
|
|
monkeypatch.setattr(srv, "_CONTACTS_CACHE_FILE", str(cache_dir / "contacts.json"))
|
|
monkeypatch.setattr(srv, "_CALENDAR_CACHE_FILE", str(cache_dir / "calendar.json"))
|
|
return state
|
|
|
|
|
|
def test_contacts_endpoint_parsea_y_cachea(client, fake_dav):
|
|
r = client.get("/api/contacts")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["status"] == "ok" and data["count"] == 2
|
|
by_uid = {c["uid"]: c for c in data["contacts"]}
|
|
maria = by_uid["maria-001"]
|
|
assert maria["nombre"] == "María Pérez"
|
|
assert maria["alias"] == "Mari"
|
|
assert maria["telefonos"] == ["+34600111222"]
|
|
assert maria["osint"] == {"dni": "12345678Z", "pais": "España"}
|
|
# Segunda llamada NO re-descarga (sirve de la caché en memoria).
|
|
reports_after_first = fake_dav["reports"]
|
|
client.get("/api/contacts")
|
|
assert fake_dav["reports"] == reports_after_first
|
|
|
|
|
|
def test_contact_by_uid_desde_cache(client, fake_dav):
|
|
r = client.get("/api/contact/maria-001")
|
|
assert r.status_code == 200
|
|
assert r.json()["contact"]["nombre"] == "María Pérez"
|
|
assert client.get("/api/contact/no-existe").status_code == 404
|
|
|
|
|
|
def test_calendar_endpoint_rango_y_cache(client, fake_dav):
|
|
# Sin rango: ambos eventos.
|
|
r = client.get("/api/calendar")
|
|
assert r.status_code == 200 and r.json()["count"] == 2
|
|
# Con rango: solo evt-001 (11 junio).
|
|
r2 = client.get("/api/calendar", params={"from": "2026-06-01", "to": "2026-06-15"})
|
|
assert [e["uid"] for e in r2.json()["events"]] == ["evt-001"]
|
|
|
|
|
|
def test_refresh_invalida_cache_dav(client, fake_dav):
|
|
client.get("/api/contacts") # llena caché
|
|
reports_before = fake_dav["reports"]
|
|
client.post("/api/refresh") # invalida + fuerza recarga
|
|
client.get("/api/contacts") # vuelve a descargar (REPORT)
|
|
assert fake_dav["reports"] > reports_before
|
|
|
|
|
|
def test_disk_cache_evita_descarga_en_proceso_nuevo(vault, fake_dav):
|
|
"""Un proceso nuevo con la caché en disco y el mismo ctag NO descarga.
|
|
|
|
Simula el reinicio del server: primer cliente descarga (1 REPORT) y escribe
|
|
la caché en disco; un segundo cliente (caché en memoria vacía) con el mismo
|
|
ctag sirve del disco sin un nuevo REPORT. Esto es el arranque instantáneo.
|
|
"""
|
|
c1 = TestClient(srv.create_app(vault))
|
|
assert c1.get("/api/contacts").json()["count"] == 2
|
|
reports_after_first = fake_dav["reports"]
|
|
assert reports_after_first >= 1 # hubo descarga al no haber disco aún
|
|
|
|
# Proceso "nuevo": estado en memoria vacío, pero la caché en disco existe y
|
|
# el ctag no cambió → debe servir del disco sin descargar.
|
|
c2 = TestClient(srv.create_app(vault))
|
|
data = c2.get("/api/contacts").json()
|
|
assert data["count"] == 2
|
|
assert {x["uid"] for x in data["contacts"]} == {"maria-001", "juan-002"}
|
|
assert fake_dav["reports"] == reports_after_first # CERO descargas nuevas
|
|
|
|
|
|
def test_disk_cache_recarga_si_cambia_ctag(vault, fake_dav):
|
|
"""Si el ctag de la colección cambia, el proceso nuevo SÍ vuelve a descargar."""
|
|
c1 = TestClient(srv.create_app(vault))
|
|
c1.get("/api/contacts")
|
|
reports_after_first = fake_dav["reports"]
|
|
|
|
fake_dav["ctag"] = "ctag-v2" # la colección cambió
|
|
c2 = TestClient(srv.create_app(vault))
|
|
c2.get("/api/contacts")
|
|
assert fake_dav["reports"] > reports_after_first # re-descargó
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CRUD de contactos: ficha .md del vault (verdad) + reflejo del vCard
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def crud_client(vault, monkeypatch):
|
|
"""Cliente con el PUT/DELETE de Xandikos mockeado (CRUD sin red).
|
|
|
|
Verifica el comportamiento sobre el vault real (la ficha .md) sin tocar
|
|
Xandikos. ``calls`` registra los PUT/DELETE que el server intentó, para
|
|
asertar que el reflejo en Xandikos se dispara.
|
|
"""
|
|
calls = {"put": [], "delete": []}
|
|
|
|
def _put(base, user, pw, coll, uid, vcard_text, **kw):
|
|
calls["put"].append({"uid": uid, "vcard": vcard_text})
|
|
return {"status": "ok", "http_status": 201, "url": coll + uid + ".vcf"}
|
|
|
|
def _delete(base, user, pw, resource_path, **kw):
|
|
calls["delete"].append(resource_path)
|
|
return {"status": "ok", "http_status": 204, "url": resource_path}
|
|
|
|
monkeypatch.setattr(srv, "carddav_put_vcard", _put)
|
|
monkeypatch.setattr(srv, "dav_delete_resource", _delete)
|
|
monkeypatch.setattr(srv, "pass_get_secret", lambda *a, **k: {"status": "ok", "value": "x"})
|
|
app = srv.create_app(vault)
|
|
client = TestClient(app)
|
|
client._crud_calls = calls # type: ignore[attr-defined]
|
|
return client
|
|
|
|
|
|
def _persona_md_path(vault_dir: str, slug: str) -> str:
|
|
return os.path.join(vault_dir, "personas", slug + ".md")
|
|
|
|
|
|
def test_crud_contact_full_cycle(crud_client, vault):
|
|
"""Golden: crear → editar → borrar un contacto (ficha .md + reflejo vCard)."""
|
|
calls = crud_client._crud_calls
|
|
|
|
# -- CREATE --
|
|
body = {
|
|
"tipo": "persona",
|
|
"nombre": "Zoé Test Crud",
|
|
"aliases": ["Zozo"],
|
|
"telefono": "+34600999888",
|
|
"email": "zoe@example.com",
|
|
"dni": "99999999Z",
|
|
"direccion": "Calle Falsa 123",
|
|
"pais": "españa",
|
|
"contexto": "prueba",
|
|
"notas": "Contacto de prueba CRUD.",
|
|
}
|
|
r = crud_client.post("/api/contact", json=body)
|
|
assert r.status_code == 201, r.text
|
|
slug = r.json()["slug"]
|
|
assert slug == "zoe-test-crud"
|
|
assert r.json()["uid"] == slug
|
|
# La ficha .md se escribió de verdad en el vault.
|
|
md = _persona_md_path(vault, slug)
|
|
assert os.path.isfile(md)
|
|
content = open(md, encoding="utf-8").read()
|
|
assert "tipo: persona" in content
|
|
assert "Zoé Test Crud" in content
|
|
assert "99999999Z" in content
|
|
assert "Contacto de prueba CRUD." in content
|
|
# Reflejo: se hizo PUT del vCard con UID=slug y los X-OSINT-*.
|
|
assert calls["put"], "debió hacer PUT del vCard"
|
|
vc = calls["put"][-1]["vcard"]
|
|
assert "UID:zoe-test-crud" in vc
|
|
assert "FN:Zoé Test Crud" in vc
|
|
assert "X-OSINT-DNI:99999999Z" in vc
|
|
assert "TEL;TYPE=CELL:+34600999888" in vc
|
|
|
|
# 409 al recrear el mismo slug.
|
|
assert crud_client.post("/api/contact", json=body).status_code == 409
|
|
|
|
# -- READ (vía /api/node, la ficha aparece en el grafo) --
|
|
nr = crud_client.get("/api/node/%s" % slug)
|
|
assert nr.status_code == 200
|
|
assert nr.json()["frontmatter"]["dni"] == "99999999Z"
|
|
|
|
# -- UPDATE --
|
|
body2 = dict(body)
|
|
body2["telefono"] = "+34611000111"
|
|
body2["notas"] = "Editado."
|
|
ur = crud_client.put("/api/contact/%s" % slug, json=body2)
|
|
assert ur.status_code == 200, ur.text
|
|
content2 = open(md, encoding="utf-8").read()
|
|
assert "+34611000111" in content2
|
|
assert "Editado." in content2
|
|
# Re-PUT del vCard con el teléfono nuevo.
|
|
assert "TEL;TYPE=CELL:+34611000111" in calls["put"][-1]["vcard"]
|
|
|
|
# -- DELETE --
|
|
dr = crud_client.delete("/api/contact/%s" % slug)
|
|
assert dr.status_code == 200, dr.text
|
|
assert dr.json()["deleted"] is True
|
|
assert not os.path.isfile(md), "la ficha .md debe desaparecer"
|
|
# Reflejo: DELETE del recurso <slug>.vcf en Xandikos.
|
|
assert any(slug + ".vcf" in p for p in calls["delete"])
|
|
|
|
|
|
def test_crud_organizacion(crud_client, vault):
|
|
"""Edge: una organización usa organizaciones/ y emite ORG en el vCard."""
|
|
calls = crud_client._crud_calls
|
|
r = crud_client.post(
|
|
"/api/contact",
|
|
json={"tipo": "organizacion", "nombre": "Acme Test Org", "pais": "españa"},
|
|
)
|
|
assert r.status_code == 201, r.text
|
|
slug = r.json()["slug"]
|
|
org_md = os.path.join(vault, "organizaciones", slug + ".md")
|
|
assert os.path.isfile(org_md)
|
|
assert "tipo: organizacion" in open(org_md, encoding="utf-8").read()
|
|
assert "ORG:Acme Test Org" in calls["put"][-1]["vcard"]
|
|
# limpieza
|
|
crud_client.delete("/api/contact/%s" % slug)
|
|
assert not os.path.isfile(org_md)
|
|
|
|
|
|
def test_crud_update_missing_404(crud_client):
|
|
"""Error: editar un contacto inexistente devuelve 404, no crash."""
|
|
r = crud_client.put(
|
|
"/api/contact/no-existe", json={"tipo": "persona", "nombre": "X"}
|
|
)
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_crud_delete_missing_404(crud_client):
|
|
"""Error: borrar un contacto inexistente devuelve 404."""
|
|
assert crud_client.delete("/api/contact/no-existe").status_code == 404
|
|
|
|
|
|
def test_crud_create_invalid_tipo_400(crud_client):
|
|
"""Error: un tipo no soportado devuelve 400."""
|
|
r = crud_client.post("/api/contact", json={"tipo": "robot", "nombre": "R2D2"})
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_crud_update_preserves_inherited_fields(crud_client, vault):
|
|
"""Edge: editar preserva campos heredados no editables (sexo, etc.)."""
|
|
# Crea y luego inyecta un campo heredado a mano (simula ficha previa).
|
|
crud_client.post(
|
|
"/api/contact", json={"tipo": "persona", "nombre": "Inés Hered"}
|
|
)
|
|
slug = "ines-hered"
|
|
md = _persona_md_path(vault, slug)
|
|
# Añade sexo + horoscopo al frontmatter como si fueran heredados.
|
|
srv.update_obsidian_note(md, set_frontmatter={"sexo": "mujer", "horoscopo": "aries"})
|
|
# Edita vía API: no toca sexo/horoscopo.
|
|
crud_client.put(
|
|
"/api/contact/%s" % slug,
|
|
json={"tipo": "persona", "nombre": "Inés Hered", "email": "i@x.com"},
|
|
)
|
|
content = open(md, encoding="utf-8").read()
|
|
assert "sexo: mujer" in content
|
|
assert "horoscopo: aries" in content
|
|
assert "i@x.com" in content
|
|
crud_client.delete("/api/contact/%s" % slug)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Contactos multi-valor: varias TEL/EMAIL/ADR + compat singular/lista
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_contactin_reconcilia_singular_y_lista():
|
|
"""ContactIn reconcilia singular↔lista: cliente viejo (singular) y nuevo (lista)."""
|
|
# Cliente viejo: solo el campo singular → se siembra la lista.
|
|
viejo = srv.ContactIn(nombre="X", telefono="111", email="a@x.com", direccion="C1")
|
|
assert viejo.telefonos == ["111"]
|
|
assert viejo.emails == ["a@x.com"]
|
|
assert viejo.direcciones == ["C1"]
|
|
# El singular se conserva = primer elemento.
|
|
assert viejo.telefono == "111"
|
|
|
|
# Cliente nuevo: listas → el singular se rellena con lista[0].
|
|
nuevo = srv.ContactIn(
|
|
nombre="X",
|
|
telefonos=["111", "222"],
|
|
emails=["a@x.com", "b@x.com"],
|
|
direcciones=["C1", "C2"],
|
|
)
|
|
assert nuevo.telefonos == ["111", "222"]
|
|
assert nuevo.telefono == "111"
|
|
assert nuevo.email == "a@x.com"
|
|
assert nuevo.direccion == "C1"
|
|
|
|
|
|
def test_build_vcard_multivalor_emite_n_lineas():
|
|
"""_build_vcard emite una TEL/EMAIL/ADR por elemento de cada lista."""
|
|
fm = {
|
|
"tipo": "persona",
|
|
"nombre": "Multi Persona",
|
|
"telefonos": ["111", "222"],
|
|
"emails": ["a@x.com", "b@x.com"],
|
|
"direcciones": ["Calle 1", "Calle 2"],
|
|
"dni": "X",
|
|
}
|
|
vc = srv._build_vcard(fm, "multi-persona")
|
|
assert vc.count("TEL;TYPE=CELL:") == 2
|
|
assert "TEL;TYPE=CELL:111" in vc and "TEL;TYPE=CELL:222" in vc
|
|
assert vc.count("EMAIL;TYPE=INTERNET:") == 2
|
|
assert vc.count("ADR;TYPE=HOME:") == 2
|
|
assert "ADR;TYPE=HOME:;;Calle 1;;;;" in vc
|
|
assert "X-OSINT-DNI:X" in vc
|
|
|
|
|
|
def test_vcard_to_json_lee_adr_multivalor():
|
|
"""_vcard_to_json reconstruye la lista de direcciones desde las líneas ADR."""
|
|
vcard = (
|
|
"BEGIN:VCARD\r\nVERSION:3.0\r\nUID:adr-1\r\nFN:Con Direcciones\r\n"
|
|
"ADR;TYPE=HOME:;;Calle Uno 1;;;;\r\n"
|
|
"ADR;TYPE=HOME:;;Calle Dos 2;Madrid;;28001;España\r\n"
|
|
"END:VCARD\r\n"
|
|
)
|
|
out = srv._vcard_to_json(vcard)
|
|
assert out["direcciones"][0] == "Calle Uno 1"
|
|
# El 2º ADR concatena street + locality/region/postal/country legibles.
|
|
assert "Calle Dos 2" in out["direcciones"][1]
|
|
assert "Madrid" in out["direcciones"][1]
|
|
|
|
|
|
def test_vcard_to_json_legacy_x_osint_direccion():
|
|
"""Compat: una dirección antigua en X-OSINT-DIRECCION sube a direcciones[]."""
|
|
vcard = (
|
|
"BEGIN:VCARD\r\nVERSION:3.0\r\nUID:legacy-1\r\nFN:Legacy\r\n"
|
|
"X-OSINT-DIRECCION:Calle Antigua 7\r\nEND:VCARD\r\n"
|
|
)
|
|
out = srv._vcard_to_json(vcard)
|
|
assert "Calle Antigua 7" in out["direcciones"]
|
|
# Se mantiene también en osint.direccion por si un lector viejo lo consulta.
|
|
assert out["osint"]["direccion"] == "Calle Antigua 7"
|
|
|
|
|
|
def test_crud_multivalor_round_trip(crud_client, vault):
|
|
"""Golden multi-valor: crear con 2 teléfonos/emails/direcciones y verlos todos."""
|
|
calls = crud_client._crud_calls
|
|
body = {
|
|
"tipo": "persona",
|
|
"nombre": "Poli Valor",
|
|
"telefonos": ["+34600000001", "+34600000002"],
|
|
"emails": ["uno@x.com", "dos@x.com"],
|
|
"direcciones": ["Calle A 1", "Calle B 2"],
|
|
}
|
|
r = crud_client.post("/api/contact", json=body)
|
|
assert r.status_code == 201, r.text
|
|
slug = r.json()["slug"]
|
|
md = os.path.join(vault, "personas", slug + ".md")
|
|
content = open(md, encoding="utf-8").read()
|
|
# El frontmatter escribe las listas multi-valor + el singular compat.
|
|
assert "+34600000001" in content and "+34600000002" in content
|
|
assert "uno@x.com" in content and "dos@x.com" in content
|
|
# El vCard emitió las dos líneas TEL/EMAIL/ADR.
|
|
vc = calls["put"][-1]["vcard"]
|
|
assert vc.count("TEL;TYPE=CELL:") == 2
|
|
assert vc.count("EMAIL;TYPE=INTERNET:") == 2
|
|
assert vc.count("ADR;TYPE=HOME:") == 2
|
|
crud_client.delete("/api/contact/%s" % slug)
|
|
|
|
|
|
def test_crud_singular_compat_sigue_funcionando(crud_client, vault):
|
|
"""Edge: un cliente viejo que envía solo el singular sigue funcionando."""
|
|
calls = crud_client._crud_calls
|
|
body = {"tipo": "persona", "nombre": "Solo Singular", "telefono": "+34611111111"}
|
|
r = crud_client.post("/api/contact", json=body)
|
|
assert r.status_code == 201, r.text
|
|
slug = r.json()["slug"]
|
|
vc = calls["put"][-1]["vcard"]
|
|
assert "TEL;TYPE=CELL:+34611111111" in vc
|
|
assert vc.count("TEL;TYPE=CELL:") == 1
|
|
crud_client.delete("/api/contact/%s" % slug)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Libretas (addressbooks) + feature flag OSINT_DB_BACKEND
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_feature_flag_off_por_defecto(monkeypatch, tmp_path):
|
|
"""El flag OSINT_DB_BACKEND está OFF por defecto (archivo ausente → False)."""
|
|
monkeypatch.setattr(srv, "_FLAGS_FILE", str(tmp_path / "no-existe.json"))
|
|
assert srv._osint_db_backend_enabled() is False
|
|
|
|
|
|
def test_feature_flag_lee_archivo(monkeypatch, tmp_path):
|
|
"""_osint_db_backend_enabled refleja el archivo dev/feature_flags.json."""
|
|
flags = tmp_path / "feature_flags.json"
|
|
flags.write_text(
|
|
'{"flags":{"OSINT_DB_BACKEND":{"enabled":true}}}', encoding="utf-8"
|
|
)
|
|
monkeypatch.setattr(srv, "_FLAGS_FILE", str(flags))
|
|
assert srv._osint_db_backend_enabled() is True
|
|
flags.write_text(
|
|
'{"flags":{"OSINT_DB_BACKEND":{"enabled":false}}}', encoding="utf-8"
|
|
)
|
|
assert srv._osint_db_backend_enabled() is False
|
|
|
|
|
|
def test_addressbooks_off_devuelve_libreta_por_defecto(crud_client):
|
|
"""Con el flag OFF, /api/addressbooks devuelve solo la libreta por defecto."""
|
|
r = crud_client.get("/api/addressbooks")
|
|
assert r.status_code == 200, r.text
|
|
data = r.json()
|
|
assert data["status"] == "ok"
|
|
assert data["count"] == 1
|
|
assert data["addressbooks"][0]["slug"] == srv.DEFAULT_ADDRESSBOOK_SLUG
|
|
assert data["default"] == srv.DEFAULT_ADDRESSBOOK_SLUG
|
|
|
|
|
|
def test_create_addressbook_off_devuelve_501(crud_client, monkeypatch, tmp_path):
|
|
"""Error: crear libreta con el flag OFF → 501 claro (requiere OSINT_DB_BACKEND)."""
|
|
monkeypatch.setattr(srv, "_FLAGS_FILE", str(tmp_path / "no-existe.json"))
|
|
r = crud_client.post("/api/addressbooks", json={"slug": "trabajo", "name": "Trabajo"})
|
|
assert r.status_code == 501
|
|
assert "OSINT_DB_BACKEND" in r.json()["detail"]
|
|
|
|
|
|
def test_contacts_flag_on_usa_osint_db(crud_client, monkeypatch, tmp_path):
|
|
"""Con el flag ON, /api/contacts lee del osint_db (mockeado), no de Xandikos."""
|
|
flags = tmp_path / "feature_flags.json"
|
|
flags.write_text(
|
|
'{"flags":{"OSINT_DB_BACKEND":{"enabled":true}}}', encoding="utf-8"
|
|
)
|
|
monkeypatch.setattr(srv, "_FLAGS_FILE", str(flags))
|
|
monkeypatch.setattr(
|
|
srv.osintdb_client,
|
|
"list_contacts",
|
|
lambda: [
|
|
{
|
|
"uid": "u1",
|
|
"collection": "addressbook",
|
|
"fn": "Desde DuckDB",
|
|
"tels": '["111", "222"]',
|
|
"emails": '["a@x.com"]',
|
|
"note_path": None,
|
|
}
|
|
],
|
|
)
|
|
r = crud_client.get("/api/contacts")
|
|
assert r.status_code == 200, r.text
|
|
contacts = r.json()["contacts"]
|
|
assert len(contacts) == 1
|
|
c = contacts[0]
|
|
assert c["nombre"] == "Desde DuckDB"
|
|
# Los JSON array de tels/emails se parsean a lista de strings.
|
|
assert c["telefonos"] == ["111", "222"]
|
|
assert c["correos"] == ["a@x.com"]
|
|
|
|
|
|
def test_contacts_flag_on_osint_db_caido_503(crud_client, monkeypatch, tmp_path):
|
|
"""Error: con el flag ON y el osint_db caído, /api/contacts degrada a 503."""
|
|
flags = tmp_path / "feature_flags.json"
|
|
flags.write_text(
|
|
'{"flags":{"OSINT_DB_BACKEND":{"enabled":true}}}', encoding="utf-8"
|
|
)
|
|
monkeypatch.setattr(srv, "_FLAGS_FILE", str(flags))
|
|
|
|
def _down():
|
|
raise srv.osintdb_client.OsintDbUnavailable("no arrancado")
|
|
|
|
monkeypatch.setattr(srv.osintdb_client, "list_contacts", _down)
|
|
r = crud_client.get("/api/contacts")
|
|
assert r.status_code == 503
|
|
assert r.json()["status"] == "error"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Calendario: parseo de fechas (TZID / UTC / todo el día) + builder VCALENDAR
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_vevent_tzid_se_normaliza_a_iso_con_offset():
|
|
"""Un DTSTART;TZID=Europe/Madrid sale como ISO con el offset de Madrid."""
|
|
vcal = (
|
|
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nUID:tz-1\r\n"
|
|
"SUMMARY:Con zona\r\n"
|
|
"DTSTART;TZID=Europe/Madrid:20260615T100000\r\n"
|
|
"DTEND;TZID=Europe/Madrid:20260615T110000\r\n"
|
|
"END:VEVENT\r\nEND:VCALENDAR\r\n"
|
|
)
|
|
evt = srv._vcalendar_to_events(vcal)[0]
|
|
# Junio: Madrid está en CEST (+02:00).
|
|
assert evt["dtstart"] == "2026-06-15T10:00:00+02:00"
|
|
assert evt["dtend"] == "2026-06-15T11:00:00+02:00"
|
|
assert evt["tz"] == "Europe/Madrid"
|
|
assert evt["all_day"] is False
|
|
|
|
|
|
def test_vevent_all_day_value_date():
|
|
"""Un DTSTART;VALUE=DATE:YYYYMMDD se marca all_day y sale como fecha ISO."""
|
|
vcal = (
|
|
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nUID:ad-1\r\n"
|
|
"SUMMARY:Todo el día\r\nDTSTART;VALUE=DATE:20260615\r\n"
|
|
"END:VEVENT\r\nEND:VCALENDAR\r\n"
|
|
)
|
|
evt = srv._vcalendar_to_events(vcal)[0]
|
|
assert evt["all_day"] is True
|
|
assert evt["dtstart"] == "2026-06-15"
|
|
assert evt["dtstart_ical"] == "20260615"
|
|
|
|
|
|
def test_build_vcalendar_tzid():
|
|
"""El builder emite DTSTART;TZID + un VTIMEZONE para una zona con DST."""
|
|
data = srv.EventIn(
|
|
cal="",
|
|
summary="Reunión, con coma",
|
|
dtstart="2026-06-15T10:00",
|
|
dtend="2026-06-15T11:00",
|
|
tz="Europe/Madrid",
|
|
all_day=False,
|
|
location="Oficina; sala 2",
|
|
)
|
|
vcal = srv._build_vcalendar(data, "uid-abc")
|
|
assert "BEGIN:VEVENT" in vcal
|
|
assert "UID:uid-abc" in vcal
|
|
assert "DTSTART;TZID=Europe/Madrid:20260615T100000" in vcal
|
|
assert "DTEND;TZID=Europe/Madrid:20260615T110000" in vcal
|
|
assert "BEGIN:VTIMEZONE" in vcal and "TZID:Europe/Madrid" in vcal
|
|
# El summary/location se escapan (coma y punto y coma).
|
|
assert "SUMMARY:Reunión\\, con coma" in vcal
|
|
assert "LOCATION:Oficina\\; sala 2" in vcal
|
|
# Round-trip: parsear lo construido reproduce la hora local de Madrid.
|
|
evt = srv._vcalendar_to_events(vcal)[0]
|
|
assert evt["dtstart"] == "2026-06-15T10:00:00+02:00"
|
|
assert evt["tz"] == "Europe/Madrid"
|
|
|
|
|
|
def test_build_vcalendar_all_day():
|
|
data = srv.EventIn(summary="Festivo", dtstart="2026-06-15", all_day=True)
|
|
vcal = srv._build_vcalendar(data, "uid-ad")
|
|
assert "DTSTART;VALUE=DATE:20260615" in vcal
|
|
assert "BEGIN:VTIMEZONE" not in vcal # all-day no necesita VTIMEZONE
|
|
|
|
|
|
def test_build_vcalendar_con_offset_va_a_utc():
|
|
"""Una entrada con offset explícito se convierte a UTC (...Z)."""
|
|
data = srv.EventIn(
|
|
summary="Con offset", dtstart="2026-06-15T10:00:00+02:00", tz="Europe/Madrid"
|
|
)
|
|
vcal = srv._build_vcalendar(data, "uid-off")
|
|
# 10:00+02:00 == 08:00 UTC.
|
|
assert "DTSTART:20260615T080000Z" in vcal
|
|
|
|
|
|
def test_build_vcalendar_fecha_invalida_lanza():
|
|
data = srv.EventIn(summary="Mala", dtstart="no-es-fecha")
|
|
with pytest.raises(ValueError):
|
|
srv._build_vcalendar(data, "uid-x")
|
|
|
|
|
|
def test_safe_event_resource_coincide_con_caldav_put():
|
|
"""El nombre .ics del DELETE coincide con el que deriva caldav_put_event."""
|
|
uid = "abc/def:ghi 123"
|
|
assert srv._safe_event_resource(uid) == "abc_def_ghi_123.ics"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Calendario: listado de colecciones + CRUD de eventos (DAV mockeado)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def cal_client(vault, monkeypatch, tmp_path):
|
|
"""Cliente con el PUT/DELETE/list/get de CalDAV mockeado (CRUD sin red).
|
|
|
|
Registra los PUT/DELETE intentados y simula un servidor con una colección de
|
|
calendario y un store de eventos en memoria (para que el GET tras crear vea
|
|
el evento). ``calls`` expone lo que el server intentó.
|
|
"""
|
|
calls = {"put": [], "delete": []}
|
|
# Store en memoria: resource_name -> VCALENDAR text.
|
|
store = {}
|
|
ctag = {"v": "cal-v1"}
|
|
|
|
def _put_event(base, user, pw, coll, uid, vcal, **kw):
|
|
calls["put"].append({"uid": uid, "vcal": vcal, "coll": coll})
|
|
store[srv._safe_event_resource(uid)] = vcal
|
|
ctag["v"] = "cal-" + str(len(calls["put"]) + len(calls["delete"]))
|
|
return {"status": "ok", "http_status": 201, "url": coll + uid + ".ics"}
|
|
|
|
def _delete(base, user, pw, resource_path, **kw):
|
|
calls["delete"].append(resource_path)
|
|
name = resource_path.rstrip("/").rsplit("/", 1)[-1]
|
|
existed = store.pop(name, None)
|
|
ctag["v"] = "cal-" + str(len(calls["put"]) + len(calls["delete"]))
|
|
if existed is None:
|
|
return {"status": "error", "http_status": 404, "error": "http 404"}
|
|
return {"status": "ok", "http_status": 204, "url": resource_path}
|
|
|
|
def _get_collection(base, user, pw, collection, content_type="ical", **kw):
|
|
res = [
|
|
{"href": collection + name, "etag": '"%s"' % name, "data": data}
|
|
for name, data in store.items()
|
|
]
|
|
return {"status": "ok", "http_status": 207, "resources": res}
|
|
|
|
def _ctag(base, user, pw, collection, **kw):
|
|
return {"status": "ok", "http_status": 207, "ctag": ctag["v"]}
|
|
|
|
def _list_calendars(base, user, pw, home, **kw):
|
|
return {
|
|
"status": "ok",
|
|
"http_status": 207,
|
|
"calendars": [
|
|
{
|
|
"href": "/enmanuel/calendars/calendar/",
|
|
"name": "calendar",
|
|
"color": None,
|
|
},
|
|
{
|
|
"href": "/enmanuel/calendars/trabajo/",
|
|
"name": "Trabajo",
|
|
"color": "#FF2968FF",
|
|
},
|
|
],
|
|
}
|
|
|
|
monkeypatch.setattr(srv, "caldav_put_event", _put_event)
|
|
monkeypatch.setattr(srv, "dav_delete_resource", _delete)
|
|
monkeypatch.setattr(srv, "dav_get_collection", _get_collection)
|
|
monkeypatch.setattr(srv, "dav_collection_ctag", _ctag)
|
|
monkeypatch.setattr(srv, "dav_list_calendars", _list_calendars)
|
|
monkeypatch.setattr(srv, "pass_get_secret", lambda *a, **k: {"status": "ok", "value": "x"})
|
|
cache_dir = tmp_path / "cal_cache"
|
|
monkeypatch.setattr(srv, "_CALENDAR_CACHE_FILE", str(cache_dir / "calendar.json"))
|
|
monkeypatch.setattr(srv, "_CACHE_DIR", str(cache_dir))
|
|
app = srv.create_app(vault)
|
|
client = TestClient(app)
|
|
client._cal_calls = calls # type: ignore[attr-defined]
|
|
return client
|
|
|
|
|
|
def test_calendars_endpoint_lista_con_color(cal_client):
|
|
r = cal_client.get("/api/calendars")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["status"] == "ok" and data["count"] == 2
|
|
by_name = {c["name"]: c for c in data["calendars"]}
|
|
assert by_name["Trabajo"]["color"] == "#FF2968FF"
|
|
assert by_name["calendar"]["color"] is None
|
|
|
|
|
|
def test_event_crud_full_cycle(cal_client):
|
|
"""Golden: crear → leer → editar → borrar un evento (VEVENT sobre CalDAV)."""
|
|
calls = cal_client._cal_calls
|
|
|
|
# -- CREATE --
|
|
body = {
|
|
"summary": "ZZ Test Event",
|
|
"dtstart": "2026-06-15T10:00",
|
|
"dtend": "2026-06-15T11:00",
|
|
"tz": "Europe/Madrid",
|
|
"location": "Sala",
|
|
"description": "Prueba CRUD",
|
|
}
|
|
r = cal_client.post("/api/event", json=body)
|
|
assert r.status_code == 201, r.text
|
|
uid = r.json()["uid"]
|
|
assert uid
|
|
assert calls["put"], "debió hacer PUT del VEVENT"
|
|
assert "SUMMARY:ZZ Test Event" in calls["put"][-1]["vcal"]
|
|
assert "DTSTART;TZID=Europe/Madrid:20260615T100000" in calls["put"][-1]["vcal"]
|
|
|
|
# -- READ (aparece en /api/calendar) --
|
|
cal_client.post("/api/refresh") # fuerza recarga desde el store mockeado
|
|
g = cal_client.get("/api/calendar", params={"from": "2026-06-01", "to": "2026-06-30"})
|
|
assert g.status_code == 200
|
|
uids = [e["uid"] for e in g.json()["events"]]
|
|
assert uid in uids
|
|
evt = next(e for e in g.json()["events"] if e["uid"] == uid)
|
|
assert evt["dtstart"] == "2026-06-15T10:00:00+02:00"
|
|
assert evt["tz"] == "Europe/Madrid"
|
|
|
|
# -- UPDATE --
|
|
body2 = dict(body)
|
|
body2["summary"] = "ZZ Test Event (editado)"
|
|
body2["dtstart"] = "2026-06-15T12:00"
|
|
ur = cal_client.put("/api/event/%s" % uid, json=body2)
|
|
assert ur.status_code == 200, ur.text
|
|
assert "SUMMARY:ZZ Test Event (editado)" in calls["put"][-1]["vcal"]
|
|
assert "20260615T120000" in calls["put"][-1]["vcal"]
|
|
# El UID se reutiliza (idempotente): mismo recurso.
|
|
assert calls["put"][-1]["uid"] == uid
|
|
|
|
# -- DELETE --
|
|
dr = cal_client.delete("/api/event/%s" % uid)
|
|
assert dr.status_code == 200, dr.text
|
|
assert dr.json()["deleted"] is True
|
|
assert any(uid in p for p in calls["delete"])
|
|
# Tras borrar, ya no aparece.
|
|
cal_client.post("/api/refresh")
|
|
g2 = cal_client.get("/api/calendar")
|
|
assert uid not in [e["uid"] for e in g2.json()["events"]]
|
|
|
|
|
|
def test_event_create_fecha_invalida_400(cal_client):
|
|
r = cal_client.post(
|
|
"/api/event", json={"summary": "X", "dtstart": "no-fecha"}
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_event_create_summary_vacio_400(cal_client):
|
|
r = cal_client.post(
|
|
"/api/event", json={"summary": " ", "dtstart": "2026-06-15T10:00"}
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_event_delete_idempotente_404_es_ok(cal_client):
|
|
"""Borrar un evento inexistente NO es error: Xandikos 404 → idempotente."""
|
|
r = cal_client.delete("/api/event/no-existe-uid")
|
|
assert r.status_code == 200
|
|
assert r.json()["deleted"] is True
|
|
|
|
|
|
def test_calendar_endpoint_degrada_sin_red(client, monkeypatch):
|
|
"""Sin Xandikos, /api/calendars y /api/calendar devuelven 503 claro."""
|
|
monkeypatch.setattr(
|
|
srv, "dav_list_calendars", lambda *a, **k: {"status": "error", "error": "sin red"}
|
|
)
|
|
monkeypatch.setattr(
|
|
srv, "dav_get_collection", lambda *a, **k: {"status": "error", "error": "sin red"}
|
|
)
|
|
monkeypatch.setattr(
|
|
srv, "dav_collection_ctag", lambda *a, **k: {"status": "error", "error": "sin red"}
|
|
)
|
|
client.app.state.vault._xandikos_password = "x"
|
|
assert client.get("/api/calendars").status_code == 503
|
|
assert client.get("/api/calendar").status_code == 503
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Smoke real opcional contra Xandikos (gateado, no corre en CI)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
os.environ.get("OSINT_WEB_DAV_SMOKE") != "1",
|
|
reason="smoke DAV real desactivado (export OSINT_WEB_DAV_SMOKE=1 para correrlo)",
|
|
)
|
|
def test_smoke_dav_real(vault):
|
|
"""Smoke contra el Xandikos real: ≥1 contacto y ≥1 evento. Requiere red + pass."""
|
|
app = srv.create_app(vault)
|
|
real_client = TestClient(app)
|
|
rc = real_client.get("/api/contacts")
|
|
assert rc.status_code == 200
|
|
assert rc.json()["status"] == "ok" and rc.json()["count"] >= 1
|
|
re_ = real_client.get("/api/calendar")
|
|
assert re_.status_code == 200 and re_.json()["status"] == "ok"
|