Add a command to test email configuration.

argos/commands.py

@@ -10,8 +10,7 @@ import uvicorn
 from alembic import command
 from alembic.config import Config
 
-from argos import logging
-from argos import VERSION
+from argos import VERSION, logging
 from argos.agent import ArgosAgent
 
 
@@ -227,7 +226,8 @@ async def watch_agents(time_without_agent, config):
     db = await get_db()
     agents = await queries.get_recent_agents_count(db, time_without_agent)
     if agents == 0:
-        click.echo(f"No agent has been seen in the last {time_without_agent} minutes.")
+        click.echo(
+            f"No agent has been seen in the last {time_without_agent} minutes.")
         sysexit(1)
 
 
@@ -305,9 +305,10 @@ async def add(config, name, password):
     os.environ["ARGOS_YAML_FILE"] = config
 
     # The imports are made here otherwise the agent will need server configuration files.
-    from argos.server import queries
     from passlib.context import CryptContext
 
+    from argos.server import queries
+
     db = await get_db()
     _user = await queries.get_user(db, name)
     if _user is not None:
@@ -339,9 +340,10 @@ async def change_password(config, name, password):
     os.environ["ARGOS_YAML_FILE"] = config
 
     # The imports are made here otherwise the agent will need server configuration files.
-    from argos.server import queries
     from passlib.context import CryptContext
 
+    from argos.server import queries
+
     db = await get_db()
     _user = await queries.get_user(db, name)
     if _user is None:
@@ -374,9 +376,10 @@ async def verify_password(config, name, password):
     os.environ["ARGOS_YAML_FILE"] = config
 
     # The imports are made here otherwise the agent will need server configuration files.
-    from argos.server import queries
     from passlib.context import CryptContext
 
+    from argos.server import queries
+
     db = await get_db()
     _user = await queries.get_user(db, name)
     if _user is None:
@@ -521,7 +524,8 @@ async def show(config):
         status = "✅"
         if _user.disabled:
             status = "❌"
-        click.echo(f"{status} {_user.username}, last login: {_user.last_login_at}")
+        click.echo(
+            f"{status} {_user.username}, last login: {_user.last_login_at}")
 
 
 @server.command(short_help="Generate a token for agents")
@@ -548,5 +552,70 @@ async def generate_config():
         print(f.read())
 
 
+@server.command()
+@click.option(
+    "--config",
+    default="argos-config.yaml",
+    help="Path of the configuration file. "
+    "If ARGOS_YAML_FILE environment variable is set, its value will be used instead.",
+    envvar="ARGOS_YAML_FILE",
+    callback=validate_config_access,
+)
+@click.option("--domain", help="Domain for the notification", default="example.org")
+@click.option("--severity", help="Severity", default="CRITICAL")
+@coroutine
+async def test_mail(config, domain, severity):
+    """Send a test email"""
+    os.environ["ARGOS_YAML_FILE"] = config
+
+    from datetime import datetime
+
+    from argos.logging import set_log_level
+    from argos.server.alerting import notify_by_mail
+    from argos.server.main import read_config
+    from argos.server.models import Result, Task
+
+    conf = read_config(config)
+
+    if not conf.general.mail:
+        click.echo("Mail is not configured, cannot test", err=True)
+        sysexit(1)
+        return
+
+    now = datetime.now()
+    task = Task(
+        url=f"https://{domain}",
+        domain=domain,
+        check="body-contains",
+        expected="foo",
+        frequency=1,
+        selected_by="test",
+        selected_at=now,
+    )
+
+    result = Result(
+        submitted_at=now,
+        status="success",
+        context={"foo": "bar"},
+        task=task,
+        agent_id="test",
+        severity="ok",
+    )
+
+    class _FalseRequest:
+        def url_for(*args, **kwargs):
+            return "/url"
+
+    set_log_level("debug")
+    notify_by_mail(
+        result,
+        task,
+        severity="SEVERITY",
+        old_severity="OLD SEVERITY",
+        config=conf.general.mail,
+        request=_FalseRequest(),
+    )
+
+
 if __name__ == "__main__":
     cli()