"""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" assert evt["dtstart"].startswith("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 error claro, no crash.""" monkeypatch.setattr( srv, "dav_list_resources", 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 == 502 assert r1.json()["status"] == "error" r2 = client.get("/api/calendar") assert r2.status_code == 502 assert r2.json()["status"] == "error"