6b25cf9c6d
Implements a full proof-of-concept for syncing IDAA's Novi AMS membership
groups to Mailman 3 mailing lists via a cron-triggered reconciliation approach.
Key changes:
- methods: rewrote sync engine around confirmed Novi API shape — group-based
member fetch (/groups/{guid}/members + /customers/{uuid}), respects
Active=false and UnsubscribeFromEmails=true flags
- methods: mirror_novi_group_to_mailman_list() diffs Novi group against
Mailman roster and subscribes/unsubscribes accordingly (full mirror)
- methods: mirror_all_configured_mappings() iterates novi_mailman_sync
config array in IDAA site cfg_json — this is the cron target
- router: replaced old /sync endpoint with POST /sync (all mappings) and
POST /sync/group/{guid} (single mapping); removed webhook endpoint
(sync is cron-based, not event-driven)
- router: added GET/POST/DELETE endpoints for list member inspection
and manual subscribe/unsubscribe
- tests: two new e2e scripts covering connection checks and full member
lifecycle; old webhook integration test archived
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
52 lines
1.4 KiB
Python
52 lines
1.4 KiB
Python
"""
|
|
ARCHIVED 2026-03-17
|
|
The webhook endpoint (/webhook/novi) was removed — sync is cron-based.
|
|
If Novi webhook support is added in future, restore the endpoint in
|
|
api_v3_actions_e_novi_mailman.py and move this file back to tests/e2e/.
|
|
The webhook secret is stored as novi_webhook_secret in IDAA site cfg_json.
|
|
|
|
Original: Integration test — send a signed Novi webhook payload to the API.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import hmac
|
|
import hashlib
|
|
import requests
|
|
|
|
BASE_URL = os.environ.get('AE_API_BASE', 'https://dev-api.oneskyit.com')
|
|
ENDPOINT = f"{BASE_URL}/v3/action/e_novi_mailman/webhook/novi"
|
|
SECRET = os.environ.get('NOVI_WEBHOOK_SECRET', 'test-secret')
|
|
|
|
payload = {
|
|
"EventType": "MembershipActivated",
|
|
"Member": {
|
|
"Email": "test+webhook@example.com",
|
|
"FirstName": "Test",
|
|
"LastName": "Webhook",
|
|
"MembershipStatus": "Active"
|
|
}
|
|
}
|
|
|
|
body = json.dumps(payload).encode('utf-8')
|
|
signature = hmac.new(SECRET.encode('utf-8'), body, hashlib.sha256).hexdigest()
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'X-Novi-Signature': signature
|
|
}
|
|
|
|
print('Posting to', ENDPOINT)
|
|
resp = requests.post(ENDPOINT, headers=headers, data=body, timeout=30)
|
|
print('Status:', resp.status_code)
|
|
try:
|
|
print('JSON:', resp.json())
|
|
except Exception:
|
|
print('Body:', resp.text)
|
|
|
|
if resp.status_code == 200:
|
|
print('\u2705 PASS: webhook accepted')
|
|
else:
|
|
print('\u274C FAIL: webhook rejected')
|
|
raise SystemExit(1)
|