144 lines
4.6 KiB
Python
144 lines
4.6 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")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_deck_board_creates_board_and_stacks_when_missing():
|
|
titles_created = []
|
|
|
|
async def fake_deck_request(method, path, json_body=None):
|
|
if (method, path) == ("GET", "/boards"):
|
|
return []
|
|
if (method, path) == ("POST", "/boards"):
|
|
return {"id": 10, "title": "Bestellungen"}
|
|
if (method, path) == ("GET", "/boards/10/stacks"):
|
|
return []
|
|
if (method, path) == ("POST", "/boards/10/stacks"):
|
|
titles_created.append(json_body["title"])
|
|
return {"id": 100 + len(titles_created), "title": json_body["title"]}
|
|
raise AssertionError(f"unexpected call {method} {path}")
|
|
|
|
with patch.object(main, "_deck_request", new=AsyncMock(side_effect=fake_deck_request)):
|
|
board = await main._ensure_deck_board()
|
|
|
|
assert board["board_id"] == 10
|
|
assert titles_created == ["Offen", "In Arbeit", "Erledigt"]
|
|
assert set(board["stacks"].keys()) == {"Offen", "In Arbeit", "Erledigt"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_deck_board_reuses_existing_board_and_stacks():
|
|
async def fake_deck_request(method, path, json_body=None):
|
|
if (method, path) == ("GET", "/boards"):
|
|
return [{"id": 10, "title": "Bestellungen"}]
|
|
if (method, path) == ("GET", "/boards/10/stacks"):
|
|
return [
|
|
{"id": 1, "title": "Offen", "cards": []},
|
|
{"id": 2, "title": "In Arbeit", "cards": []},
|
|
{"id": 3, "title": "Erledigt", "cards": []},
|
|
]
|
|
raise AssertionError(f"unexpected call {method} {path}")
|
|
|
|
with patch.object(main, "_deck_request", new=AsyncMock(side_effect=fake_deck_request)) as mock_req:
|
|
board = await main._ensure_deck_board()
|
|
|
|
assert board == {"board_id": 10, "stacks": {"Offen": 1, "In Arbeit": 2, "Erledigt": 3}}
|
|
assert mock_req.call_count == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_deck_board_caches_after_first_lookup():
|
|
main._deck_cache["board_id"] = 10
|
|
main._deck_cache["stacks"] = {"Offen": 1, "In Arbeit": 2, "Erledigt": 3}
|
|
|
|
with patch.object(main, "_deck_request", new=AsyncMock()) as mock_req:
|
|
board = await main._ensure_deck_board()
|
|
|
|
mock_req.assert_not_called()
|
|
assert board["board_id"] == 10
|