Add backend monitor and notification tests
This commit is contained in:
@@ -0,0 +1,51 @@
|
||||
from collections.abc import Generator
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.auth.dependencies import get_current_user
|
||||
from app.db.session import get_db
|
||||
from app.main import app
|
||||
from app.models import Base, User
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_session() -> Generator[Session, None, None]:
|
||||
engine = create_engine(
|
||||
"sqlite://",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
testing_session_local = sessionmaker(bind=engine, autoflush=False, autocommit=False)
|
||||
|
||||
with testing_session_local() as session:
|
||||
yield session
|
||||
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
engine.dispose()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(db_session: Session) -> Generator[TestClient, None, None]:
|
||||
def override_get_db() -> Generator[Session, None, None]:
|
||||
yield db_session
|
||||
|
||||
def override_current_user() -> User:
|
||||
return User(
|
||||
id=1,
|
||||
email="owner@example.com",
|
||||
display_name="Test Owner",
|
||||
hashed_password="not-used",
|
||||
role="owner",
|
||||
is_active=True,
|
||||
)
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
app.dependency_overrides[get_current_user] = override_current_user
|
||||
with TestClient(app) as test_client:
|
||||
yield test_client
|
||||
app.dependency_overrides.clear()
|
||||
@@ -0,0 +1,74 @@
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models import AlertRule, Asset, Monitor
|
||||
|
||||
|
||||
def test_create_website_monitor_creates_asset_and_alert_rule(client: TestClient, db_session: Session) -> None:
|
||||
response = client.post(
|
||||
"/monitors/website",
|
||||
json={
|
||||
"name": "Example Site",
|
||||
"url": "https://example.com",
|
||||
"expected_status": 200,
|
||||
"expected_text": "Example Domain",
|
||||
"unexpected_text": None,
|
||||
"timeout_seconds": 7,
|
||||
"check_tls_expiry": True,
|
||||
"tls_warning_days": 45,
|
||||
"interval_seconds": 60,
|
||||
"create_asset": True,
|
||||
"alert_enabled": True,
|
||||
"alert_severity": "critical",
|
||||
"failure_threshold": 2,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["name"] == "Example Site"
|
||||
assert body["monitor_type"] == "http"
|
||||
assert body["target"] == "https://example.com"
|
||||
assert body["config"]["check_tls_expiry"] is True
|
||||
assert body["config"]["tls_warning_days"] == 45
|
||||
|
||||
monitor = db_session.get(Monitor, body["id"])
|
||||
assert monitor is not None
|
||||
assert monitor.asset_id is not None
|
||||
|
||||
asset = db_session.get(Asset, monitor.asset_id)
|
||||
assert asset is not None
|
||||
assert asset.name == "Example Site"
|
||||
assert asset.asset_type == "website"
|
||||
assert asset.address == "https://example.com"
|
||||
|
||||
rule = db_session.scalar(select(AlertRule).where(AlertRule.monitor_id == monitor.id))
|
||||
assert rule is not None
|
||||
assert rule.name == "Example Site website failure"
|
||||
assert rule.severity == "critical"
|
||||
assert rule.condition == {"type": "status_not_up"}
|
||||
assert rule.failure_threshold == 2
|
||||
assert rule.cooldown_seconds == 300
|
||||
assert rule.is_enabled is True
|
||||
|
||||
|
||||
def test_create_website_monitor_can_skip_default_alert_rule(client: TestClient, db_session: Session) -> None:
|
||||
response = client.post(
|
||||
"/monitors/website",
|
||||
json={
|
||||
"name": "Status Only",
|
||||
"url": "https://status.example.com",
|
||||
"interval_seconds": 120,
|
||||
"create_asset": False,
|
||||
"alert_enabled": False,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
|
||||
monitor = db_session.get(Monitor, body["id"])
|
||||
assert monitor is not None
|
||||
assert monitor.asset_id is None
|
||||
assert db_session.scalars(select(AlertRule).where(AlertRule.monitor_id == monitor.id)).all() == []
|
||||
@@ -0,0 +1,74 @@
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.secrets import decrypt_secret
|
||||
from app.models import NotificationChannel
|
||||
|
||||
|
||||
def test_notification_channel_does_not_return_saved_secret(client: TestClient, db_session: Session) -> None:
|
||||
response = client.post(
|
||||
"/notifications/channels",
|
||||
json={
|
||||
"name": "Operations Webhook",
|
||||
"channel_type": "generic_webhook",
|
||||
"settings": {"username": "OrbitalWard"},
|
||||
"secret": "https://hooks.example.test/orbitalward",
|
||||
"is_enabled": True,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["has_secret"] is True
|
||||
assert "secret" not in body
|
||||
assert "encrypted_secret" not in body
|
||||
|
||||
channel = db_session.get(NotificationChannel, body["id"])
|
||||
assert channel is not None
|
||||
assert channel.encrypted_secret != "https://hooks.example.test/orbitalward"
|
||||
assert decrypt_secret(channel.encrypted_secret) == "https://hooks.example.test/orbitalward"
|
||||
|
||||
list_response = client.get("/notifications/channels")
|
||||
assert list_response.status_code == 200
|
||||
listed_channel = list_response.json()[0]
|
||||
assert listed_channel["has_secret"] is True
|
||||
assert "secret" not in listed_channel
|
||||
assert "encrypted_secret" not in listed_channel
|
||||
|
||||
|
||||
def test_notification_channel_update_without_secret_preserves_existing_secret(client: TestClient, db_session: Session) -> None:
|
||||
create_response = client.post(
|
||||
"/notifications/channels",
|
||||
json={
|
||||
"name": "Mattermost",
|
||||
"channel_type": "mattermost",
|
||||
"settings": {"username": "OrbitalWard"},
|
||||
"secret": "https://hooks.example.test/mattermost",
|
||||
"is_enabled": True,
|
||||
},
|
||||
)
|
||||
channel_id = create_response.json()["id"]
|
||||
original_secret = db_session.get(NotificationChannel, channel_id).encrypted_secret
|
||||
|
||||
update_response = client.patch(
|
||||
f"/notifications/channels/{channel_id}",
|
||||
json={
|
||||
"name": "Mattermost Alerts",
|
||||
"settings": {"username": "OrbitalWard Alerts"},
|
||||
"is_enabled": False,
|
||||
},
|
||||
)
|
||||
|
||||
assert update_response.status_code == 200
|
||||
body = update_response.json()
|
||||
assert body["name"] == "Mattermost Alerts"
|
||||
assert body["settings"]["username"] == "OrbitalWard Alerts"
|
||||
assert body["is_enabled"] is False
|
||||
assert body["has_secret"] is True
|
||||
assert "secret" not in body
|
||||
assert "encrypted_secret" not in body
|
||||
|
||||
channel = db_session.get(NotificationChannel, channel_id)
|
||||
assert channel is not None
|
||||
assert channel.encrypted_secret == original_secret
|
||||
assert decrypt_secret(channel.encrypted_secret) == "https://hooks.example.test/mattermost"
|
||||
+8
-2
@@ -50,6 +50,12 @@ Implemented alerting management slice:
|
||||
- Existing simple alert conditions are shown in friendly language instead of raw condition data.
|
||||
- Worker honors alert rule cooldown before opening a new incident for a recently-triggered rule.
|
||||
|
||||
Implemented backend test coverage:
|
||||
|
||||
- Test fixtures isolate API tests with an in-memory database and authenticated owner override.
|
||||
- Website monitor tests cover asset creation, default alert rule creation, TLS config persistence, and disabled default alerts.
|
||||
- Notification channel tests verify saved webhook URLs are encrypted and are not returned by create, list, or update responses.
|
||||
|
||||
## Known Gaps
|
||||
|
||||
- Credential vault UI and real credential encryption workflows are not complete.
|
||||
@@ -63,7 +69,7 @@ Implemented alerting management slice:
|
||||
- Email/SMTP notifications are not implemented yet.
|
||||
- Graphing exists only as placeholders; metric visualization is not implemented.
|
||||
- Worker scheduling is simple polling, not a Redis queue yet.
|
||||
- Tests are still minimal and need meaningful backend/worker/frontend coverage.
|
||||
- Tests still need worker notification delivery, alert evaluation, and frontend coverage.
|
||||
- Production deployment hardening is not done.
|
||||
|
||||
## Recommended Next Work
|
||||
@@ -78,7 +84,7 @@ Implemented alerting management slice:
|
||||
8. Add user administration UI.
|
||||
9. Add graphs for website response time and monitor status history.
|
||||
10. Add richer alert condition editing.
|
||||
11. Add backend and worker tests for the website-monitor and notification flows.
|
||||
11. Add worker tests for alert evaluation and notification delivery.
|
||||
|
||||
## Operational Notes
|
||||
|
||||
|
||||
Reference in New Issue
Block a user