a bundle is noncompliant on a field only Evaluations names

"""Report the exact fields that make a regulatory Bundle noncompliant.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. Creating a fresh evaluation is a write, so
this reads the most recent one and prints what it found; the repair is printed
for a human to run.
"""
import argparse
import datetime
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("twilio_bundle_evaluation_audit")

NUMBERS = "https://numbers.twilio.com/v2"

CHECKABLE = ("draft", "twilio-rejected")


def parse_date(value):
    """Parse an ISO 8601 timestamp from the numbers v2 API into aware UTC.

    fromisoformat on Python 3.9 rejects a trailing Z, and comparing a naive
    datetime against an aware one raises, so both are normalised here rather
    than at every call site.
    """
    text = str(value or "").strip()
    if not text:
        return None
    try:
        parsed = datetime.datetime.fromisoformat(text.replace("Z", "+00:00"))
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=datetime.timezone.utc)
    return parsed.astimezone(datetime.timezone.utc)


def latest_evaluation(evaluations):
    """The most recent run by date_created, or None.

    Chosen by date rather than by position in the response. A bundle can carry a
    dozen evaluations, and trusting the API's ordering means the report changes
    meaning if that ordering ever changes.
    """
    dated = [(parse_date(e.get("date_created")), e) for e in evaluations or []]
    dated = [(d, e) for d, e in dated if d is not None]
    if not dated:
        return (evaluations or [None])[0]
    return max(dated, key=lambda pair: pair[0])[1]


def failures(evaluation):
    """Flatten one Evaluation into the attributes that did not pass. Pure, so
    the nesting can be tested without a network.

    Two levels matter. results[] is per requirement; results[].invalid[] is per
    attribute, and object_field in there is the only thing that names what to
    correct. A failed requirement with an empty invalid[] is the missing-document
    case: there is no attribute to blame, and dropping it would hide the most
    basic failure the evaluation reports.

    Returns a list of (requirement, object_type, field, reason).
    """
    out = []
    for result in (evaluation or {}).get("results") or []:
        if result.get("passed"):
            continue
        requirement = (result.get("requirement_friendly_name")
                       or result.get("requirement_name")
                       or "unnamed requirement")
        object_type = result.get("object_type") or "unknown object type"
        invalid = result.get("invalid") or []
        if not invalid:
            reason = (result.get("failure_reason")
                      or ("error %s" % result["error_code"]
                          if result.get("error_code") is not None else None)
                      or "no reason given at requirement level")
            out.append((requirement, object_type, "(no field named)", str(reason)))
            continue
        for field in invalid:
            name = (field.get("object_field") or field.get("friendly_name")
                    or "(unnamed field)")
            out.append((requirement, object_type, name,
                        str(field.get("failure_reason") or "no reason given")))
    return out


def verdict(evaluation):
    """Classify the most recent evaluation of one bundle. Pure.

    Returns (state, detail).
    """
    if not evaluation:
        return ("never-evaluated",
                "no evaluation has ever been run on this bundle. The check is "
                "free and exhaustive, and nothing has asked for it.")

    status = str(evaluation.get("status") or "").strip().lower()
    bad = failures(evaluation)

    if status == "compliant":
        return ("compliant",
                "the run passed every requirement in the regulation. That is a "
                "statement about the moment it ran, not a live status.")

    if status == "noncompliant":
        return ("noncompliant",
                "%d attribute(s) failed. The names below are the fields to "
                "correct on the assigned End-User or Supporting Document."
                % len(bad))

    return ("unknown",
            "evaluation status is %s, which this script does not classify. %d "
            "attribute(s) are marked failed regardless."
            % (status or "unset", len(bad)))


def staleness(evaluation, bundle):
    """Whether the evaluation predates the bundle's last edit, or None.

    An evaluation is a snapshot. Edit an End-User afterwards and the record does
    not move, so a compliant run older than date_updated is evidence about a
    version of the bundle that no longer exists.
    """
    if not evaluation:
        return None
    ran = parse_date(evaluation.get("date_created"))
    edited = parse_date(bundle.get("date_updated"))
    if ran is None or edited is None or ran >= edited:
        return None
    return ("this evaluation ran %s, before the bundle was last updated at %s: "
            "it describes an earlier version of the bundle and only a fresh run "
            "can say what is true now."
            % (ran.isoformat(), edited.isoformat()))


def get(session, url, **params):
    r = session.get(url, params=params, timeout=30)
    if r.status_code in (401, 403):
        raise SystemExit("%d from Twilio: check TWILIO_ACCOUNT_SID and that the "
                         "API key belongs to that account with read access"
                         % r.status_code)
    r.raise_for_status()
    return r.json()


def list_v2(session, url, limit, **params):
    """Page a numbers v2 collection: rows under `results`, next page absolute in
    `meta.next_page_url`."""
    params = dict(params, PageSize=50)
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("results", []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out[:limit]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--all", action="store_true",
                    help="check every bundle rather than only draft and rejected ones")
    ap.add_argument("--max-bundles", type=int, default=200,
                    help="stop after this many bundles")
    args = ap.parse_args()

    account = os.environ.get("TWILIO_ACCOUNT_SID")
    key = os.environ.get("TWILIO_API_KEY")
    secret = os.environ.get("TWILIO_API_SECRET")
    if not (account and key and secret):
        log.error("set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET "
                  "(an API Key with read access, not the auth token)")
        return 2

    session = requests.Session()
    session.auth = (key, secret)

    bundles = list_v2(session, "%s/RegulatoryCompliance/Bundles" % NUMBERS,
                      args.max_bundles)
    if not args.all:
        bundles = [b for b in bundles
                   if str(b.get("status") or "").strip().lower() in CHECKABLE]
    if not bundles:
        log.info("no bundles in a state worth evaluating")
        return 0

    noncompliant = unevaluated = 0
    for bundle in bundles:
        sid = bundle.get("sid", "?")
        runs = list_v2(session, "%s/RegulatoryCompliance/Bundles/%s/Evaluations"
                       % (NUMBERS, sid), 100)
        evaluation = latest_evaluation(runs)
        state, detail = verdict(evaluation)
        label = "%s/%s" % (bundle.get("iso_country") or "??",
                           bundle.get("number_type") or "?")
        line = "%-15s %s  %s  %s" % (state, sid, label, detail)

        if state == "compliant":
            log.info(line)
            note = staleness(evaluation, bundle)
            if note:
                log.warning("  %s", note)
            continue

        if state == "never-evaluated":
            unevaluated += 1
        else:
            noncompliant += 1
        log.warning(line)

        for requirement, object_type, field, reason in failures(evaluation):
            log.warning("  %s [%s] %s: %s", requirement, object_type, field, reason)
        note = staleness(evaluation, bundle)
        if note:
            log.warning("  %s", note)
        log.warning("  repair: correct the named object_field on the assigned "
                    "End-User or Supporting Document, reassign it, then ask for a "
                    "fresh evaluation at %s/RegulatoryCompliance/Bundles/%s/"
                    "Evaluations before submitting", NUMBERS, sid)

    log.info("%d bundle(s) checked, %d noncompliant, %d never evaluated",
             len(bundles), noncompliant, unevaluated)
    return 1 if (noncompliant or unevaluated) else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Report the exact fields that make a regulatory Bundle noncompliant.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. Creating a fresh evaluation is a
 * write, so this reads the most recent one and prints what it found.
 */
const NUMBERS = 'https://numbers.twilio.com/v2';

const CHECKABLE = ['draft', 'twilio-rejected'];

/** Parse an ISO 8601 timestamp from the numbers v2 API. */
export function parseDate(value) {
  const text = String(value ?? '').trim();
  if (!text) return null;
  const ms = Date.parse(text);
  return Number.isNaN(ms) ? null : new Date(ms);
}

/**
 * The most recent run by date_created, or null. Chosen by date rather than by
 * position, so the report does not change meaning if the API's ordering does.
 */
export function latestEvaluation(evaluations) {
  const dated = (evaluations ?? [])
    .map((e) => [parseDate(e.date_created), e])
    .filter(([d]) => d !== null);
  if (dated.length === 0) return (evaluations ?? [])[0] ?? null;
  return dated.reduce((best, cur) => (cur[0] > best[0] ? cur : best))[1];
}

/**
 * Flatten one Evaluation into the attributes that did not pass. Pure, so the
 * nesting can be tested without a network.
 *
 * results[] is per requirement; results[].invalid[] is per attribute, and
 * object_field in there names what to correct. A failed requirement with an
 * empty invalid[] is the missing-document case and must still be reported.
 *
 * Returns an array of [requirement, objectType, field, reason].
 */
export function failures(evaluation) {
  const out = [];
  for (const result of evaluation?.results ?? []) {
    if (result.passed) continue;
    const requirement = result.requirement_friendly_name
      ?? result.requirement_name ?? 'unnamed requirement';
    const objectType = result.object_type ?? 'unknown object type';
    const invalid = result.invalid ?? [];
    if (invalid.length === 0) {
      const reason = result.failure_reason
        ?? (result.error_code != null ? `error ${result.error_code}` : null)
        ?? 'no reason given at requirement level';
      out.push([requirement, objectType, '(no field named)', String(reason)]);
      continue;
    }
    for (const field of invalid) {
      const name = field.object_field ?? field.friendly_name ?? '(unnamed field)';
      out.push([requirement, objectType, name,
                String(field.failure_reason ?? 'no reason given')]);
    }
  }
  return out;
}

/** Classify the most recent evaluation of one bundle. Returns [state, detail]. */
export function verdict(evaluation) {
  if (!evaluation) {
    return ['never-evaluated',
      'no evaluation has ever been run on this bundle. The check is free and ' +
      'exhaustive, and nothing has asked for it.'];
  }

  const status = String(evaluation.status ?? '').trim().toLowerCase();
  const bad = failures(evaluation);

  if (status === 'compliant') {
    return ['compliant',
      'the run passed every requirement in the regulation. That is a statement ' +
      'about the moment it ran, not a live status.'];
  }

  if (status === 'noncompliant') {
    return ['noncompliant',
      `${bad.length} attribute(s) failed. The names below are the fields to ` +
      'correct on the assigned End-User or Supporting Document.'];
  }

  return ['unknown',
    `evaluation status is ${status || 'unset'}, which this script does not ` +
    `classify. ${bad.length} attribute(s) are marked failed regardless.`];
}

/** Whether the evaluation predates the bundle's last edit, or null. */
export function staleness(evaluation, bundle) {
  if (!evaluation) return null;
  const ran = parseDate(evaluation.date_created);
  const edited = parseDate(bundle.date_updated);
  if (ran === null || edited === null || ran >= edited) return null;
  return `this evaluation ran ${ran.toISOString()}, before the bundle was last ` +
         `updated at ${edited.toISOString()}: it describes an earlier version of ` +
         'the bundle and only a fresh run can say what is true now.';
}

function authHeader(key, secret) {
  return `Basic ${Buffer.from(`${key}:${secret}`).toString('base64')}`;
}

async function get(auth, url, params = {}) {
  const u = new URL(url);
  for (const [k, v] of Object.entries(params)) u.searchParams.set(k, v);
  const res = await fetch(u, { headers: { Authorization: auth } });
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from Twilio: check TWILIO_ACCOUNT_SID and ` +
                    'that the API key belongs to that account with read access');
  }
  if (!res.ok) throw new Error(`${res.status} from ${u.pathname}`);
  return res.json();
}

export async function listV2(auth, url, limit = 200) {
  let params = { PageSize: 50 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.results ?? []));
    url = page.meta?.next_page_url ?? null;
    params = {};
  }
  return out.slice(0, limit);
}

async function main() {
  const account = process.env.TWILIO_ACCOUNT_SID;
  const key = process.env.TWILIO_API_KEY;
  const secret = process.env.TWILIO_API_SECRET;
  if (!account || !key || !secret) {
    console.error('set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET ' +
                  '(an API Key with read access, not the auth token)');
    process.exitCode = 2;
    return;
  }
  const auth = authHeader(key, secret);
  const all = process.argv.includes('--all');

  let bundles = await listV2(auth, `${NUMBERS}/RegulatoryCompliance/Bundles`, 200);
  if (!all) {
    bundles = bundles.filter(
      (b) => CHECKABLE.includes(String(b.status ?? '').trim().toLowerCase()));
  }
  if (bundles.length === 0) {
    console.log('no bundles in a state worth evaluating');
    return;
  }

  let noncompliant = 0;
  let unevaluated = 0;
  for (const bundle of bundles) {
    const sid = bundle.sid ?? '?';
    const runs = await listV2(
      auth, `${NUMBERS}/RegulatoryCompliance/Bundles/${sid}/Evaluations`, 100);
    const evaluation = latestEvaluation(runs);
    const [state, detail] = verdict(evaluation);
    const label = `${bundle.iso_country ?? '??'}/${bundle.number_type ?? '?'}`;
    const line = `${state.padEnd(15)} ${sid}  ${label}  ${detail}`;

    if (state === 'compliant') {
      console.log(line);
      const fresh = staleness(evaluation, bundle);
      if (fresh) console.warn(`  ${fresh}`);
      continue;
    }

    if (state === 'never-evaluated') unevaluated += 1; else noncompliant += 1;
    console.warn(line);

    for (const [requirement, objectType, field, reason] of failures(evaluation)) {
      console.warn(`  ${requirement} [${objectType}] ${field}: ${reason}`);
    }
    const note = staleness(evaluation, bundle);
    if (note) console.warn(`  ${note}`);
    console.warn('  repair: correct the named object_field on the assigned ' +
                 'End-User or Supporting Document, reassign it, then ask for a ' +
                 `fresh evaluation at ${NUMBERS}/RegulatoryCompliance/Bundles/` +
                 `${sid}/Evaluations before submitting`);
  }

  console.log(`${bundles.length} bundle(s) checked, ${noncompliant} noncompliant, ` +
              `${unevaluated} never evaluated`);
  process.exitCode = (noncompliant || unevaluated) ? 1 : 0;
}

// Only run when invoked directly. The test file imports this module, and without
// the guard main() would run there too, fail on the missing credentials and set a
// non-zero exit code that fails the whole test file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from twilio_bundle_evaluation_audit import (failures, latest_evaluation,
                                              parse_date, staleness, verdict)

BUNDLE = {"sid": "BU00000000000000000000000000000002",
          "iso_country": "FR",
          "number_type": "national",
          "status": "draft",
          "date_updated": "2026-08-20T10:00:00Z"}

NONCOMPLIANT = {
    "sid": "EL00000000000000000000000000000001",
    "status": "noncompliant",
    "date_created": "2026-08-25T09:00:00Z",
    "results": [
        {"requirement_friendly_name": "Business Name",
         "requirement_name": "business_name_info",
         "object_type": "business",
         "passed": True,
         "invalid": []},
        {"requirement_friendly_name": "Business Identity",
         "requirement_name": "business_identity_info",
         "object_type": "business",
         "passed": False,
         "failure_reason": "one or more attributes are invalid",
         "invalid": [
             {"friendly_name": "Business Registration Number",
              "object_field": "business_registration_number",
              "failure_reason": "value does not match the expected format"},
             {"friendly_name": "Business Address Country",
              "object_field": "iso_country",
              "failure_reason": "address country does not match the regulation"},
         ]},
    ],
}


def test_failed_attributes_are_listed_one_per_field():
    rows = failures(NONCOMPLIANT)
    assert len(rows) == 2
    assert [r[2] for r in rows] == ["business_registration_number", "iso_country"]
    assert rows[0][0] == "Business Identity"
    assert "expected format" in rows[0][3]


def test_a_passing_requirement_is_not_reported():
    assert all(r[0] != "Business Name" for r in failures(NONCOMPLIANT))


def test_a_failure_with_no_invalid_entries_still_produces_a_row():
    # The missing-document case: nothing to name, and the most basic failure
    # there is. A reader that only walks invalid[] shows an empty report.
    evaluation = {"status": "noncompliant",
                  "results": [{"requirement_friendly_name": "Address",
                               "object_type": "supporting_document",
                               "passed": False,
                               "error_code": 22215,
                               "invalid": []}]}
    rows = failures(evaluation)
    assert len(rows) == 1
    assert rows[0][2] == "(no field named)"
    assert "22215" in rows[0][3]


def test_verdict_counts_the_attributes_rather_than_the_requirements():
    state, detail = verdict(NONCOMPLIANT)
    assert state == "noncompliant"
    assert "2 attribute(s)" in detail


def test_a_bundle_with_no_evaluation_is_its_own_state():
    state, detail = verdict(None)
    assert state == "never-evaluated"
    assert "free and exhaustive" in detail


def test_the_latest_run_is_chosen_by_date_not_by_position():
    old = {"sid": "EL1", "date_created": "2026-01-01T00:00:00Z"}
    new = {"sid": "EL2", "date_created": "2026-08-25T09:00:00Z"}
    assert latest_evaluation([new, old])["sid"] == "EL2"
    assert latest_evaluation([old, new])["sid"] == "EL2"
    assert latest_evaluation([]) is None


def test_a_compliant_run_older_than_the_last_edit_is_flagged_as_stale():
    compliant = {"status": "compliant", "date_created": "2026-08-01T00:00:00Z"}
    note = staleness(compliant, BUNDLE)
    assert note is not None
    assert "earlier version" in note
    fresh = {"status": "compliant", "date_created": "2026-08-25T09:00:00Z"}
    assert staleness(fresh, BUNDLE) is None


def test_dates_parse_with_a_trailing_z():
    assert parse_date("2026-08-25T09:00:00Z").hour == 9
    assert parse_date("") is None
    assert parse_date("not a date") is None

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  failures, latestEvaluation, parseDate, staleness, verdict,
} from './twilio-bundle-evaluation-audit.mjs';

const BUNDLE = {
  sid: 'BU00000000000000000000000000000002',
  iso_country: 'FR',
  number_type: 'national',
  status: 'draft',
  date_updated: '2026-08-20T10:00:00Z',
};

const NONCOMPLIANT = {
  sid: 'EL00000000000000000000000000000001',
  status: 'noncompliant',
  date_created: '2026-08-25T09:00:00Z',
  results: [
    {
      requirement_friendly_name: 'Business Name',
      requirement_name: 'business_name_info',
      object_type: 'business',
      passed: true,
      invalid: [],
    },
    {
      requirement_friendly_name: 'Business Identity',
      requirement_name: 'business_identity_info',
      object_type: 'business',
      passed: false,
      failure_reason: 'one or more attributes are invalid',
      invalid: [
        {
          friendly_name: 'Business Registration Number',
          object_field: 'business_registration_number',
          failure_reason: 'value does not match the expected format',
        },
        {
          friendly_name: 'Business Address Country',
          object_field: 'iso_country',
          failure_reason: 'address country does not match the regulation',
        },
      ],
    },
  ],
};

test('failed attributes are listed one per field', () => {
  const rows = failures(NONCOMPLIANT);
  assert.equal(rows.length, 2);
  assert.deepEqual(rows.map((r) => r[2]),
                   ['business_registration_number', 'iso_country']);
  assert.equal(rows[0][0], 'Business Identity');
  assert.match(rows[0][3], /expected format/);
});

test('a passing requirement is not reported', () => {
  assert.ok(failures(NONCOMPLIANT).every((r) => r[0] !== 'Business Name'));
});

test('a failure with no invalid entries still produces a row', () => {
  const evaluation = {
    status: 'noncompliant',
    results: [{
      requirement_friendly_name: 'Address',
      object_type: 'supporting_document',
      passed: false,
      error_code: 22215,
      invalid: [],
    }],
  };
  const rows = failures(evaluation);
  assert.equal(rows.length, 1);
  assert.equal(rows[0][2], '(no field named)');
  assert.match(rows[0][3], /22215/);
});

test('verdict counts the attributes rather than the requirements', () => {
  const [state, detail] = verdict(NONCOMPLIANT);
  assert.equal(state, 'noncompliant');
  assert.match(detail, /2 attribute\(s\)/);
});

test('a bundle with no evaluation is its own state', () => {
  const [state, detail] = verdict(null);
  assert.equal(state, 'never-evaluated');
  assert.match(detail, /free and exhaustive/);
});

test('the latest run is chosen by date not by position', () => {
  const old = { sid: 'EL1', date_created: '2026-01-01T00:00:00Z' };
  const fresh = { sid: 'EL2', date_created: '2026-08-25T09:00:00Z' };
  assert.equal(latestEvaluation([fresh, old]).sid, 'EL2');
  assert.equal(latestEvaluation([old, fresh]).sid, 'EL2');
  assert.equal(latestEvaluation([]), null);
});

test('a compliant run older than the last edit is flagged as stale', () => {
  const note = staleness(
    { status: 'compliant', date_created: '2026-08-01T00:00:00Z' }, BUNDLE);
  assert.notEqual(note, null);
  assert.match(note, /earlier version/);
  assert.equal(
    staleness({ status: 'compliant', date_created: '2026-08-25T09:00:00Z' }, BUNDLE),
    null);
});

test('dates parse with a trailing z', () => {
  assert.equal(parseDate('2026-08-25T09:00:00Z').getUTCHours(), 9);
  assert.equal(parseDate(''), null);
  assert.equal(parseDate('not a date'), null);
});