206 lines
8.1 KiB
Python
206 lines
8.1 KiB
Python
"""
|
||
Tests for BATON-ARCH-013: Keep-alive mechanism / health endpoint.
|
||
|
||
Acceptance criteria:
|
||
1. GET /health returns HTTP 200 OK.
|
||
2. Response body contains JSON with {"status": "ok"}.
|
||
3. Endpoint does not require authorization (no token, no secret header needed).
|
||
4. Keep-alive loop is started when APP_URL is set, and NOT started when APP_URL is unset.
|
||
5. deploy/ contains valid systemd .service and .timer config files.
|
||
6. README documents both hosting scenarios and keep-alive instructions.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from pathlib import Path
|
||
|
||
os.environ.setdefault("BOT_TOKEN", "test-bot-token")
|
||
os.environ.setdefault("CHAT_ID", "-1001234567890")
|
||
os.environ.setdefault("WEBHOOK_SECRET", "test-webhook-secret")
|
||
os.environ.setdefault("WEBHOOK_URL", "https://example.com/api/webhook/telegram")
|
||
os.environ.setdefault("FRONTEND_ORIGIN", "http://localhost:3000")
|
||
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
import pytest
|
||
|
||
from tests.conftest import make_app_client, temp_db
|
||
|
||
PROJECT_ROOT = Path(__file__).parent.parent
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Criterion 1 & 2 & 3 — GET /health → 200 OK, {"status": "ok"}, no auth
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_health_returns_200_ok():
|
||
"""GET /health должен вернуть HTTP 200 без какого-либо заголовка авторизации."""
|
||
async with make_app_client() as client:
|
||
response = await client.get("/health")
|
||
|
||
assert response.status_code == 200
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_health_returns_status_ok():
|
||
"""GET /health должен вернуть JSON содержащий {"status": "ok"}."""
|
||
async with make_app_client() as client:
|
||
response = await client.get("/health")
|
||
|
||
data = response.json()
|
||
assert data.get("status") == "ok"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_health_returns_timestamp():
|
||
"""GET /health должен вернуть поле timestamp в JSON."""
|
||
async with make_app_client() as client:
|
||
response = await client.get("/health")
|
||
|
||
data = response.json()
|
||
assert "timestamp" in data
|
||
assert isinstance(data["timestamp"], int)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_health_no_auth_header_required():
|
||
"""GET /health без заголовков авторизации должен вернуть 200 (не 401/403)."""
|
||
async with make_app_client() as client:
|
||
response = await client.get("/health")
|
||
|
||
assert response.status_code not in (401, 403)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Criterion 4 — keep-alive task lifecycle
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_keepalive_started_when_app_url_set():
|
||
"""Keep-alive задача должна стартовать при наличии APP_URL."""
|
||
from backend.main import app
|
||
|
||
with temp_db():
|
||
with patch("backend.telegram.set_webhook", new_callable=AsyncMock):
|
||
with patch("backend.config.APP_URL", "https://example.com"):
|
||
with patch("backend.main._keep_alive_loop", new_callable=AsyncMock) as mock_loop:
|
||
async with app.router.lifespan_context(app):
|
||
pass
|
||
|
||
# asyncio.create_task вызывается с корутиной _keep_alive_loop — проверяем что она была вызвана
|
||
assert mock_loop.called
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_keepalive_not_started_when_app_url_unset():
|
||
"""Keep-alive задача НЕ должна стартовать при отсутствии APP_URL."""
|
||
from backend.main import app
|
||
|
||
with temp_db():
|
||
with patch("backend.telegram.set_webhook", new_callable=AsyncMock):
|
||
with patch("backend.config.APP_URL", None):
|
||
with patch("backend.main._keep_alive_loop", new_callable=AsyncMock) as mock_loop:
|
||
async with app.router.lifespan_context(app):
|
||
pass
|
||
|
||
assert not mock_loop.called
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_keepalive_called_with_app_url_value():
|
||
"""Keep-alive задача должна получить значение APP_URL в качестве аргумента."""
|
||
from backend.main import app
|
||
|
||
with temp_db():
|
||
with patch("backend.telegram.set_webhook", new_callable=AsyncMock):
|
||
with patch("backend.config.APP_URL", "https://my-app.fly.dev"):
|
||
with patch("backend.main._keep_alive_loop", new_callable=AsyncMock) as mock_loop:
|
||
async with app.router.lifespan_context(app):
|
||
pass
|
||
|
||
mock_loop.assert_called_once_with("https://my-app.fly.dev")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Criterion 5 — systemd config files in deploy/
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_keepalive_service_file_exists():
|
||
"""Файл deploy/baton-keepalive.service должен существовать."""
|
||
assert (PROJECT_ROOT / "deploy" / "baton-keepalive.service").exists()
|
||
|
||
|
||
def test_keepalive_timer_file_exists():
|
||
"""Файл deploy/baton-keepalive.timer должен существовать."""
|
||
assert (PROJECT_ROOT / "deploy" / "baton-keepalive.timer").exists()
|
||
|
||
|
||
def test_keepalive_service_has_oneshot_type():
|
||
"""baton-keepalive.service должен содержать Type=oneshot."""
|
||
content = (PROJECT_ROOT / "deploy" / "baton-keepalive.service").read_text()
|
||
assert "Type=oneshot" in content
|
||
|
||
|
||
def test_keepalive_service_pings_health():
|
||
"""baton-keepalive.service должен вызывать curl с /health endpoint."""
|
||
content = (PROJECT_ROOT / "deploy" / "baton-keepalive.service").read_text()
|
||
assert "/health" in content
|
||
assert "curl" in content
|
||
|
||
|
||
def test_keepalive_timer_has_unit_active_sec():
|
||
"""baton-keepalive.timer должен содержать OnUnitActiveSec."""
|
||
content = (PROJECT_ROOT / "deploy" / "baton-keepalive.timer").read_text()
|
||
assert "OnUnitActiveSec" in content
|
||
|
||
|
||
def test_keepalive_timer_has_install_section():
|
||
"""baton-keepalive.timer должен содержать секцию [Install] с WantedBy=timers.target."""
|
||
content = (PROJECT_ROOT / "deploy" / "baton-keepalive.timer").read_text()
|
||
assert "[Install]" in content
|
||
assert "timers.target" in content
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Criterion 6 — README documents hosting scenarios and keep-alive
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_readme_documents_selfhosting_scenario():
|
||
"""README должен описывать вариант self-hosting (VPS)."""
|
||
content = (PROJECT_ROOT / "README.md").read_text()
|
||
assert "самохост" in content.lower() or "vps" in content.lower() or "Self" in content
|
||
|
||
|
||
def test_readme_documents_fly_io_scenario():
|
||
"""README должен описывать вариант хостинга Fly.io."""
|
||
content = (PROJECT_ROOT / "README.md").read_text()
|
||
assert "fly.io" in content.lower()
|
||
|
||
|
||
def test_readme_documents_cron_keepalive():
|
||
"""README должен содержать инструкцию по настройке cron для keep-alive."""
|
||
content = (PROJECT_ROOT / "README.md").read_text()
|
||
assert "cron" in content.lower() or "crontab" in content.lower()
|
||
|
||
|
||
def test_readme_documents_systemd_keepalive():
|
||
"""README должен содержать инструкцию по настройке systemd timer для keep-alive."""
|
||
content = (PROJECT_ROOT / "README.md").read_text()
|
||
assert "systemd" in content.lower()
|
||
|
||
|
||
def test_readme_documents_uptimerobot():
|
||
"""README должен содержать секцию UptimeRobot как внешний watchdog."""
|
||
content = (PROJECT_ROOT / "README.md").read_text()
|
||
assert "uptimerobot" in content.lower()
|
||
|
||
|
||
def test_readme_documents_app_url_env_var():
|
||
"""README должен упоминать переменную APP_URL для keep-alive."""
|
||
content = (PROJECT_ROOT / "README.md").read_text()
|
||
assert "APP_URL" in content
|