139 lines
4.7 KiB
Python
139 lines
4.7 KiB
Python
import os
|
|
import sys
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_setting_returns_stored_value():
|
|
with patch.object(main, "db_query", new=AsyncMock(return_value={"value": "Custom GmbH"})) as mock_query:
|
|
value = await main.get_setting("company_name")
|
|
|
|
assert value == "Custom GmbH"
|
|
args, kwargs = mock_query.call_args
|
|
assert "SELECT value FROM settings" in args[0]
|
|
assert args[1] == ("company_name",)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_setting_falls_back_to_default_when_missing():
|
|
with patch.object(main, "db_query", new=AsyncMock(return_value=None)):
|
|
value = await main.get_setting("assistant_name")
|
|
|
|
assert value == "JARVIS"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_all_settings_merges_defaults_with_stored_overrides():
|
|
rows = [{"key": "assistant_name", "value": "FRIDAY"}]
|
|
with patch.object(main, "db_query", new=AsyncMock(return_value=rows)):
|
|
settings = await main.get_all_settings()
|
|
|
|
assert settings == {
|
|
"assistant_name": "FRIDAY",
|
|
"company_name": "MBO-Tech-IT",
|
|
"contact_email": "kontakt@mbo-tech-it.de",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_all_settings_returns_defaults_when_table_empty():
|
|
with patch.object(main, "db_query", new=AsyncMock(return_value=[])):
|
|
settings = await main.get_all_settings()
|
|
|
|
assert settings == main.SETTINGS_DEFAULTS
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_setting_upserts_value():
|
|
with patch.object(main, "db_query", new=AsyncMock(return_value=None)) as mock_query:
|
|
await main.set_setting("company_name", "Neue Firma GmbH")
|
|
|
|
args, kwargs = mock_query.call_args
|
|
assert "INSERT INTO settings" in args[0]
|
|
assert "ON CONFLICT (key) DO UPDATE" in args[0]
|
|
assert args[1] == ("company_name", "Neue Firma GmbH")
|
|
|
|
|
|
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_get_settings_endpoint_returns_merged_settings(client):
|
|
settings = {"assistant_name": "JARVIS", "company_name": "MBO-Tech-IT", "contact_email": "kontakt@mbo-tech-it.de"}
|
|
with patch.object(main, "get_all_settings", new=AsyncMock(return_value=settings)):
|
|
response = client.get("/api/v1/settings", headers=HEADERS)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == settings
|
|
|
|
|
|
def test_get_settings_endpoint_requires_admin_key(client):
|
|
response = client.get("/api/v1/settings")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_put_settings_endpoint_updates_and_returns_merged_settings(client):
|
|
updated = {"assistant_name": "FRIDAY", "company_name": "MBO-Tech-IT", "contact_email": "kontakt@mbo-tech-it.de"}
|
|
with patch.object(main, "set_setting", new=AsyncMock()) as mock_set, patch.object(
|
|
main, "get_all_settings", new=AsyncMock(return_value=updated)
|
|
):
|
|
response = client.put("/api/v1/settings", headers=HEADERS, json={"assistant_name": "FRIDAY"})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == updated
|
|
mock_set.assert_called_once_with("assistant_name", "FRIDAY")
|
|
|
|
|
|
def test_put_settings_endpoint_rejects_unknown_key(client):
|
|
with patch.object(main, "set_setting", new=AsyncMock()) as mock_set:
|
|
response = client.put("/api/v1/settings", headers=HEADERS, json={"nonsense_key": "x"})
|
|
|
|
assert response.status_code == 400
|
|
mock_set.assert_not_called()
|
|
|
|
|
|
def test_put_settings_endpoint_requires_admin_key(client):
|
|
response = client.put("/api/v1/settings", json={"assistant_name": "X"})
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_identity_system_prompt_uses_settings():
|
|
settings = {"assistant_name": "FRIDAY", "company_name": "Stark Industries", "contact_email": "info@stark.example"}
|
|
with patch.object(main, "get_all_settings", new=AsyncMock(return_value=settings)):
|
|
prompt = await main._identity_system_prompt()
|
|
|
|
assert "FRIDAY" in prompt
|
|
assert "Stark Industries" in prompt
|
|
assert "info@stark.example" in prompt
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_identity_system_prompt_falls_back_to_defaults_on_db_error():
|
|
with patch.object(
|
|
main, "get_all_settings", new=AsyncMock(side_effect=RuntimeError("relation \"settings\" does not exist"))
|
|
):
|
|
prompt = await main._identity_system_prompt()
|
|
|
|
assert main.SETTINGS_DEFAULTS["assistant_name"] in prompt
|
|
assert main.SETTINGS_DEFAULTS["company_name"] in prompt
|
|
assert main.SETTINGS_DEFAULTS["contact_email"] in prompt
|