61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
"""Тесты feed_quality.notify (флаги notify_*)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime
|
|
|
|
from app import create_app, db
|
|
from app.models.feed_alert import FeedAlert
|
|
from app.services.feed_quality.notify import notify_feed_quality_alerts
|
|
from app.services.feed_quality.rules import EVENT_OVERLOAD, SEVERITY_ERROR, SEVERITY_WARNING
|
|
from app.services.feed_quality.settings_store import save_feed_quality_settings
|
|
from app.models import ZootechNotification
|
|
from tests.helpers.zootech_test_helpers import ZootechTestConfig
|
|
|
|
|
|
class FeedQualityNotifyTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.app = create_app(ZootechTestConfig)
|
|
self.ctx = self.app.app_context()
|
|
self.ctx.push()
|
|
db.create_all()
|
|
|
|
def tearDown(self) -> None:
|
|
db.session.remove()
|
|
db.drop_all()
|
|
self.ctx.pop()
|
|
|
|
def _alert(self, *, severity: str) -> FeedAlert:
|
|
return FeedAlert(
|
|
id="fq-nt-1",
|
|
event_type=EVENT_OVERLOAD,
|
|
severity=severity,
|
|
loading_report_id="lr-1",
|
|
recipe_id="r-1",
|
|
recipe_name="Рейс",
|
|
component_name="C1",
|
|
detail="перегруз",
|
|
created_at=datetime.now(),
|
|
)
|
|
|
|
def test_notify_warning_and_critical_by_default(self) -> None:
|
|
n1 = notify_feed_quality_alerts([self._alert(severity=SEVERITY_WARNING)])
|
|
n2 = notify_feed_quality_alerts([self._alert(severity=SEVERITY_ERROR)])
|
|
self.assertEqual(n1, 1)
|
|
self.assertEqual(n2, 1)
|
|
self.assertEqual(db.session.query(ZootechNotification).count(), 2)
|
|
|
|
def test_notify_warning_disabled(self) -> None:
|
|
save_feed_quality_settings({"loading": {"notify_warning": False}})
|
|
count = notify_feed_quality_alerts([self._alert(severity=SEVERITY_WARNING)])
|
|
self.assertEqual(count, 0)
|
|
|
|
def test_notify_critical_disabled(self) -> None:
|
|
save_feed_quality_settings({"loading": {"notify_critical": False}})
|
|
count = notify_feed_quality_alerts([self._alert(severity=SEVERITY_ERROR)])
|
|
self.assertEqual(count, 0)
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|