Testing Lambdas

Layout

Location Scope

lambdas/tests/unit/

Cross-function unit tests — BigQuery contact sync, campaign re-engagement, email categorisation, Twilio lookup, the generic handler

lambdas/tests/integration/

API Gateway integration test

lambdas/tests/conftest.py

Shared fixtures

lambdas/morpheus_hubspot/tests/

Handler tests co-located with the function — re-engagement segments, dialer filters, Five9, lifecycle promotion, contact backfill, webhook processor, call-list refresh

lambdas/enrichment_scraper/tests/

SmartScout provider and auth, plus a live integration test

libs/<lib>/tests/

Library tests

Two conventions coexist: morpheus_hubspot and enrichment_scraper keep tests inside the function directory (with their own conftest.py), while other functions' tests live in lambdas/tests/unit/.

Running

cd lambdas && uv run pytest tests/ -v                       # cross-function
cd lambdas && uv run pytest morpheus_hubspot/tests -v       # hubspot handlers
cd lambdas/enrichment_scraper && uv run pytest tests -v     # scraper

cd libs/morpheus-common && uv run pytest tests -v           # event models
cd libs/morpheus-data-layer && uv run pytest tests -v       # repositories
cd libs/pyb-hubspot && uv run pytest tests -v               # HubSpot helpers

make test                                                    # agentcore runtime

libs/morpheus-data-layer/tests/integration/ needs a reachable database and reads its connection string from libs/morpheus-data-layer/.env. enrichment_scraper/tests/integration/test_smartscout.py calls the live SmartScout API and needs credentials.

The deferred-import pattern

Lambda layer packages are not installed in a test environment. Handler modules therefore defer their imports into @lru_cache functions, so the module imports cleanly and the test patches at the function boundary:

@lru_cache(maxsize=1)
def _get_agent():
    from morpheus_llm.providers import ClaudeAgentService      # deferred
    from morpheus_llm.types import AgentConfig
    from morpheus_llm.utils import AgentModelIds
    return ClaudeAgentService(config=AgentConfig(...), tools=lambda: [])

The same applies to load_secrets() and create_bigquery_client() — every scheduled handler imports morpheus_bigquery inside handle() rather than at module scope.

Preserve this pattern when adding an agent or a BigQuery-backed handler. A module-level from morpheus_llm…​ import …​ makes the whole module unimportable without the layer, which breaks every test that touches it.

@lru_cache also means the agent is built once per container. Call _get_agent.cache_clear() in a test that needs a different configuration.

What handler tests assert

Handlers are mostly decision code, so tests target the decisions rather than the API calls:

Test file Asserts

test_email_click_reengagement.py

The _should_skip suppression rule and the priority passthrough

test_lost_deal_reengagement.py, test_unattended_meeting_reengagement.py

Score-to-priority rounding, the called-today guard, ownership branches

test_dialer_filters.py

_effective_disposition() and every _skip_reason() branch

test_lifecycle_stage_promotion.py

Rank guard — promotion happens only from a lower stage

test_ensure_local_contact_row.py

The four-step short-circuit and foreign-key repair

test_newsletter_opt_out_processor.py

applies_to() matching, including the subscription-type id check

test_sales_rep_call_list_refresh.py

Delete-then-create ordering and the id write-back

test_five9_call_sync.py

Duration and timestamp parsing, disposition mapping, contact resolution

test_bigquery_engagement_sync.py

Watermark advance and activity upsert columns

test_campaign_reengagement.py

Priority computation and the abort-on-no-rep behaviour

Handlers that only orchestrate — crm_update_activity.py, crm_dialer_call_ended.py — have thin or no direct tests; the logic they delegate to is tested instead.

Writing a handler test

The shape that works with this codebase:

from unittest.mock import MagicMock, patch

def test_skips_when_called_today():
    row = {"contact_id": "123", "last_call_date": _today_utc(), "priority_score": 3.4}

    with patch("events.scheduled.supplier_account_reengagement.get_session"), \
         patch("events.scheduled.supplier_account_reengagement.update_contact_properties") as upd, \
         patch("events.scheduled.supplier_account_reengagement._query_supplier_targets",
               return_value=[row]), \
         patch("events.scheduled.supplier_account_reengagement.load_secrets") as secrets:
        secrets.return_value.bigquery_credentials = "{}"
        SupplierAccountReEngagementHandler().handle(ScheduledEvent(), MagicMock())

    upd.assert_not_called()

Patch the module-level name the handler resolves (events.scheduled.x.get_session), not the definition site — the handler imported it into its own namespace at import time.

Pure functions — _score_to_call_priority, _compute_priority, _should_skip, normalize_phone, _effective_disposition, _skip_reason — are the highest-value test targets and need no patching at all.

Local invocation

cd lambdas
sam build
sam local invoke MorpheusHubspotFunction \
  --event morpheus_hubspot/events/samples/contact_update.json

Sample payloads live in <function>/events/samples/, each directory with a README.md describing when to use which. For a scheduled handler, hand-write the envelope:

{"detail-type": "Hubspot.EmailClickReEngagement", "source": "morpheus.hubspot", "detail": {}}

sam local invoke reaches real AWS for Secrets Manager, SQS, Bedrock, and the database. A local run of a re-engagement handler will PATCH production HubSpot contacts. Treat every local invocation as production traffic.

Safer alternatives: invoke the deployed function with a narrow payload (for example Contact.ResolveHubSpotIds with batch_limit: 1), or unit-test the decision function directly.

Sample payloads

morpheus_hubspot/events/samples/ covers contact_update, contact_unsubscribe, contact_resolve_hubspot_ids, crm_custom_event_agent_email, crm_custom_event_agent_email_clicked, crm_dialer_call_ended, crm_hubspot_webhook_newsletter_optout, crm_new_activity, crm_new_activity_incoming, crm_task_create, crm_update_activity, and seller_crm_sync.

email_agent/events/samples/ covers the two Email.IncomingCategorise resolution paths. phone_agent/events/samples/ has phone_line_intel_pending.json. lambdas/events/ holds two older top-level fixtures (event.json, sellers_sqs.json).

Add a sample with every new event type — it doubles as documentation and as a sam local invoke fixture.

CI

.github/workflows/security-review.yml runs a security review over pull-request diffs. There is no test-running workflow — tests are run locally.