1083 lines
45 KiB
Python
1083 lines
45 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import inspect, text
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import SQLModel, Session, create_engine, select
|
|
|
|
|
|
@pytest.fixture
|
|
def api_client(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "notifications_test.db"
|
|
monkeypatch.setenv("APPOPS_DB_PATH", str(db_path))
|
|
monkeypatch.setenv("APP_OPS_VOICE_TOKEN", "test-voice-token")
|
|
|
|
from backend.app.db import session
|
|
from backend.app.db.init_db import init_db
|
|
from backend.app.main import app
|
|
|
|
engine = create_engine(f"sqlite:///{Path(db_path)}", connect_args={"check_same_thread": False})
|
|
session.engine = engine
|
|
SQLModel.metadata.drop_all(engine)
|
|
SQLModel.metadata.create_all(engine)
|
|
init_db()
|
|
|
|
with TestClient(app) as client:
|
|
yield client
|
|
|
|
|
|
def test_default_notification_rules_and_dashboard_channel_are_seeded(api_client):
|
|
rules = api_client.get('/api/notifications/rules').json()
|
|
channels = api_client.get('/api/notifications/channels').json()
|
|
|
|
rule_ids = {rule['id'] for rule in rules}
|
|
assert 'notify-critical-safety-finding' in rule_ids
|
|
assert 'notify-coverage-incomplete' in rule_ids
|
|
assert 'notify-connector-missing-config' in rule_ids
|
|
assert all(rule['cooldownSeconds'] > 0 for rule in rules)
|
|
assert any(channel['id'] == 'dashboard' and channel['enabled'] for channel in channels)
|
|
assert {'discord', 'telegram', 'email'}.issubset({channel['id'] for channel in channels})
|
|
|
|
serialized = str({'rules': rules, 'channels': channels})
|
|
assert 'TOKEN_VALUE' not in serialized
|
|
assert 'Bearer ' not in serialized
|
|
|
|
|
|
def test_external_notification_credentials_are_write_only_and_delivery_is_bounded(api_client, monkeypatch):
|
|
from backend.app.services.notification_channels import external
|
|
|
|
calls = []
|
|
monkeypatch.setattr(external, '_http_json', lambda url, payload, timeout: calls.append((url, payload, timeout)) or {'success': True, 'status': 'sent', 'httpStatus': 204})
|
|
|
|
updated = api_client.patch('/api/notifications/channels/discord', json={
|
|
'enabled': True,
|
|
'token': 'https://discord.com/api/webhooks/123/write-only-value',
|
|
'config': {'username': 'AppOps private', 'timeoutSeconds': 9, 'unexpected': 'discard-me'},
|
|
}).json()
|
|
tested = api_client.post('/api/notifications/channels/discord/test').json()
|
|
listed = next(item for item in api_client.get('/api/notifications/channels').json() if item['id'] == 'discord')
|
|
|
|
assert updated['hasSecret'] is True
|
|
assert updated['safeConfig']['username'] == 'AppOps private'
|
|
assert 'unexpected' not in updated['safeConfig']
|
|
assert tested['status'] == 'sent'
|
|
assert calls and calls[0][2] == 9
|
|
assert 'write-only-value' not in str(updated)
|
|
assert 'write-only-value' not in str(listed)
|
|
|
|
|
|
def test_external_delivery_retries_are_bounded_and_dead_lettered(api_client, monkeypatch):
|
|
from backend.app.db.models import NotificationChannel
|
|
from backend.app.db.session import engine
|
|
from backend.app.services.notification_channels import external
|
|
|
|
api_client.patch('/api/notifications/channels/discord', json={
|
|
'enabled': True,
|
|
'token': 'https://discord.com/api/webhooks/123/write-only-retry',
|
|
'config': {'maxAttempts': 3, 'retryBackoffSeconds': 0},
|
|
})
|
|
calls = []
|
|
monkeypatch.setattr(external, '_http_json', lambda *args: calls.append(args) or {'success': False, 'status': 'failed', 'reason': 'connection_failed'})
|
|
|
|
with Session(engine) as session:
|
|
channel = session.get(NotificationChannel, 'discord')
|
|
result = external.send_notification(session, channel, {'title': 'test', 'message': 'safe'})
|
|
|
|
assert len(calls) == 3
|
|
assert result['attempts'] == 3
|
|
assert result['deadLetter'] is True
|
|
|
|
|
|
def test_email_starttls_mode_is_supported(api_client, monkeypatch):
|
|
from backend.app.services.notification_channels import external
|
|
|
|
actions = []
|
|
|
|
class FakeSmtp:
|
|
def __init__(self, host, port, timeout):
|
|
actions.append(('connect', host, port, timeout))
|
|
def __enter__(self):
|
|
return self
|
|
def __exit__(self, *_):
|
|
return None
|
|
def ehlo(self):
|
|
actions.append(('ehlo',))
|
|
def starttls(self, context):
|
|
actions.append(('starttls', bool(context)))
|
|
def login(self, username, password):
|
|
actions.append(('login', username, bool(password)))
|
|
def send_message(self, message):
|
|
actions.append(('send', message['To']))
|
|
|
|
monkeypatch.setattr(external.smtplib, 'SMTP', FakeSmtp)
|
|
updated = api_client.patch('/api/notifications/channels/email', json={
|
|
'enabled': True,
|
|
'token': 'write-only-smtp-password',
|
|
'config': {'host': 'smtp.internal', 'port': 587, 'username': 'appops', 'fromAddress': 'appops@internal', 'toAddress': 'ops@internal', 'smtpMode': 'starttls'},
|
|
}).json()
|
|
result = api_client.post('/api/notifications/channels/email/test').json()
|
|
|
|
assert updated['safeConfig']['smtpMode'] == 'starttls'
|
|
assert result['status'] == 'sent'
|
|
assert any(action[0] == 'starttls' for action in actions)
|
|
assert 'write-only-smtp-password' not in str(result)
|
|
|
|
|
|
def test_scheduled_notification_cycle_runs_retries_before_evaluation(api_client, monkeypatch):
|
|
from backend.app.services import maintenance_scheduler, notification_service
|
|
|
|
calls = []
|
|
monkeypatch.setattr(notification_service, 'retry_failed_deliveries', lambda session: calls.append('retry') or {})
|
|
monkeypatch.setattr(notification_service, 'evaluate_source', lambda session, source: calls.append(f'evaluate:{source}') or {})
|
|
|
|
maintenance_scheduler.evaluate_notifications_and_retries()
|
|
|
|
assert calls == ['retry', 'evaluate:all']
|
|
|
|
|
|
def test_notification_evaluation_prefers_specific_rule_over_generic_duplicate(api_client):
|
|
from backend.app.db.session import engine
|
|
from backend.app.services import notification_service
|
|
|
|
event = {
|
|
'source': 'media_pipeline',
|
|
'event_type': 'media.issue',
|
|
'severity': 'medium',
|
|
'title': 'Deluge stalled downloads',
|
|
'message': '6 stalled download(s).',
|
|
'category': 'downloads',
|
|
'target': 'deluge',
|
|
'dedupe_key': 'media.issue:deluge:stalled',
|
|
'payload': {'count': 6},
|
|
}
|
|
|
|
with Session(engine) as session:
|
|
result = notification_service.evaluate_rules(session, [event])
|
|
|
|
assert result['created'] == 1
|
|
assert result['events'][0]['ruleId'] == 'notify-deluge-stalled-downloads'
|
|
|
|
|
|
def test_notification_upgrade_dismisses_existing_generic_media_overlap(api_client):
|
|
from datetime import UTC, datetime, timedelta
|
|
from backend.app.db.init_db import _upgrade_duplicate_notification_events
|
|
from backend.app.db.models import NotificationEvent
|
|
from backend.app.db.session import engine
|
|
|
|
created_at = datetime.now(UTC).replace(tzinfo=None)
|
|
with Session(engine) as session:
|
|
session.add(NotificationEvent(
|
|
id='notif-generic-duplicate', rule_id='notify-media-pipeline-issue', source='media_pipeline',
|
|
event_type='media.issue', severity='medium', title='Media Pipeline issue',
|
|
message='Deluge has stalled downloads', category='media', target='media-pipeline',
|
|
dedupe_key='media_pipeline:media.issue:media-pipeline:media', status='sent', created_at=created_at,
|
|
))
|
|
session.add(NotificationEvent(
|
|
id='notif-specific-duplicate', rule_id='notify-deluge-stalled-downloads', source='media_pipeline',
|
|
event_type='media.issue', severity='medium', title='Deluge stalled downloads',
|
|
message='6 stalled download(s).', category='downloads', target='deluge',
|
|
dedupe_key='media_pipeline:media.issue:deluge:downloads', status='sent', created_at=created_at + timedelta(milliseconds=10),
|
|
))
|
|
session.commit()
|
|
|
|
assert _upgrade_duplicate_notification_events(session) is True
|
|
session.commit()
|
|
|
|
generic = session.get(NotificationEvent, 'notif-generic-duplicate')
|
|
specific = session.get(NotificationEvent, 'notif-specific-duplicate')
|
|
assert generic is not None
|
|
assert generic.status == 'dismissed'
|
|
assert generic.reason == 'superseded_by_specific_rule'
|
|
assert generic.read is True
|
|
assert specific is not None
|
|
assert specific.status == 'sent'
|
|
|
|
|
|
def test_notification_upgrade_compacts_only_routine_suppression_flood(api_client):
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from backend.app.db.init_db import _ensure_notification_suppression_index, _upgrade_suppressed_notification_events
|
|
from backend.app.db.models import NotificationEvent
|
|
from backend.app.db.session import engine
|
|
|
|
created_at = datetime.now(UTC).replace(tzinfo=None)
|
|
with engine.begin() as connection:
|
|
connection.execute(text('DROP INDEX IF EXISTS uq_notification_routine_suppression'))
|
|
with Session(engine) as session:
|
|
for index in range(3):
|
|
row_created_at = created_at + timedelta(minutes=index)
|
|
session.add(NotificationEvent(
|
|
id=f'notif-cooldown-{index}', rule_id='notify-deluge-stalled-downloads', source='media_pipeline',
|
|
event_type='media.issue', severity='medium', title='Deluge stalled downloads',
|
|
message='6 stalled download(s).', category='downloads', target='deluge',
|
|
dedupe_key='media.issue:deluge:legacy-flood', channels_json=['dashboard'],
|
|
status='suppressed', reason='cooldown', read=False,
|
|
payload_json={
|
|
'suppressionCount': 1,
|
|
'firstSuppressedAt': (created_at - timedelta(minutes=5) if index == 0 else row_created_at).isoformat(),
|
|
'lastSuppressedAt': (created_at + timedelta(minutes=9) if index == 2 else row_created_at).isoformat(),
|
|
},
|
|
created_at=row_created_at,
|
|
))
|
|
session.add(NotificationEvent(
|
|
id='notif-unsupported-channel', rule_id='manual-test', source='audit',
|
|
event_type='audit.notice', severity='low', title='Unsupported delivery evidence',
|
|
message='No channel accepted this event.', category='delivery', target='custom',
|
|
dedupe_key='audit.notice:unsupported', channels_json=['custom'],
|
|
status='suppressed', reason=None, read=False, created_at=created_at,
|
|
))
|
|
session.commit()
|
|
|
|
assert _upgrade_suppressed_notification_events(session) is True
|
|
session.commit()
|
|
_ensure_notification_suppression_index()
|
|
|
|
routine_rows = session.exec(
|
|
select(NotificationEvent)
|
|
.where(NotificationEvent.dedupe_key == 'media.issue:deluge:legacy-flood')
|
|
).all()
|
|
unsupported = session.get(NotificationEvent, 'notif-unsupported-channel')
|
|
|
|
assert len(routine_rows) == 1
|
|
assert routine_rows[0].id == 'notif-cooldown-2'
|
|
assert routine_rows[0].read is True
|
|
assert routine_rows[0].payload_json['suppressionCount'] == 3
|
|
assert routine_rows[0].payload_json['firstSuppressedAt'] == (created_at - timedelta(minutes=5)).isoformat()
|
|
assert routine_rows[0].payload_json['lastSuppressedAt'] == (created_at + timedelta(minutes=9)).isoformat()
|
|
assert unsupported is not None
|
|
assert unsupported.read is True
|
|
assert 'uq_notification_routine_suppression' in {index['name'] for index in inspect(engine).get_indexes('notification_events')}
|
|
|
|
|
|
def test_notification_summary_is_lightweight_db_read(api_client, monkeypatch):
|
|
from backend.app.services import notification_service
|
|
|
|
def fail_reconcile(_session):
|
|
raise AssertionError("summary must not run live reconciliation")
|
|
|
|
monkeypatch.setattr(notification_service, "_reconcile_current_state", fail_reconcile)
|
|
|
|
response = api_client.get('/api/notifications/summary')
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload['activeRulesCount'] > 0
|
|
assert 'channels' in payload
|
|
|
|
|
|
def test_notification_event_lists_are_lightweight_db_reads(api_client, monkeypatch):
|
|
from datetime import UTC, datetime
|
|
|
|
from backend.app.db.models import NotificationEvent
|
|
from backend.app.db.session import engine
|
|
from backend.app.services import notification_service
|
|
|
|
with Session(engine) as session:
|
|
session.add(
|
|
NotificationEvent(
|
|
id='event-lightweight-read',
|
|
rule_id='manual-test',
|
|
source='audit',
|
|
event_type='audit.blocked_action',
|
|
severity='medium',
|
|
title='Stored event',
|
|
message='Stored event must be returned without live reconciliation',
|
|
category='ops',
|
|
target='container.stop',
|
|
dedupe_key='test:event-lightweight-read',
|
|
channels_json=['dashboard'],
|
|
status='sent',
|
|
read=False,
|
|
created_at=datetime.now(UTC),
|
|
)
|
|
)
|
|
session.add(
|
|
NotificationEvent(
|
|
id='event-dismissed-read',
|
|
rule_id='manual-test',
|
|
source='audit',
|
|
event_type='audit.blocked_action',
|
|
severity='medium',
|
|
title='Dismissed event',
|
|
message='Dismissed unread events stay out of the unread list',
|
|
category='ops',
|
|
target='container.restart',
|
|
dedupe_key='test:event-dismissed-read',
|
|
channels_json=['dashboard'],
|
|
status='dismissed',
|
|
read=False,
|
|
created_at=datetime.now(UTC),
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
def fail_reconcile(_session):
|
|
raise AssertionError("event list reads must not run live reconciliation")
|
|
|
|
monkeypatch.setattr(notification_service, "_reconcile_current_state", fail_reconcile)
|
|
|
|
events_response = api_client.get('/api/notifications/events')
|
|
unread_response = api_client.get('/api/notifications/events/unread')
|
|
|
|
assert events_response.status_code == 200
|
|
assert unread_response.status_code == 200
|
|
event_ids = {event['id'] for event in events_response.json()}
|
|
unread_ids = {event['id'] for event in unread_response.json()}
|
|
assert 'event-lightweight-read' in event_ids
|
|
assert 'event-lightweight-read' in unread_ids
|
|
assert 'event-dismissed-read' not in unread_ids
|
|
|
|
|
|
def test_home_assistant_channel_seeded_and_default_rules_are_spam_safe(api_client):
|
|
rules = {rule['id']: rule for rule in api_client.get('/api/notifications/rules').json()}
|
|
channels = {channel['id']: channel for channel in api_client.get('/api/notifications/channels').json()}
|
|
|
|
home_assistant = channels['home_assistant']
|
|
|
|
assert home_assistant['type'] == 'home_assistant'
|
|
assert home_assistant['name'] == 'Home Assistant'
|
|
assert home_assistant['enabled'] is False
|
|
assert home_assistant['status'] == 'missing_config'
|
|
assert home_assistant['hasSecret'] is False
|
|
assert set(home_assistant['missing']) >= {'HOME_ASSISTANT_URL', 'HOME_ASSISTANT_TOKEN'}
|
|
assert home_assistant['safeConfig']['notifyDomain'] == 'notify'
|
|
assert home_assistant['safeConfig']['notifyService'] == 'notify'
|
|
|
|
for rule_id in [
|
|
'notify-critical-safety-finding',
|
|
'notify-high-safety-finding',
|
|
'notify-npm-certificate-expiring',
|
|
'notify-public-host-without-auth',
|
|
'notify-blocked-dangerous-action',
|
|
]:
|
|
assert 'dashboard' in rules[rule_id]['channels']
|
|
assert 'home_assistant' in rules[rule_id]['channels']
|
|
|
|
for rule_id in [
|
|
'notify-coverage-incomplete',
|
|
'notify-connector-missing-config',
|
|
'notify-media-pipeline-issue',
|
|
'notify-deluge-stalled-downloads',
|
|
'notify-voice-auth-failure',
|
|
]:
|
|
assert rules[rule_id]['channels'] == ['dashboard']
|
|
|
|
serialized = str({'rules': rules, 'channels': channels})
|
|
assert 'HOME_ASSISTANT_TOKEN_VALUE' not in serialized
|
|
assert 'Bearer ' not in serialized
|
|
|
|
|
|
def test_home_assistant_connector_test_uses_notification_channel_missing_config(api_client):
|
|
payload = api_client.post('/api/connectors/home_assistant/test').json()
|
|
|
|
assert payload['success'] is False
|
|
assert payload['status'] == 'missing_config'
|
|
assert payload['missing_config'] is True
|
|
assert payload['required_env_vars'] == ['HOME_ASSISTANT_URL', 'HOME_ASSISTANT_TOKEN']
|
|
assert payload['secretRef'] == 'HOME_ASSISTANT_TOKEN'
|
|
assert payload['message'] != 'test_not_implemented'
|
|
assert 'HOME_ASSISTANT_TOKEN_VALUE' not in str(payload)
|
|
|
|
|
|
def test_home_assistant_connector_test_uses_notification_channel_connection(api_client):
|
|
api_client.patch('/api/notifications/channels/home_assistant', json={
|
|
'enabled': False,
|
|
'token': 'HOME_ASSISTANT_TOKEN_VALUE',
|
|
'config': {
|
|
'url': 'http://homeassistant.local:8123',
|
|
'notifyDomain': 'notify',
|
|
'notifyService': 'persistent_notification',
|
|
'verifyTls': False,
|
|
'timeoutSeconds': 5,
|
|
'dryRun': True,
|
|
},
|
|
})
|
|
|
|
payload = api_client.post('/api/connectors/home_assistant/test').json()
|
|
|
|
assert payload['success'] is True
|
|
assert payload['status'] == 'connected'
|
|
assert payload['missing_config'] is False
|
|
assert payload['required_env_vars'] == ['HOME_ASSISTANT_URL', 'HOME_ASSISTANT_TOKEN']
|
|
assert payload['secretRef'] == 'HOME_ASSISTANT_TOKEN'
|
|
assert 'HOME_ASSISTANT_TOKEN_VALUE' not in str(payload)
|
|
|
|
|
|
def test_home_assistant_channel_patch_stores_secret_and_test_notification_is_audited(api_client):
|
|
from backend.app.db.models import AuditEvent
|
|
from backend.app.db.session import engine
|
|
from backend.app.services.secret_store import get_secret
|
|
|
|
patched = api_client.patch('/api/notifications/channels/home_assistant', json={
|
|
'enabled': True,
|
|
'token': 'HOME_ASSISTANT_TOKEN_VALUE',
|
|
'config': {
|
|
'url': 'http://homeassistant.local:8123',
|
|
'notifyDomain': 'notify',
|
|
'notifyService': 'persistent_notification',
|
|
'target': 'operator_phone',
|
|
'verifyTls': False,
|
|
'timeoutSeconds': 5,
|
|
'dryRun': True,
|
|
},
|
|
}).json()
|
|
|
|
assert patched['enabled'] is True
|
|
assert patched['hasSecret'] is True
|
|
assert patched['missing'] == []
|
|
assert patched['safeConfig']['url'] == 'http://homeassistant.local:8123'
|
|
assert patched['safeConfig']['target'] == 'operator_phone'
|
|
assert 'HOME_ASSISTANT_TOKEN_VALUE' not in str(patched)
|
|
|
|
with Session(engine) as session:
|
|
assert get_secret(session, 'HOME_ASSISTANT_TOKEN') == 'HOME_ASSISTANT_TOKEN_VALUE'
|
|
|
|
test_result = api_client.post('/api/notifications/channels/home_assistant/test').json()
|
|
|
|
assert test_result['success'] is True
|
|
assert test_result['status'] == 'sent'
|
|
assert test_result['channel']['hasSecret'] is True
|
|
assert test_result['channel']['lastTestAt']
|
|
|
|
with Session(engine) as session:
|
|
audit = session.exec(
|
|
select(AuditEvent).where(AuditEvent.action_id == 'notification.channel.test')
|
|
).first()
|
|
assert audit is not None
|
|
assert audit.target == 'home_assistant'
|
|
assert audit.result == 'success'
|
|
|
|
serialized = str({'patch': patched, 'test': test_result})
|
|
assert 'HOME_ASSISTANT_TOKEN_VALUE' not in serialized
|
|
assert 'Bearer ' not in serialized
|
|
|
|
|
|
def test_home_assistant_delivery_records_channel_results_and_preserves_dashboard(api_client):
|
|
from backend.app.db.session import engine
|
|
from backend.app.services import notification_service
|
|
|
|
api_client.patch('/api/notifications/channels/home_assistant', json={
|
|
'enabled': True,
|
|
'token': 'HOME_ASSISTANT_TOKEN_VALUE',
|
|
'config': {
|
|
'url': 'http://homeassistant.local:8123',
|
|
'notifyDomain': 'notify',
|
|
'notifyService': 'persistent_notification',
|
|
'verifyTls': False,
|
|
'dryRun': True,
|
|
},
|
|
})
|
|
|
|
event = {
|
|
'source': 'safety',
|
|
'event_type': 'safety.finding',
|
|
'severity': 'critical',
|
|
'title': 'Public host without auth coverage',
|
|
'message': 'Public endpoint needs auth coverage.',
|
|
'category': 'auth',
|
|
'target': 'npm:widefrog.example',
|
|
'dedupe_key': 'safety.finding:public-host-auth-test',
|
|
'payload': {'secretToken': 'HOME_ASSISTANT_TOKEN_VALUE'},
|
|
}
|
|
|
|
with Session(engine) as session:
|
|
first = notification_service.evaluate_rules(session, [event])
|
|
second = notification_service.evaluate_rules(session, [event])
|
|
|
|
assert first['created'] == 1
|
|
assert first['events'][0]['ruleId'] == 'notify-public-host-without-auth'
|
|
assert all(created['status'] == 'sent' for created in first['events'])
|
|
assert all(created['channelResults']['dashboard']['status'] == 'sent' for created in first['events'])
|
|
assert all(created['channelResults']['home_assistant']['status'] == 'sent' for created in first['events'])
|
|
assert 'HOME_ASSISTANT_TOKEN_VALUE' not in str(first)
|
|
|
|
assert second['created'] == 0
|
|
assert second['suppressed'] == 1
|
|
assert all(event['channelResults']['home_assistant']['status'] == 'suppressed' for event in second['suppressedEvents'])
|
|
|
|
|
|
def test_repeated_cooldown_suppression_stays_bounded_and_out_of_unread(api_client):
|
|
from backend.app.db.models import NotificationEvent
|
|
from backend.app.db.session import engine
|
|
from backend.app.services import notification_service
|
|
|
|
event = {
|
|
'source': 'media_pipeline',
|
|
'event_type': 'media.issue',
|
|
'severity': 'medium',
|
|
'title': 'Deluge stalled downloads',
|
|
'message': '6 stalled download(s).',
|
|
'category': 'downloads',
|
|
'target': 'deluge',
|
|
'dedupe_key': 'media.issue:deluge:bounded-suppression',
|
|
'payload': {'count': 6},
|
|
}
|
|
|
|
with Session(engine) as session:
|
|
first = notification_service.evaluate_rules(session, [event])
|
|
repeated = [notification_service.evaluate_rules(session, [event]) for _ in range(7)]
|
|
stored_dedupe_key = first['events'][0]['dedupeKey']
|
|
rows = session.exec(
|
|
select(NotificationEvent)
|
|
.where(NotificationEvent.dedupe_key == stored_dedupe_key)
|
|
.order_by(NotificationEvent.created_at.asc())
|
|
).all()
|
|
|
|
assert first['created'] == 1
|
|
assert all(result['created'] == 0 and result['suppressed'] == 1 for result in repeated)
|
|
assert len({result['suppressedEvents'][0]['suppressedUntil'] for result in repeated}) == 1
|
|
assert len(rows) == 2
|
|
assert [row.status for row in rows] == ['sent', 'suppressed']
|
|
assert rows[1].read is True
|
|
assert rows[1].payload_json['suppressionCount'] == 7
|
|
|
|
summary = api_client.get('/api/notifications/summary').json()
|
|
unread = api_client.get('/api/notifications/events/unread').json()
|
|
assert summary['unreadCount'] == 1
|
|
assert {item['status'] for item in unread} == {'sent'}
|
|
|
|
|
|
def test_concurrent_routine_suppression_updates_one_atomic_counter(api_client):
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import timedelta
|
|
|
|
from backend.app.db.models import NotificationEvent, NotificationRule
|
|
from backend.app.db.session import engine
|
|
from backend.app.services import notification_service
|
|
|
|
event = {
|
|
'source': 'media_pipeline',
|
|
'event_type': 'media.issue',
|
|
'severity': 'medium',
|
|
'title': 'Concurrent stalled downloads',
|
|
'message': 'The same routine signal arrived concurrently.',
|
|
'category': 'downloads',
|
|
'target': 'deluge',
|
|
'dedupe_key': 'media.issue:deluge:concurrent-suppression',
|
|
'payload': {'count': 4},
|
|
}
|
|
|
|
def record(_: int) -> str:
|
|
with Session(engine) as session:
|
|
rule = session.get(NotificationRule, 'notify-deluge-stalled-downloads')
|
|
assert rule is not None
|
|
row = notification_service._record_suppression(
|
|
session,
|
|
rule,
|
|
event,
|
|
'cooldown',
|
|
notification_service._now() + timedelta(minutes=15),
|
|
)
|
|
return row.id
|
|
|
|
with ThreadPoolExecutor(max_workers=4) as executor:
|
|
row_ids = list(executor.map(record, range(4)))
|
|
|
|
with Session(engine) as session:
|
|
rows = session.exec(
|
|
select(NotificationEvent)
|
|
.where(NotificationEvent.reason == 'cooldown')
|
|
.where(NotificationEvent.title == 'Concurrent stalled downloads')
|
|
).all()
|
|
|
|
assert len(set(row_ids)) == 1
|
|
assert len(rows) == 1
|
|
assert rows[0].payload_json['suppressionCount'] == 4
|
|
|
|
|
|
def test_quiet_hours_use_the_exact_window_end_and_release_at_boundary():
|
|
from datetime import datetime
|
|
|
|
from backend.app.db.models import NotificationRule
|
|
from backend.app.services import notification_service
|
|
|
|
rule = NotificationRule(
|
|
id='quiet-hours-test',
|
|
name='Quiet hours test',
|
|
source='audit',
|
|
event_type='audit.notice',
|
|
quiet_hours_json={
|
|
'enabled': True,
|
|
'startHour': 22,
|
|
'endHour': 7,
|
|
'allowCriticalOverride': True,
|
|
},
|
|
)
|
|
medium_event = {'severity': 'medium'}
|
|
|
|
assert notification_service._quiet_hours_until(
|
|
rule,
|
|
medium_event,
|
|
datetime(2026, 8, 23, 23, 30),
|
|
) == datetime(2026, 8, 24, 7, 0)
|
|
assert notification_service._quiet_hours_until(
|
|
rule,
|
|
medium_event,
|
|
datetime(2026, 8, 24, 6, 59),
|
|
) == datetime(2026, 8, 24, 7, 0)
|
|
assert notification_service._quiet_hours_until(
|
|
rule,
|
|
medium_event,
|
|
datetime(2026, 8, 24, 7, 0),
|
|
) is None
|
|
assert notification_service._quiet_hours_until(
|
|
rule,
|
|
{'severity': 'critical'},
|
|
datetime(2026, 8, 23, 23, 30),
|
|
) is None
|
|
|
|
|
|
def test_retry_query_selects_old_pending_delivery_before_recent_noise(api_client, monkeypatch):
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from backend.app.db.models import NotificationEvent
|
|
from backend.app.db.session import engine
|
|
from backend.app.services import notification_service
|
|
|
|
created_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=1)
|
|
with Session(engine) as session:
|
|
session.add(NotificationEvent(
|
|
id='notif-pending-old', rule_id='manual-test', source='audit', event_type='audit.notice',
|
|
severity='high', title='Pending delivery', message='Retry this delivery.',
|
|
dedupe_key='audit.notice:pending-old', channels_json=['dashboard'], status='sent',
|
|
payload_json={'deliveryPending': True, 'deliveryRetryCount': 1}, created_at=created_at,
|
|
))
|
|
for index in range(125):
|
|
session.add(NotificationEvent(
|
|
id=f'notif-noise-{index}', rule_id='manual-test', source='audit', event_type='audit.notice',
|
|
severity='low', title='Recent notification', message='No retry required.',
|
|
dedupe_key=f'audit.notice:noise-{index}', channels_json=['dashboard'], status='sent',
|
|
payload_json={'deliveryPending': False}, created_at=created_at + timedelta(minutes=index + 1),
|
|
))
|
|
session.commit()
|
|
|
|
retried_ids: list[str] = []
|
|
|
|
def deliver(_session, row):
|
|
retried_ids.append(row.id)
|
|
row.payload_json = {**row.payload_json, 'deliveryPending': False}
|
|
return row
|
|
|
|
monkeypatch.setattr(notification_service, '_deliver_event', deliver)
|
|
result = notification_service.retry_failed_deliveries(session, limit=1)
|
|
|
|
assert retried_ids == ['notif-pending-old']
|
|
assert result == {'retried': 1, 'recovered': 1, 'deadLettered': 0}
|
|
|
|
|
|
def test_notification_event_limit_is_bounded_by_api(api_client):
|
|
assert api_client.get('/api/notifications/events?limit=1').status_code == 200
|
|
assert api_client.get('/api/notifications/events?limit=501').status_code == 422
|
|
assert api_client.get('/api/notifications/events?offset=-1').status_code == 422
|
|
assert api_client.get('/api/notifications/events/unread?limit=501').status_code == 422
|
|
|
|
|
|
def test_bounded_event_inbox_prioritizes_unread_over_newer_dismissed_history(api_client):
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from backend.app.db.models import NotificationEvent
|
|
from backend.app.db.session import engine
|
|
|
|
created_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=1)
|
|
with Session(engine) as session:
|
|
session.add(NotificationEvent(
|
|
id='notif-unread-priority', rule_id='manual-test', source='safety', event_type='safety.finding',
|
|
severity='critical', title='Unread safety signal', message='This must stay visible.',
|
|
dedupe_key='safety.finding:unread-priority', channels_json=['dashboard'], status='sent',
|
|
read=False, created_at=created_at,
|
|
))
|
|
for index in range(260):
|
|
session.add(NotificationEvent(
|
|
id=f'notif-dismissed-history-{index}', rule_id='manual-test', source='audit', event_type='audit.notice',
|
|
severity='low', title='Dismissed history', message='Historical event.',
|
|
dedupe_key=f'audit.notice:dismissed-history-{index}', channels_json=['dashboard'], status='dismissed',
|
|
read=True, created_at=created_at + timedelta(minutes=index + 1),
|
|
))
|
|
session.commit()
|
|
|
|
bounded = api_client.get('/api/notifications/events?limit=10').json()
|
|
active_only = api_client.get('/api/notifications/events?limit=10&include_dismissed=false').json()
|
|
older_page = api_client.get('/api/notifications/events?limit=10&offset=250').json()
|
|
|
|
assert bounded[0]['id'] == 'notif-unread-priority'
|
|
assert any(event['id'] == 'notif-unread-priority' for event in active_only)
|
|
assert all(event['status'] != 'dismissed' for event in active_only)
|
|
assert len(older_page) == 10
|
|
|
|
|
|
def test_unread_event_pages_remain_reachable_with_stable_offset(api_client):
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from backend.app.db.models import NotificationEvent
|
|
from backend.app.db.session import engine
|
|
|
|
created_at = datetime.now(UTC).replace(tzinfo=None)
|
|
with Session(engine) as session:
|
|
for index in range(3):
|
|
session.add(NotificationEvent(
|
|
id=f'notif-unread-page-{index}', rule_id='manual-test', source='safety', event_type='safety.finding',
|
|
severity='high', title=f'Unread page {index}', message='Paged unread event.',
|
|
dedupe_key=f'safety.finding:unread-page-{index}', channels_json=['dashboard'], status='sent',
|
|
read=False, created_at=created_at + timedelta(seconds=index),
|
|
))
|
|
session.commit()
|
|
|
|
first = api_client.get('/api/notifications/events/unread?limit=1&offset=0').json()
|
|
second = api_client.get('/api/notifications/events/unread?limit=1&offset=1').json()
|
|
|
|
assert [event['id'] for event in first] == ['notif-unread-page-2']
|
|
assert [event['id'] for event in second] == ['notif-unread-page-1']
|
|
|
|
|
|
def test_home_assistant_failure_does_not_break_dashboard_delivery(api_client, monkeypatch):
|
|
from backend.app.db.session import engine
|
|
from backend.app.services import notification_service
|
|
from backend.app.services.notification_channels import home_assistant
|
|
|
|
api_client.patch('/api/notifications/channels/home_assistant', json={
|
|
'enabled': True,
|
|
'token': 'HOME_ASSISTANT_TOKEN_VALUE',
|
|
'config': {
|
|
'url': 'http://homeassistant.local:8123',
|
|
'verifyTls': False,
|
|
'dryRun': False,
|
|
},
|
|
})
|
|
|
|
monkeypatch.setattr(
|
|
home_assistant,
|
|
'send_notification',
|
|
lambda session, channel, event: {'status': 'failed', 'reason': 'auth_failed', 'detail': 'auth_failed'},
|
|
)
|
|
|
|
with Session(engine) as session:
|
|
result = notification_service.evaluate_rules(session, [{
|
|
'source': 'audit',
|
|
'event_type': 'audit.blocked_action',
|
|
'severity': 'high',
|
|
'title': 'Blocked dangerous action attempt',
|
|
'message': 'volume.delete was blocked.',
|
|
'category': 'policy',
|
|
'target': 'array',
|
|
'dedupe_key': 'audit.blocked_action:ha-failure-test',
|
|
'payload': {},
|
|
}])
|
|
|
|
assert result['created'] == 1
|
|
created = result['events'][0]
|
|
assert created['status'] == 'sent'
|
|
assert created['channelResults']['dashboard']['status'] == 'sent'
|
|
assert created['channelResults']['home_assistant']['status'] == 'failed'
|
|
assert created['channelResults']['home_assistant']['reason'] == 'auth_failed'
|
|
|
|
|
|
def test_safety_evaluation_creates_events_and_dedupes_coverage(api_client):
|
|
first = api_client.post('/api/notifications/evaluate/safety').json()
|
|
second = api_client.post('/api/notifications/evaluate/safety').json()
|
|
events = api_client.get('/api/notifications/events').json()
|
|
|
|
assert first['created'] >= 1
|
|
assert first['suppressed'] >= 0
|
|
assert second['created'] == 0
|
|
assert second['suppressed'] >= 1
|
|
assert any(event['source'] == 'safety' and event['eventType'] == 'safety.finding' for event in events)
|
|
assert all(event['dedupeKey'] for event in events)
|
|
|
|
|
|
def test_global_safety_score_alone_does_not_create_notification(api_client, monkeypatch):
|
|
from backend.app.services import notification_service
|
|
|
|
def score_only_events(session):
|
|
return [{
|
|
'source': 'safety',
|
|
'event_type': 'safety.score',
|
|
'severity': 'critical',
|
|
'title': 'Global score critical',
|
|
'message': 'Score is low without concrete finding.',
|
|
'category': 'score',
|
|
'target': 'safety-center',
|
|
'dedupe_key': 'safety.score:critical',
|
|
'payload': {'score': 40},
|
|
}]
|
|
|
|
monkeypatch.setattr(notification_service, '_safety_events', score_only_events)
|
|
|
|
result = api_client.post('/api/notifications/evaluate/safety').json()
|
|
|
|
assert result['created'] == 0
|
|
assert result['suppressed'] == 0
|
|
|
|
|
|
def test_connector_media_and_audit_evaluation_create_rule_bound_events(api_client):
|
|
from backend.app.db.models import AuditEvent
|
|
from backend.app.db.session import engine
|
|
|
|
with Session(engine) as session:
|
|
session.add(AuditEvent(
|
|
id='audit-blocked-dangerous',
|
|
source='voice',
|
|
identity='operator',
|
|
action_id='volume.delete',
|
|
target='array',
|
|
target_type='volume',
|
|
risk='critical',
|
|
policy_decision='blocked',
|
|
result='blocked',
|
|
reason='Dangerous action blocked.',
|
|
))
|
|
session.commit()
|
|
|
|
connector_result = api_client.post('/api/notifications/evaluate/connectors').json()
|
|
media_result = api_client.post('/api/notifications/evaluate/media').json()
|
|
all_result = api_client.post('/api/notifications/evaluate').json()
|
|
|
|
assert connector_result['created'] >= 1
|
|
assert media_result['created'] >= 0
|
|
assert all_result['created'] >= 1 or all_result['suppressed'] >= 1
|
|
|
|
events = api_client.get('/api/notifications/events').json()
|
|
assert any(event['eventType'] == 'connector.missing_config' for event in events)
|
|
assert any(event['eventType'] == 'audit.blocked_action' for event in events)
|
|
|
|
|
|
def test_media_notification_evaluation_passes_session_to_media_summary(api_client, monkeypatch):
|
|
from backend.app.services import media_pipeline_service
|
|
|
|
observed = {'session': None}
|
|
|
|
def media_summary(session):
|
|
observed['session'] = session
|
|
return {
|
|
'success': True,
|
|
'status': 'warning',
|
|
'risk': {'level': 'medium', 'reasons': ['Deluge has stalled downloads']},
|
|
'counts': {'stalledDownloads': 1, 'failedWidefrogJobs': 0},
|
|
}
|
|
|
|
monkeypatch.setattr(media_pipeline_service, 'summary', media_summary)
|
|
|
|
response = api_client.post('/api/notifications/evaluate/media')
|
|
|
|
assert response.status_code == 200
|
|
assert observed['session'] is not None
|
|
payload = response.json()
|
|
assert payload['evaluated'] == 1
|
|
assert payload['created'] == 1
|
|
assert payload['events'][0]['ruleId'] == 'notify-deluge-stalled-downloads'
|
|
|
|
|
|
def test_connector_evaluation_auto_resolves_stale_connector_events(api_client):
|
|
from backend.app.db.models import Connector, NotificationEvent
|
|
from backend.app.db.session import engine
|
|
|
|
api_client.post('/api/notifications/evaluate/connectors')
|
|
|
|
with Session(engine) as session:
|
|
deluge = session.get(Connector, 'deluge')
|
|
deluge.status = 'connected'
|
|
session.add(deluge)
|
|
session.commit()
|
|
|
|
api_client.post('/api/notifications/evaluate/connectors')
|
|
|
|
with Session(engine) as session:
|
|
stale = session.exec(
|
|
select(NotificationEvent)
|
|
.where(NotificationEvent.source == 'connector')
|
|
.where(NotificationEvent.target == 'deluge')
|
|
.order_by(NotificationEvent.created_at.desc())
|
|
).first()
|
|
|
|
assert stale.status == 'dismissed'
|
|
assert stale.reason == 'auto_resolved'
|
|
assert stale.read is True
|
|
|
|
|
|
def test_safety_evaluation_auto_resolves_stale_safety_events(api_client, monkeypatch):
|
|
from backend.app.db.models import NotificationEvent
|
|
from backend.app.db.session import engine
|
|
from backend.app.services import notification_service
|
|
|
|
def stale_safety_event(session):
|
|
return [{
|
|
'source': 'safety',
|
|
'event_type': 'safety.finding',
|
|
'severity': 'high',
|
|
'title': 'Critical discovered app needs review: NginxProxyManager',
|
|
'message': 'A critical-looking discovered app still needs review or mapping.',
|
|
'category': 'inventory',
|
|
'target': 'NginxProxyManager',
|
|
'dedupe_key': 'safety.finding:ai-sec-inventory-critical-review-dc-old-npm',
|
|
'payload': {'findingId': 'ai-sec-inventory-critical-review-dc-old-npm'},
|
|
}]
|
|
|
|
monkeypatch.setattr(notification_service, '_safety_events', stale_safety_event)
|
|
api_client.post('/api/notifications/evaluate/safety')
|
|
monkeypatch.setattr(notification_service, '_safety_events', lambda session: [])
|
|
result = api_client.post('/api/notifications/evaluate/safety').json()
|
|
|
|
with Session(engine) as session:
|
|
stale = session.exec(
|
|
select(NotificationEvent)
|
|
.where(NotificationEvent.source == 'safety')
|
|
.where(NotificationEvent.event_type == 'safety.finding')
|
|
.where(NotificationEvent.target == 'NginxProxyManager')
|
|
).first()
|
|
|
|
assert result['resolved'] == 1
|
|
assert stale.status == 'dismissed'
|
|
assert stale.reason == 'auto_resolved'
|
|
assert stale.read is True
|
|
|
|
|
|
def test_notification_lifecycle_rule_disable_and_secret_safe_channel_test(api_client):
|
|
rules = api_client.get('/api/notifications/rules').json()
|
|
high_rule = next(rule for rule in rules if rule['id'] == 'notify-high-safety-finding')
|
|
patched = api_client.patch(f"/api/notifications/rules/{high_rule['id']}", json={'enabled': False}).json()
|
|
assert patched['enabled'] is False
|
|
|
|
api_client.post('/api/notifications/evaluate/safety')
|
|
event = api_client.get('/api/notifications/events').json()[0]
|
|
read = api_client.post(f"/api/notifications/events/{event['id']}/mark-read").json()
|
|
dismissed = api_client.post(f"/api/notifications/events/{event['id']}/dismiss").json()
|
|
|
|
assert read['status'] == 'sent'
|
|
assert read['read'] is True
|
|
assert dismissed['status'] == 'dismissed'
|
|
|
|
channel = api_client.post('/api/notifications/channels', json={
|
|
'id': 'ha-main',
|
|
'type': 'home_assistant',
|
|
'name': 'Home Assistant',
|
|
'enabled': True,
|
|
'config': {'url': 'http://homeassistant.local:8123', 'token': 'TOKEN_VALUE', 'dryRun': True},
|
|
'secretRefs': ['HOME_ASSISTANT_TOKEN'],
|
|
}).json()
|
|
test_result = api_client.post(f"/api/notifications/channels/{channel['id']}/test").json()
|
|
|
|
serialized = str({'channel': channel, 'test': test_result})
|
|
assert 'TOKEN_VALUE' not in serialized
|
|
assert 'Bearer ' not in serialized
|
|
assert test_result['success'] is True
|
|
assert test_result['status'] in {'prepared', 'sent', 'missing_config'}
|
|
assert test_result['channel']['lastTestAt']
|
|
listed_channel = next(item for item in api_client.get('/api/notifications/channels').json() if item['id'] == channel['id'])
|
|
assert listed_channel['lastTestAt'] == test_result['channel']['lastTestAt']
|
|
|
|
|
|
def test_notification_evaluation_does_not_create_jobs_or_approvals(api_client):
|
|
before_jobs = api_client.get('/api/jobs').json()
|
|
before_approvals = api_client.get('/api/approvals').json()
|
|
|
|
api_client.post('/api/notifications/evaluate')
|
|
|
|
after_jobs = api_client.get('/api/jobs').json()
|
|
after_approvals = api_client.get('/api/approvals').json()
|
|
assert before_jobs == after_jobs
|
|
assert before_approvals == after_approvals
|
|
|
|
|
|
def test_notification_tables_store_redacted_payloads(api_client):
|
|
api_client.post('/api/notifications/evaluate/safety')
|
|
|
|
from backend.app.db.models import NotificationEvent
|
|
from backend.app.db.session import engine
|
|
|
|
with Session(engine) as session:
|
|
rows = session.exec(select(NotificationEvent)).all()
|
|
|
|
serialized = str([row.payload_json for row in rows])
|
|
assert 'TOKEN_VALUE' not in serialized
|
|
assert 'Bearer ' not in serialized
|
|
assert 'api_key' not in serialized.lower()
|
|
|
|
|
|
def test_init_db_upgrades_existing_production_db_without_resetting_setup_or_secrets(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "upgrade_existing.db"
|
|
monkeypatch.setenv("APPOPS_DB_PATH", str(db_path))
|
|
monkeypatch.setenv("APPOPS_DATA_DIR", str(tmp_path / "data"))
|
|
monkeypatch.setenv("APPOPS_ENV", "production")
|
|
monkeypatch.setenv("APPOPS_SECRET_KEY", "promotion-upgrade-test-master-key")
|
|
monkeypatch.delenv("APPOPS_SECRET_KEY_FILE", raising=False)
|
|
|
|
from backend.app.core.config import settings
|
|
from backend.app.db import session as db_session
|
|
from backend.app.db.init_db import init_db
|
|
from backend.app.db.models import AppSetting, NotificationChannel, NotificationRule, Secret
|
|
from backend.app.services.secret_store import get_secret, set_secret
|
|
from backend.app.services.setup_service import get_setup_status, serialize_connector_settings
|
|
|
|
previous = {
|
|
"db_path": settings.db_path,
|
|
"data_dir": settings.data_dir,
|
|
"environment": settings.environment,
|
|
"secret_key": settings.secret_key,
|
|
"secret_key_source": settings.secret_key_source,
|
|
"secret_key_file": settings.secret_key_file,
|
|
"apis": dict(settings.apis),
|
|
}
|
|
settings.db_path = str(db_path)
|
|
settings.data_dir = str(tmp_path / "data")
|
|
settings.environment = "production"
|
|
settings.secret_key = "promotion-upgrade-test-master-key"
|
|
settings.secret_key_source = "env"
|
|
settings.secret_key_file = None
|
|
for key in settings.apis:
|
|
if key.startswith("PORTAINER_") or key.startswith("UNRAID_"):
|
|
settings.apis[key] = None
|
|
|
|
engine = create_engine(f"sqlite:///{Path(db_path)}", connect_args={"check_same_thread": False})
|
|
db_session.engine = engine
|
|
SQLModel.metadata.drop_all(engine)
|
|
SQLModel.metadata.create_all(engine)
|
|
|
|
with Session(engine) as session:
|
|
session.add(AppSetting(key="setup:first_run_completed", value_json={"value": True}, is_secret=False))
|
|
session.add(AppSetting(key="connector:portainer:config", value_json={"url": "http://portainer.local", "endpointId": "1", "verifyTls": False}, is_secret=False))
|
|
set_secret(session, "PORTAINER_API_KEY", "PORTAINER_UPGRADE_SECRET", used_by=["portainer"], commit=False)
|
|
session.commit()
|
|
|
|
with engine.begin() as connection:
|
|
connection.execute(text("DROP TABLE notification_channels"))
|
|
connection.execute(text("""
|
|
CREATE TABLE notification_channels (
|
|
id VARCHAR NOT NULL PRIMARY KEY,
|
|
type VARCHAR NOT NULL,
|
|
name VARCHAR NOT NULL,
|
|
enabled BOOLEAN NOT NULL,
|
|
config_json JSON,
|
|
secret_refs_json JSON,
|
|
created_at DATETIME,
|
|
updated_at DATETIME
|
|
)
|
|
"""))
|
|
connection.execute(text("""
|
|
INSERT INTO notification_channels (id, type, name, enabled, config_json, secret_refs_json, created_at, updated_at)
|
|
VALUES ('dashboard', 'dashboard', 'Dashboard', 1, '{"delivery":"in_app"}', '[]', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
|
"""))
|
|
|
|
try:
|
|
init_db()
|
|
init_db()
|
|
|
|
inspector = inspect(engine)
|
|
assert "notification_rules" in inspector.get_table_names()
|
|
assert "notification_events" in inspector.get_table_names()
|
|
assert "last_test_at" in {column["name"] for column in inspector.get_columns("notification_channels")}
|
|
|
|
with Session(engine) as session:
|
|
assert get_setup_status(session)["firstRunCompleted"] is True
|
|
assert serialize_connector_settings(session, "portainer")["hasSecret"] is True
|
|
assert get_secret(session, "PORTAINER_API_KEY") == "PORTAINER_UPGRADE_SECRET"
|
|
assert len(session.exec(select(NotificationChannel).where(NotificationChannel.id == "dashboard")).all()) == 1
|
|
assert len(session.exec(select(NotificationRule).where(NotificationRule.id == "notify-high-safety-finding")).all()) == 1
|
|
assert len(session.exec(select(Secret).where(Secret.id == "PORTAINER_API_KEY")).all()) == 1
|
|
finally:
|
|
settings.db_path = previous["db_path"]
|
|
settings.data_dir = previous["data_dir"]
|
|
settings.environment = previous["environment"]
|
|
settings.secret_key = previous["secret_key"]
|
|
settings.secret_key_source = previous["secret_key_source"]
|
|
settings.secret_key_file = previous["secret_key_file"]
|
|
settings.apis = previous["apis"]
|