103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
import time
|
|
|
|
from app.models import KIND_PROBLEM, Event
|
|
from tests.test_web import (ADMIN, VIEWER, admin, app, client, csrf_from, login_ok, post, # noqa: F401
|
|
users, viewer)
|
|
|
|
TOKEN = "123456:SECRET-token_abc"
|
|
|
|
|
|
def telegram_form(**overrides):
|
|
data = {
|
|
"telegram_enabled": "1",
|
|
"telegram_token": TOKEN,
|
|
"telegram_chat_id": "-100123",
|
|
"telegram_allowlist": "111, 222",
|
|
"telegram_api_base": "",
|
|
}
|
|
data.update(overrides)
|
|
return data
|
|
|
|
|
|
def add_alert(app, **kwargs):
|
|
event = Event(ts=int(time.time()), host="web01", service="Disk <b>", kind=KIND_PROBLEM, state=2,
|
|
output="voll <script>x</script>", **kwargs)
|
|
app.state.runtime.alerts.insert_alerts([event], "g1")
|
|
|
|
|
|
def test_alerts_page_requires_login(client):
|
|
assert client.get("/alerts").status_code == 303
|
|
|
|
|
|
def test_alerts_page_reachable_for_viewer_and_escapes(client, app, viewer):
|
|
add_alert(app, suppressed=True, suppress_reason="downtime")
|
|
login_ok(client, VIEWER)
|
|
response = client.get("/alerts")
|
|
assert response.status_code == 200
|
|
assert "web01" in response.text
|
|
assert "unterdrückt (Downtime)" in response.text
|
|
assert "<script>x</script>" not in response.text
|
|
assert "Verlauf" in response.text
|
|
|
|
|
|
def test_dashboard_shows_counts_and_open_problems(client, app, viewer):
|
|
add_alert(app)
|
|
app.state.db.execute(
|
|
"INSERT INTO service_states(host, service, state, output, last_state_change, in_downtime) "
|
|
"VALUES ('web01', 'Disk', 2, 'voll', 1000, 1)")
|
|
login_ok(client, VIEWER)
|
|
response = client.get("/")
|
|
assert "1 Alarme in 24 h" in response.text
|
|
assert "web01" in response.text and "Downtime" in response.text
|
|
|
|
|
|
def test_telegram_settings_saved_with_encrypted_token(client, app, admin):
|
|
login_ok(client)
|
|
response = post(client, "/settings/telegram", "/settings/telegram", **telegram_form())
|
|
assert response.status_code == 303
|
|
store = app.state.store
|
|
assert store.get("telegram.token") == TOKEN
|
|
raw = app.state.db.query("SELECT value, is_secret FROM settings WHERE key='telegram.token'")[0]
|
|
assert raw["is_secret"] == 1 and TOKEN not in raw["value"]
|
|
assert store.get("telegram.chat_id") == "-100123"
|
|
page = client.get("/settings/telegram")
|
|
assert TOKEN not in page.text
|
|
assert TOKEN not in str(app.state.db.query("SELECT old_value, new_value FROM audit"))
|
|
|
|
|
|
def test_telegram_invalid_values_rejected(client, app, admin):
|
|
login_ok(client)
|
|
for override in ({"telegram_chat_id": "abc"}, {"telegram_allowlist": "12,ab"},
|
|
{"telegram_token": "kein-token"}, {"telegram_api_base": "ftp://x"}):
|
|
response = post(client, "/settings/telegram", "/settings/telegram", **telegram_form(**override))
|
|
assert response.status_code == 422
|
|
assert app.state.store.get("telegram.chat_id") is None
|
|
|
|
|
|
def test_telegram_enabled_requires_token_and_chat_id(client, admin):
|
|
login_ok(client)
|
|
response = post(client, "/settings/telegram", "/settings/telegram",
|
|
**telegram_form(telegram_token="", telegram_chat_id=""))
|
|
assert response.status_code == 422
|
|
assert "Bot-Token fehlt" in response.text
|
|
|
|
|
|
def test_telegram_optional_for_setup(client, admin):
|
|
login_ok(client)
|
|
page = client.get("/setup")
|
|
assert page.status_code == 200 and "Telegram" in page.text and "optional" in page.text
|
|
|
|
|
|
def test_telegram_test_message_requires_admin(client, viewer):
|
|
login_ok(client, VIEWER)
|
|
token = csrf_from(client, "/")
|
|
response = client.post("/settings/telegram/message", data={"csrf": token})
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_telegram_test_message_without_config_reports_error(client, admin):
|
|
login_ok(client)
|
|
response = post(client, "/settings/telegram/message", "/settings/telegram",
|
|
telegram_enabled="1", telegram_chat_id="42")
|
|
assert response.status_code == 200 and response.json()["ok"] is False
|