egutierrez
73f41a3474
dav_get_collection trae TODOS los recursos de una coleccion CardDAV/CalDAV en UNA peticion REPORT (addressbook-query / calendar-query) con el contenido vCard / VCALENDAR inline, evitando el patron N+1 (PROPFIND + un GET por recurso). Para 1064 contactos baja de ~9s a ~1s. dav_collection_ctag lee el ctag de la coleccion (PROPFIND Depth:0 barato) para validar caches sin descargar cuando nada cambio. Ambas: solo stdlib, basic auth, verify_tls, error-safe, tests que mockean el multistatus. Grupo dav, verificadas contra Xandikos real. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
204 lines
6.7 KiB
Python
204 lines
6.7 KiB
Python
"""Tests para dav_get_collection.
|
|
|
|
Smoke deterministas: monkeypatchean urllib.request.urlopen para capturar el
|
|
Request (method REPORT, Depth, auth, cuerpo del query) y devolver un XML
|
|
multistatus simulado con contenido inline. Cubren ambos perfiles (vcard / ical),
|
|
el des-escapado de entidades XML, los sinonimos de content_type, el content_type
|
|
invalido y el path de error HTTP.
|
|
"""
|
|
|
|
import base64
|
|
import sys
|
|
|
|
import infra.dav_get_collection # noqa: F401
|
|
|
|
mod = sys.modules["infra.dav_get_collection"]
|
|
dav_get_collection = mod.dav_get_collection
|
|
|
|
# Multistatus de un addressbook-query: 2 vCards inline. El segundo contiene una
|
|
# entidad XML (<) que debe des-escaparse a '<' en el campo data.
|
|
_VCARD_XML = (
|
|
'<?xml version="1.0"?>'
|
|
'<ns0:multistatus xmlns:ns0="DAV:" xmlns:ns1="urn:ietf:params:xml:ns:carddav">'
|
|
"<ns0:response><ns0:href>/enmanuel/contacts/addressbook/ada.vcf</ns0:href>"
|
|
"<ns0:propstat><ns0:status>HTTP/1.1 200 OK</ns0:status><ns0:prop>"
|
|
'<ns0:getetag>"etag-ada"</ns0:getetag>'
|
|
"<ns1:address-data>BEGIN:VCARD\nFN:Ada\nUID:ada\nEND:VCARD\n</ns1:address-data>"
|
|
"</ns0:prop></ns0:propstat></ns0:response>"
|
|
"<ns0:response><ns0:href>/enmanuel/contacts/addressbook/alan.vcf</ns0:href>"
|
|
"<ns0:propstat><ns0:status>HTTP/1.1 200 OK</ns0:status><ns0:prop>"
|
|
'<ns0:getetag>"etag-alan"</ns0:getetag>'
|
|
"<ns1:address-data>BEGIN:VCARD\nFN:Alan <turing>\nUID:alan\nEND:VCARD\n</ns1:address-data>"
|
|
"</ns0:prop></ns0:propstat></ns0:response>"
|
|
"</ns0:multistatus>"
|
|
)
|
|
|
|
# Multistatus de un calendar-query: 1 VEVENT inline.
|
|
_ICAL_XML = (
|
|
'<?xml version="1.0"?>'
|
|
'<ns0:multistatus xmlns:ns0="DAV:" xmlns:ns1="urn:ietf:params:xml:ns:caldav">'
|
|
"<ns0:response><ns0:href>/enmanuel/calendars/calendar/e1.ics</ns0:href>"
|
|
"<ns0:propstat><ns0:status>HTTP/1.1 200 OK</ns0:status><ns0:prop>"
|
|
'<ns0:getetag>"etag-e1"</ns0:getetag>'
|
|
"<ns1:calendar-data>BEGIN:VCALENDAR\nBEGIN:VEVENT\nSUMMARY:Cita\nUID:e1\nEND:VEVENT\nEND:VCALENDAR\n</ns1:calendar-data>"
|
|
"</ns0:prop></ns0:propstat></ns0:response>"
|
|
"</ns0:multistatus>"
|
|
)
|
|
|
|
|
|
class _FakeResp:
|
|
def __init__(self, payload: str):
|
|
self._payload = payload
|
|
self.status = 207
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
def read(self):
|
|
return self._payload.encode("utf-8")
|
|
|
|
|
|
def _capture(monkeypatch, payload: str):
|
|
captured = {}
|
|
|
|
def fake_urlopen(req, timeout=None, context=None):
|
|
captured["url"] = req.full_url
|
|
captured["method"] = req.get_method()
|
|
captured["headers"] = {k.lower(): v for k, v in req.header_items()}
|
|
captured["body"] = req.data.decode("utf-8") if req.data else ""
|
|
return _FakeResp(payload)
|
|
|
|
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
|
|
return captured
|
|
|
|
|
|
def test_vcard_construye_report_addressbook_query(monkeypatch):
|
|
cap = _capture(monkeypatch, _VCARD_XML)
|
|
dav_get_collection(
|
|
"https://dav.example.com",
|
|
"enmanuel",
|
|
"secret-pw",
|
|
"/enmanuel/contacts/addressbook/",
|
|
content_type="vcard",
|
|
)
|
|
assert cap["method"] == "REPORT"
|
|
assert cap["headers"]["depth"] == "1"
|
|
assert "addressbook-query" in cap["body"]
|
|
assert "address-data" in cap["body"]
|
|
assert "urn:ietf:params:xml:ns:carddav" in cap["body"]
|
|
|
|
|
|
def test_ical_construye_report_calendar_query_con_filtro(monkeypatch):
|
|
cap = _capture(monkeypatch, _ICAL_XML)
|
|
dav_get_collection(
|
|
"https://dav.example.com",
|
|
"enmanuel",
|
|
"secret-pw",
|
|
"/enmanuel/calendars/calendar/",
|
|
content_type="ical",
|
|
)
|
|
assert cap["method"] == "REPORT"
|
|
assert "calendar-query" in cap["body"]
|
|
assert 'comp-filter name="VCALENDAR"' in cap["body"]
|
|
assert "urn:ietf:params:xml:ns:caldav" in cap["body"]
|
|
|
|
|
|
def test_basic_auth_header_correcto(monkeypatch):
|
|
cap = _capture(monkeypatch, _VCARD_XML)
|
|
dav_get_collection(
|
|
"https://dav.example.com",
|
|
"enmanuel",
|
|
"secret-pw",
|
|
"/enmanuel/contacts/addressbook/",
|
|
)
|
|
expected = "Basic " + base64.b64encode(b"enmanuel:secret-pw").decode("ascii")
|
|
assert cap["headers"]["authorization"] == expected
|
|
|
|
|
|
def test_parsea_resources_con_data_inline(monkeypatch):
|
|
_capture(monkeypatch, _VCARD_XML)
|
|
res = dav_get_collection(
|
|
"https://dav.example.com",
|
|
"enmanuel",
|
|
"secret-pw",
|
|
"/enmanuel/contacts/addressbook/",
|
|
)
|
|
assert res["status"] == "ok"
|
|
assert len(res["resources"]) == 2
|
|
by_href = {r["href"]: r for r in res["resources"]}
|
|
ada = by_href["/enmanuel/contacts/addressbook/ada.vcf"]
|
|
assert ada["etag"] == '"etag-ada"'
|
|
assert "BEGIN:VCARD" in ada["data"]
|
|
assert "FN:Ada" in ada["data"]
|
|
|
|
|
|
def test_desescapa_entidades_xml_del_data(monkeypatch):
|
|
_capture(monkeypatch, _VCARD_XML)
|
|
res = dav_get_collection(
|
|
"https://dav.example.com",
|
|
"enmanuel",
|
|
"secret-pw",
|
|
"/enmanuel/contacts/addressbook/",
|
|
)
|
|
alan = next(r for r in res["resources"] if r["href"].endswith("alan.vcf"))
|
|
# <turing> debe quedar des-escapado a <turing>.
|
|
assert "FN:Alan <turing>" in alan["data"]
|
|
|
|
|
|
def test_ical_parsea_calendar_data(monkeypatch):
|
|
_capture(monkeypatch, _ICAL_XML)
|
|
res = dav_get_collection(
|
|
"https://dav.example.com",
|
|
"enmanuel",
|
|
"secret-pw",
|
|
"/enmanuel/calendars/calendar/",
|
|
content_type="ical",
|
|
)
|
|
assert res["status"] == "ok"
|
|
assert len(res["resources"]) == 1
|
|
assert "BEGIN:VEVENT" in res["resources"][0]["data"]
|
|
|
|
|
|
def test_acepta_sinonimos_de_content_type(monkeypatch):
|
|
_capture(monkeypatch, _VCARD_XML)
|
|
for alias in ("contacts", "carddav", "addressbook"):
|
|
res = dav_get_collection(
|
|
"https://dav.example.com",
|
|
"enmanuel",
|
|
"secret-pw",
|
|
"/enmanuel/contacts/addressbook/",
|
|
content_type=alias,
|
|
)
|
|
assert res["status"] == "ok"
|
|
|
|
|
|
def test_content_type_invalido_devuelve_error(monkeypatch):
|
|
_capture(monkeypatch, _VCARD_XML)
|
|
res = dav_get_collection(
|
|
"https://dav.example.com",
|
|
"enmanuel",
|
|
"secret-pw",
|
|
"/enmanuel/contacts/addressbook/",
|
|
content_type="json",
|
|
)
|
|
assert res["status"] == "error"
|
|
assert res["http_status"] is None
|
|
|
|
|
|
def test_httperror_devuelve_status_error(monkeypatch):
|
|
def fake_urlopen(req, timeout=None, context=None):
|
|
raise mod.urllib.error.HTTPError(req.full_url, 401, "Unauthorized", {}, None)
|
|
|
|
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
|
|
res = dav_get_collection(
|
|
"https://dav.example.com",
|
|
"enmanuel",
|
|
"secret-pw",
|
|
"/enmanuel/contacts/addressbook/",
|
|
)
|
|
assert res["status"] == "error"
|
|
assert res["http_status"] == 401
|