88 lines
2.4 KiB
Python
88 lines
2.4 KiB
Python
import os
|
|
import sys
|
|
import json
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
# Mock psycopg2 before importing main (see test_memory.py for rationale).
|
|
sys.modules["psycopg2"] = MagicMock()
|
|
sys.modules["psycopg2.pool"] = MagicMock()
|
|
sys.modules["psycopg2.extras"] = MagicMock()
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
import pytest
|
|
|
|
import main
|
|
|
|
main.NEXTCLOUD_CALDAV_URL = "https://cloud.ffw-onza.de/remote.php/dav/calendars/jonny/ffw-onza-alle/"
|
|
main.NEXTCLOUD_USER = "jonny"
|
|
main.NEXTCLOUD_APP_PASSWORD = "test-app-password"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_deck_cache():
|
|
main._deck_cache.clear()
|
|
yield
|
|
main._deck_cache.clear()
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, status, payload):
|
|
self.status = status
|
|
self._payload = payload
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
async def json(self):
|
|
return self._payload
|
|
|
|
async def text(self):
|
|
return json.dumps(self._payload)
|
|
|
|
|
|
class _FakeSession:
|
|
def __init__(self, response):
|
|
self._response = response
|
|
self.request_calls = []
|
|
|
|
def request(self, method, url, **kwargs):
|
|
self.request_calls.append((method, url, kwargs))
|
|
return self._response
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
|
|
def test_deck_base_url_derives_host_from_caldav_url():
|
|
assert main._deck_base_url() == "https://cloud.ffw-onza.de/index.php/apps/deck/api/v1.0"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deck_request_returns_parsed_json_on_success():
|
|
fake_session = _FakeSession(_FakeResponse(200, {"id": 5}))
|
|
|
|
with patch.object(main.aiohttp, "ClientSession", return_value=fake_session):
|
|
result = await main._deck_request("GET", "/boards")
|
|
|
|
assert result == {"id": 5}
|
|
method, url, kwargs = fake_session.request_calls[0]
|
|
assert method == "GET"
|
|
assert url == "https://cloud.ffw-onza.de/index.php/apps/deck/api/v1.0/boards"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deck_request_raises_on_error_status():
|
|
fake_session = _FakeSession(_FakeResponse(500, {"message": "boom"}))
|
|
|
|
with patch.object(main.aiohttp, "ClientSession", return_value=fake_session):
|
|
with pytest.raises(RuntimeError):
|
|
await main._deck_request("GET", "/boards")
|