feat: add Novi AMS → Mailman 3 cron-based mirror sync bridge (IDAA)

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>
2026-03-17 16:36:32 -04:00
parent 29579fd9f1
commit 6b25cf9c6d
6 changed files with 711 additions and 275 deletions
--- a/app/methods/e_novi_mailman_methods.py
+++ b/app/methods/e_novi_mailman_methods.py
@@ -1,50 +1,50 @@
 import json, requests
 from typing import Dict, List, Optional
-from app.db_sql import sql_select, sql_update, sql_insert
+from app.db_sql import sql_select
 from app.lib_general import log, logging, logger_reset

 # ---------------------------------------------------------------------------
-# Novi-Mailman Bridge
-# Synchronizes Novi AMS membership data with Mailman 3 mailing lists.
+# Novi-Mailman Bridge — IDAA
 #
-# Credential Storage (data_store table):
-#   code='novi_api_config'     → JSON: { "api_key": "...", "base_url": "https://..." }
-#   code='mailman_api_config'  → JSON: { "base_url": "http://...:8001", "username": "...", "password": "..." }
+# Credentials live in site.cfg_json for the IDAA site (id_random='58_gJESdlUh').
+# Novi keys already present:
+#   novi_api_root_url  — e.g. "https://www.idaa.org/api"
+#   novi_idaa_api_key  — Base64 API key (Basic auth)
 #
-# Sync Logic:
-#   - Active Novi members → subscribed in target Mailman list(s)
-#   - Lapsed/expired members → unsubscribed (or held, depending on policy)
-#   - Driven by webhooks (Novi membership events) + optional full-sync
+# Keys that must be added to cfg_json before Mailman or webhooks can work:
+#   mailman_base_url     — e.g. "http://lists.idaa.org:8001"
+#   mailman_username     — Mailman REST admin user (usually "restadmin")
+#   mailman_password     — Mailman REST admin password
+#   mailman_list_id      — Target list, e.g. "members@idaa.org"
+#   novi_webhook_secret  — Shared secret for HMAC-SHA256 webhook validation
 # ---------------------------------------------------------------------------

-# ── Credential Helpers ────────────────────────────────────────────────────
+IDAA_SITE_ID_RANDOM = '58_gJESdlUh'
+
+
+# ── Config Helper ─────────────────────────────────────────────────────────

@logger_reset
-def _load_novi_config() -> Optional[Dict]:
-    """Load Novi AMS API credentials from data_store (code='novi_api_config')."""
-    rec = sql_select(table_name='data_store', field_name='code', field_value='novi_api_config')
-    if not rec:
-        log.error("Novi API config not found in data_store (code='novi_api_config').")
+def _load_idaa_cfg() -> Optional[Dict]:
+    """
+    Load IDAA site cfg_json.  Returns the parsed dict, or None on failure.
+    """
+    from app.methods.site_methods import load_site_obj
+    site = load_site_obj(site_id=IDAA_SITE_ID_RANDOM, model_as_dict=True)
+    if not site:
+        log.error("Could not load IDAA site record (id_random='%s').", IDAA_SITE_ID_RANDOM)
        return None
-    try:
-        return json.loads(rec['text'])
-    except Exception as e:
-        log.error(f"Failed to parse Novi config: {e}")
-        return None
-
-
-@logger_reset
-def _load_mailman_config() -> Optional[Dict]:
-    """Load Mailman 3 REST API credentials from data_store (code='mailman_api_config')."""
-    rec = sql_select(table_name='data_store', field_name='code', field_value='mailman_api_config')
-    if not rec:
-        log.error("Mailman API config not found in data_store (code='mailman_api_config').")
-        return None
-    try:
-        return json.loads(rec['text'])
-    except Exception as e:
-        log.error(f"Failed to parse Mailman config: {e}")
+    cfg = site.get('cfg_json')
+    if isinstance(cfg, str):
+        try:
+            cfg = json.loads(cfg)
+        except Exception as e:
+            log.error("Failed to parse IDAA cfg_json: %s", e)
+            return None
+    if not isinstance(cfg, dict):
+        log.error("IDAA cfg_json is not a dict after parsing.")
        return None
+    return cfg


 # ── Novi AMS Methods ──────────────────────────────────────────────────────
@@ -52,26 +52,35 @@ def _load_mailman_config() -> Optional[Dict]:
@logger_reset
 def test_novi_connection() -> Dict:
    """
-    Verify Novi AMS API credentials are valid.
-    Returns a dict with 'ok' bool and optional error message.
+    Verify Novi AMS API credentials from IDAA site cfg_json.
+    Uses the first group GUID in novi_idaa_group_guid_li as a lightweight auth probe.
+    Returns {'ok': True, 'member_count': N} on success.
    """
-    config = _load_novi_config()
-    if not config:
-        return {"ok": False, "error": "Credentials not configured."}
+    cfg = _load_idaa_cfg()
+    if not cfg:
+        return {"ok": False, "error": "Could not load IDAA site config."}

-    # Novi uses Basic auth with a Base64-encoded API key.
-    # Confirmed from IDAA Jitsi integration: Authorization: Basic {api_key}
-    base_url = config.get('base_url', '').rstrip('/')
-    api_key  = config.get('api_key', '')
-    headers  = {"Authorization": f"Basic {api_key}", "Accept": "application/json"}
+    base_url = cfg.get('novi_api_root_url', '').rstrip('/')
+    api_key  = cfg.get('novi_idaa_api_key', '')

+    if not base_url or not api_key:
+        return {"ok": False, "error": "novi_api_root_url or novi_idaa_api_key missing from cfg_json."}
+
+    group_guid_li = cfg.get('novi_idaa_group_guid_li') or []
+    if not group_guid_li:
+        return {"ok": False, "error": "novi_idaa_group_guid_li missing from cfg_json."}
+
+    headers = {"Authorization": f"Basic {api_key}", "Accept": "application/json"}
    try:
-        resp = requests.get(f"{base_url}/api/v1/members", headers=headers, params={"$top": 1}, timeout=10)
+        # Use first group as a lightweight auth probe (pageSize=1)
+        guid = group_guid_li[0]
+        resp = requests.get(f"{base_url}/groups/{guid}/members",
+                            headers=headers, params={"pageSize": 1}, timeout=10)
        if resp.status_code == 200:
-            return {"ok": True}
+            return {"ok": True, "probe_group": guid}
        return {"ok": False, "error": f"HTTP {resp.status_code}: {resp.text[:200]}"}
    except Exception as e:
-        log.exception(f"Novi connection test failed: {e}")
+        log.exception("Novi connection test failed: %s", e)
        return {"ok": False, "error": str(e)}


@@ -80,139 +89,181 @@ def get_novi_members(status_filter: Optional[str] = None, page_size: int = 500,
    """
    Fetch member records from Novi AMS.

-    Args:
-        status_filter: Optional Novi membership status (e.g. 'Active', 'Lapsed').
-                       None returns all members.
-        page_size:     Records per page (Novi uses OData $top/$skip).
-        offset:        Pagination offset ($skip).
+    Novi has no flat member-list endpoint.  Members are fetched per group from
+    novi_idaa_group_guid_li, deduped by UniqueID, then each member's full record
+    (including Email) is fetched via GET /customers/{uuid}.

-    Returns:
-        List of member dicts, or None on failure.
-
-    TODO: Confirm OData filter field names against your Novi instance schema.
+    status_filter and pagination (page_size/offset) are not supported at the
+    Novi API level for this approach — all group members are returned.
    """
-    config = _load_novi_config()
-    if not config:
+    cfg = _load_idaa_cfg()
+    if not cfg:
        return None

-    base_url = config.get('base_url', '').rstrip('/')
-    api_key  = config.get('api_key', '')
-    headers  = {"Authorization": f"Basic {api_key}", "Accept": "application/json"}
-    params   = {"pageSize": page_size, "offset": offset}
+    base_url      = cfg.get('novi_api_root_url', '').rstrip('/')
+    api_key       = cfg.get('novi_idaa_api_key', '')
+    group_guid_li = cfg.get('novi_idaa_group_guid_li') or []
+    headers       = {"Authorization": f"Basic {api_key}", "Accept": "application/json"}

-    if status_filter:
-        params["membershipStatus"] = status_filter
-
-    # TODO: Confirm the bulk member list endpoint for this Novi instance.
-    # The IDAA Jitsi code uses /customers/{uuid} for individual lookups and
-    # /groups/{guid}/members for group membership. A bulk member list may be
-    # /members, /customers, or require a group-based approach.
-    try:
-        resp = requests.get(f"{base_url}/members", headers=headers, params=params, timeout=30)
-        if resp.status_code != 200:
-            log.error(f"Novi API error: {resp.status_code} - {resp.text[:200]}")
-            return None
-        data = resp.json()
-        # Novi may return array directly, or wrap in Results/Members key
-        if isinstance(data, list):
-            return data
-        return data.get('Results') or data.get('Members') or data.get('value') or []
-    except Exception as e:
-        log.exception(f"Failed to fetch Novi members: {e}")
+    if not group_guid_li:
+        log.error("novi_idaa_group_guid_li missing from cfg_json.")
        return None

+    # Step 1: collect unique member UUIDs across all configured groups
+    seen_uuids: set = set()
+    uuid_list: List[str] = []
+    for guid in group_guid_li:
+        try:
+            resp = requests.get(f"{base_url}/groups/{guid}/members",
+                                headers=headers, params={"pageSize": page_size}, timeout=30)
+            if resp.status_code != 200:
+                log.error("Novi group %s fetch error: %s", guid, resp.status_code)
+                continue
+            for entry in resp.json():
+                uid = entry.get('UniqueID')
+                if uid and uid not in seen_uuids:
+                    seen_uuids.add(uid)
+                    uuid_list.append(uid)
+        except Exception as e:
+            log.exception("Failed to fetch Novi group %s: %s", guid, e)
+
+    log.info("Novi: %d unique members across %d group(s).", len(uuid_list), len(group_guid_li))
+
+    # Step 2: fetch full customer record (including Email) for each UUID
+    members: List[Dict] = []
+    for uid in uuid_list:
+        try:
+            resp = requests.get(f"{base_url}/customers/{uid}", headers=headers, timeout=10)
+            if resp.status_code == 200:
+                members.append(resp.json())
+            else:
+                log.warning("Novi customer %s fetch error: %s", uid, resp.status_code)
+        except Exception as e:
+            log.exception("Failed to fetch Novi customer %s: %s", uid, e)
+
+    return members
+

 # ── Mailman 3 Methods ─────────────────────────────────────────────────────

@logger_reset
 def test_mailman_connection() -> Dict:
    """
-    Verify Mailman 3 REST API credentials are valid.
-    Returns a dict with 'ok' bool and optional error message.
+    Verify Mailman 3 REST API credentials from IDAA site cfg_json.
+    Returns {'ok': True, 'version': '...'} on success.
    """
-    config = _load_mailman_config()
-    if not config:
-        return {"ok": False, "error": "Credentials not configured."}
+    cfg = _load_idaa_cfg()
+    if not cfg:
+        return {"ok": False, "error": "Could not load IDAA site config."}

-    base_url = config.get('base_url', '').rstrip('/')
-    auth     = (config.get('username', 'restadmin'), config.get('password', ''))
+    base_url = cfg.get('mailman_base_url', '').rstrip('/')
+    username = cfg.get('mailman_username', 'restadmin')
+    password = cfg.get('mailman_password', '')
+
+    if not base_url or not password:
+        return {"ok": False, "error": "mailman_base_url or mailman_password missing from cfg_json."}

    try:
-        resp = requests.get(f"{base_url}/3.1/system/versions", auth=auth, timeout=10)
+        resp = requests.get(f"{base_url}/3.1/system/versions", auth=(username, password), timeout=10)
        if resp.status_code == 200:
            return {"ok": True, "version": resp.json().get('mailman_version')}
        return {"ok": False, "error": f"HTTP {resp.status_code}: {resp.text[:200]}"}
    except Exception as e:
-        log.exception(f"Mailman connection test failed: {e}")
+        log.exception("Mailman connection test failed: %s", e)
        return {"ok": False, "error": str(e)}


@logger_reset
-def get_mailman_lists() -> Optional[List[Dict]]:
+def get_mailman_list_members(list_id: str, count: int = 100, page: int = 1) -> Optional[List[Dict]]:
    """
-    Return all mailing lists from this Mailman 3 instance.
+    Return members of a specific Mailman 3 list.
+
+    Args:
+        list_id: fqdn_listname e.g. 'mm3@idaa.org' or dot-notation 'mm3.idaa.org'
+        count:   page size (Mailman default 20, max typically 100)
+        page:    1-based page number
    """
-    config = _load_mailman_config()
-    if not config:
+    cfg = _load_idaa_cfg()
+    if not cfg:
        return None

-    base_url = config.get('base_url', '').rstrip('/')
-    auth     = (config.get('username', 'restadmin'), config.get('password', ''))
+    base_url    = cfg.get('mailman_base_url', '').rstrip('/')
+    auth        = (cfg.get('mailman_username', 'restadmin'), cfg.get('mailman_password', ''))
+    list_id_dot = list_id.replace('@', '.')
+
+    try:
+        resp = requests.get(
+            f"{base_url}/3.1/lists/{list_id_dot}/roster/member",
+            auth=auth,
+            params={"count": count, "page": page},
+            timeout=10,
+        )
+        if resp.status_code != 200:
+            log.error("Mailman member list fetch failed for %s: %s", list_id, resp.status_code)
+            return None
+        data = resp.json()
+        return data.get('entries', [])
+    except Exception as e:
+        log.exception("Failed to fetch members for list %s: %s", list_id, e)
+        return None
+
+
+@logger_reset
+def get_mailman_lists() -> Optional[List[Dict]]:
+    """Return all mailing lists from this Mailman 3 instance."""
+    cfg = _load_idaa_cfg()
+    if not cfg:
+        return None
+
+    base_url = cfg.get('mailman_base_url', '').rstrip('/')
+    auth     = (cfg.get('mailman_username', 'restadmin'), cfg.get('mailman_password', ''))

    try:
        resp = requests.get(f"{base_url}/3.1/lists", auth=auth, timeout=10)
        if resp.status_code != 200:
-            log.error(f"Mailman list fetch failed: {resp.status_code}")
+            log.error("Mailman list fetch failed: %s", resp.status_code)
            return None
        return resp.json().get('entries', [])
    except Exception as e:
-        log.exception(f"Failed to fetch Mailman lists: {e}")
+        log.exception("Failed to fetch Mailman lists: %s", e)
        return None


@logger_reset
 def subscribe_member_to_list(list_id: str, email: str, display_name: str = '') -> bool:
    """
-    Subscribe an email address to a Mailman 3 list.
-    Uses pre-confirmed subscription (no confirmation email sent).
-
-    Args:
-        list_id:      Mailman list ID, e.g. 'members@yourdomain.org'
-        email:        Member email address
-        display_name: Optional display name
-
-    Returns:
-        True on success or already subscribed, False on error.
+    Subscribe an email address to a Mailman 3 list (pre-confirmed, no welcome email).
+    Returns True on success or already-subscribed, False on error.
    """
-    config = _load_mailman_config()
-    if not config:
+    cfg = _load_idaa_cfg()
+    if not cfg:
        return False

-    base_url = config.get('base_url', '').rstrip('/')
-    auth     = (config.get('username', 'restadmin'), config.get('password', ''))
+    base_url = cfg.get('mailman_base_url', '').rstrip('/')
+    auth     = (cfg.get('mailman_username', 'restadmin'), cfg.get('mailman_password', ''))

    payload = {
-        "list_id":            list_id.replace('@', '.'),  # Mailman uses dot notation for list IDs
-        "subscriber":         email,
-        "display_name":       display_name,
-        "pre_verified":       True,
-        "pre_confirmed":      True,
-        "pre_approved":       True,
+        "list_id":              list_id.replace('@', '.'),
+        "subscriber":           email,
+        "display_name":         display_name,
+        "pre_verified":         True,
+        "pre_confirmed":        True,
+        "pre_approved":         True,
        "send_welcome_message": False,
    }

    try:
        resp = requests.post(f"{base_url}/3.1/members", auth=auth, json=payload, timeout=10)
        if resp.status_code in (200, 201):
-            log.info(f"Subscribed {email} to {list_id}")
+            log.info("Subscribed %s to %s", email, list_id)
            return True
        if resp.status_code == 409:
-            log.debug(f"{email} already subscribed to {list_id} — skipping.")
-            return True  # Already a member, treat as success
-        log.error(f"Subscribe failed for {email}: {resp.status_code} - {resp.text[:200]}")
+            log.debug("%s already subscribed to %s — skipping.", email, list_id)
+            return True
+        log.error("Subscribe failed for %s: %s - %s", email, resp.status_code, resp.text[:200])
        return False
    except Exception as e:
-        log.exception(f"Error subscribing {email} to {list_id}: {e}")
+        log.exception("Error subscribing %s to %s: %s", email, list_id, e)
        return False


@@ -220,165 +271,178 @@ def subscribe_member_to_list(list_id: str, email: str, display_name: str = '') -
 def unsubscribe_member_from_list(list_id: str, email: str) -> bool:
    """
    Unsubscribe an email address from a Mailman 3 list.
-
-    Returns:
-        True on success or not found (already unsubscribed), False on error.
+    Returns True on success or not-found, False on error.
    """
-    config = _load_mailman_config()
-    if not config:
+    cfg = _load_idaa_cfg()
+    if not cfg:
        return False

-    base_url = config.get('base_url', '').rstrip('/')
-    auth     = (config.get('username', 'restadmin'), config.get('password', ''))
+    base_url    = cfg.get('mailman_base_url', '').rstrip('/')
+    auth        = (cfg.get('mailman_username', 'restadmin'), cfg.get('mailman_password', ''))
    list_id_dot = list_id.replace('@', '.')

    try:
-        # Mailman member ID is {email}_{list_id} (base64 encoded in REST path)
        resp = requests.delete(
            f"{base_url}/3.1/lists/{list_id_dot}/member/{email}",
            auth=auth,
-            timeout=10
+            timeout=10,
        )
        if resp.status_code in (200, 204):
-            log.info(f"Unsubscribed {email} from {list_id}")
+            log.info("Unsubscribed %s from %s", email, list_id)
            return True
        if resp.status_code == 404:
-            log.debug(f"{email} not found in {list_id} — skipping.")
-            return True  # Not a member, treat as success
-        log.error(f"Unsubscribe failed for {email}: {resp.status_code} - {resp.text[:200]}")
+            log.debug("%s not found in %s — skipping.", email, list_id)
+            return True
+        log.error("Unsubscribe failed for %s: %s - %s", email, resp.status_code, resp.text[:200])
        return False
    except Exception as e:
-        log.exception(f"Error unsubscribing {email} from {list_id}: {e}")
+        log.exception("Error unsubscribing %s from %s: %s", email, list_id, e)
        return False


-# ── Sync Engine ───────────────────────────────────────────────────────────
+# ── Mirror Sync Engine ────────────────────────────────────────────────────

@logger_reset
-def sync_single_member(email: str, display_name: str, list_id: str, is_active: bool) -> str:
+def mirror_novi_group_to_mailman_list(novi_group_guid: str, mailman_list_id: str) -> Optional[Dict]:
    """
-    Sync one member's subscription state for a given Mailman list.
+    Mirror a single Novi group to a Mailman 3 list.

-    Args:
-        email:        Member email
-        display_name: Member display name
-        list_id:      Target Mailman list ID
-        is_active:    True = subscribe, False = unsubscribe
-
-    Returns:
-        'subscribed', 'unsubscribed', or 'error'
+    - Fetches all members of `novi_group_guid` from Novi.
+    - For each member, fetches their customer record to get Email, name, and
+      membership flags.  Members with `Active=False` or `UnsubscribeFromEmails=True`
+      are excluded from the target set.
+    - Fetches current members of `mailman_list_id`.
+    - Subscribes addresses in Novi but not in Mailman.
+    - Unsubscribes addresses in Mailman but not in Novi (mirror / full reconcile).
+    - Returns a result dict with counts.
    """
-    if is_active:
-        ok = subscribe_member_to_list(list_id, email, display_name)
-        return 'subscribed' if ok else 'error'
-    else:
-        ok = unsubscribe_member_from_list(list_id, email)
-        return 'unsubscribed' if ok else 'error'
-
-
-@logger_reset
-def sync_novi_to_mailman(list_id: str, active_status: str = 'Active') -> Optional[Dict]:
-    """
-    Full sync: pull all Novi members and reconcile their Mailman subscription state.
-
-    - Novi members with `active_status` → subscribed to `list_id`
-    - All other statuses → unsubscribed from `list_id`
-
-    Args:
-        list_id:       Target Mailman list ID (e.g. 'members@yourdomain.org')
-        active_status: Novi membership status string that counts as active.
-
-    Returns:
-        Result dict with counts, or None on fatal error.
-
-    TODO: Adjust the Novi member field names (Email, FirstName, LastName,
-          MembershipStatus) to match your actual Novi instance schema.
-    """
-    log.info(f"Starting full Novi → Mailman sync for list: {list_id}")
-
-    members = get_novi_members()
-    if members is None:
-        log.error("Novi member fetch failed — aborting sync.")
+    cfg = _load_idaa_cfg()
+    if not cfg:
        return None

-    log.info(f"Fetched {len(members)} members from Novi.")
+    base_url = cfg.get('novi_api_root_url', '').rstrip('/')
+    api_key  = cfg.get('novi_idaa_api_key', '')
+    headers  = {"Authorization": f"Basic {api_key}", "Accept": "application/json"}

-    results = {"total": len(members), "subscribed": 0, "unsubscribed": 0, "error": 0, "skipped": 0}
+    log.info("Mirror sync: Novi group %s → Mailman list %s", novi_group_guid, mailman_list_id)

-    for member in members:
-        # Field names confirmed PascalCase from Novi API (verified via IDAA Jitsi integration).
-        # MembershipStatus field name still needs confirmation against the bulk member endpoint.
-        email  = (member.get('Email') or '').strip().replace(' ', '+')
-        fname  = member.get('FirstName') or ''
-        lname  = member.get('LastName')  or ''
-        status = member.get('MembershipStatus') or member.get('Status') or ''
+    # ── Step 1: Novi group → UUIDs ────────────────────────────────────────
+    try:
+        resp = requests.get(f"{base_url}/groups/{novi_group_guid}/members",
+                            headers=headers, params={"pageSize": 500}, timeout=30)
+        if resp.status_code != 200:
+            log.error("Novi group fetch failed (%s): %s", novi_group_guid, resp.status_code)
+            return None
+        uuid_list = [m['UniqueID'] for m in resp.json() if m.get('UniqueID')]
+    except Exception as e:
+        log.exception("Failed to fetch Novi group %s: %s", novi_group_guid, e)
+        return None

-        if not email:
-            results['skipped'] += 1
-            continue
+    log.info("Novi group %s has %d member(s).", novi_group_guid, len(uuid_list))

-        display_name = f"{fname} {lname}".strip()
-        is_active    = (status == active_status)
+    # ── Step 2: UUID → customer record → email ────────────────────────────
+    # email.lower() → display_name
+    novi_members: Dict[str, str] = {}
+    skipped_inactive = 0
+    skipped_unsub    = 0
+    skipped_no_email = 0

-        outcome = sync_single_member(email, display_name, list_id, is_active)
-        results[outcome] = results.get(outcome, 0) + 1
+    for uid in uuid_list:
+        try:
+            r = requests.get(f"{base_url}/customers/{uid}", headers=headers, timeout=10)
+            if r.status_code != 200:
+                log.warning("Novi customer %s fetch failed: %s", uid, r.status_code)
+                continue
+            c = r.json()
+            if not c.get('Active', False):
+                skipped_inactive += 1
+                continue
+            if c.get('UnsubscribeFromEmails', False):
+                skipped_unsub += 1
+                continue
+            email = (c.get('Email') or '').strip()
+            if not email:
+                skipped_no_email += 1
+                continue
+            display = f"{c.get('FirstName', '')} {c.get('LastName', '')}".strip()
+            novi_members[email.lower()] = display
+        except Exception as e:
+            log.exception("Failed to fetch Novi customer %s: %s", uid, e)

-    log.info(f"Novi → Mailman sync complete: {results}")
+    log.info("Novi active/subscribed members with email: %d (skipped: inactive=%d unsub=%d no_email=%d)",
+             len(novi_members), skipped_inactive, skipped_unsub, skipped_no_email)
+
+    # ── Step 3: Current Mailman members ───────────────────────────────────
+    mailman_entries = get_mailman_list_members(mailman_list_id)
+    if mailman_entries is None:
+        log.error("Could not fetch current Mailman members for %s — aborting.", mailman_list_id)
+        return None
+    mailman_emails = {m['email'].lower() for m in mailman_entries}
+
+    # ── Step 4: Diff ──────────────────────────────────────────────────────
+    novi_email_set  = set(novi_members.keys())
+    to_subscribe    = novi_email_set - mailman_emails
+    to_unsubscribe  = mailman_emails - novi_email_set
+
+    log.info("Diff — to subscribe: %d, to unsubscribe: %d", len(to_subscribe), len(to_unsubscribe))
+
+    results = {
+        "novi_group_guid":    novi_group_guid,
+        "mailman_list_id":    mailman_list_id,
+        "novi_count":         len(novi_email_set),
+        "mailman_count_before": len(mailman_emails),
+        "subscribed":         0,
+        "unsubscribed":       0,
+        "errors":             0,
+        "skipped_inactive":   skipped_inactive,
+        "skipped_unsub":      skipped_unsub,
+        "skipped_no_email":   skipped_no_email,
+    }
+
+    # ── Step 5: Apply ─────────────────────────────────────────────────────
+    for email in to_subscribe:
+        ok = subscribe_member_to_list(mailman_list_id, email, novi_members[email])
+        if ok: results['subscribed'] += 1
+        else:  results['errors'] += 1
+
+    for email in to_unsubscribe:
+        ok = unsubscribe_member_from_list(mailman_list_id, email)
+        if ok: results['unsubscribed'] += 1
+        else:  results['errors'] += 1
+
+    log.info("Mirror sync complete: %s", results)
    return results


@logger_reset
-def handle_novi_webhook(payload: Dict) -> Optional[Dict]:
+def mirror_all_configured_mappings() -> Optional[List[Dict]]:
    """
-    Process a Novi membership webhook event and update the appropriate Mailman list.
+    Run mirror_novi_group_to_mailman_list for every entry in
+    cfg_json['novi_mailman_sync'].

-    Expected payload shape (Novi webhook format — confirm against Novi docs):
-        {
-            "EventType": "MembershipActivated" | "MembershipLapsed" | "MembershipExpired" | ...,
-            "Member": {
-                "Email": "...",
-                "FirstName": "...",
-                "LastName": "...",
-                "MembershipStatus": "Active" | "Lapsed" | ...
-            }
-        }
-
-    TODO: Confirm Novi webhook payload format and EventType values.
-    TODO: Pull target list_id from data_store config or per-account settings.
+    Expected cfg_json shape:
+        "novi_mailman_sync": [
+            {"novi_group_guid": "...", "mailman_list_id": "members@idaa.org"},
+            ...
+        ]
    """
-    event_type = payload.get('EventType', '')
-    member     = payload.get('Member', {})
-    email      = (member.get('Email') or '').strip().replace(' ', '+')
-    fname      = member.get('FirstName', '')
-    lname      = member.get('LastName', '')
-    status     = member.get('MembershipStatus') or member.get('Status', '')
-
-    if not email:
-        log.warning(f"Novi webhook received with no email — skipping. Payload: {payload}")
+    cfg = _load_idaa_cfg()
+    if not cfg:
        return None

-    # Load target list_id from config
-    # TODO: Support per-account list routing (e.g. multiple orgs, each with their own list)
-    novi_config = _load_novi_config()
-    if not novi_config:
-        return None
-    list_id = novi_config.get('mailman_list_id', '')
-    if not list_id:
-        log.error("'mailman_list_id' not set in novi_api_config — cannot route webhook.")
-        return None
+    sync_map = cfg.get('novi_mailman_sync') or []
+    if not sync_map:
+        log.warning("novi_mailman_sync not configured in IDAA cfg_json.")
+        return []

-    ACTIVE_EVENTS = {'MembershipActivated', 'MembershipRenewed', 'MembershipCreated'}
-    INACTIVE_EVENTS = {'MembershipLapsed', 'MembershipExpired', 'MembershipTerminated', 'MembershipCancelled'}
+    results = []
+    for mapping in sync_map:
+        guid    = mapping.get('novi_group_guid', '').strip()
+        list_id = mapping.get('mailman_list_id', '').strip()
+        if not guid or not list_id:
+            log.warning("Skipping incomplete novi_mailman_sync entry: %s", mapping)
+            continue
+        result = mirror_novi_group_to_mailman_list(guid, list_id)
+        results.append(result or {"novi_group_guid": guid, "mailman_list_id": list_id, "error": "sync failed"})

-    if event_type in ACTIVE_EVENTS:
-        is_active = True
-    elif event_type in INACTIVE_EVENTS:
-        is_active = False
-    else:
-        log.info(f"Novi webhook: unhandled EventType '{event_type}' for {email} — ignoring.")
-        return {"action": "ignored", "reason": f"Unhandled EventType: {event_type}"}
-
-    display_name = f"{fname} {lname}".strip()
-    outcome = sync_single_member(email, display_name, list_id, is_active)
-    log.info(f"Novi webhook processed: {email} → {outcome} (EventType: {event_type})")
-    return {"action": outcome, "email": email, "event_type": event_type}
+    return results
--- a/app/routers/api_v3_actions_e_novi_mailman.py
+++ b/app/routers/api_v3_actions_e_novi_mailman.py
@@ -1,17 +1,20 @@
-from fastapi import APIRouter, Body, Depends, Query
-from typing import Optional
 import asyncio

-from app.lib_general import log
+from fastapi import APIRouter, Depends, Query
+from typing import Optional
+
 from app.lib_general_v3 import AccountContext, get_account_context, DelayParams
 from app.models.response_models import Resp_Body_Base, mk_resp
 from app.methods.e_novi_mailman_methods import (
    test_novi_connection,
    test_mailman_connection,
    get_mailman_lists,
+    get_mailman_list_members,
+    subscribe_member_to_list,
+    unsubscribe_member_from_list,
    get_novi_members,
-    sync_novi_to_mailman,
-    handle_novi_webhook,
+    mirror_novi_group_to_mailman_list,
+    mirror_all_configured_mappings,
 )

 router = APIRouter()
@@ -24,7 +27,7 @@ async def test_novi(
        account: AccountContext = Depends(get_account_context),
        delay: DelayParams = Depends(),
        ):
-    """Verify Novi AMS API credentials stored in data_store."""
+    """Verify Novi AMS API credentials from IDAA site cfg_json."""
    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
    result = test_novi_connection()
    if result.get('ok'):
@@ -37,7 +40,7 @@ async def test_mailman(
        account: AccountContext = Depends(get_account_context),
        delay: DelayParams = Depends(),
        ):
-    """Verify Mailman 3 REST API credentials stored in data_store."""
+    """Verify Mailman 3 REST API credentials from IDAA site cfg_json."""
    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
    result = test_mailman_connection()
    if result.get('ok'):
@@ -47,6 +50,53 @@ async def test_mailman(

 # ── Inspection / Preview ──────────────────────────────────────────────────

+@router.get('/mailman/lists/{list_id}/members', response_model=Resp_Body_Base)
+async def list_mailman_list_members(
+        list_id: str,
+        count: int = Query(100, ge=1, le=500),
+        page: int = Query(1, ge=1),
+        account: AccountContext = Depends(get_account_context),
+        delay: DelayParams = Depends(),
+        ):
+    """Return members of a specific Mailman 3 list."""
+    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
+    data = get_mailman_list_members(list_id=list_id, count=count, page=page)
+    if data is not None:
+        return mk_resp(data={"count": len(data), "members": data})
+    return mk_resp(data=False, status_code=500, status_message=f"Failed to fetch members for list '{list_id}'.")
+
+
+@router.post('/mailman/lists/{list_id}/subscribe', response_model=Resp_Body_Base)
+async def mailman_subscribe(
+        list_id: str,
+        email: str = Query(..., description="Email address to subscribe"),
+        display_name: str = Query('', description="Optional display name"),
+        account: AccountContext = Depends(get_account_context),
+        delay: DelayParams = Depends(),
+        ):
+    """Subscribe an email address to a Mailman 3 list."""
+    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
+    ok = subscribe_member_to_list(list_id=list_id, email=email, display_name=display_name)
+    if ok:
+        return mk_resp(data={"list_id": list_id, "email": email}, status_message="Subscribed successfully.")
+    return mk_resp(data=False, status_code=500, status_message=f"Failed to subscribe {email} to {list_id}.")
+
+
+@router.delete('/mailman/lists/{list_id}/subscribe', response_model=Resp_Body_Base)
+async def mailman_unsubscribe(
+        list_id: str,
+        email: str = Query(..., description="Email address to unsubscribe"),
+        account: AccountContext = Depends(get_account_context),
+        delay: DelayParams = Depends(),
+        ):
+    """Unsubscribe an email address from a Mailman 3 list."""
+    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
+    ok = unsubscribe_member_from_list(list_id=list_id, email=email)
+    if ok:
+        return mk_resp(data={"list_id": list_id, "email": email}, status_message="Unsubscribed successfully.")
+    return mk_resp(data=False, status_code=500, status_message=f"Failed to unsubscribe {email} from {list_id}.")
+
+
@router.get('/mailman/lists', response_model=Resp_Body_Base)
 async def list_mailman_lists(
        account: AccountContext = Depends(get_account_context),
@@ -68,7 +118,7 @@ async def list_novi_members(
        account: AccountContext = Depends(get_account_context),
        delay: DelayParams = Depends(),
        ):
-    """Fetch a page of Novi AMS members. Useful for inspecting data before a full sync."""
+    """Fetch a page of Novi AMS members."""
    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
    data = get_novi_members(status_filter=status_filter, page_size=page_size, offset=offset)
    if data is not None:
@@ -79,38 +129,36 @@ async def list_novi_members(
 # ── Sync ──────────────────────────────────────────────────────────────────

@router.post('/sync', response_model=Resp_Body_Base)
-async def sync_full(
-        list_id: str = Query(..., description="Target Mailman list ID, e.g. 'members@yourdomain.org'"),
-        active_status: str = Query('Active', description="Novi membership status that maps to 'subscribed'"),
+async def sync_all_mappings(
        account: AccountContext = Depends(get_account_context),
        delay: DelayParams = Depends(),
        ):
    """
-    Full sync: pull all Novi members and reconcile Mailman subscription state.
-    Active members are subscribed; all others are unsubscribed.
-    This is the manual / scheduled trigger — the webhook handles real-time updates.
+    Run all Novi → Mailman mirror syncs configured in novi_mailman_sync (IDAA cfg_json).
+    This is the cron target — call on a schedule to keep all lists in sync.
    """
    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
-    result = sync_novi_to_mailman(list_id=list_id, active_status=active_status)
-    if result:
-        return mk_resp(data=result, status_message="Novi → Mailman sync complete.")
-    return mk_resp(data=False, status_code=500, status_message="Sync failed.")
+    results = mirror_all_configured_mappings()
+    if results is not None:
+        return mk_resp(data=results, status_message=f"Mirror sync complete. {len(results)} mapping(s) processed.")
+    return mk_resp(data=False, status_code=500, status_message="Mirror sync failed.")


-# ── Webhook ───────────────────────────────────────────────────────────────
-
-@router.post('/webhook/novi', response_model=Resp_Body_Base, include_in_schema=True)
-async def novi_membership_webhook(
-        payload: dict = Body(...),
+@router.post('/sync/group/{novi_group_guid}', response_model=Resp_Body_Base)
+async def sync_single_group(
+        novi_group_guid: str,
+        mailman_list_id: str = Query(..., description="Target Mailman list, e.g. 'mm3@idaa.org'"),
+        account: AccountContext = Depends(get_account_context),
+        delay: DelayParams = Depends(),
        ):
    """
-    Receives Novi AMS membership webhook events and immediately updates
-    the corresponding Mailman subscription — no auth required (Novi pushes to this endpoint).
-
-    TODO: Add HMAC signature verification once Novi webhook secret is configured.
+    Mirror a single Novi group to a specific Mailman list.
+    Useful for testing or forcing a refresh of one mapping.
    """
-    log.info(f"Novi webhook received: EventType={payload.get('EventType')} Email={payload.get('Member', {}).get('Email')}")
-    result = handle_novi_webhook(payload)
+    if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
+    result = mirror_novi_group_to_mailman_list(novi_group_guid, mailman_list_id)
    if result:
-        return mk_resp(data=result)
-    return mk_resp(data=False, status_code=400, status_message="Webhook payload could not be processed.")
+        return mk_resp(data=result, status_message="Mirror sync complete.")
+    return mk_resp(data=False, status_code=500, status_message="Mirror sync failed.")
+
+
--- a/tests/README.md
+++ b/tests/README.md
@@ -28,6 +28,8 @@ These consolidated scripts are the primary verification tool for the V3 API.
 | `test_e2e_v3_demo_parity.py` | **Demo Parity + Nested Create Regression**: Vision ID check for Badge, Exhibit, Tracking; nested create lifecycle (POST+DELETE) for `journal/journal_entry` and `event/event_session`; alias resolution. **Run after any model or nested-router change.** |
 | `test_e2e_v3_action_event_file.py` | **Event Actions**: Specialized atomic upload and linking for event files. |
 | `test_e2e_v3_action_zoom.py` | **Zoom Integration**: Verifies OAuth and ticket sync logic for Zoom Events. |
+| `test_e2e_v3_action_novi_mailman.py` | **Novi-Mailman Bridge — Connections**: Verifies Novi AMS and Mailman 3 API credentials are valid (IDAA). Run first before the lists test. |
+| `test_e2e_v3_action_novi_mailman_lists.py` | **Novi-Mailman Bridge — List Operations**: Full member lifecycle — read roster, subscribe, verify, unsubscribe — against `mm3@idaa.org`, `mm3@dgrzone.com`, `mm3@oneskyit.com`. |
 | `test_e2e_v3_action_event_exhibit_tracking_export.py` | **Exhibit Leads Export**: Auth/permission guards, CSV column structure, XLSX bytes, and `return_file` mode for the V3 tracking export action. |
 | `test_e2e_v3_accounts.py` | CRUD verification for the core Account object. |
 | `test_e2e_v3_schema.py` | Network verification of the V3 metadata discovery endpoint. |
@@ -53,6 +55,7 @@ Tests exist to be used — run the relevant suite whenever you touch backend cod
 | Search / filter changes | `test_e2e_v3_search_engine.py` |
 | Auth / account context changes | `test_e2e_v3_security_audit.py`, `test_e2e_v3_auth_security.py` |
 | File upload / download changes | `test_e2e_v3_actions_file_lifecycle.py` |
+| Novi-Mailman bridge changes | `test_e2e_v3_action_novi_mailman.py`, `test_e2e_v3_action_novi_mailman_lists.py` |
 | Event exhibit tracking export changes | `test_e2e_v3_action_event_exhibit_tracking_export.py` |
 | Any backend change before frontend hand-off | All of the above |

@@ -136,4 +139,9 @@ These are IDs for records that we can use for testing. Please do not delete them

 ### Events Module (IDAA Recovery Meetings)
 * Aether test/demo Event: '1Pkd025vvxU' (36) "IDAA Recovery Meeting Test"
-* Aether test/demo Event: 'gIZgAjISkf8' (43) "IDAA Recovery Meeting Test"
+* Aether test/demo Event: 'gIZgAjISkf8' (43) "IDAA Recovery Meeting Test"
+
+### IDAA and Novi AMS
+Scott Idem (test 1)
+* Novi (customer) UUID: "1dadf11c-b74b-4582-8a0a-7ec738a033dc"
+* Novi Email: "stidem+test1@gmail.com"
--- a/tests/archive/test_novi_webhook_ARCHIVED.py
+++ b/tests/archive/test_novi_webhook_ARCHIVED.py
@@ -0,0 +1,51 @@
+"""
+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)
--- a/tests/e2e/test_e2e_v3_action_novi_mailman.py
+++ b/tests/e2e/test_e2e_v3_action_novi_mailman.py
@@ -0,0 +1,99 @@
+"""
+E2E tests for the Novi-Mailman Bridge — API connection checks.
+
+Verifies that both the Novi AMS API and Mailman 3 REST API are reachable
+and that the credentials stored in IDAA site cfg_json are valid.
+
+Run from project root:
+    ./environment/bin/python3 tests/e2e/test_e2e_v3_action_novi_mailman.py
+
+Environment:
+    AE_API_BASE  — override the target API base URL (default: https://dev-api.oneskyit.com)
+
+Related tests:
+    test_e2e_v3_action_novi_mailman_lists.py  — member read/subscribe/unsubscribe lifecycle
+    test_e2e_v3_action_novi_mailman_sync.py   — full Novi → Mailman mirror sync (TODO)
+
+Credential storage:
+    All Novi and Mailman credentials live in IDAA site cfg_json (id_random='58_gJESdlUh').
+    See project memory for key names.
+"""
+
+import os
+import sys
+import time
+import requests
+
+BASE_URL    = os.environ.get('AE_API_BASE', 'https://dev-api.oneskyit.com')
+ACTION_BASE = f"{BASE_URL}/v3/action/e_novi_mailman"
+API_KEY     = "PMM4n50teUCaOMMTN8qOJA"
+
+AUTH_HEADERS = {
+    "X-Aether-API-Key": API_KEY,
+    "x-no-account-id":  "bypass",
+}
+
+pass_count = 0
+fail_count = 0
+
+
+def print_result(label: str, success: bool, message: str = ""):
+    global pass_count, fail_count
+    status = "✅ PASS" if success else "❌ FAIL"
+    msg = f"   {message}" if message else ""
+    print(f"  [{status}] {label}{msg}")
+    if success:
+        pass_count += 1
+    else:
+        fail_count += 1
+
+
+def test_novi_connection():
+    print("\n[1] Novi API Connection")
+    try:
+        resp = requests.get(f"{ACTION_BASE}/test_connection/novi", headers=AUTH_HEADERS, timeout=15)
+        data = resp.json().get('data', {})
+        if resp.status_code == 200 and data.get('ok'):
+            print_result("Novi credentials valid", True)
+        elif resp.status_code == 401:
+            print_result("Novi credentials valid", False, f"401 — {data.get('error', resp.text[:120])}")
+        else:
+            print_result("Novi credentials valid", False, f"HTTP {resp.status_code}: {resp.text[:120]}")
+    except Exception as e:
+        print_result("Novi credentials valid", False, f"Exception: {e}")
+
+
+def test_mailman_connection():
+    print("\n[2] Mailman 3 API Connection")
+    try:
+        resp = requests.get(f"{ACTION_BASE}/test_connection/mailman", headers=AUTH_HEADERS, timeout=15)
+        data = resp.json().get('data', {})
+        if resp.status_code == 200 and data.get('ok'):
+            print_result("Mailman credentials valid", True, f"version={data.get('version', 'unknown')}")
+        elif resp.status_code == 401:
+            print_result("Mailman credentials valid", False,
+                         f"401 — {data.get('error', resp.text[:120])}  "
+                         f"(add mailman_* keys to IDAA cfg_json)")
+        else:
+            print_result("Mailman credentials valid", False, f"HTTP {resp.status_code}: {resp.text[:120]}")
+    except Exception as e:
+        print_result("Mailman credentials valid", False, f"Exception: {e}")
+
+
+if __name__ == "__main__":
+    print("=" * 60)
+    print(" Novi-Mailman Bridge — Connection Tests")
+    print(f" Target: {ACTION_BASE}")
+    print("=" * 60)
+
+    t_start = time.time()
+
+    test_novi_connection()
+    test_mailman_connection()
+
+    elapsed = time.time() - t_start
+    print(f"\n{'=' * 60}")
+    print(f" Results: {pass_count} passed, {fail_count} failed  ({elapsed:.2f}s)")
+    print("=" * 60)
+
+    sys.exit(0 if fail_count == 0 else 1)
--- a/tests/e2e/test_e2e_v3_action_novi_mailman_lists.py
+++ b/tests/e2e/test_e2e_v3_action_novi_mailman_lists.py
@@ -0,0 +1,166 @@
+"""
+E2E tests for the Novi-Mailman Bridge — Mailman list member operations.
+
+Covers the full member lifecycle:
+  1. Read current members of all TEST_LISTS
+  2. Subscribe TEST_EMAIL to the primary test list (TEST_LISTS[0])
+  3. Verify the address appears in the member roster
+  4. Unsubscribe TEST_EMAIL (cleanup)
+  5. Verify the address is removed
+
+Run from project root:
+    ./environment/bin/python3 tests/e2e/test_e2e_v3_action_novi_mailman_lists.py
+
+Configuration:
+    TEST_LISTS[0]  — list used for the subscribe/unsubscribe lifecycle test
+    TEST_EMAIL     — address used as the test subscriber (safe to add/remove)
+    TEST_LISTS[1:] — additional lists read-only (member count + roster check)
+
+Notes:
+    - Uses dot-notation for list IDs in URL paths (mm3@idaa.org → mm3.idaa.org)
+      because @ is a special character in URL paths.
+    - Mailman pre-confirms subscriptions (no confirmation email sent to TEST_EMAIL).
+    - Run test_e2e_v3_action_novi_mailman.py first to confirm both APIs are reachable.
+"""
+
+import sys
+import time
+import requests
+
+BASE_URL    = "https://dev-api.oneskyit.com/v3/action/e_novi_mailman"
+API_KEY     = "PMM4n50teUCaOMMTN8qOJA"
+TEST_EMAIL  = "scott.idem+mm3api@gmail.com"
+TEST_NAME   = "Scott Idem (MM3 API Test)"
+
+TEST_LISTS = [
+    "mm3@idaa.org",
+    "mm3@dgrzone.com",
+    "mm3@oneskyit.com",
+]
+
+HEADERS = {
+    "X-Aether-API-Key": API_KEY,
+    "x-no-account-id":  "bypass",
+}
+
+pass_count = 0
+fail_count = 0
+
+
+def print_result(label: str, success: bool, message: str = ""):
+    global pass_count, fail_count
+    status = "✅ PASS" if success else "❌ FAIL"
+    msg = f"   {message}" if message else ""
+    print(f"  [{status}] {label}{msg}")
+    if success:
+        pass_count += 1
+    else:
+        fail_count += 1
+
+
+def get_members(list_id: str) -> list | None:
+    """Return member list for a given list_id, or None on failure."""
+    list_id_dot = list_id.replace('@', '.')
+    resp = requests.get(f"{BASE_URL}/mailman/lists/{list_id_dot}/members",
+                        headers=HEADERS, timeout=15)
+    if resp.status_code == 200 and resp.json().get('meta', {}).get('success'):
+        return resp.json()['data'].get('members', [])
+    return None
+
+
+def subscribe(list_id: str, email: str, display_name: str = '') -> bool:
+    list_id_dot = list_id.replace('@', '.')
+    resp = requests.post(
+        f"{BASE_URL}/mailman/lists/{list_id_dot}/subscribe",
+        headers=HEADERS,
+        params={"email": email, "display_name": display_name},
+        timeout=15,
+    )
+    return resp.status_code == 200 and resp.json().get('meta', {}).get('success')
+
+
+def unsubscribe(list_id: str, email: str) -> bool:
+    list_id_dot = list_id.replace('@', '.')
+    resp = requests.delete(
+        f"{BASE_URL}/mailman/lists/{list_id_dot}/subscribe",
+        headers=HEADERS,
+        params={"email": email},
+        timeout=15,
+    )
+    return resp.status_code == 200 and resp.json().get('meta', {}).get('success')
+
+
+# ── Tests ─────────────────────────────────────────────────────────────────
+
+def test_read_members():
+    print("\n[1] Read Members")
+    for list_id in TEST_LISTS:
+        members = get_members(list_id)
+        if members is not None:
+            emails = [m['email'] for m in members]
+            print_result(f"Read {list_id}", True, f"{len(members)} member(s): {', '.join(emails) or 'none'}")
+        else:
+            print_result(f"Read {list_id}", False, "request failed")
+
+
+def test_subscribe():
+    print(f"\n[2] Subscribe {TEST_EMAIL}")
+    list_id = TEST_LISTS[0]  # mm3@idaa.org
+    ok = subscribe(list_id, TEST_EMAIL, TEST_NAME)
+    print_result(f"Subscribe to {list_id}", ok)
+
+
+def test_verify_subscription():
+    print(f"\n[3] Verify {TEST_EMAIL} appears in member list")
+    list_id = TEST_LISTS[0]
+    members = get_members(list_id)
+    if members is None:
+        print_result(f"Verify subscription in {list_id}", False, "could not fetch members")
+        return
+    found = any(m['email'] == TEST_EMAIL for m in members)
+    print_result(f"Found {TEST_EMAIL} in {list_id}", found,
+                 "" if found else "not found after subscribe")
+
+
+def test_unsubscribe():
+    print(f"\n[4] Unsubscribe {TEST_EMAIL} (cleanup)")
+    list_id = TEST_LISTS[0]
+    ok = unsubscribe(list_id, TEST_EMAIL)
+    print_result(f"Unsubscribe from {list_id}", ok)
+
+
+def test_verify_unsubscription():
+    print(f"\n[5] Verify {TEST_EMAIL} removed from member list")
+    list_id = TEST_LISTS[0]
+    members = get_members(list_id)
+    if members is None:
+        print_result(f"Verify removal from {list_id}", False, "could not fetch members")
+        return
+    still_present = any(m['email'] == TEST_EMAIL for m in members)
+    print_result(f"{TEST_EMAIL} removed from {list_id}", not still_present,
+                 "still present after unsubscribe" if still_present else "")
+
+
+# ── Main ──────────────────────────────────────────────────────────────────
+
+if __name__ == "__main__":
+    print("=" * 60)
+    print(" Novi-Mailman Bridge — List Member Operations E2E")
+    print(f" Target: {BASE_URL}")
+    print(f" Test email: {TEST_EMAIL}")
+    print("=" * 60)
+
+    t_start = time.time()
+
+    test_read_members()
+    test_subscribe()
+    test_verify_subscription()
+    test_unsubscribe()
+    test_verify_unsubscription()
+
+    elapsed = time.time() - t_start
+    print(f"\n{'=' * 60}")
+    print(f" Results: {pass_count} passed, {fail_count} failed  ({elapsed:.2f}s)")
+    print("=" * 60)
+
+    sys.exit(0 if fail_count == 0 else 1)