tfaour
31adc34a19
64 tests covering: auth/setup and the CSRF gate, the "owner adds their own data, anyone linked can mute it" permission pattern shared across calendar-select/tasks-source/whiteboard-source, migration correctness (fresh install, idempotent re-run, expected columns), battery estimate outlier rejection, calendar_feed's fetch/merge/partial-failure handling, webdav_client's fetch/list-directory, the whiteboard force-refresh throttle bypass and browse endpoint, and render-size invariants across calendar views/orientations. No DB/HTTP fixtures need Docker, Node, or a real Immich/CalDAV/WebDAV server -- a fresh temp SQLite file plus a couple of small local HTTP servers as test doubles cover it all. Table data is wiped and reseeded between tests rather than relying on SQLAlchemy's transaction-rollback isolation pattern, which needs a pysqlite event-listener workaround app/db.py's engine doesn't have and has no reason to gain just for tests. Wired into .gitea/workflows/server-docker-build.yml as its own job that build-and-push now depends on, so a failing suite blocks the image push rather than just running alongside it for show.
139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
"""webdav_client.py against a real local WebDAV-ish HTTP server (Basic
|
|
auth + PROPFIND) -- no ORM, no FastAPI, pure protocol-level fetch/list
|
|
functions used by whiteboard frame mode."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
import pytest
|
|
|
|
from app.webdav_client import WebDavError, fetch_file, list_directory, parent_directory_url
|
|
|
|
_USERNAME = "alice"
|
|
_PASSWORD = "secret"
|
|
_FILE_BYTES = b'{"elements": [], "appState": {}}'
|
|
|
|
_PROPFIND_RESPONSE = b"""<?xml version="1.0"?>
|
|
<d:multistatus xmlns:d="DAV:">
|
|
<d:response>
|
|
<d:href>/dav/files/alice/Boards/</d:href>
|
|
<d:propstat>
|
|
<d:prop>
|
|
<d:displayname>Boards</d:displayname>
|
|
<d:resourcetype><d:collection/></d:resourcetype>
|
|
</d:prop>
|
|
<d:status>HTTP/1.1 200 OK</d:status>
|
|
</d:propstat>
|
|
</d:response>
|
|
<d:response>
|
|
<d:href>/dav/files/alice/Boards/Family%20Board.whiteboard</d:href>
|
|
<d:propstat>
|
|
<d:prop>
|
|
<d:displayname>Family Board.whiteboard</d:displayname>
|
|
<d:resourcetype/>
|
|
</d:prop>
|
|
<d:status>HTTP/1.1 200 OK</d:status>
|
|
</d:propstat>
|
|
</d:response>
|
|
<d:response>
|
|
<d:href>/dav/files/alice/Boards/Archive/</d:href>
|
|
<d:propstat>
|
|
<d:prop>
|
|
<d:displayname>Archive</d:displayname>
|
|
<d:resourcetype><d:collection/></d:resourcetype>
|
|
</d:prop>
|
|
<d:status>HTTP/1.1 200 OK</d:status>
|
|
</d:propstat>
|
|
</d:response>
|
|
</d:multistatus>
|
|
"""
|
|
|
|
_AUTH_HEADER = "Basic " + base64.b64encode(f"{_USERNAME}:{_PASSWORD}".encode()).decode()
|
|
|
|
|
|
class _Handler(BaseHTTPRequestHandler):
|
|
def _authed(self) -> bool:
|
|
return self.headers.get("Authorization", "") == _AUTH_HEADER
|
|
|
|
def do_GET(self):
|
|
if self.path == "/dav/files/alice/Boards/Family%20Board.whiteboard":
|
|
if not self._authed():
|
|
self.send_response(401)
|
|
self.end_headers()
|
|
return
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(_FILE_BYTES)
|
|
return
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
|
|
def do_PROPFIND(self):
|
|
if not self._authed():
|
|
self.send_response(401)
|
|
self.end_headers()
|
|
return
|
|
self.send_response(207)
|
|
self.send_header("Content-Type", "application/xml")
|
|
self.end_headers()
|
|
self.wfile.write(_PROPFIND_RESPONSE)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def webdav_server():
|
|
server = HTTPServer(("127.0.0.1", 0), _Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
port = server.server_address[1]
|
|
try:
|
|
yield f"http://127.0.0.1:{port}"
|
|
finally:
|
|
server.shutdown()
|
|
thread.join()
|
|
|
|
|
|
def test_fetch_file_success(webdav_server):
|
|
data = fetch_file(f"{webdav_server}/dav/files/alice/Boards/Family%20Board.whiteboard", _USERNAME, _PASSWORD)
|
|
assert data == _FILE_BYTES
|
|
|
|
|
|
def test_fetch_file_wrong_password_raises(webdav_server):
|
|
with pytest.raises(WebDavError):
|
|
fetch_file(f"{webdav_server}/dav/files/alice/Boards/Family%20Board.whiteboard", _USERNAME, "wrong")
|
|
|
|
|
|
def test_fetch_file_missing_raises(webdav_server):
|
|
with pytest.raises(WebDavError):
|
|
fetch_file(f"{webdav_server}/dav/files/alice/Boards/nope.whiteboard", _USERNAME, _PASSWORD)
|
|
|
|
|
|
def test_list_directory_excludes_self_and_sorts_dirs_first(webdav_server):
|
|
entries = list_directory(f"{webdav_server}/dav/files/alice/Boards/", _USERNAME, _PASSWORD)
|
|
assert [e["name"] for e in entries] == ["Archive", "Family Board.whiteboard"]
|
|
assert entries[0]["is_dir"] is True
|
|
assert entries[1]["is_dir"] is False
|
|
|
|
|
|
def test_list_directory_wrong_password_raises(webdav_server):
|
|
with pytest.raises(WebDavError):
|
|
list_directory(f"{webdav_server}/dav/files/alice/Boards/", _USERNAME, "wrong")
|
|
|
|
|
|
def test_list_directory_urls_are_absolute(webdav_server):
|
|
entries = list_directory(f"{webdav_server}/dav/files/alice/Boards/", _USERNAME, _PASSWORD)
|
|
archive = next(e for e in entries if e["name"] == "Archive")
|
|
assert archive["url"] == f"{webdav_server}/dav/files/alice/Boards/Archive/"
|
|
|
|
|
|
def test_parent_directory_url_stops_at_base():
|
|
base = "https://cloud.example.com/dav/files/alice/"
|
|
assert parent_directory_url(base, base) is None
|
|
assert parent_directory_url(base, base + "Boards/") == base
|
|
assert parent_directory_url(base, base + "Boards/Family/") == base + "Boards/"
|