agent
e792bc6e17
Backend (server/main.py): - GET /api/calendars: lista las colecciones de calendario bajo el calendar-home con nombre y color (compone dav_list_calendars del registry). - GET /api/calendar?cal=&from=&to=: eventos de una colección concreta (caché por colección validada por ctag). dtstart/dtend ahora en ISO con offset + tz original + all_day; parseo robusto de TZID/UTC/todo-el-día con zoneinfo. - POST/PUT/DELETE /api/event[/<uid>]: CRUD de VEVENT contra Xandikos (fuente de verdad). Construye el VCALENDAR (con VTIMEZONE para zonas con DST), reutiliza el UID al editar (idempotente), trata 404 del DELETE como idempotente, invalida la caché de la colección tras escribir. Frontend: - CalendarView reescrita: conmutador Mes/Semana/Día con rejilla horaria propia (Mantine + dayjs, sin react-big-calendar para evitar fricción con React 19), mini-calendario de navegación, selector de calendario (con color), selector de zona horaria que recoloca los eventos, colores por evento (del VEVENT o del calendario). - EventModal: alta/edición/borrado con summary, inicio/fin, todo-el-día, TZ, calendario, color, ubicación y descripción. Fechas en formato local 24h. - calendar.ts: helpers de TZ (dayjs utc+timezone), posicionado por hora, semana empezando en lunes, locale es. api.ts: tipos y funciones de eventos/calendarios. Verificado: ciclo real crear→editar→borrar contra Xandikos (cero residuo), render del calendario en navegador (React 19 + Mantine v9 montan), pnpm build verde, 40 tests verdes (+ smoke gateado). MKCALENDAR queda fuera (documentado). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
896 lines
34 KiB
Python
896 lines
34 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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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"
|