tobast/matrix-alertbot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

auto extend silence without duration

Browse source
This commit is contained in:
HgO 2022-08-08 00:28:36 +02:00
parent 6781bc82fa
commit f381eac689
10 changed files with 804 additions and 467 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

54
matrix_alertbot/alertmanager.py
View file

@ -12,10 +12,12 @@ from matrix_alertbot.errors import (
AlertmanagerServerError,
AlertNotFoundError,
SilenceExpiredError,
SilenceExtendError,
SilenceNotFoundError,
)
MAX_DURATION_DAYS = 3652
DEFAULT_DURATION = timedelta(hours=3)
MAX_DURATION = timedelta(days=3652)
class AlertmanagerClient:
@ -60,7 +62,6 @@ class AlertmanagerClient:
fingerprint: str,
user: str,
duration_seconds: Optional[int] = None,
silence_id: Optional[str] = None,
) -> str:
alert = await self.get_alert(fingerprint)
@ -69,13 +70,50 @@ class AlertmanagerClient:
for label, value in alert["labels"].items()
]
start_time = datetime.now()
max_duration = timedelta(days=MAX_DURATION_DAYS)
if duration_seconds is None or duration_seconds > max_duration.total_seconds():
end_time = start_time + max_duration
return await self._create_or_update_silence(
fingerprint, silence_matchers, user, duration_seconds
)
async def update_silence(self, fingerprint: str) -> str:
try:
silence_id: Optional[str]
expire_time: Optional[int]
silence_id, expire_time = self.cache.get(fingerprint, expire_time=True)
except TypeError:
silence_id = None
if silence_id is None:
raise SilenceNotFoundError(
f"Cannot find silence for alert with fingerprint {fingerprint} in cache."
)
if expire_time is not None:
raise SilenceExtendError(
f"Cannot extend silence ID {silence_id} with static duration."
)
silence = await self.get_silence(silence_id)
user = silence["createdBy"]
silence_matchers = silence["matchers"]
return await self._create_or_update_silence(fingerprint, silence_matchers, user)
async def _create_or_update_silence(
self,
fingerprint: str,
silence_matchers: List,
user: str,
duration_seconds: Optional[int] = None,
silence_id: Optional[str] = None,
) -> str:
if duration_seconds is None:
duration_delta = DEFAULT_DURATION
elif duration_seconds > MAX_DURATION.total_seconds():
duration_delta = MAX_DURATION
else:
duration_delta = timedelta(seconds=duration_seconds)
end_time = start_time + duration_delta
start_time = datetime.now()
end_time = start_time + duration_delta
silence = {
"id": silence_id,
@ -97,6 +135,8 @@ class AlertmanagerClient:
f"Cannot create silence for alert fingerprint {fingerprint}"
) from e
self.cache.set(fingerprint, data["silenceID"], expire=duration_seconds)
return data["silenceID"]
async def delete_silence(self, silence_id: str) -> None:

34
matrix_alertbot/callback.py
View file

@ -24,8 +24,8 @@ logger = logging.getLogger(__name__)
class Callbacks:
def __init__(
self,
client: AsyncClient,
alertmanager: AlertmanagerClient,
matrix_client: AsyncClient,
alertmanager_client: AlertmanagerClient,
cache: Cache,
config: Config,
):
@ -39,9 +39,9 @@ class Callbacks:
config: Bot configuration parameters.
"""
self.client = client
self.matrix_client = matrix_client
self.cache = cache
self.alertmanager = alertmanager
self.alertmanager_client = alertmanager_client
self.config = config
self.command_prefix = config.command_prefix
@ -54,7 +54,7 @@ class Callbacks:
event: The event defining the message.
"""
# Ignore messages from ourselves
if event.sender == self.client.user:
if event.sender == self.matrix_client.user:
return
# Ignore messages from unauthorized room
@ -91,9 +91,9 @@ class Callbacks:
try:
command = CommandFactory.create(
cmd,
self.client,
self.matrix_client,
self.cache,
self.alertmanager,
self.alertmanager_client,
self.config,
room,
event.sender,
@ -122,7 +122,7 @@ class Callbacks:
# Attempt to join 3 times before giving up
for attempt in range(3):
result = await self.client.join(room.room_id)
result = await self.matrix_client.join(room.room_id)
if type(result) == JoinError:
logger.error(
f"Error joining room {room.room_id} (attempt %d): %s",
@ -146,7 +146,7 @@ class Callbacks:
not actually our own invite event (such as the inviter's membership).
This makes sure we only call `callbacks.invite` with our own invite events.
"""
if event.state_key == self.client.user_id:
if event.state_key == self.matrix_client.user_id:
# This is our own membership (invite) event
await self.invite(room, event)
@ -167,7 +167,7 @@ class Callbacks:
return
# Ignore reactions from ourselves
if event.sender == self.client.user:
if event.sender == self.matrix_client.user:
return
reaction = event.source.get("content", {}).get("m.relates_to", {}).get("key")
@ -178,7 +178,9 @@ class Callbacks:
return
# Get the original event that was reacted to
event_response = await self.client.room_get_event(room.room_id, alert_event_id)
event_response = await self.matrix_client.room_get_event(
room.room_id, alert_event_id
)
if isinstance(event_response, RoomGetEventError):
logger.warning(
f"Error getting event that was reacted to ({alert_event_id})"
@ -192,9 +194,9 @@ class Callbacks:
# Send a message acknowledging the reaction
command = AckAlertCommand(
self.client,
self.matrix_client,
self.cache,
self.alertmanager,
self.alertmanager_client,
self.config,
room,
event.sender,
@ -210,15 +212,15 @@ class Callbacks:
return
# Ignore redactions from ourselves
if event.sender == self.client.user:
if event.sender == self.matrix_client.user:
return
logger.debug(f"Received event to remove event ID {event.redacts}")
command = UnackAlertCommand(
self.client,
self.matrix_client,
self.cache,
self.alertmanager,
self.alertmanager_client,
self.config,
room,
event.sender,

55
matrix_alertbot/command.py
View file

@ -21,9 +21,9 @@ logger = logging.getLogger(__name__)
class BaseCommand:
def __init__(
self,
client: AsyncClient,
matrix_client: AsyncClient,
cache: Cache,
alertmanager: AlertmanagerClient,
alertmanager_client: AlertmanagerClient,
config: Config,
room: MatrixRoom,
sender: str,
@ -49,9 +49,9 @@ class BaseCommand:
event_id: The ID of the event describing the command.
"""
self.client = client
self.matrix_client = matrix_client
self.cache = cache
self.alertmanager = alertmanager
self.alertmanager_client = alertmanager_client
self.config = config
self.room = room
self.sender = sender
@ -98,7 +98,7 @@ class AckAlertCommand(BaseAlertCommand):
if duration_seconds is None:
logger.error(f"Unable to create silence: Invalid duration '{duration}'")
await send_text_to_room(
self.client,
self.matrix_client,
self.room.room_id,
f"I tried really hard, but I can't convert the duration '{duration}' to a number of seconds.",
)
@ -108,16 +108,13 @@ class AckAlertCommand(BaseAlertCommand):
f"Unable to create silence: Duration must be positive, got '{duration}'"
)
await send_text_to_room(
self.client,
self.matrix_client,
self.room.room_id,
"I can't create a silence with a negative duration!",
)
return
cache_expire_time = duration_seconds
else:
duration_seconds = None
cache_expire_time = self.config.cache_expire_time
logger.debug(
"Receiving a command to create a silence for an indefinite period"
)
@ -133,27 +130,14 @@ class AckAlertCommand(BaseAlertCommand):
)
return
cached_silence_id = self.cache.get(alert_fingerprint)
if cached_silence_id is None:
logger.debug(
f"Creating silence for alert with fingerprint {alert_fingerprint}."
)
else:
logger.debug(
f"Updating silence with ID {cached_silence_id} for alert with fingerprint {alert_fingerprint}."
)
try:
silence_id = await self.alertmanager.create_silence(
alert_fingerprint,
self.room.user_name(self.sender),
duration_seconds,
cached_silence_id,
silence_id = await self.alertmanager_client.create_silence(
alert_fingerprint, self.room.user_name(self.sender), duration_seconds
)
except AlertNotFoundError as e:
logger.warning(f"Unable to create silence: {e}")
await send_text_to_room(
self.client,
self.matrix_client,
self.room.room_id,
f"Sorry, I couldn't create silence for alert with fingerprint {alert_fingerprint}: {e}",
)
@ -161,18 +145,17 @@ class AckAlertCommand(BaseAlertCommand):
except AlertmanagerError as e:
logger.exception(f"Unable to create silence: {e}", exc_info=e)
await send_text_to_room(
self.client,
self.matrix_client,
self.room.room_id,
f"Sorry, I couldn't create silence for alert with fingerprint {alert_fingerprint} "
f"because something went wrong with Alertmanager: {e}",
)
return
self.cache.set(self.event_id, alert_fingerprint, expire=cache_expire_time)
self.cache.set(alert_fingerprint, silence_id, expire=duration_seconds)
self.cache.set(self.event_id, alert_fingerprint, expire=duration_seconds)
await send_text_to_room(
self.client,
self.matrix_client,
self.room.room_id,
f"Created silence with ID {silence_id}.",
)
@ -212,11 +195,11 @@ class UnackAlertCommand(BaseAlertCommand):
)
try:
await self.alertmanager.delete_silence(silence_id)
await self.alertmanager_client.delete_silence(silence_id)
except (SilenceNotFoundError, SilenceExpiredError) as e:
logger.error(f"Unable to delete silence: {e}")
await send_text_to_room(
self.client,
self.matrix_client,
self.room.room_id,
f"Sorry, I couldn't remove silence for alert with fingerprint {alert_fingerprint}: {e}",
)
@ -224,7 +207,7 @@ class UnackAlertCommand(BaseAlertCommand):
except AlertmanagerError as e:
logger.exception(f"Unable to delete silence: {e}", exc_info=e)
await send_text_to_room(
self.client,
self.matrix_client,
self.room.room_id,
f"Sorry, I couldn't remove silence for alert with fingerprint {alert_fingerprint} "
f"because something went wrong with Alertmanager: {e}",
@ -234,7 +217,7 @@ class UnackAlertCommand(BaseAlertCommand):
self.cache.delete(alert_fingerprint)
await send_text_to_room(
self.client,
self.matrix_client,
self.room.room_id,
f"Removed silence with ID {silence_id}.",
)
@ -249,7 +232,7 @@ class HelpCommand(BaseCommand):
"Hello, I am a bot made with matrix-nio! Use `help commands` to view "
"available commands."
)
await send_text_to_room(self.client, self.room.room_id, text)
await send_text_to_room(self.matrix_client, self.room.room_id, text)
return
topic = self.args[0]
@ -259,7 +242,7 @@ class HelpCommand(BaseCommand):
text = "Available commands: ..."
else:
text = "Unknown help topic!"
await send_text_to_room(self.client, self.room.room_id, text)
await send_text_to_room(self.matrix_client, self.room.room_id, text)
class UnknownCommand(BaseCommand):
@ -268,7 +251,7 @@ class UnknownCommand(BaseCommand):
f"Sending unknown command response to room {self.room.display_name}"
)
await send_text_to_room(
self.client,
self.matrix_client,
self.room.room_id,
"Unknown command. Try the 'help' command for more information.",
)

6
matrix_alertbot/errors.py
View file

@ -49,6 +49,12 @@ class SilenceExpiredError(AlertmanagerError):
pass
class SilenceExtendError(AlertmanagerError):
"""An error encountered when a silence cannot be extended."""
pass
class AlertmanagerServerError(AlertmanagerError):
"""An error encountered with Alertmanager server."""

50
matrix_alertbot/main.py
View file

@ -28,7 +28,7 @@ logger = logging.getLogger(__name__)
def create_matrix_client(config: Config) -> AsyncClient:
# Configuration options for the AsyncClient
client_config = AsyncClientConfig(
matrix_client_config = AsyncClientConfig(
max_limit_exceeded=0,
max_timeouts=0,
store_sync_tokens=True,
@ -36,38 +36,38 @@ def create_matrix_client(config: Config) -> AsyncClient:
)
# Initialize the matrix client
client = AsyncClient(
matrix_client = AsyncClient(
config.homeserver_url,
config.user_id,
device_id=config.device_id,
store_path=config.store_dir,
config=client_config,
config=matrix_client_config,
)
if config.user_token:
client.access_token = config.user_token
client.user_id = config.user_id
matrix_client.access_token = config.user_token
matrix_client.user_id = config.user_id
return client
return matrix_client
async def start_matrix_client(
client: AsyncClient, cache: Cache, config: Config
matrix_client: AsyncClient, cache: Cache, config: Config
) -> bool:
# Keep trying to reconnect on failure (with some time in-between)
while True:
try:
if config.user_token:
# Use token to log in
client.load_store()
matrix_client.load_store()
# Sync encryption keys with the server
if client.should_upload_keys:
await client.keys_upload()
if matrix_client.should_upload_keys:
await matrix_client.keys_upload()
else:
# Try to login with the configured username/password
try:
login_response = await client.login(
login_response = await matrix_client.login(
password=config.user_password,
device_name=config.device_name,
)
@ -90,14 +90,14 @@ async def start_matrix_client(
# Login succeeded!
logger.info(f"Logged in as {config.user_id}")
await client.sync_forever(timeout=30000, full_state=True)
await matrix_client.sync_forever(timeout=30000, full_state=True)
except (ClientConnectionError, ServerDisconnectedError, TimeoutError):
logger.warning("Unable to connect to homeserver, retrying in 15s...")
# Sleep so we don't bombard the server with login requests
await asyncio.sleep(15)
finally:
await client.close()
await matrix_client.close()
def main() -> None:
@ -113,30 +113,30 @@ def main() -> None:
# Read the parsed config file and create a Config object
config = Config(config_path)
client = create_matrix_client(config)
matrix_client = create_matrix_client(config)
# Configure the cache
cache = Cache(config.cache_dir)
# Configure Alertmanager client
alertmanager = AlertmanagerClient(config.alertmanager_url, cache)
alertmanager_client = AlertmanagerClient(config.alertmanager_url, cache)
# Set up event callbacks
callbacks = Callbacks(client, alertmanager, cache, config)
client.add_event_callback(callbacks.message, (RoomMessageText,))
client.add_event_callback(
callbacks = Callbacks(matrix_client, alertmanager_client, cache, config)
matrix_client.add_event_callback(callbacks.message, (RoomMessageText,))
matrix_client.add_event_callback(
callbacks.invite_event_filtered_callback, (InviteMemberEvent,)
)
client.add_event_callback(callbacks.decryption_failure, (MegolmEvent,))
client.add_event_callback(callbacks.unknown, (UnknownEvent,))
client.add_event_callback(callbacks.redaction, (RedactionEvent,))
matrix_client.add_event_callback(callbacks.decryption_failure, (MegolmEvent,))
matrix_client.add_event_callback(callbacks.unknown, (UnknownEvent,))
matrix_client.add_event_callback(callbacks.redaction, (RedactionEvent,))
# Configure webhook server
webhook_server = Webhook(client, cache, config)
webhook_server = Webhook(matrix_client, alertmanager_client, cache, config)
loop = asyncio.get_event_loop()
loop.create_task(webhook_server.start())
loop.create_task(start_matrix_client(client, cache, config))
loop.create_task(start_matrix_client(matrix_client, cache, config))
try:
loop.run_forever()
@ -144,6 +144,6 @@ def main() -> None:
logger.error(e)
finally:
loop.run_until_complete(webhook_server.close())
loop.run_until_complete(alertmanager.close())
loop.run_until_complete(client.close())
loop.run_until_complete(alertmanager_client.close())
loop.run_until_complete(matrix_client.close())
cache.close()

89
matrix_alertbot/webhook.py
View file

@ -10,8 +10,14 @@ from diskcache import Cache
from nio import AsyncClient, LocalProtocolError
from matrix_alertbot.alert import Alert, AlertRenderer
from matrix_alertbot.alertmanager import AlertmanagerClient
from matrix_alertbot.chat_functions import send_text_to_room
from matrix_alertbot.config import Config
from matrix_alertbot.errors import (
AlertmanagerError,
SilenceExtendError,
SilenceNotFoundError,
)
logger = logging.getLogger(__name__)
@ -28,10 +34,7 @@ async def create_alerts(request: web_request.Request) -> web.Response:
data = await request.json()
room_id = request.match_info["room_id"]
client: AsyncClient = request.app["client"]
config: Config = request.app["config"]
cache: Cache = request.app["cache"]
alert_renderer: AlertRenderer = request.app["alert_renderer"]
if room_id not in config.allowed_rooms:
logger.error("Cannot send alerts to room ID {room_id}.")
@ -43,33 +46,39 @@ async def create_alerts(request: web_request.Request) -> web.Response:
logger.error("Received data without 'alerts' key")
return web.Response(status=400, body="Data must contain 'alerts' key.")
alerts = data["alerts"]
alert_dicts = data["alerts"]
if not isinstance(data["alerts"], list):
alerts_type = type(alerts).__name__
alerts_type = type(alert_dicts).__name__
logger.error(f"Received data with invalid alerts type '{alerts_type}'.")
return web.Response(
status=400, body=f"Alerts must be a list, got '{alerts_type}'."
)
logger.info(f"Received {len(alerts)} alerts for room ID {room_id}: {data}")
logger.info(f"Received {len(alert_dicts)} alerts for room ID {room_id}: {data}")
if len(data["alerts"]) == 0:
return web.Response(status=400, body="Alerts cannot be empty.")
for alert in alerts:
alerts = []
for alert in alert_dicts:
try:
alert = Alert.from_dict(alert)
except KeyError as e:
logger.error(f"Cannot parse alert dict: {e}")
return web.Response(status=400, body=f"Invalid alert: {alert}.")
alerts.append(alert)
plaintext = alert_renderer.render(alert, html=False)
html = alert_renderer.render(alert, html=True)
for alert in alerts:
try:
event = await send_text_to_room(
client, room_id, plaintext, html, notice=False
await create_alert(alert, room_id, request)
except AlertmanagerError as e:
logger.error(
f"An error occured with Alertmanager when handling alert with fingerprint {alert.fingerprint}: {e}"
)
return web.Response(
status=500,
body=f"An error occured with Alertmanager when handling alert with fingerprint {alert.fingerprint}.",
)
except (LocalProtocolError, ClientError) as e:
logger.error(
@ -80,18 +89,60 @@ async def create_alerts(request: web_request.Request) -> web.Response:
body=f"An error occured when sending alert with fingerprint '{alert.fingerprint}' to Matrix room.",
)
if alert.firing:
cache.set(
event.event_id, alert.fingerprint, expire=config.cache_expire_time
)
return web.Response(status=200)
async def create_alert(
alert: Alert, room_id: str, request: web_request.Request
) -> None:
alertmanager_client: AlertmanagerClient = request.app["alertmanager_client"]
alert_renderer: AlertRenderer = request.app["alert_renderer"]
matrix_client: AsyncClient = request.app["matrix_client"]
cache: Cache = request.app["cache"]
config: Config = request.app["config"]
if alert.firing:
try:
silence_id = await alertmanager_client.update_silence(alert.fingerprint)
logger.debug(
f"Extended silence ID {silence_id} for alert with fingerprint {alert.fingerprint}"
)
return
except SilenceNotFoundError as e:
logger.debug(
f"Unable to extend silence for alert with fingerprint {alert.fingerprint}: {e}"
)
cache.delete(alert.fingerprint)
except SilenceExtendError as e:
logger.debug(
f"Unable to extend silence for alert with fingerprint {alert.fingerprint}: {e}"
)
plaintext = alert_renderer.render(alert, html=False)
html = alert_renderer.render(alert, html=True)
event = await send_text_to_room(
matrix_client, room_id, plaintext, html, notice=False
)
if alert.firing:
cache.set(event.event_id, alert.fingerprint, expire=config.cache_expire_time)
else:
cache.delete(event.event_id)
cache.delete(alert.fingerprint)
class Webhook:
def __init__(self, client: AsyncClient, cache: Cache, config: Config) -> None:
def __init__(
self,
matrix_client: AsyncClient,
alertmanager_client: AlertmanagerClient,
cache: Cache,
config: Config,
) -> None:
self.app = web.Application(logger=logger)
self.app["client"] = client
self.app["matrix_client"] = matrix_client
self.app["alertmanager_client"] = alertmanager_client
self.app["config"] = config
self.app["cache"] = cache
self.app["alert_renderer"] = AlertRenderer(config.template_dir)

489
tests/test_alertmanager.py
View file

@ -3,8 +3,8 @@ from __future__ import annotations
import json
import unittest
from datetime import datetime, timedelta
from typing import Any
from unittest.mock import MagicMock, Mock
from typing import Any, Dict, Optional, Tuple
from unittest.mock import Mock
import aiohttp
import aiohttp.test_utils
@ -18,16 +18,25 @@ from matrix_alertbot.errors import (
AlertmanagerServerError,
AlertNotFoundError,
SilenceExpiredError,
SilenceExtendError,
SilenceNotFoundError,
)
class FakeTimeDelta:
def __init__(self, seconds: int) -> None:
self.seconds = seconds
class FakeCache:
def __init__(self, cache_dict: Optional[Dict] = None) -> None:
if cache_dict is None:
cache_dict = {}
self.cache = cache_dict
def __radd__(self, other: Any) -> datetime:
return datetime.utcfromtimestamp(self.seconds)
def get(
self, key: str, expire_time: bool = False
) -> Optional[Tuple[str, Optional[int]] | str]:
return self.cache.get(key)
def set(self, key: str, value: str, expire: int) -> None:
self.cache[key] = value, expire
print(self.cache)
class AbstractFakeAlertmanagerServer:
@ -42,8 +51,18 @@ class AbstractFakeAlertmanagerServer:
]
)
self.app["silences"] = [
{"id": "silence1", "status": {"state": "active"}},
{"id": "silence2", "status": {"state": "expired"}},
{
"id": "silence1",
"createdBy": "user1",
"status": {"state": "active"},
"matchers": [],
},
{
"id": "silence2",
"createdBy": "user2",
"status": {"state": "expired"},
"matchers": [],
},
]
self.runner = web.AppRunner(self.app)
@ -111,8 +130,9 @@ class FakeAlertmanagerServer(AbstractFakeAlertmanagerServer):
silences = self.app["silences"]
silence = await request.json()
if silence["id"] is None:
silence["id"] = "silence1"
silence_count = len(silences) + 1
silence["id"] = f"silence{silence_count}"
silence["status"] = {"state": "active"}
silences.append(silence)
@ -165,17 +185,17 @@ class FakeAlertmanagerServerWithErrorDeleteSilence(FakeAlertmanagerServer):
class AlertmanagerClientTestCase(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self) -> None:
self.fake_fingerprints = Mock(return_value=["fingerprint1", "fingerprint2"])
self.fake_cache = MagicMock(spec=Cache)
self.fake_cache.__getitem__ = self.fake_fingerprints
async def test_get_alerts_happy(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServer() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
alerts = await alertmanager.get_alerts()
async with aiotools.closing_async(alertmanager_client):
alerts = await alertmanager_client.get_alerts()
self.assertEqual(
[
@ -197,72 +217,94 @@ class AlertmanagerClientTestCase(unittest.IsolatedAsyncioTestCase):
)
async def test_get_alerts_empty(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
alerts = await alertmanager.get_alerts()
async with aiotools.closing_async(alertmanager_client):
alerts = await alertmanager_client.get_alerts()
self.assertEqual([], alerts)
async def test_get_alerts_raise_alertmanager_error(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServerWithErrorAlerts() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
async with aiotools.closing_async(alertmanager_client):
with self.assertRaises(AlertmanagerServerError):
await alertmanager.get_alerts()
await alertmanager_client.get_alerts()
async def test_get_silences_happy(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServer() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
silences = await alertmanager.get_silences()
async with aiotools.closing_async(alertmanager_client):
silences = await alertmanager_client.get_silences()
self.assertEqual(
[
{"id": "silence1", "status": {"state": "active"}},
{"id": "silence2", "status": {"state": "expired"}},
{
"id": "silence1",
"createdBy": "user1",
"status": {"state": "active"},
"matchers": [],
},
{
"id": "silence2",
"createdBy": "user2",
"status": {"state": "expired"},
"matchers": [],
},
],
silences,
)
async def test_get_silences_empty(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
silences = await alertmanager.get_silences()
async with aiotools.closing_async(alertmanager_client):
silences = await alertmanager_client.get_silences()
self.assertEqual([], silences)
async def test_get_silences_raise_alertmanager_error(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServerWithErrorSilences() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
async with aiotools.closing_async(alertmanager_client):
with self.assertRaises(AlertmanagerServerError):
await alertmanager.get_silences()
await alertmanager_client.get_silences()
async def test_get_alert_happy(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServer() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
alert = await alertmanager.get_alert("fingerprint1")
async with aiotools.closing_async(alertmanager_client):
alert = await alertmanager_client.get_alert("fingerprint1")
self.assertEqual(
{
@ -274,76 +316,98 @@ class AlertmanagerClientTestCase(unittest.IsolatedAsyncioTestCase):
)
async def test_get_alert_raise_alert_not_found(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
async with aiotools.closing_async(alertmanager_client):
with self.assertRaises(AlertNotFoundError):
await alertmanager.get_alert("fingerprint1")
await alertmanager_client.get_alert("fingerprint1")
async def test_get_alert_raise_alertmanager_error(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServerWithErrorAlerts() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
async with aiotools.closing_async(alertmanager_client):
with self.assertRaises(AlertmanagerServerError):
await alertmanager.get_alert("fingerprint1")
await alertmanager_client.get_alert("fingerprint1")
async def test_get_silence_happy(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServer() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
silence1 = await alertmanager.get_silence("silence1")
silence2 = await alertmanager.get_silence("silence2")
async with aiotools.closing_async(alertmanager_client):
silence1 = await alertmanager_client.get_silence("silence1")
silence2 = await alertmanager_client.get_silence("silence2")
self.assertEqual(
{"id": "silence1", "status": {"state": "active"}},
{
"id": "silence1",
"createdBy": "user1",
"status": {"state": "active"},
"matchers": [],
},
silence1,
)
self.assertEqual(
{"id": "silence2", "status": {"state": "expired"}},
{
"id": "silence2",
"createdBy": "user2",
"status": {"state": "expired"},
"matchers": [],
},
silence2,
)
async def test_get_silence_raise_silence_not_found(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
async with aiotools.closing_async(alertmanager_client):
with self.assertRaises(SilenceNotFoundError):
await alertmanager.get_silence("silence1")
await alertmanager_client.get_silence("silence1")
async def test_get_silence_raise_alertmanager_error(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServerWithErrorSilences() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
async with aiotools.closing_async(alertmanager_client):
with self.assertRaises(AlertmanagerServerError):
await alertmanager.get_silence("silence1")
await alertmanager_client.get_silence("silence1")
@freeze_time(datetime.utcfromtimestamp(0))
async def test_create_silence(self) -> None:
async def test_create_silence_with_duration(self) -> None:
fake_cache = Mock(return_value=FakeCache())
async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
silence_id = await alertmanager.create_silence(
async with aiotools.closing_async(alertmanager_client):
silence_id = await alertmanager_client.create_silence(
"fingerprint1", "user", 86400
)
silence = await alertmanager.get_silence("silence1")
silence = await alertmanager_client.get_silence("silence1")
self.assertEqual("silence1", silence_id)
self.assertEqual(
@ -365,21 +429,81 @@ class AlertmanagerClientTestCase(unittest.IsolatedAsyncioTestCase):
},
silence,
)
fake_cache.set.assert_called_once_with("fingerprint1", "silence1", expire=86400)
@freeze_time(datetime.utcfromtimestamp(0))
async def test_create_silence_with_id(self) -> None:
async def test_update_silence_with_duration(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
silence_id = await alertmanager.create_silence(
"fingerprint1", "user", 86400, "silence2"
)
silence = await alertmanager.get_silence("silence2")
async with aiotools.closing_async(alertmanager_client):
await alertmanager_client.create_silence("fingerprint1", "user", 86400)
with self.assertRaises(SilenceExtendError):
await alertmanager_client.update_silence("fingerprint1")
with self.assertRaises(SilenceNotFoundError):
await alertmanager_client.get_silence("silence2")
self.assertEqual({"fingerprint1": ("silence1", 86400)}, fake_cache.cache)
self.assertEqual("silence2", silence_id)
@freeze_time(datetime.utcfromtimestamp(0))
async def test_create_silence_without_duration(self) -> None:
fake_cache = Mock(spec=Cache)
fake_cache.get.return_value = None
async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager_client):
silence_id = await alertmanager_client.create_silence(
"fingerprint1", "user"
)
silence = await alertmanager_client.get_silence("silence1")
self.assertEqual("silence1", silence_id)
self.assertEqual(
{
"id": "silence1",
"status": {"state": "active"},
"matchers": [
{
"name": "alertname",
"value": "alert1",
"isRegex": False,
"isEqual": True,
}
],
"createdBy": "user",
"startsAt": "1970-01-01T00:00:00",
"endsAt": "1970-01-01T03:00:00",
"comment": "Acknowledge alert from Matrix",
},
silence,
)
fake_cache.set.assert_called_once_with("fingerprint1", "silence1", expire=None)
@freeze_time(datetime.utcfromtimestamp(0))
async def test_update_silence_without_duration(self) -> None:
fake_cache = FakeCache()
async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager_client):
silence1_id = await alertmanager_client.create_silence(
"fingerprint1", "user"
)
silence2_id = await alertmanager_client.update_silence("fingerprint1")
silence = await alertmanager_client.get_silence("silence2")
self.assertEqual("silence1", silence1_id)
self.assertEqual("silence2", silence2_id)
self.assertEqual(
{
"id": "silence2",
@ -394,56 +518,27 @@ class AlertmanagerClientTestCase(unittest.IsolatedAsyncioTestCase):
],
"createdBy": "user",
"startsAt": "1970-01-01T00:00:00",
"endsAt": "1970-01-02T00:00:00",
"comment": "Acknowledge alert from Matrix",
},
silence,
)
@freeze_time(datetime.utcfromtimestamp(0))
async def test_create_silence_with_indefinite_duration(self) -> None:
async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
)
async with aiotools.closing_async(alertmanager):
silence_id = await alertmanager.create_silence("fingerprint1", "user")
silence = await alertmanager.get_silence("silence1")
self.assertEqual("silence1", silence_id)
self.assertEqual(
{
"id": "silence1",
"status": {"state": "active"},
"matchers": [
{
"name": "alertname",
"value": "alert1",
"isRegex": False,
"isEqual": True,
}
],
"createdBy": "user",
"startsAt": "1970-01-01T00:00:00",
"endsAt": "1980-01-01T00:00:00",
"endsAt": "1970-01-01T03:00:00",
"comment": "Acknowledge alert from Matrix",
},
silence,
)
self.assertEqual({"fingerprint1": ("silence2", None)}, fake_cache.cache)
@freeze_time(datetime.utcfromtimestamp(0))
async def test_create_silence_with_max_duration(self) -> None:
fake_cache = Mock(spec=Cache)
fake_cache.get.return_value = None
async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
silence_id = await alertmanager.create_silence(
"fingerprint1", "user", int(timedelta.max.total_seconds())
async with aiotools.closing_async(alertmanager_client):
silence_id = await alertmanager_client.create_silence(
"fingerprint1", "user", int(timedelta.max.total_seconds()) + 1
)
silence = await alertmanager.get_silence("silence1")
silence = await alertmanager_client.get_silence("silence1")
self.assertEqual("silence1", silence_id)
self.assertEqual(
@ -467,99 +562,165 @@ class AlertmanagerClientTestCase(unittest.IsolatedAsyncioTestCase):
)
async def test_create_silence_raise_alert_not_found(self) -> None:
fake_cache = Mock(spec=Cache)
fake_cache.get.return_value = None
async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
async with aiotools.closing_async(alertmanager_client):
with self.assertRaises(AlertNotFoundError):
await alertmanager.create_silence("fingerprint1", "user")
await alertmanager_client.create_silence("fingerprint1", "user")
async def test_create_silence_raise_alertmanager_error(self) -> None:
fake_cache = Mock(spec=Cache)
fake_cache.get.return_value = None
async with FakeAlertmanagerServerWithErrorCreateSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
await alertmanager.get_alert("fingerprint1")
async with aiotools.closing_async(alertmanager_client):
await alertmanager_client.get_alert("fingerprint1")
with self.assertRaises(AlertmanagerServerError):
await alertmanager.create_silence("fingerprint1", "user")
await alertmanager_client.create_silence("fingerprint1", "user")
async def test_update_silence_raise_silence_not_found(self) -> None:
fake_cache = FakeCache({"fingerprint1": ("silence1", None)})
async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager_client):
with self.assertRaises(SilenceNotFoundError):
await alertmanager_client.update_silence("fingerprint1")
with self.assertRaises(SilenceNotFoundError):
await alertmanager_client.update_silence("fingerprint2")
async def test_update_silence_raise_silence_extend_error(self) -> None:
fake_cache = FakeCache({"fingerprint1": ("silence1", 86400)})
async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager_client):
with self.assertRaises(SilenceExtendError):
await alertmanager_client.update_silence("fingerprint1")
async def test_update_silence_raise_alertmanager_error(self) -> None:
fake_cache = FakeCache({"fingerprint1": ("silence1", None)})
async with FakeAlertmanagerServerWithErrorCreateSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager_client):
await alertmanager_client.get_alert("fingerprint1")
with self.assertRaises(AlertmanagerServerError):
await alertmanager_client.update_silence("fingerprint1")
async def test_delete_silence(self) -> None:
fake_cache = Mock(spec=Cache)
async with FakeAlertmanagerServer() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
await alertmanager.delete_silence("silence1")
silences = await alertmanager.get_silences()
self.assertEqual([{"id": "silence2", "status": {"state": "expired"}}], silences)
async def test_delete_silence_raise_silence_expired(self) -> None:
async with FakeAlertmanagerServer() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
)
async with aiotools.closing_async(alertmanager):
with self.assertRaises(SilenceExpiredError):
await alertmanager.delete_silence("silence2")
silences = await alertmanager.get_silences()
async with aiotools.closing_async(alertmanager_client):
await alertmanager_client.delete_silence("silence1")
silences = await alertmanager_client.get_silences()
self.assertEqual(
[
{"id": "silence1", "status": {"state": "active"}},
{"id": "silence2", "status": {"state": "expired"}},
{
"id": "silence2",
"createdBy": "user2",
"status": {"state": "expired"},
"matchers": [],
}
],
silences,
)
async def test_delete_silence_raise_silence_expired(self) -> None:
fake_cache = Mock(spec=Cache)
async with FakeAlertmanagerServer() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager_client):
with self.assertRaises(SilenceExpiredError):
await alertmanager_client.delete_silence("silence2")
silences = await alertmanager_client.get_silences()
self.assertEqual(2, len(silences))
async def test_delete_silence_raise_alertmanager_error(self) -> None:
fake_cache = Mock(spec=Cache)
async with FakeAlertmanagerServerWithErrorDeleteSilence() as fake_alertmanager_server:
port = fake_alertmanager_server.port
alertmanager = AlertmanagerClient(
f"http://localhost:{port}", self.fake_cache
alertmanager_client = AlertmanagerClient(
f"http://localhost:{port}", fake_cache
)
async with aiotools.closing_async(alertmanager):
await alertmanager.get_alert("fingerprint1")
async with aiotools.closing_async(alertmanager_client):
await alertmanager_client.get_alert("fingerprint1")
with self.assertRaises(AlertmanagerServerError):
await alertmanager.delete_silence("silence1")
await alertmanager_client.delete_silence("silence1")
async def test_find_alert_happy(self) -> None:
alertmanager = AlertmanagerClient("http://localhost", self.fake_cache)
alert = alertmanager._find_alert(
fake_cache = Mock(spec=Cache)
alertmanager_client = AlertmanagerClient("http://localhost", fake_cache)
alert = alertmanager_client._find_alert(
"fingerprint1", [{"fingerprint": "fingerprint1"}]
)
self.assertEqual({"fingerprint": "fingerprint1"}, alert)
async def test_find_alert_raise_alert_not_found(self) -> None:
alertmanager = AlertmanagerClient("http://localhost", self.fake_cache)
fake_cache = Mock(spec=Cache)
alertmanager_client = AlertmanagerClient("http://localhost", fake_cache)
with self.assertRaises(AlertNotFoundError):
alertmanager._find_alert("fingerprint1", [])
alertmanager_client._find_alert("fingerprint1", [])
with self.assertRaises(AlertNotFoundError):
alertmanager._find_alert("fingerprint2", [{"fingerprint": "fingerprint1"}])
alertmanager_client._find_alert(
"fingerprint2", [{"fingerprint": "fingerprint1"}]
)
async def test_find_silence_happy(self) -> None:
alertmanager = AlertmanagerClient("http://localhost", self.fake_cache)
silence = alertmanager._find_silence("silence1", [{"id": "silence1"}])
fake_cache = Mock(spec=Cache)
alertmanager_client = AlertmanagerClient("http://localhost", fake_cache)
silence = alertmanager_client._find_silence("silence1", [{"id": "silence1"}])
self.assertEqual({"id": "silence1"}, silence)
async def test_find_silence_raise_silence_not_found(self) -> None:
alertmanager = AlertmanagerClient("http://localhost", self.fake_cache)
fake_cache = Mock(spec=Cache)
alertmanager_client = AlertmanagerClient("http://localhost", fake_cache)
with self.assertRaises(SilenceNotFoundError):
alertmanager._find_silence("silence1", [])
alertmanager_client._find_silence("silence1", [])
with self.assertRaises(SilenceNotFoundError):
alertmanager._find_silence("silence2", [{"id": "silence1"}])
alertmanager_client._find_silence("silence2", [{"id": "silence1"}])
if __name__ == "__main__":

63
tests/test_callback.py
View file

@ -17,11 +17,11 @@ from tests.utils import make_awaitable
class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
def setUp(self) -> None:
# Create a Callbacks object and give it some Mock'd objects to use
self.fake_client = Mock(spec=nio.AsyncClient)
self.fake_client.user = "@fake_user:example.com"
self.fake_matrix_client = Mock(spec=nio.AsyncClient)
self.fake_matrix_client.user = "@fake_user:example.com"
self.fake_cache = MagicMock(spec=Cache)
self.fake_alertmanager = Mock(spec=AlertmanagerClient)
self.fake_alertmanager_client = Mock(spec=AlertmanagerClient)
# Create a fake room to play with
self.fake_room = Mock(spec=nio.MatrixRoom)
@ -35,7 +35,10 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_config.command_prefix = "!alert "
self.callbacks = Callbacks(
self.fake_client, self.fake_alertmanager, self.fake_cache, self.fake_config
self.fake_matrix_client,
self.fake_alertmanager_client,
self.fake_cache,
self.fake_config,
)
async def test_invite(self) -> None:
@ -45,13 +48,13 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
fake_invite_event.sender = "@some_other_fake_user:example.com"
# Pretend that attempting to join a room is always successful
self.fake_client.join.return_value = make_awaitable(None)
self.fake_matrix_client.join.return_value = make_awaitable(None)
# Pretend that we received an invite event
await self.callbacks.invite(self.fake_room, fake_invite_event)
# Check that we attempted to join the room
self.fake_client.join.assert_called_once_with(self.fake_room.room_id)
self.fake_matrix_client.join.assert_called_once_with(self.fake_room.room_id)
@patch.object(matrix_alertbot.callback.CommandFactory, "create", autospec=True)
async def test_message_without_prefix(self, fake_command_create: Mock) -> None:
@ -84,9 +87,9 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Check that the command was not executed
fake_command.assert_called_with(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
fake_message_event.sender,
@ -115,9 +118,9 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Check that we attempted to execute the command
fake_command.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
fake_message_event.sender,
@ -132,7 +135,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Tests that the bot process messages in the room that contain a command
fake_message_event = Mock(spec=nio.RoomMessageText)
fake_message_event.sender = self.fake_client.user
fake_message_event.sender = self.fake_matrix_client.user
# Pretend that we received a text message event
await self.callbacks.message(self.fake_room, fake_message_event)
@ -195,9 +198,9 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Check that the command was not executed
fake_command.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
fake_message_event.sender,
@ -244,9 +247,9 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Check that the command was not executed
fake_command.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
fake_message_event.sender,
@ -280,7 +283,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
fake_event_response = Mock(spec=nio.RoomGetEventResponse)
fake_event_response.event = fake_alert_event
self.fake_client.room_get_event.return_value = make_awaitable(
self.fake_matrix_client.room_get_event.return_value = make_awaitable(
fake_event_response
)
@ -289,9 +292,9 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Check that we attempted to execute the command
fake_command.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
fake_reaction_event.sender,
@ -299,7 +302,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
"some alert event id",
)
fake_command.return_value.process.assert_called_once()
self.fake_client.room_get_event.assert_called_once_with(
self.fake_matrix_client.room_get_event.assert_called_once_with(
self.fake_room.room_id, fake_alert_event.event_id
)
@ -324,7 +327,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
}
fake_event_response = Mock(spec=nio.RoomGetEventError)
self.fake_client.room_get_event.return_value = make_awaitable(
self.fake_matrix_client.room_get_event.return_value = make_awaitable(
fake_event_response
)
@ -334,7 +337,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Check that we attempted to execute the command
fake_command.assert_not_called()
self.fake_cache.set.assert_not_called()
self.fake_client.room_get_event.assert_called_once_with(
self.fake_matrix_client.room_get_event.assert_called_once_with(
self.fake_room.room_id, fake_alert_event_id
)
@ -364,7 +367,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
fake_event_response = Mock(spec=nio.RoomGetEventResponse)
fake_event_response.event = fake_alert_event
self.fake_client.room_get_event.return_value = make_awaitable(
self.fake_matrix_client.room_get_event.return_value = make_awaitable(
fake_event_response
)
@ -374,7 +377,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Check that we attempted to execute the command
fake_command.assert_not_called()
self.fake_cache.set.assert_not_called()
self.fake_client.room_get_event.assert_called_once_with(
self.fake_matrix_client.room_get_event.assert_called_once_with(
self.fake_room.room_id, fake_alert_event.event_id
)
@ -403,7 +406,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Check that we attempted to execute the command
fake_command.assert_not_called()
self.fake_client.room_get_event.assert_not_called()
self.fake_matrix_client.room_get_event.assert_not_called()
@patch.object(matrix_alertbot.callback, "AckAlertCommand", autospec=True)
async def test_ignore_reaction_sent_by_bot_user(self, fake_command: Mock) -> None:
@ -414,7 +417,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
fake_reaction_event = Mock(spec=nio.UnknownEvent)
fake_reaction_event.type = "m.reaction"
fake_reaction_event.event_id = "some event id"
fake_reaction_event.sender = self.fake_client.user
fake_reaction_event.sender = self.fake_matrix_client.user
fake_reaction_event.source = {
"content": {
"m.relates_to": {
@ -433,7 +436,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Check that we attempted to execute the command
fake_command.assert_not_called()
self.fake_client.room_get_event.assert_not_called()
self.fake_matrix_client.room_get_event.assert_not_called()
@patch.object(matrix_alertbot.callback, "AckAlertCommand", autospec=True)
async def test_ignore_reaction_in_unauthorized_room(
@ -467,7 +470,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Check that we attempted to execute the command
fake_command.assert_not_called()
self.fake_client.room_get_event.assert_not_called()
self.fake_matrix_client.room_get_event.assert_not_called()
@patch.object(matrix_alertbot.callback, "UnackAlertCommand", autospec=True)
async def test_redaction(self, fake_command: Mock) -> None:
@ -488,9 +491,9 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
# Check that we attempted to execute the command
fake_command.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
fake_redaction_event.sender,
@ -504,7 +507,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
"""Tests the callback for RoomMessageText with the command prefix"""
# Tests that the bot process messages in the room that contain a command
fake_redaction_event = Mock(spec=nio.RedactionEvent)
fake_redaction_event.sender = self.fake_client.user
fake_redaction_event.sender = self.fake_matrix_client.user
fake_cache_dict: Dict = {}
self.fake_cache.__getitem__.side_effect = fake_cache_dict.__getitem__

248
tests/test_command.py
View file

@ -32,10 +32,7 @@ def cache_get_item(key: str) -> str:
async def create_silence(
fingerprint: str,
user: str,
seconds: Optional[int] = None,
silence_id: Optional[str] = None,
fingerprint: str, user: str, seconds: Optional[int] = None
) -> str:
if fingerprint == "fingerprint1":
return "silence1"
@ -45,10 +42,7 @@ async def create_silence(
async def create_silence_raise_alertmanager_error(
fingerprint: str,
user: str,
seconds: Optional[int] = None,
silence_id: Optional[str] = None,
fingerprint: str, user: str, seconds: Optional[int] = None
) -> str:
if fingerprint == "fingerprint1":
raise AlertmanagerError
@ -56,10 +50,7 @@ async def create_silence_raise_alertmanager_error(
async def create_silence_raise_alert_not_found_error(
fingerprint: str,
user: str,
seconds: Optional[int] = None,
silence_id: Optional[str] = None,
fingerprint: str, user: str, seconds: Optional[int] = None
) -> str:
if fingerprint == "fingerprint1":
raise AlertNotFoundError
@ -79,17 +70,17 @@ async def delete_silence_raise_silence_not_found_error(silence_id: str) -> None:
class CommandTestCase(unittest.IsolatedAsyncioTestCase):
def setUp(self) -> None:
# Create a Command object and give it some Mock'd objects to use
self.fake_client = Mock(spec=nio.AsyncClient)
self.fake_client.user = "@fake_user:example.com"
self.fake_matrix_client = Mock(spec=nio.AsyncClient)
self.fake_matrix_client.user = "@fake_user:example.com"
# Pretend that attempting to send a message is always successful
self.fake_client.room_send.return_value = make_awaitable(None)
self.fake_matrix_client.room_send.return_value = make_awaitable(None)
self.fake_cache = MagicMock(spec=Cache)
self.fake_cache.__getitem__.side_effect = cache_get_item
self.fake_cache.__contains__.return_value = True
self.fake_alertmanager = Mock(spec=AlertmanagerClient)
self.fake_alertmanager.create_silence.side_effect = create_silence
self.fake_alertmanager_client = Mock(spec=AlertmanagerClient)
self.fake_alertmanager_client.create_silence.side_effect = create_silence
# Create a fake room to play with
self.fake_room = Mock(spec=nio.MatrixRoom)
@ -113,9 +104,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
command = CommandFactory.create(
"ack",
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -134,9 +125,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
command = CommandFactory.create(
"ack 1w 3d",
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -156,9 +147,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
for unack_cmd in ("unack", "nack"):
command = CommandFactory.create(
unack_cmd,
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -177,9 +168,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
command = CommandFactory.create(
"help",
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -197,9 +188,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
command = CommandFactory.create(
"",
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -219,12 +210,11 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
}
self.fake_cache.__getitem__.side_effect = fake_cache_dict.__getitem__
self.fake_cache.get.side_effect = fake_cache_dict.get
command = AckAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -234,25 +224,17 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.create_silence.assert_called_once_with(
"fingerprint1", self.fake_sender, None, None
self.fake_alertmanager_client.create_silence.assert_called_once_with(
"fingerprint1", self.fake_sender, None
)
fake_send_text_to_room.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_room.room_id,
"Created silence with ID silence1.",
)
self.fake_cache.__getitem__.assert_called_once_with(self.fake_alert_event_id)
self.fake_cache.get.assert_called_once_with("fingerprint1")
self.fake_cache.set.assert_has_calls(
[
call(
"some event id",
"fingerprint1",
expire=self.fake_config.cache_expire_time,
),
call("fingerprint1", "silence1", expire=None),
]
self.fake_cache.set.assert_called_once_with(
"some event id", "fingerprint1", expire=None
)
@patch.object(matrix_alertbot.command, "send_text_to_room")
@ -264,12 +246,11 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
}
self.fake_cache.__getitem__.side_effect = fake_cache_dict.__getitem__
self.fake_cache.get.side_effect = fake_cache_dict.get
command = AckAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -280,21 +261,17 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.create_silence.assert_called_once_with(
"fingerprint1", self.fake_sender, 864000, None
self.fake_alertmanager_client.create_silence.assert_called_once_with(
"fingerprint1", self.fake_sender, 864000
)
fake_send_text_to_room.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_room.room_id,
"Created silence with ID silence1.",
)
self.fake_cache.__getitem__.assert_called_once_with(self.fake_alert_event_id)
self.fake_cache.get.assert_called_once_with("fingerprint1")
self.fake_cache.set.assert_has_calls(
[
call("some event id", "fingerprint1", expire=864000),
call("fingerprint1", "silence1", expire=864000),
]
self.fake_cache.set.assert_called_once_with(
"some event id", "fingerprint1", expire=864000
)
@patch.object(matrix_alertbot.command, "send_text_to_room")
@ -308,12 +285,11 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
}
self.fake_cache.__getitem__.side_effect = fake_cache_dict.__getitem__
self.fake_cache.get.side_effect = fake_cache_dict.get
command = AckAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -321,22 +297,21 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_alert_event_id,
)
self.fake_alertmanager.create_silence.side_effect = (
self.fake_alertmanager_client.create_silence.side_effect = (
create_silence_raise_alertmanager_error
)
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.create_silence.assert_called_once_with(
"fingerprint1", self.fake_sender, None, None
self.fake_alertmanager_client.create_silence.assert_called_once_with(
"fingerprint1", self.fake_sender, None
)
fake_send_text_to_room.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_room.room_id,
"Sorry, I couldn't create silence for alert with fingerprint fingerprint1 because something went wrong with Alertmanager: ",
)
self.fake_cache.__getitem__.assert_called_once_with(self.fake_alert_event_id)
self.fake_cache.get.assert_called_once_with("fingerprint1")
self.fake_cache.set.assert_not_called()
@patch.object(matrix_alertbot.command, "send_text_to_room")
@ -350,12 +325,11 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
}
self.fake_cache.__getitem__.side_effect = fake_cache_dict.__getitem__
self.fake_cache.get.side_effect = fake_cache_dict.get
command = AckAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -363,22 +337,21 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_alert_event_id,
)
self.fake_alertmanager.create_silence.side_effect = (
self.fake_alertmanager_client.create_silence.side_effect = (
create_silence_raise_alert_not_found_error
)
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.create_silence.assert_called_once_with(
"fingerprint1", self.fake_sender, None, None
self.fake_alertmanager_client.create_silence.assert_called_once_with(
"fingerprint1", self.fake_sender, None
)
fake_send_text_to_room.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_room.room_id,
"Sorry, I couldn't create silence for alert with fingerprint fingerprint1: ",
)
self.fake_cache.__getitem__.assert_called_once_with(self.fake_alert_event_id)
self.fake_cache.get.assert_called_once_with("fingerprint1")
self.fake_cache.set.assert_not_called()
@patch.object(matrix_alertbot.command, "send_text_to_room")
@ -388,9 +361,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
"""Tests the callback for InviteMemberEvents"""
# Tests that the bot attempts to join a room after being invited to it
command = AckAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -402,9 +375,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.create_silence.assert_not_called()
self.fake_alertmanager_client.create_silence.assert_not_called()
fake_send_text_to_room.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_room.room_id,
"I tried really hard, but I can't convert the duration 'invalid duration' to a number of seconds.",
)
@ -419,9 +392,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
"""Tests the callback for InviteMemberEvents"""
# Tests that the bot attempts to join a room after being invited to it
command = AckAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -433,9 +406,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.create_silence.assert_not_called()
self.fake_alertmanager_client.create_silence.assert_not_called()
fake_send_text_to_room.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_room.room_id,
"I can't create a silence with a negative duration!",
)
@ -455,9 +428,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_cache.get.side_effect = fake_cache_dict.get
command = AckAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -468,61 +441,12 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.create_silence.assert_not_called()
self.fake_alertmanager_client.create_silence.assert_not_called()
fake_send_text_to_room.assert_not_called()
self.fake_cache.__getitem__.assert_called_once_with("some alert event id")
self.fake_cache.get.assert_not_called()
self.fake_cache.set.assert_not_called()
@patch.object(matrix_alertbot.command, "send_text_to_room")
async def test_ack_with_silence_in_cache(
self, fake_send_text_to_room: Mock
) -> None:
"""Tests the callback for InviteMemberEvents"""
# Tests that the bot attempts to join a room after being invited to it
fake_cache_dict = {
self.fake_alert_event_id: "fingerprint1",
"fingerprint1": "silence2",
}
self.fake_cache.__getitem__.side_effect = fake_cache_dict.__getitem__
self.fake_cache.get.side_effect = fake_cache_dict.get
command = AckAlertCommand(
self.fake_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_config,
self.fake_room,
self.fake_sender,
self.fake_event_id,
self.fake_alert_event_id,
)
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.create_silence.assert_called_once_with(
"fingerprint1", self.fake_sender, None, "silence2"
)
fake_send_text_to_room.assert_called_once_with(
self.fake_client,
self.fake_room.room_id,
"Created silence with ID silence1.",
)
self.fake_cache.__getitem__.assert_called_once_with("some alert event id")
self.fake_cache.get.assert_called_once_with("fingerprint1")
self.fake_cache.set.assert_has_calls(
[
call(
self.fake_event_id,
"fingerprint1",
expire=self.fake_config.cache_expire_time,
),
call("fingerprint1", "silence1", expire=None),
]
)
@patch.object(matrix_alertbot.command, "send_text_to_room")
async def test_unack(self, fake_send_text_to_room: Mock) -> None:
"""Tests the callback for InviteMemberEvents"""
@ -535,9 +459,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_cache.__getitem__.side_effect = fake_cache_dict.__getitem__
command = UnackAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -547,9 +471,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.delete_silence.assert_called_once_with("silence1")
self.fake_alertmanager_client.delete_silence.assert_called_once_with("silence1")
fake_send_text_to_room.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_room.room_id,
"Removed silence with ID silence1.",
)
@ -571,9 +495,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_cache.__getitem__.side_effect = fake_cache_dict.__getitem__
command = UnackAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -581,15 +505,15 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_alert_event_id,
)
self.fake_alertmanager.delete_silence.side_effect = (
self.fake_alertmanager_client.delete_silence.side_effect = (
delete_silence_raise_alertmanager_error
)
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.delete_silence.assert_called_once_with("silence1")
self.fake_alertmanager_client.delete_silence.assert_called_once_with("silence1")
fake_send_text_to_room.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_room.room_id,
"Sorry, I couldn't remove silence for alert with fingerprint fingerprint1 because something went wrong with Alertmanager: ",
)
@ -611,9 +535,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_cache.__getitem__.side_effect = fake_cache_dict.__getitem__
command = UnackAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -621,15 +545,15 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_alert_event_id,
)
self.fake_alertmanager.delete_silence.side_effect = (
self.fake_alertmanager_client.delete_silence.side_effect = (
delete_silence_raise_silence_not_found_error
)
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.delete_silence.assert_called_once_with("silence1")
self.fake_alertmanager_client.delete_silence.assert_called_once_with("silence1")
fake_send_text_to_room.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_room.room_id,
"Sorry, I couldn't remove silence for alert with fingerprint fingerprint1: ",
)
@ -648,9 +572,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_cache.__getitem__.side_effect = fake_cache_dict.__getitem__
command = UnackAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -661,7 +585,7 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.delete_silence.assert_not_called()
self.fake_alertmanager_client.delete_silence.assert_not_called()
fake_send_text_to_room.assert_not_called()
self.fake_cache.__getitem__.assert_called_once_with(self.fake_alert_event_id)
@ -676,9 +600,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_cache.__getitem__.side_effect = fake_cache_dict.__getitem__
command = UnackAlertCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -689,7 +613,7 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
await command.process()
# Check that we attempted to create silences
self.fake_alertmanager.delete_silence.assert_not_called()
self.fake_alertmanager_client.delete_silence.assert_not_called()
fake_send_text_to_room.assert_not_called()
self.fake_cache.__getitem__.assert_has_calls(
[call(self.fake_alert_event_id), call("fingerprint1")]
@ -701,9 +625,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
# Tests that the bot attempts to join a room after being invited to it
command = HelpCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -723,9 +647,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
# Tests that the bot attempts to join a room after being invited to it
command = HelpCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -746,9 +670,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
# Tests that the bot attempts to join a room after being invited to it
command = HelpCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -769,9 +693,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
# Tests that the bot attempts to join a room after being invited to it
command = HelpCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -792,9 +716,9 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
# Tests that the bot attempts to join a room after being invited to it
command = UnknownCommand(
self.fake_client,
self.fake_matrix_client,
self.fake_cache,
self.fake_alertmanager,
self.fake_alertmanager_client,
self.fake_config,
self.fake_room,
self.fake_sender,
@ -805,7 +729,7 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
# Check that we attempted to create silences
fake_send_text_to_room.assert_called_once_with(
self.fake_client,
self.fake_matrix_client,
self.fake_room.room_id,
"Unknown command. Try the 'help' command for more information.",
)

183
tests/test_webhook.py
View file

@ -9,7 +9,13 @@ from diskcache import Cache
from nio import LocalProtocolError, RoomSendResponse
import matrix_alertbot.webhook
from matrix_alertbot.alertmanager import AlertmanagerClient
from matrix_alertbot.config import Config
from matrix_alertbot.errors import (
AlertmanagerError,
SilenceExtendError,
SilenceNotFoundError,
)
from matrix_alertbot.webhook import Webhook
@ -19,9 +25,22 @@ def send_text_to_room_raise_error(
raise LocalProtocolError()
def update_silence_raise_silence_not_found(fingerprint: str) -> str:
raise SilenceNotFoundError
def update_silence_raise_silence_extend_error(fingerprint: str) -> str:
raise SilenceExtendError
def update_silence_raise_alertmanager_error(fingerprint: str) -> str:
raise AlertmanagerError
class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
async def get_application(self) -> web.Application:
self.fake_client = Mock(spec=nio.AsyncClient)
self.fake_matrix_client = Mock(spec=nio.AsyncClient)
self.fake_alertmanager_client = Mock(spec=AlertmanagerClient)
self.fake_cache = Mock(spec=Cache)
self.fake_room_id = "!abcdefg:example.com"
@ -61,21 +80,36 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
]
}
webhook = Webhook(self.fake_client, self.fake_cache, self.fake_config)
webhook = Webhook(
self.fake_matrix_client,
self.fake_alertmanager_client,
self.fake_cache,
self.fake_config,
)
return webhook.app
@patch.object(matrix_alertbot.webhook, "send_text_to_room")
async def test_post_alerts(self, fake_send_text_to_room: Mock) -> None:
async def test_post_alerts_with_silence_not_found(
self, fake_send_text_to_room: Mock
) -> None:
self.fake_alertmanager_client.update_silence.side_effect = (
update_silence_raise_silence_not_found
)
data = self.fake_alerts
async with self.client.request(
"POST", f"/alerts/{self.fake_room_id}", json=data
) as response:
self.assertEqual(200, response.status)
self.fake_alertmanager_client.update_silence.assert_called_once_with(
"fingerprint1"
)
self.assertEqual(2, fake_send_text_to_room.call_count)
fake_send_text_to_room.assert_has_calls(
[
call(
self.fake_client,
self.fake_matrix_client,
self.fake_room_id,
"[🔥 CRITICAL] alert1: some description1",
'<font color="#dc3545">\n <b>[🔥 CRITICAL]</b>\n</font> '
@ -84,7 +118,7 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
notice=False,
),
call(
self.fake_client,
self.fake_matrix_client,
self.fake_room_id,
"[🥦 RESOLVED] alert2: some description2",
'<font color="#33cc33">\n <b>[🥦 RESOLVED]</b>\n</font> '
@ -99,6 +133,118 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
"fingerprint1",
expire=self.fake_config.cache_expire_time,
)
self.assertEqual(3, self.fake_cache.delete.call_count)
self.fake_cache.delete.assert_has_calls(
[
call("fingerprint1"),
call(fake_send_text_to_room.return_value.event_id),
call("fingerprint2"),
]
)
@patch.object(matrix_alertbot.webhook, "send_text_to_room")
async def test_post_alerts_with_silence_extend_error(
self, fake_send_text_to_room: Mock
) -> None:
self.fake_alertmanager_client.update_silence.side_effect = (
update_silence_raise_silence_extend_error
)
data = self.fake_alerts
async with self.client.request(
"POST", f"/alerts/{self.fake_room_id}", json=data
) as response:
self.assertEqual(200, response.status)
self.fake_alertmanager_client.update_silence.assert_called_once_with(
"fingerprint1"
)
self.assertEqual(2, fake_send_text_to_room.call_count)
fake_send_text_to_room.assert_has_calls(
[
call(
self.fake_matrix_client,
self.fake_room_id,
"[🔥 CRITICAL] alert1: some description1",
'<font color="#dc3545">\n <b>[🔥 CRITICAL]</b>\n</font> '
'<a href="http://example.com/alert1">alert1</a>\n (job1)<br/>\n'
"some description1",
notice=False,
),
call(
self.fake_matrix_client,
self.fake_room_id,
"[🥦 RESOLVED] alert2: some description2",
'<font color="#33cc33">\n <b>[🥦 RESOLVED]</b>\n</font> '
'<a href="http://example.com/alert2">alert2</a>\n (job2)<br/>\n'
"some description2",
notice=False,
),
]
)
self.fake_cache.set.assert_called_once_with(
fake_send_text_to_room.return_value.event_id,
"fingerprint1",
expire=self.fake_config.cache_expire_time,
)
self.assertEqual(2, self.fake_cache.delete.call_count)
self.fake_cache.delete.assert_has_calls(
[call(fake_send_text_to_room.return_value.event_id), call("fingerprint2")]
)
@patch.object(matrix_alertbot.webhook, "send_text_to_room")
async def test_post_alerts_with_alertmanager_error(
self, fake_send_text_to_room: Mock
) -> None:
self.fake_alertmanager_client.update_silence.side_effect = (
update_silence_raise_alertmanager_error
)
data = self.fake_alerts
async with self.client.request(
"POST", f"/alerts/{self.fake_room_id}", json=data
) as response:
self.assertEqual(500, response.status)
self.fake_alertmanager_client.update_silence.assert_called_once_with(
"fingerprint1"
)
fake_send_text_to_room.assert_not_called()
self.fake_cache.set.assert_not_called()
self.fake_cache.delete.assert_not_called()
@patch.object(matrix_alertbot.webhook, "send_text_to_room")
async def test_post_alerts_with_existing_silence(
self, fake_send_text_to_room: Mock
) -> None:
self.fake_alertmanager_client.update_silence.return_value = "silence1"
data = self.fake_alerts
async with self.client.request(
"POST", f"/alerts/{self.fake_room_id}", json=data
) as response:
self.assertEqual(200, response.status)
self.fake_alertmanager_client.update_silence.assert_called_once_with(
"fingerprint1"
)
fake_send_text_to_room.assert_called_once_with(
self.fake_matrix_client,
self.fake_room_id,
"[🥦 RESOLVED] alert2: some description2",
'<font color="#33cc33">\n <b>[🥦 RESOLVED]</b>\n</font> '
'<a href="http://example.com/alert2">alert2</a>\n (job2)<br/>\n'
"some description2",
notice=False,
)
self.fake_cache.set.assert_not_called()
self.assertEqual(2, self.fake_cache.delete.call_count)
self.fake_cache.delete.assert_has_calls(
[
call(fake_send_text_to_room.return_value.event_id),
call("fingerprint2"),
]
)
@patch.object(matrix_alertbot.webhook, "send_text_to_room")
async def test_post_alerts_in_unauthorized_room(
@ -116,6 +262,7 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
)
fake_send_text_to_room.assert_not_called()
self.fake_cache.set.assert_not_called()
self.fake_cache.delete.assert_not_called()
@patch.object(matrix_alertbot.webhook, "send_text_to_room")
async def test_post_alerts_with_empty_data(
@ -130,6 +277,7 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
self.assertEqual("Data must contain 'alerts' key.", error_msg)
fake_send_text_to_room.assert_not_called()
self.fake_cache.set.assert_not_called()
self.fake_cache.delete.assert_not_called()
@patch.object(matrix_alertbot.webhook, "send_text_to_room")
async def test_post_empty_alerts(self, fake_send_text_to_room: Mock) -> None:
@ -143,6 +291,7 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
self.assertEqual("Alerts cannot be empty.", error_msg)
fake_send_text_to_room.assert_not_called()
self.fake_cache.set.assert_not_called()
self.fake_cache.delete.assert_not_called()
@patch.object(matrix_alertbot.webhook, "send_text_to_room")
async def test_post_invalid_alerts(self, fake_send_text_to_room: Mock) -> None:
@ -156,6 +305,7 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
self.assertEqual("Alerts must be a list, got 'str'.", error_msg)
fake_send_text_to_room.assert_not_called()
self.fake_cache.set.assert_not_called()
self.fake_cache.delete.assert_not_called()
@patch.object(matrix_alertbot.webhook, "send_text_to_room")
async def test_post_alerts_with_empty_items(
@ -171,6 +321,7 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
self.assertEqual("Invalid alert: {}.", error_msg)
fake_send_text_to_room.assert_not_called()
self.fake_cache.set.assert_not_called()
self.fake_cache.delete.assert_not_called()
@patch.object(
matrix_alertbot.webhook,
@ -180,6 +331,10 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
async def test_post_alerts_raise_send_error(
self, fake_send_text_to_room: Mock
) -> None:
self.fake_alertmanager_client.update_silence.side_effect = (
update_silence_raise_silence_not_found
)
data = self.fake_alerts
async with self.client.request(
"POST", f"/alerts/{self.fake_room_id}", json=data
@ -193,6 +348,7 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
)
fake_send_text_to_room.assert_called_once()
self.fake_cache.set.assert_not_called()
self.fake_cache.delete.assert_called_once_with("fingerprint1")
async def test_health(self) -> None:
async with self.client.request("GET", "/health") as response:
@ -205,7 +361,8 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
class WebhookServerTestCase(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self) -> None:
self.fake_client = Mock(spec=nio.AsyncClient)
self.fake_matrix_client = Mock(spec=nio.AsyncClient)
self.fake_alertmanager_client = Mock(spec=AlertmanagerClient)
self.fake_cache = Mock(spec=Cache)
self.fake_config = Mock(spec=Config)
@ -217,7 +374,12 @@ class WebhookServerTestCase(unittest.IsolatedAsyncioTestCase):
@patch.object(matrix_alertbot.webhook.web, "TCPSite", autospec=True)
async def test_webhook_start_address_port(self, fake_tcp_site: Mock) -> None:
webhook = Webhook(self.fake_client, self.fake_cache, self.fake_config)
webhook = Webhook(
self.fake_matrix_client,
self.fake_alertmanager_client,
self.fake_cache,
self.fake_config,
)
await webhook.start()
fake_tcp_site.assert_called_once_with(
@ -231,7 +393,12 @@ class WebhookServerTestCase(unittest.IsolatedAsyncioTestCase):
self.fake_config.address = None
self.fake_config.port = None
webhook = Webhook(self.fake_client, self.fake_cache, self.fake_config)
webhook = Webhook(
self.fake_matrix_client,
self.fake_alertmanager_client,
self.fake_cache,
self.fake_config,
)
await webhook.start()
fake_unix_site.assert_called_once_with(webhook.runner, self.fake_config.socket)