49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
import os
|
|
import sys
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import main
|
|
|
|
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_weather_returns_503_when_no_data(client):
|
|
with patch.object(main, "pg_pool", "not-none"), patch.object(
|
|
main, "get_latest_weather", new=AsyncMock(return_value=None)
|
|
):
|
|
response = client.get("/api/v1/weather", headers=HEADERS)
|
|
assert response.status_code == 503
|
|
|
|
|
|
def test_weather_returns_latest_row(client):
|
|
row = {
|
|
"location": "Crailsheim",
|
|
"temperature_c": 18.4,
|
|
"condition_text": "Bewoelkt",
|
|
"fetched_at": datetime(2026, 9, 12, 16, 0, 3, 123456),
|
|
}
|
|
with patch.object(main, "pg_pool", "not-none"), patch.object(
|
|
main, "get_latest_weather", new=AsyncMock(return_value=row)
|
|
):
|
|
response = client.get("/api/v1/weather", headers=HEADERS)
|
|
assert response.status_code == 200
|
|
assert response.json()["location"] == "Crailsheim"
|
|
|
|
|
|
def test_weather_requires_admin_key(client):
|
|
response = client.get("/api/v1/weather")
|
|
assert response.status_code == 401
|