diff --git a/backend/app/services/orchestrator/__init__.py b/backend/app/services/orchestrator/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/services/orchestrator/alerts.py b/backend/app/services/orchestrator/alerts.py new file mode 100644 index 0000000..1634c42 --- /dev/null +++ b/backend/app/services/orchestrator/alerts.py @@ -0,0 +1,21 @@ +import logging + +from app.config import settings +from app.services.email import get_email_backend + +logger = logging.getLogger(__name__) + + +def send_admin_alert(subject: str, body: str) -> None: + if not settings.ADMIN_EMAIL: + logger.warning("ADMIN_EMAIL not set; dropping alert: %s", subject) + return + try: + get_email_backend().send( + to=settings.ADMIN_EMAIL, + subject=f"[MealPlanner ALERT] {subject}", + html=f"
{body}",
+ text=body,
+ )
+ except Exception:
+ logger.exception("Failed to send admin alert: %s", subject)
diff --git a/backend/tests/test_orchestrator.py b/backend/tests/test_orchestrator.py
new file mode 100644
index 0000000..8e3e3c3
--- /dev/null
+++ b/backend/tests/test_orchestrator.py
@@ -0,0 +1,170 @@
+"""Tests for the Phase 5 weekly orchestration."""
+import pytest
+from datetime import date, datetime, timezone
+from uuid import uuid4
+
+pytestmark = pytest.mark.requires_postgres
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+WEEK = date(2026, 5, 8) # a Friday
+
+
+@pytest.fixture()
+def family(db):
+ from app.models import FamilyProfile
+ fp = FamilyProfile(
+ id=uuid4(),
+ name="Test Family",
+ household_size=4,
+ adult_count=2,
+ child_count=2,
+ pending_approval_policy="approve",
+ )
+ db.add(fp)
+ db.flush()
+ return fp
+
+
+@pytest.fixture()
+def member(db, family):
+ from app.models import FamilyMember, FamilyMemberRole
+ m = FamilyMember(
+ id=uuid4(),
+ family_profile_id=family.id,
+ name="Alice",
+ email="alice@example.com",
+ role=FamilyMemberRole.ADULT,
+ )
+ db.add(m)
+ db.flush()
+ return m
+
+
+@pytest.fixture()
+def weekly_run(db, family):
+ from app.models import WeeklyRun
+ r = WeeklyRun(family_id=family.id, week_start_date=WEEK, status="pending")
+ db.add(r)
+ db.flush()
+ return r
+
+
+@pytest.fixture()
+def weekly_run_scraped(db, family):
+ from app.models import WeeklyRun
+ r = WeeklyRun(
+ family_id=family.id,
+ week_start_date=WEEK,
+ status="running",
+ scraped_at=datetime.now(timezone.utc),
+ )
+ db.add(r)
+ db.flush()
+ return r
+
+
+@pytest.fixture()
+def weekly_run_generated(db, family):
+ from app.models import WeeklyRun
+ r = WeeklyRun(
+ family_id=family.id,
+ week_start_date=WEEK,
+ status="running",
+ scraped_at=datetime.now(timezone.utc),
+ generated_at=datetime.now(timezone.utc),
+ )
+ db.add(r)
+ db.flush()
+ return r
+
+
+@pytest.fixture()
+def meal_plan(db, family):
+ from app.models import MealPlan, MealPlanStatus
+ mp = MealPlan(
+ id=uuid4(),
+ family_profile_id=family.id,
+ week_start_date=WEEK,
+ status=MealPlanStatus.PENDING_APPROVAL,
+ )
+ db.add(mp)
+ db.flush()
+ return mp
+
+
+@pytest.fixture()
+def pending_item(db, meal_plan):
+ from app.models import MealPlanItem, MealPlanItemStatus, MealType, Recipe
+ recipe = Recipe(
+ id=uuid4(),
+ name="Test Pasta",
+ servings=4,
+ ingredients=[{"name": "pasta", "qty": "400", "unit": "g"}],
+ instructions=["boil water", "cook pasta"],
+ )
+ db.add(recipe)
+ db.flush()
+ item = MealPlanItem(
+ id=uuid4(),
+ meal_plan_id=meal_plan.id,
+ recipe_id=recipe.id,
+ day_of_week=5,
+ meal_type=MealType.DINNER,
+ approval_status=MealPlanItemStatus.PENDING,
+ )
+ db.add(item)
+ db.flush()
+ return item
+
+
+@pytest.fixture()
+def weekly_run_emailed(db, family):
+ from app.models import WeeklyRun
+ r = WeeklyRun(
+ family_id=family.id,
+ week_start_date=WEEK,
+ status="running",
+ scraped_at=datetime.now(timezone.utc),
+ generated_at=datetime.now(timezone.utc),
+ emailed_at=datetime.now(timezone.utc),
+ )
+ db.add(r)
+ db.flush()
+ return r
+
+
+# ---------------------------------------------------------------------------
+# alerts.py tests
+# ---------------------------------------------------------------------------
+
+def test_send_admin_alert_noop_when_email_unset(monkeypatch):
+ monkeypatch.setattr(
+ "app.services.orchestrator.alerts.settings",
+ type("S", (), {"ADMIN_EMAIL": ""})(),
+ )
+ from app.services.orchestrator.alerts import send_admin_alert
+ send_admin_alert("test subject", "test body") # must not raise
+
+
+def test_send_admin_alert_calls_backend(monkeypatch):
+ sent = []
+
+ class FakeBackend:
+ def send(self, **kwargs):
+ sent.append(kwargs)
+
+ monkeypatch.setattr(
+ "app.services.orchestrator.alerts.settings",
+ type("S", (), {"ADMIN_EMAIL": "admin@example.com"})(),
+ )
+ monkeypatch.setattr(
+ "app.services.orchestrator.alerts.get_email_backend",
+ lambda: FakeBackend(),
+ )
+ from app.services.orchestrator.alerts import send_admin_alert
+ send_admin_alert("boom", "details here")
+ assert len(sent) == 1
+ assert "[MealPlanner ALERT] boom" in sent[0]["subject"]
+ assert sent[0]["to"] == "admin@example.com"