61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
import logging
|
|
|
|
from app.db import Database
|
|
from app.models import KIND_FLAPPING, KIND_PROBLEM, KIND_RECOVERY
|
|
from app.routing import ZPOOL_MARKER, RouteRepository, seed_default_rules
|
|
|
|
|
|
def test_zpool_rule_is_created_with_specified_fields():
|
|
db = Database(":memory:")
|
|
assert seed_default_rules(db) is True
|
|
(rule,) = RouteRepository(db).list_all()
|
|
assert rule.name == "ZFS Zpool" and rule.enabled
|
|
assert (rule.match_host, rule.match_service, rule.min_state) == ("*", "*zpool*", 1)
|
|
assert set(rule.kinds) == {KIND_PROBLEM, KIND_RECOVERY, KIND_FLAPPING}
|
|
assert rule.use_default_targets and rule.targets == ()
|
|
assert rule.delay_minutes == 0 and rule.ai_enabled and rule.reminder_hours == 4 and rule.loud
|
|
assert "zpool status -v" in rule.ai_hint and "nicht verifiziert" in rule.ai_hint
|
|
assert len(rule.ai_hint) <= 800
|
|
|
|
|
|
def test_seed_runs_only_once():
|
|
db = Database(":memory:")
|
|
assert seed_default_rules(db) is True
|
|
assert seed_default_rules(db) is False
|
|
assert len(RouteRepository(db).list_all()) == 1
|
|
|
|
|
|
def test_deleted_rule_does_not_come_back():
|
|
db = Database(":memory:")
|
|
seed_default_rules(db)
|
|
repo = RouteRepository(db)
|
|
repo.delete(repo.list_all()[0].id)
|
|
assert seed_default_rules(db) is False
|
|
assert repo.list_all() == []
|
|
|
|
|
|
def test_seed_is_appended_after_existing_rules_and_audited():
|
|
db = Database(":memory:")
|
|
db.execute("INSERT INTO routes(name, position) VALUES ('alt', 7)")
|
|
seed_default_rules(db)
|
|
assert [r.name for r in RouteRepository(db).list_all()] == ["alt", "ZFS Zpool"]
|
|
assert db.query("SELECT user, action FROM audit")[0]["action"] == "route.seed"
|
|
assert db.query("SELECT 1 FROM app_state WHERE key=?", (ZPOOL_MARKER,))
|
|
|
|
|
|
def test_seed_failure_is_logged_and_never_raises(caplog):
|
|
class BrokenDb:
|
|
def query(self, *args):
|
|
raise RuntimeError("db kaputt")
|
|
|
|
with caplog.at_level(logging.ERROR):
|
|
assert seed_default_rules(BrokenDb()) is False
|
|
assert "Standardregel" in caplog.text
|
|
|
|
|
|
def test_failed_write_leaves_no_marker_so_next_start_retries():
|
|
db = Database(":memory:")
|
|
db.execute("DROP TABLE routes")
|
|
assert seed_default_rules(db) is False
|
|
assert db.query("SELECT 1 FROM app_state WHERE key=?", (ZPOOL_MARKER,)) == []
|