100 lines
3.6 KiB
Python
100 lines
3.6 KiB
Python
import sys
|
|
import os
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from fastapi.testclient import TestClient
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
import pytest
|
|
from main import app
|
|
|
|
# We need to mock the bot before importing or using the app fully,
|
|
# although patching it contextually is better.
|
|
# Because 'bot' is a global instance in main.py, we should patch it where it is used.
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
# Mock the bot instance methods including login/close for lifespan
|
|
with patch("main.bot") as mock_bot:
|
|
# Use AsyncMock for async methods
|
|
|
|
mock_bot.login = AsyncMock()
|
|
mock_bot.close = AsyncMock()
|
|
mock_bot.send_message = AsyncMock()
|
|
|
|
with TestClient(app) as c:
|
|
yield c
|
|
|
|
def test_health_check(client):
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "ok"}
|
|
|
|
def test_notify_success(client):
|
|
# We need to verify that bot.send_message was called.
|
|
|
|
with patch("main.bot.send_message", new_callable=AsyncMock) as mock_send:
|
|
# Updated payload
|
|
payload = {
|
|
"service_name": "TestService",
|
|
"content": "System is down"
|
|
}
|
|
response = client.post("/notify", json=payload)
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "queued"}
|
|
|
|
# Check formatted message
|
|
mock_send.assert_called_once()
|
|
args, kwargs = mock_send.call_args
|
|
# args[0] is plain_message, args[1] is html_message, args[2] is room_id
|
|
assert args[0] == "[TestService]\nSystem is down"
|
|
assert args[1] == "<b>[TestService]</b><br>System is down"
|
|
|
|
def test_jellyfin_movie(client):
|
|
with patch("main.bot.send_message", new_callable=AsyncMock) as mock_send:
|
|
payload = {
|
|
"notification_type": "ItemAdded",
|
|
"item_type": "Movie",
|
|
"name": "The Matrix",
|
|
"year": 1999,
|
|
"overview": "A computer hacker..."
|
|
}
|
|
response = client.post("/jellyfin", json=payload)
|
|
assert response.status_code == 200
|
|
|
|
mock_send.assert_called_once()
|
|
args, _ = mock_send.call_args
|
|
assert "[Jellyfin]\nNew Movie: The Matrix (1999)\nA computer hacker..." in args[0]
|
|
assert "<b>[Jellyfin]</b><br>New Movie: The Matrix (1999)<br>A computer hacker..." in args[1]
|
|
|
|
def test_jellyfin_episode(client):
|
|
with patch("main.bot.send_message", new_callable=AsyncMock) as mock_send:
|
|
payload = {
|
|
"notification_type": "ItemAdded",
|
|
"item_type": "Episode",
|
|
"name": "Pilot",
|
|
"series_name": "Lost",
|
|
"season": 1,
|
|
"episode": 1
|
|
}
|
|
response = client.post("/jellyfin", json=payload)
|
|
assert response.status_code == 200
|
|
|
|
mock_send.assert_called_once()
|
|
args, _ = mock_send.call_args
|
|
assert "[Jellyfin]\nNew Episode: Lost - S01E01 - Pilot" in args[0]
|
|
|
|
def test_notify_custom_room(client):
|
|
with patch("main.bot.send_message", new_callable=AsyncMock) as mock_send:
|
|
payload = {
|
|
"service_name": "TestService",
|
|
"content": "msg",
|
|
"room_id": "!custom:room"
|
|
}
|
|
response = client.post("/notify", json=payload)
|
|
assert response.status_code == 200
|
|
mock_send.assert_called_once_with("[TestService]\nmsg", "<b>[TestService]</b><br>msg", "!custom:room")
|
|
|
|
def test_notify_validation_error(client):
|
|
response = client.post("/notify", json={"service_name": "JustServiceNoContent"})
|
|
assert response.status_code == 422
|