Aegis/backend/tests/test_notification_email_coverage.py

"""Coverage for the remaining notification-email hooks wired this session:
stale coverage alerts, campaign assignment, generic test-state-change,
all-team validations, webhook delivery failures, new user registration,
and background-job system errors (see notification_service.notify_roles_by_email
and its call sites).
"""

import uuid
from types import SimpleNamespace
from unittest.mock import patch

from app.services.notification_service import notify_roles_by_email


class _FakeUser:
    def __init__(self, user_id=None, email="user@test.com", full_name="A User", role="admin", prefs=None):
        self.id = user_id or uuid.uuid4()
        self.email = email
        self.full_name = full_name
        self.role = role
        self.notification_preferences = prefs


class _FakeQuery:
    def __init__(self, rows):
        self._rows = rows

    def filter(self, *args, **kwargs):
        return self

    def all(self):
        return self._rows


def test_notify_roles_by_email_filters_by_role_and_preference(db):
    lead = _FakeUser(role="red_lead", email="lead@test.com")
    other_role = _FakeUser(role="viewer", email="viewer@test.com")
    opted_out = _FakeUser(role="blue_lead", email="opted-out@test.com", prefs={"email_on_all_team_validations": False})

    with patch.object(db, "query", return_value=_FakeQuery([lead, other_role, opted_out])), \
         patch("app.services.webhook_email_service.send_webhook_email") as mock_send:
        notify_roles_by_email(
            db, roles=["red_lead", "blue_lead"],
            preference_key="email_on_all_team_validations",
            subject="Vote cast",
            message="Someone voted.",
        )
    # Only `lead` matches role filter (query already scopes it) *and* is opted in;
    # `opted_out` matched the fake query's role filter (a no-op in the stub) but
    # is excluded by preference; `other_role` is excluded by role in a real query
    # (the stub returns all rows regardless, so assert on what was actually sent).
    sent_to = {c.kwargs["to"] for c in mock_send.call_args_list}
    assert "lead@test.com" in sent_to
    assert "opted-out@test.com" not in sent_to


def test_notify_roles_by_email_excludes_actor(db):
    actor = _FakeUser(email="actor@test.com")
    other = _FakeUser(email="other@test.com")

    with patch.object(db, "query", return_value=_FakeQuery([actor, other])), \
         patch("app.services.webhook_email_service.send_webhook_email") as mock_send:
        notify_roles_by_email(
            db, roles=["admin"],
            preference_key="email_on_all_team_validations",
            subject="x", message="y",
            exclude_user_id=actor.id,
        )
    sent_to = {c.kwargs["to"] for c in mock_send.call_args_list}
    assert sent_to == {"other@test.com"}


def test_stale_technique_alert_dispatches_email_but_other_rule_types_dont():
    from app.services.operational_alert_service import _dispatch_inapp_notifications

    class _FakeAlertQuery:
        def filter(self, *a, **kw):
            return self

        def all(self):
            return []

    rule = SimpleNamespace(rule_type="stale_technique")
    instance = SimpleNamespace(id=uuid.uuid4(), title="Stale coverage", message="5 techniques are stale.")
    fake_db = SimpleNamespace(query=lambda *a, **kw: _FakeAlertQuery())

    with patch("app.services.notification_service.notify_roles_by_email") as mock_notify:
        _dispatch_inapp_notifications(fake_db, rule, instance)
    mock_notify.assert_called_once()
    assert mock_notify.call_args.kwargs["preference_key"] == "email_on_stale_coverage"

    rule.rule_type = "high_risk"
    with patch("app.services.notification_service.notify_roles_by_email") as mock_notify:
        _dispatch_inapp_notifications(fake_db, rule, instance)
    mock_notify.assert_not_called()


def test_webhook_failure_emails_admins_on_third_consecutive_failure():
    from app.services.webhook_service import _send_webhook

    wh = SimpleNamespace(name="my-hook", url="https://example.com/hook", secret=None, failure_count=2,
                         last_triggered_at=None)

    class _FakeDb:
        def commit(self):
            pass

        def rollback(self):
            pass

    with patch("app.services.webhook_service.requests.post", side_effect=Exception("boom")), \
         patch("app.services.notification_service.notify_roles_by_email") as mock_notify:
        _send_webhook(_FakeDb(), wh, "test_validated", {"foo": "bar"})

    assert wh.failure_count == 3
    mock_notify.assert_called_once()
    assert mock_notify.call_args.kwargs["preference_key"] == "email_on_webhook_failures"


def test_webhook_failure_does_not_email_before_threshold():
    from app.services.webhook_service import _send_webhook

    wh = SimpleNamespace(name="my-hook", url="https://example.com/hook", secret=None, failure_count=0,
                         last_triggered_at=None)

    class _FakeDb:
        def commit(self):
            pass

        def rollback(self):
            pass

    with patch("app.services.webhook_service.requests.post", side_effect=Exception("boom")), \
         patch("app.services.notification_service.notify_roles_by_email") as mock_notify:
        _send_webhook(_FakeDb(), wh, "test_validated", {"foo": "bar"})

    assert wh.failure_count == 1
    mock_notify.assert_not_called()


def test_create_user_notifies_other_admins_but_not_the_actor(api, auth_headers, db):
    from app.models.user import User
    from app.auth import hash_password

    other_admin = User(
        username="secondadmin@test.com", email="secondadmin@test.com",
        hashed_password=hash_password("Whatever123!@#"), role="admin",
    )
    db.add(other_admin)
    db.commit()

    with patch("app.services.webhook_email_service.send_webhook_email") as mock_send:
        resp = api(
            "post", "/api/v1/users", auth_headers,
            json={"full_name": "Fresh User", "email": "freshuser@test.com", "role": "viewer"},
        )
    assert resp.status_code == 201, resp.text

    sent_to = {c.kwargs["to"] for c in mock_send.call_args_list}
    assert "secondadmin@test.com" in sent_to
    assert mock_send.call_args_list[0].kwargs["subject"].startswith("New User Created")