diff --git a/backend/app/assets/email_logo.png b/backend/app/assets/email_logo.png new file mode 100644 index 0000000..9529485 Binary files /dev/null and b/backend/app/assets/email_logo.png differ diff --git a/backend/app/jobs/mitre_sync_job.py b/backend/app/jobs/mitre_sync_job.py index a152592..27caf2e 100644 --- a/backend/app/jobs/mitre_sync_job.py +++ b/backend/app/jobs/mitre_sync_job.py @@ -16,6 +16,7 @@ from datetime import datetime, timedelta, timezone # Import BackgroundScheduler from apscheduler.schedulers.background from apscheduler.schedulers.background import BackgroundScheduler +from apscheduler.events import EVENT_JOB_ERROR # Import SessionLocal from app.database from app.database import SessionLocal @@ -60,6 +61,27 @@ logger = logging.getLogger(__name__) scheduler = BackgroundScheduler() +def _on_job_error(event) -> None: + """Email admins (opted-in) whenever a scheduled background job raises.""" + db = SessionLocal() + try: + from app.services.notification_service import notify_roles_by_email + notify_roles_by_email( + db, roles=["admin"], + preference_key="email_on_system_errors", + subject=f"Aegis Background Job Failed: {event.job_id}", + message=( + f'The scheduled job "{event.job_id}" raised an exception and did ' + f"not complete:\n\n{event.exception}" + ), + ) + db.commit() + except Exception: + logger.exception("Failed to dispatch system-error notification") + finally: + db.close() + + # --------------------------------------------------------------------------- # Job functions # --------------------------------------------------------------------------- @@ -440,6 +462,7 @@ def start_scheduler() -> None: Neither job fires immediately on startup. """ + scheduler.add_listener(_on_job_error, EVENT_JOB_ERROR) # Call scheduler.add_job() scheduler.add_job( _run_mitre_sync, diff --git a/backend/app/routers/campaigns.py b/backend/app/routers/campaigns.py index d74dca0..c8cb7c9 100644 --- a/backend/app/routers/campaigns.py +++ b/backend/app/routers/campaigns.py @@ -124,7 +124,7 @@ from app.services.campaign_crud_service import ( from app.services.audit_service import log_action # Import notify_role from app.services.notification_service -from app.services.notification_service import notify_role +from app.services.notification_service import notify_role, notify_roles_by_email from app.services.webhook_service import dispatch_webhook # Assign logger = logging.getLogger(__name__) @@ -543,6 +543,12 @@ def approve_campaign_endpoint( entity_id=campaign.id, details={"start_date": payload.start_date}, ) + notify_roles_by_email( + db, roles=["red_tech"], + preference_key="email_on_assigned_to_campaign", + subject=f"Campaign Activated: {campaign.name}", + message=f'Campaign "{campaign.name}" has been approved and activated. You may have tests assigned.', + ) uow.commit() db.refresh(campaign) @@ -868,6 +874,12 @@ def activate_campaign( # Keyword argument: entity_id entity_id=campaign.id, ) + notify_roles_by_email( + db, roles=["red_tech"], + preference_key="email_on_assigned_to_campaign", + subject=f"Campaign Activated: {campaign.name}", + message=f'Campaign "{campaign.name}" has been activated. You may have tests assigned.', + ) # Call log_action() log_action( db, diff --git a/backend/app/routers/users.py b/backend/app/routers/users.py index 655e046..fffe67a 100644 --- a/backend/app/routers/users.py +++ b/backend/app/routers/users.py @@ -185,6 +185,18 @@ def create_user_route( # Reload ORM object attributes from the database db.refresh(user) + from app.services.notification_service import notify_roles_by_email + notify_roles_by_email( + db, roles=["admin"], + preference_key="email_on_new_users", + subject=f"New User Created: {user.full_name or user.email}", + message=( + f'A new user account was created: {user.full_name or user.email} ' + f"({user.email}), role: {user.role}." + ), + exclude_user_id=current_user.id, + ) + # Return user return user diff --git a/backend/app/services/notification_service.py b/backend/app/services/notification_service.py index c637bbd..5b07e7e 100644 --- a/backend/app/services/notification_service.py +++ b/backend/app/services/notification_service.py @@ -291,6 +291,30 @@ def notify_all_users_by_email(db: Session, *, preference_key: str, subject: str, pass # nosec B110 — one user's failure must not skip the rest +def notify_roles_by_email( + db: Session, *, roles: list[str], preference_key: str, subject: str, message: str, + exclude_user_id=None, +) -> None: + """Email every active, opted-in user holding any of *roles*. + + For workflow-wide callouts to a team (e.g. every lead vote) rather than + a single actor's own action — no in-app notification is created here, + only the email; pair with ``create_notification``/``notify_role_with_email`` + if an in-app entry is also needed. + """ + users = db.query(User).filter(User.role.in_(roles), User.is_active == True).all() # noqa: E712 + for user in users: + if exclude_user_id and user.id == exclude_user_id: + continue + if not _preference_allows(user, preference_key): + continue + try: + from app.services.webhook_email_service import send_webhook_email + send_webhook_email(db, to=user.email, subject=subject, message=message, full_name=user.full_name) + except Exception: + pass # nosec B110 — one user's failure must not skip the rest + + def notify_role_with_email( db: Session, *, @@ -362,6 +386,12 @@ def notify_test_state_change(db: Session, test, new_state: str) -> None: # Keyword argument: entity_id entity_id=test_id, ) + notify_user_by_email( + db, creator_id, + preference_key="email_on_test_state_change", + subject=f"Test Execution Started: {test_name}", + message=f'Your test "{test_name}" has moved to the execution phase.', + ) # Alternative: new_state == "blue_evaluating" elif new_state == "blue_evaluating": @@ -385,6 +415,12 @@ def notify_test_state_change(db: Session, test, new_state: str) -> None: # Keyword argument: entity_id entity_id=test_id, ) + notify_user_by_email( + db, user.id, + preference_key="email_on_test_state_change", + subject=f"Test Ready for Blue Evaluation: {test_name}", + message=f'Test "{test_name}" needs blue team evaluation.', + ) # Alternative: new_state == "in_review" elif new_state == "in_review": @@ -414,6 +450,12 @@ def notify_test_state_change(db: Session, test, new_state: str) -> None: # Keyword argument: entity_id entity_id=test_id, ) + notify_user_by_email( + db, user.id, + preference_key="email_on_test_state_change", + subject=f"Test Ready for Validation: {test_name}", + message=f'Test "{test_name}" is awaiting your review.', + ) # Alternative: new_state == "rejected" and creator_id elif new_state == "rejected" and creator_id: diff --git a/backend/app/services/operational_alert_service.py b/backend/app/services/operational_alert_service.py index 69c16cf..d3c7069 100644 --- a/backend/app/services/operational_alert_service.py +++ b/backend/app/services/operational_alert_service.py @@ -29,7 +29,7 @@ log = logging.getLogger(__name__) def _dispatch_inapp_notifications(db: Session, rule: AlertRule, instance: AlertInstance) -> None: """Create in-app Notification rows for all admins and leads.""" - from app.services.notification_service import create_notification + from app.services.notification_service import create_notification, notify_roles_by_email admin_roles = {"admin", "red_lead", "blue_lead"} users = db.query(User).filter( @@ -47,6 +47,14 @@ def _dispatch_inapp_notifications(db: Session, rule: AlertRule, instance: AlertI entity_id = instance.id, ) + if rule.rule_type == AlertRuleType.stale_technique.value: + notify_roles_by_email( + db, roles=list(admin_roles), + preference_key="email_on_stale_coverage", + subject=instance.title, + message=instance.message, + ) + def _dispatch_webhooks(rule: AlertRule, instance: AlertInstance) -> None: """Fire webhook(s) for a triggered alert (all exceptions caught).""" diff --git a/backend/app/services/test_workflow_service.py b/backend/app/services/test_workflow_service.py index 0615bac..f8e78ef 100644 --- a/backend/app/services/test_workflow_service.py +++ b/backend/app/services/test_workflow_service.py @@ -39,6 +39,7 @@ from app.services.notification_service import ( notify_test_state_change, create_notification, notify_role_with_email, + notify_roles_by_email, ) # Assign logger = logging.getLogger(__name__) @@ -1048,6 +1049,13 @@ def validate_as_red_lead( }, ) + notify_roles_by_email( + db, roles=["red_lead", "blue_lead"], + preference_key="email_on_all_team_validations", + subject=f"Red Lead Vote: {test.name}", + message=f'{user.full_name or user.username} cast a "{validation_status}" vote as Red Lead on test "{test.name}".', + exclude_user_id=user.id, + ) _dispatch_dual_validation_effects(db, test, entity, actor=user) return test @@ -1113,6 +1121,13 @@ def validate_as_blue_lead( }, ) + notify_roles_by_email( + db, roles=["red_lead", "blue_lead"], + preference_key="email_on_all_team_validations", + subject=f"Blue Lead Vote: {test.name}", + message=f'{user.full_name or user.username} cast a "{validation_status}" vote as Blue Lead on test "{test.name}".', + exclude_user_id=user.id, + ) _dispatch_dual_validation_effects(db, test, entity, actor=user) return test diff --git a/backend/app/services/webhook_email_service.py b/backend/app/services/webhook_email_service.py index 81d1086..461e118 100644 --- a/backend/app/services/webhook_email_service.py +++ b/backend/app/services/webhook_email_service.py @@ -5,8 +5,9 @@ platform (password setup/reset, test validated, campaign completed, new MITRE techniques synced, etc.) goes through ``send_webhook_email`` here, which POSTs a ``{to, subject, body}`` JSON payload to a single configured webhook URL — intended for a Power Automate flow that actually delivers -the email. An optional API key, if configured, is sent as an -``x-api-key`` header. +the email (the ``body`` is full HTML; the flow's "send email" step must +have its "Is HTML" option enabled). An optional API key, if configured, +is sent as an ``x-api-key`` header. SMTP (``app.services.email_service``) is intentionally left in place but unused and unreachable from the UI — see that module's docstring. @@ -14,7 +15,11 @@ unused and unreachable from the UI — see that module's docstring. from __future__ import annotations +import base64 +import html as html_lib import logging +import re +from pathlib import Path import requests from sqlalchemy.orm import Session @@ -24,22 +29,91 @@ logger = logging.getLogger(__name__) _WEBHOOK_URL_CONFIG_KEY = "email_webhook.url" _WEBHOOK_API_KEY_CONFIG_KEY = "email_webhook.api_key" -# Standard greeting + footer wrapped around every notification email's -# actual message, matching the platform's established template. -_EMAIL_SIGNATURE = ( - "\n\nRegards,\n" - "AEGIS Security Platform\n" - "Purple Team Engineering\n" - "Owned and operated by Enterprise Corp.\n\n" - "Assume breach. Validate controls. Improve continuously.\n\n" - "This is an automated notification. Please do not reply." -) +_LOGO_PATH = Path(__file__).resolve().parent.parent / "assets" / "email_logo.png" +_URL_RE = re.compile(r"^https?://\S+$") + +_logo_b64_cache: str | None = None + + +def _get_logo_base64() -> str: + """Read + cache the inline email logo as a base64 string (empty if missing).""" + global _logo_b64_cache + if _logo_b64_cache is None: + try: + _logo_b64_cache = base64.b64encode(_LOGO_PATH.read_bytes()).decode("ascii") + except OSError: + logger.warning("Email logo asset missing at %s", _LOGO_PATH) + _logo_b64_cache = "" + return _logo_b64_cache + + +def _message_to_html(message: str) -> str: + """Turn a plain-text message (paragraphs separated by blank lines) into + escaped HTML, rendering any lone-URL paragraph as a call-to-action button. + """ + parts: list[str] = [] + for para in message.strip().split("\n\n"): + para = para.strip() + if not para: + continue + if _URL_RE.match(para): + url = html_lib.escape(para, quote=True) + parts.append( + f'

' + f'Open Link →' + f'

' + ) + else: + escaped = html_lib.escape(para).replace("\n", "
") + parts.append(f'

{escaped}

') + return "".join(parts) def build_email_body(full_name: str | None, message: str) -> str: - """Wrap *message* in the standard greeting + signature template.""" - greeting = full_name or "there" - return f"HI {greeting}\n\n{message}{_EMAIL_SIGNATURE}" + """Wrap *message* in Aegis's branded HTML email template (inline logo).""" + greeting = html_lib.escape(full_name) if full_name else "there" + message_html = _message_to_html(message) + logo_b64 = _get_logo_base64() + logo_tag = ( + f'Aegis' + if logo_b64 + else "" + ) + return f""" + + + + + + +
+ + + + + + + + + + +
+ {logo_tag} +
AEGIS SECURITY PLATFORM
+
Purple Team Engineering
+
+

Hi {greeting},

+
{message_html}
+
+

Assume breach. Validate controls. Improve continuously.

+

Owned and operated by Enterprise Corp. This is an automated notification — please do not reply.

+
+
+ +""" def _read_system_config(db: Session, key: str) -> str | None: diff --git a/backend/app/services/webhook_service.py b/backend/app/services/webhook_service.py index bf2afeb..84266ee 100644 --- a/backend/app/services/webhook_service.py +++ b/backend/app/services/webhook_service.py @@ -184,6 +184,19 @@ def _send_webhook(db, wh: WebhookConfig, event_type: str, payload: dict) -> None "Webhook '%s' (%s) failed for event=%s: %s (failure_count=%d)", wh.name, wh.url, event_type, exc, wh.failure_count, ) + if wh.failure_count == 3: + # Email once on the 3rd consecutive failure — signals "this looks + # broken", without alerting on every transient single failure. + from app.services.notification_service import notify_roles_by_email + notify_roles_by_email( + db, roles=["admin"], + preference_key="email_on_webhook_failures", + subject=f"Webhook Delivery Failing: {wh.name}", + message=( + f'The webhook "{wh.name}" ({wh.url}) has failed {wh.failure_count} times ' + f'in a row for event "{event_type}". Last error: {exc}' + ), + ) # --------------------------------------------------------------------------- diff --git a/backend/tests/test_integration_v2.py b/backend/tests/test_integration_v2.py index 4c47b4c..fec34aa 100644 --- a/backend/tests/test_integration_v2.py +++ b/backend/tests/test_integration_v2.py @@ -97,11 +97,17 @@ if "apscheduler.schedulers.background" not in sys.modules: _apsched = ModuleType("apscheduler.schedulers.background") class _FakeBGScheduler: def add_job(self, *a, **kw): pass + def add_listener(self, *a, **kw): pass def start(self): pass def shutdown(self, **kw): pass _apsched.BackgroundScheduler = _FakeBGScheduler sys.modules["apscheduler.schedulers.background"] = _apsched +if "apscheduler.events" not in sys.modules: + _apsched_events = ModuleType("apscheduler.events") + _apsched_events.EVENT_JOB_ERROR = 1 + sys.modules["apscheduler.events"] = _apsched_events + if "taxii2client" not in sys.modules: sys.modules["taxii2client"] = ModuleType("taxii2client") if "taxii2client.v20" not in sys.modules: diff --git a/backend/tests/test_metrics_v2.py b/backend/tests/test_metrics_v2.py index 7a151a5..843d18d 100644 --- a/backend/tests/test_metrics_v2.py +++ b/backend/tests/test_metrics_v2.py @@ -76,6 +76,7 @@ for _mod in [ "apscheduler", "apscheduler.schedulers", "apscheduler.schedulers.background", "apscheduler.triggers", "apscheduler.triggers.cron", + "apscheduler.events", ]: if _mod not in sys.modules: m = ModuleType(_mod) @@ -85,6 +86,7 @@ for _mod in [ elif _mod == "botocore.exceptions": m.ClientError = Exception elif _mod == "apscheduler.schedulers.background": m.BackgroundScheduler = MagicMock elif _mod == "apscheduler.triggers.cron": m.CronTrigger = MagicMock + elif _mod == "apscheduler.events": m.EVENT_JOB_ERROR = 1 sys.modules[_mod] = m # --------------------------------------------------------------------------- diff --git a/backend/tests/test_notification_email_coverage.py b/backend/tests/test_notification_email_coverage.py new file mode 100644 index 0000000..8b87ad3 --- /dev/null +++ b/backend/tests/test_notification_email_coverage.py @@ -0,0 +1,161 @@ +"""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") diff --git a/backend/tests/test_password_setup_flow.py b/backend/tests/test_password_setup_flow.py index cae4b7b..dcb51d1 100644 --- a/backend/tests/test_password_setup_flow.py +++ b/backend/tests/test_password_setup_flow.py @@ -74,7 +74,7 @@ def test_send_password_email_posts_to_webhook_and_issues_token(api, db, auth_hea payload = call_kwargs.kwargs["json"] assert payload["to"] == "setpw@test.com" assert payload["subject"] == "Set Your Password" - assert "HI Set Password User" in payload["body"] + assert "Hi Set Password User" in payload["body"] assert "token=" in payload["body"] assert "Purple Team Engineering" in payload["body"] diff --git a/backend/tests/test_t107_status_service.py b/backend/tests/test_t107_status_service.py index 0297d4c..ba2b446 100644 --- a/backend/tests/test_t107_status_service.py +++ b/backend/tests/test_t107_status_service.py @@ -103,6 +103,8 @@ if "apscheduler" not in sys.modules: sys.modules["apscheduler.triggers"] = ModuleType("apscheduler.triggers") sys.modules["apscheduler.triggers.cron"] = ModuleType("apscheduler.triggers.cron") sys.modules["apscheduler.triggers.cron"].CronTrigger = MagicMock + sys.modules["apscheduler.events"] = ModuleType("apscheduler.events") + sys.modules["apscheduler.events"].EVENT_JOB_ERROR = 1 # --------------------------------------------------------------------------- # Imports diff --git a/backend/tests/test_t108_atomic_import.py b/backend/tests/test_t108_atomic_import.py index 214224d..36c351d 100644 --- a/backend/tests/test_t108_atomic_import.py +++ b/backend/tests/test_t108_atomic_import.py @@ -100,6 +100,8 @@ if "apscheduler" not in sys.modules: sys.modules["apscheduler.triggers"] = ModuleType("apscheduler.triggers") sys.modules["apscheduler.triggers.cron"] = ModuleType("apscheduler.triggers.cron") sys.modules["apscheduler.triggers.cron"].CronTrigger = MagicMock + sys.modules["apscheduler.events"] = ModuleType("apscheduler.events") + sys.modules["apscheduler.events"].EVENT_JOB_ERROR = 1 # --------------------------------------------------------------------------- # Imports diff --git a/backend/tests/test_t109_tests_router.py b/backend/tests/test_t109_tests_router.py index 45de66e..2b660e4 100644 --- a/backend/tests/test_t109_tests_router.py +++ b/backend/tests/test_t109_tests_router.py @@ -76,6 +76,7 @@ for mod_name in [ "apscheduler", "apscheduler.schedulers", "apscheduler.schedulers.background", "apscheduler.triggers", "apscheduler.triggers.cron", + "apscheduler.events", ]: if mod_name not in sys.modules: m = ModuleType(mod_name) @@ -92,6 +93,8 @@ for mod_name in [ m.BackgroundScheduler = MagicMock elif mod_name == "apscheduler.triggers.cron": m.CronTrigger = MagicMock + elif mod_name == "apscheduler.events": + m.EVENT_JOB_ERROR = 1 sys.modules[mod_name] = m # --------------------------------------------------------------------------- diff --git a/backend/tests/test_t110_evidence_router.py b/backend/tests/test_t110_evidence_router.py index fee04cd..b716a91 100644 --- a/backend/tests/test_t110_evidence_router.py +++ b/backend/tests/test_t110_evidence_router.py @@ -74,6 +74,7 @@ for mod_name in [ "apscheduler", "apscheduler.schedulers", "apscheduler.schedulers.background", "apscheduler.triggers", "apscheduler.triggers.cron", + "apscheduler.events", ]: if mod_name not in sys.modules: m = ModuleType(mod_name) @@ -83,6 +84,7 @@ for mod_name in [ elif mod_name == "botocore.exceptions": m.ClientError = Exception elif mod_name == "apscheduler.schedulers.background": m.BackgroundScheduler = MagicMock elif mod_name == "apscheduler.triggers.cron": m.CronTrigger = MagicMock + elif mod_name == "apscheduler.events": m.EVENT_JOB_ERROR = 1 sys.modules[mod_name] = m # --------------------------------------------------------------------------- diff --git a/backend/tests/test_t111_test_templates_router.py b/backend/tests/test_t111_test_templates_router.py index fb42c81..e04bb08 100644 --- a/backend/tests/test_t111_test_templates_router.py +++ b/backend/tests/test_t111_test_templates_router.py @@ -74,6 +74,7 @@ for mod_name in [ "apscheduler", "apscheduler.schedulers", "apscheduler.schedulers.background", "apscheduler.triggers", "apscheduler.triggers.cron", + "apscheduler.events", ]: if mod_name not in sys.modules: m = ModuleType(mod_name) @@ -83,6 +84,7 @@ for mod_name in [ elif mod_name == "botocore.exceptions": m.ClientError = Exception elif mod_name == "apscheduler.schedulers.background": m.BackgroundScheduler = MagicMock elif mod_name == "apscheduler.triggers.cron": m.CronTrigger = MagicMock + elif mod_name == "apscheduler.events": m.EVENT_JOB_ERROR = 1 sys.modules[mod_name] = m # --------------------------------------------------------------------------- diff --git a/backend/tests/test_t112_system_import.py b/backend/tests/test_t112_system_import.py index 20a41d1..2e09500 100644 --- a/backend/tests/test_t112_system_import.py +++ b/backend/tests/test_t112_system_import.py @@ -78,6 +78,7 @@ for mod_name in [ "apscheduler", "apscheduler.schedulers", "apscheduler.schedulers.background", "apscheduler.triggers", "apscheduler.triggers.cron", + "apscheduler.events", ]: if mod_name not in sys.modules: m = ModuleType(mod_name) @@ -87,6 +88,7 @@ for mod_name in [ elif mod_name == "botocore.exceptions": m.ClientError = Exception elif mod_name == "apscheduler.schedulers.background": m.BackgroundScheduler = MagicMock elif mod_name == "apscheduler.triggers.cron": m.CronTrigger = MagicMock + elif mod_name == "apscheduler.events": m.EVENT_JOB_ERROR = 1 sys.modules[mod_name] = m # --------------------------------------------------------------------------- diff --git a/backend/tests/test_templates_crud.py b/backend/tests/test_templates_crud.py index 2653d10..5b46a8d 100644 --- a/backend/tests/test_templates_crud.py +++ b/backend/tests/test_templates_crud.py @@ -77,6 +77,7 @@ for _mod in [ "apscheduler", "apscheduler.schedulers", "apscheduler.schedulers.background", "apscheduler.triggers", "apscheduler.triggers.cron", + "apscheduler.events", ]: if _mod not in sys.modules: m = ModuleType(_mod) @@ -86,6 +87,7 @@ for _mod in [ elif _mod == "botocore.exceptions": m.ClientError = Exception elif _mod == "apscheduler.schedulers.background": m.BackgroundScheduler = MagicMock elif _mod == "apscheduler.triggers.cron": m.CronTrigger = MagicMock + elif _mod == "apscheduler.events": m.EVENT_JOB_ERROR = 1 sys.modules[_mod] = m # --------------------------------------------------------------------------- diff --git a/backend/tests/test_workflow.py b/backend/tests/test_workflow.py index 558031f..a755274 100644 --- a/backend/tests/test_workflow.py +++ b/backend/tests/test_workflow.py @@ -83,6 +83,7 @@ for _mod in [ "apscheduler", "apscheduler.schedulers", "apscheduler.schedulers.background", "apscheduler.triggers", "apscheduler.triggers.cron", + "apscheduler.events", ]: if _mod not in sys.modules: m = ModuleType(_mod) @@ -92,6 +93,7 @@ for _mod in [ elif _mod == "botocore.exceptions": m.ClientError = Exception elif _mod == "apscheduler.schedulers.background": m.BackgroundScheduler = MagicMock elif _mod == "apscheduler.triggers.cron": m.CronTrigger = MagicMock + elif _mod == "apscheduler.events": m.EVENT_JOB_ERROR = 1 sys.modules[_mod] = m # ---------------------------------------------------------------------------