453 lines
17 KiB
Python
453 lines
17 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", "Storniert"]
|
|
assert set(board["stacks"].keys()) == {"Offen", "In Arbeit", "Erledigt", "Storniert"}
|
|
|
|
|
|
@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": []},
|
|
{"id": 4, "title": "Storniert", "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, "Storniert": 4}}
|
|
assert mock_req.call_count == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_deck_board_creates_missing_stack_on_existing_board():
|
|
"""Regression: a board created before 'Storniert' existed only has the
|
|
original three stacks - the fourth must be added, not just assumed."""
|
|
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": []},
|
|
]
|
|
if (method, path) == ("POST", "/boards/10/stacks"):
|
|
assert json_body["title"] == "Storniert"
|
|
return {"id": 4, "title": "Storniert"}
|
|
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["stacks"] == {"Offen": 1, "In Arbeit": 2, "Erledigt": 3, "Storniert": 4}
|
|
|
|
|
|
@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, "Storniert": 4}
|
|
|
|
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
|
|
|
|
|
|
_BOARD = {"board_id": 10, "stacks": {"Offen": 1, "In Arbeit": 2, "Erledigt": 3}}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_order_creates_card_in_offen_stack():
|
|
with patch.object(main, "_ensure_deck_board", new=AsyncMock(return_value=_BOARD)), patch.object(
|
|
main, "_deck_request", new=AsyncMock(return_value={"id": 55})
|
|
) as mock_req:
|
|
order = await main.create_order("Feuerwehr Onza", "10x Handschuhe Groesse L")
|
|
|
|
assert order == {
|
|
"id": 55,
|
|
"customer": "Feuerwehr Onza",
|
|
"description": "10x Handschuhe Groesse L",
|
|
"status": "Offen",
|
|
"due_date": None,
|
|
}
|
|
method, path, body = mock_req.call_args[0]
|
|
assert method == "POST"
|
|
assert path == "/boards/10/stacks/1/cards"
|
|
assert body["title"] == "Feuerwehr Onza"
|
|
assert body["description"] == "10x Handschuhe Groesse L"
|
|
assert "duedate" not in body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_order_localizes_due_date():
|
|
with patch.object(main, "_ensure_deck_board", new=AsyncMock(return_value=_BOARD)), patch.object(
|
|
main, "_deck_request", new=AsyncMock(return_value={"id": 55})
|
|
) as mock_req:
|
|
await main.create_order("Feuerwehr Onza", "Helme", due_date="2026-09-20")
|
|
|
|
_, _, body = mock_req.call_args[0]
|
|
assert body["duedate"] == datetime(2026, 9, 20, tzinfo=main.CALENDAR_TIMEZONE).isoformat()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_orders_flattens_cards_across_stacks():
|
|
stacks_response = [
|
|
{"id": 1, "title": "Offen", "cards": [
|
|
{"id": 55, "title": "Feuerwehr Onza", "description": "Helme", "duedate": None}
|
|
]},
|
|
{"id": 2, "title": "In Arbeit", "cards": []},
|
|
{"id": 3, "title": "Erledigt", "cards": [
|
|
{"id": 56, "title": "Musterfirma", "description": "Stiefel", "duedate": "2026-09-20T00:00:00+02:00"}
|
|
]},
|
|
]
|
|
with patch.object(main, "_ensure_deck_board", new=AsyncMock(return_value=_BOARD)), patch.object(
|
|
main, "_deck_request", new=AsyncMock(return_value=stacks_response)
|
|
):
|
|
orders = await main.list_orders()
|
|
|
|
assert len(orders) == 2
|
|
assert orders[0] == {
|
|
"id": 55, "customer": "Feuerwehr Onza", "description": "Helme", "status": "Offen", "due_date": None
|
|
}
|
|
assert orders[1]["status"] == "Erledigt"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_orders_filters_by_status():
|
|
stacks_response = [
|
|
{"id": 1, "title": "Offen", "cards": [{"id": 55, "title": "A", "description": "", "duedate": None}]},
|
|
{"id": 2, "title": "In Arbeit", "cards": []},
|
|
{"id": 3, "title": "Erledigt", "cards": [{"id": 56, "title": "B", "description": "", "duedate": None}]},
|
|
]
|
|
with patch.object(main, "_ensure_deck_board", new=AsyncMock(return_value=_BOARD)), patch.object(
|
|
main, "_deck_request", new=AsyncMock(return_value=stacks_response)
|
|
):
|
|
orders = await main.list_orders(status="Erledigt")
|
|
|
|
assert [o["id"] for o in orders] == [56]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_order_status_finds_current_stack_and_reorders():
|
|
stacks_response = [
|
|
{"id": 1, "title": "Offen", "cards": [{"id": 55, "title": "A"}]},
|
|
{"id": 2, "title": "In Arbeit", "cards": []},
|
|
{"id": 3, "title": "Erledigt", "cards": []},
|
|
]
|
|
|
|
async def fake_deck_request(method, path, json_body=None):
|
|
if (method, path) == ("GET", "/boards/10/stacks"):
|
|
return stacks_response
|
|
if method == "PUT":
|
|
return {"id": 55}
|
|
raise AssertionError(f"unexpected call {method} {path}")
|
|
|
|
with patch.object(main, "_ensure_deck_board", new=AsyncMock(return_value=_BOARD)), patch.object(
|
|
main, "_deck_request", new=AsyncMock(side_effect=fake_deck_request)
|
|
) as mock_req:
|
|
result = await main.update_order_status(55, "In Arbeit")
|
|
|
|
assert result == {"id": 55, "status": "In Arbeit"}
|
|
put_call = [c for c in mock_req.call_args_list if c.args[0] == "PUT"][0]
|
|
_, path, body = put_call.args
|
|
assert path == "/boards/10/stacks/1/cards/55/reorder"
|
|
assert body == {"stackId": 2, "order": 999}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_order_status_raises_for_unknown_status():
|
|
with patch.object(main, "_ensure_deck_board", new=AsyncMock(return_value=_BOARD)):
|
|
with pytest.raises(ValueError):
|
|
await main.update_order_status(55, "Storniert")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_order_status_raises_when_card_not_found():
|
|
stacks_response = [
|
|
{"id": 1, "title": "Offen", "cards": []},
|
|
{"id": 2, "title": "In Arbeit", "cards": []},
|
|
{"id": 3, "title": "Erledigt", "cards": []},
|
|
]
|
|
with patch.object(main, "_ensure_deck_board", new=AsyncMock(return_value=_BOARD)), patch.object(
|
|
main, "_deck_request", new=AsyncMock(return_value=stacks_response)
|
|
):
|
|
with pytest.raises(ValueError):
|
|
await main.update_order_status(999, "Erledigt")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_tool_dispatches_order_tools():
|
|
with patch.object(main, "create_order", new=AsyncMock(return_value={"id": 1, "customer": "X"})):
|
|
result = await main.execute_tool(
|
|
"create_order", {"customer": "X", "description": "Y"}
|
|
)
|
|
assert json.loads(result) == {"id": 1, "customer": "X"}
|
|
|
|
with patch.object(main, "list_orders", new=AsyncMock(return_value=[{"id": 1}])):
|
|
result = await main.execute_tool("list_orders", {})
|
|
assert json.loads(result) == [{"id": 1}]
|
|
|
|
with patch.object(main, "update_order_status", new=AsyncMock(return_value={"id": 1, "status": "Erledigt"})):
|
|
result = await main.execute_tool("update_order_status", {"order_id": 1, "status": "Erledigt"})
|
|
assert json.loads(result) == {"id": 1, "status": "Erledigt"}
|
|
|
|
with patch.object(main, "update_order", new=AsyncMock(return_value={"id": 1, "customer": "Neu"})) as mock_update:
|
|
result = await main.execute_tool("update_order", {"order_id": 1, "customer": "Neu"})
|
|
assert json.loads(result) == {"id": 1, "customer": "Neu"}
|
|
mock_update.assert_called_once_with(1, "Neu", None, None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_chat_completion_includes_order_tools_and_instructions():
|
|
completion = MagicMock()
|
|
completion.stop_reason = "end_turn"
|
|
text_block = MagicMock()
|
|
text_block.type = "text"
|
|
text_block.text = "Bestellung angelegt."
|
|
completion.content = [text_block]
|
|
usage = MagicMock()
|
|
usage.input_tokens = 10
|
|
usage.output_tokens = 5
|
|
completion.usage = usage
|
|
|
|
main.claude_client = MagicMock()
|
|
main.claude_client.messages.create.return_value = completion
|
|
|
|
with patch.object(main, "build_memory_context", new=AsyncMock(return_value="")):
|
|
await main.run_chat_completion(
|
|
[{"role": "user", "content": "Neue Bestellung fuer Feuerwehr Onza"}], conversation_id=1
|
|
)
|
|
|
|
_, kwargs = main.claude_client.messages.create.call_args
|
|
assert kwargs["tools"] == main.CALENDAR_TOOLS + main.MEMORY_TOOLS + main.ORDER_TOOLS
|
|
assert "create_order" in kwargs["system"] or "Bestellungen" in kwargs["system"]
|
|
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
main.API_KEY_ADMIN = "test-secret"
|
|
HEADERS = {"X-Admin-Key": "test-secret"}
|
|
|
|
|
|
@pytest.fixture()
|
|
def client():
|
|
with TestClient(main.app) as c:
|
|
yield c
|
|
|
|
|
|
def test_orders_endpoint_returns_list(client):
|
|
orders = [{"id": 55, "customer": "A", "description": "", "status": "Offen", "due_date": None}]
|
|
with patch.object(main, "list_orders", new=AsyncMock(return_value=orders)) as mock_list:
|
|
response = client.get("/api/v1/orders", headers=HEADERS)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"orders": orders}
|
|
mock_list.assert_called_once_with(None)
|
|
|
|
|
|
def test_orders_endpoint_filters_by_status_query_param(client):
|
|
with patch.object(main, "list_orders", new=AsyncMock(return_value=[])) as mock_list:
|
|
response = client.get("/api/v1/orders?status=Erledigt", headers=HEADERS)
|
|
assert response.status_code == 200
|
|
mock_list.assert_called_once_with("Erledigt")
|
|
|
|
|
|
def test_orders_endpoint_requires_admin_key(client):
|
|
response = client.get("/api/v1/orders")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_orders_endpoint_returns_503_on_deck_error(client):
|
|
with patch.object(main, "list_orders", new=AsyncMock(side_effect=RuntimeError("Deck down"))):
|
|
response = client.get("/api/v1/orders", headers=HEADERS)
|
|
assert response.status_code == 503
|
|
|
|
|
|
_STACKS_WITH_CARD_55 = [
|
|
{"id": 1, "title": "Offen", "cards": [
|
|
{"id": 55, "title": "Feuerwehr Onza", "description": "Helme", "duedate": None, "type": "plain", "order": 999}
|
|
]},
|
|
{"id": 2, "title": "In Arbeit", "cards": []},
|
|
{"id": 3, "title": "Erledigt", "cards": []},
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_order_merges_description_and_keeps_other_fields():
|
|
with patch.object(main, "_ensure_deck_board", new=AsyncMock(return_value=_BOARD)), patch.object(
|
|
main, "_deck_request", new=AsyncMock(return_value=_STACKS_WITH_CARD_55)
|
|
) as mock_req:
|
|
order = await main.update_order(55, description="20x Handschuhe Groesse L")
|
|
|
|
assert order == {
|
|
"id": 55,
|
|
"customer": "Feuerwehr Onza",
|
|
"description": "20x Handschuhe Groesse L",
|
|
"status": "Offen",
|
|
"due_date": None,
|
|
}
|
|
put_calls = [c for c in mock_req.call_args_list if c.args[0] == "PUT"]
|
|
assert len(put_calls) == 1
|
|
_, path, body = put_calls[0].args
|
|
assert path == "/boards/10/stacks/1/cards/55"
|
|
assert body["title"] == "Feuerwehr Onza"
|
|
assert body["description"] == "20x Handschuhe Groesse L"
|
|
assert "duedate" not in body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_order_localizes_new_due_date():
|
|
with patch.object(main, "_ensure_deck_board", new=AsyncMock(return_value=_BOARD)), patch.object(
|
|
main, "_deck_request", new=AsyncMock(return_value=_STACKS_WITH_CARD_55)
|
|
) as mock_req:
|
|
order = await main.update_order(55, due_date="2026-10-01")
|
|
|
|
assert order["due_date"] == datetime(2026, 10, 1, tzinfo=main.CALENDAR_TIMEZONE).isoformat()
|
|
put_calls = [c for c in mock_req.call_args_list if c.args[0] == "PUT"]
|
|
_, _, body = put_calls[0].args
|
|
assert body["duedate"] == datetime(2026, 10, 1, tzinfo=main.CALENDAR_TIMEZONE).isoformat()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_order_keeps_existing_due_date_when_not_given():
|
|
stacks_with_due_date = [
|
|
{"id": 1, "title": "Offen", "cards": [
|
|
{"id": 55, "title": "A", "description": "B", "duedate": "2026-09-20T00:00:00+02:00", "type": "plain", "order": 999}
|
|
]},
|
|
{"id": 2, "title": "In Arbeit", "cards": []},
|
|
{"id": 3, "title": "Erledigt", "cards": []},
|
|
]
|
|
with patch.object(main, "_ensure_deck_board", new=AsyncMock(return_value=_BOARD)), patch.object(
|
|
main, "_deck_request", new=AsyncMock(return_value=stacks_with_due_date)
|
|
) as mock_req:
|
|
order = await main.update_order(55, customer="Neuer Name")
|
|
|
|
assert order["due_date"] == "2026-09-20T00:00:00+02:00"
|
|
put_calls = [c for c in mock_req.call_args_list if c.args[0] == "PUT"]
|
|
_, _, body = put_calls[0].args
|
|
assert body["duedate"] == "2026-09-20T00:00:00+02:00"
|
|
assert body["title"] == "Neuer Name"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_order_raises_when_card_not_found():
|
|
with patch.object(main, "_ensure_deck_board", new=AsyncMock(return_value=_BOARD)), patch.object(
|
|
main, "_deck_request", new=AsyncMock(return_value=_STACKS_WITH_CARD_55)
|
|
):
|
|
with pytest.raises(ValueError):
|
|
await main.update_order(999, description="X")
|