Tax rate wrong when shipping address differs from customer default

pip install requests

export MAGENTO_URL="https://yourstore.example.com"
export MAGENTO_ADMIN_USERNAME="admin"
export MAGENTO_ADMIN_PASSWORD="change-me"
export ORDER_IDS="1001,1002,1003"
export RATE_EPSILON="0.05"
export DRY_RUN="true"   # report only by default
export REPAIR_CONFIRM="false"   # set true only with DRY_RUN=false to post a review comment

// Node 18+ has fetch built in, no dependencies needed

export MAGENTO_URL="https://yourstore.example.com"
export MAGENTO_ADMIN_USERNAME="admin"
export MAGENTO_ADMIN_PASSWORD="change-me"
export ORDER_IDS="1001,1002,1003"
export RATE_EPSILON="0.05"
export DRY_RUN="true"   // report only by default
export REPAIR_CONFIRM="false"   // set true only with DRY_RUN=false to post a review comment

import os, requests

MAGENTO_URL = os.environ["MAGENTO_URL"].rstrip("/")

def get_token(username, password):
    r = requests.post(
        f"{MAGENTO_URL}/rest/V1/integration/admin/token",
        json={"username": username, "password": password},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()

def get_store_tax_based_on(token):
    r = requests.get(
        f"{MAGENTO_URL}/rest/V1/store/storeConfigs",
        headers={"Authorization": f"Bearer {token}"},
        timeout=30,
    )
    r.raise_for_status()
    configs = r.json()
    # Adobe Commerce exposes this as a store config extension attribute or a
    # dedicated config value depending on version; fall back to "shipping".
    first = configs[0] if configs else {}
    return (first.get("extension_attributes", {}) or {}).get(
        "tax_calculation_based_on", "shipping"
    )

def get_order(token, order_id):
    r = requests.get(
        f"{MAGENTO_URL}/rest/V1/orders/{order_id}",
        headers={"Authorization": f"Bearer {token}"},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()

const MAGENTO_URL = (process.env.MAGENTO_URL || "").replace(/\/$/, "");

async function getToken(username, password) {
  const res = await fetch(`${MAGENTO_URL}/rest/V1/integration/admin/token`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ username, password }),
  });
  if (!res.ok) throw new Error(`Magento ${res.status}`);
  return res.json();
}

async function getStoreTaxBasedOn(token) {
  const res = await fetch(`${MAGENTO_URL}/rest/V1/store/storeConfigs`, {
    headers: { Authorization: `Bearer ${token}` },
  });
  if (!res.ok) throw new Error(`Magento ${res.status}`);
  const configs = await res.json();
  const first = configs[0] || {};
  return (first.extension_attributes || {}).tax_calculation_based_on || "shipping";
}

async function getOrder(token, orderId) {
  const res = await fetch(`${MAGENTO_URL}/rest/V1/orders/${orderId}`, {
    headers: { Authorization: `Bearer ${token}` },
  });
  if (!res.ok) throw new Error(`Magento ${res.status}`);
  return res.json();
}

def search_all(token, path, page_size=100):
    items = []
    page = 1
    while True:
        params = {
            "searchCriteria[pageSize]": page_size,
            "searchCriteria[currentPage]": page,
        }
        r = requests.get(
            f"{MAGENTO_URL}/rest/V1/{path}",
            params=params,
            headers={"Authorization": f"Bearer {token}"},
            timeout=30,
        )
        r.raise_for_status()
        body = r.json()
        batch = body.get("items", [])
        items.extend(batch)
        if len(batch) < page_size:
            return items
        page += 1

def get_tax_rules(token):
    return search_all(token, "taxRules/search")

def get_tax_rates(token):
    return search_all(token, "taxRates/search")

async function searchAll(token, path, pageSize = 100) {
  const items = [];
  let page = 1;
  while (true) {
    const params = new URLSearchParams({
      "searchCriteria[pageSize]": String(pageSize),
      "searchCriteria[currentPage]": String(page),
    });
    const res = await fetch(`${MAGENTO_URL}/rest/V1/${path}?${params}`, {
      headers: { Authorization: `Bearer ${token}` },
    });
    if (!res.ok) throw new Error(`Magento ${res.status}`);
    const body = await res.json();
    const batch = body.items || [];
    items.push(...batch);
    if (batch.length < pageSize) return items;
    page += 1;
  }
}

async function getTaxRules(token) {
  return searchAll(token, "taxRules/search");
}

async function getTaxRates(token) {
  return searchAll(token, "taxRates/search");
}

def _rate_matches_address(rate, address):
    if str(rate.get("tax_country_id")) != str(address.get("country_id")):
        return False
    region_id = rate.get("tax_region_id")
    if region_id not in (None, 0, "0") and str(region_id) != str(address.get("region_id")):
        return False
    postcode = rate.get("tax_postcode") or "*"
    if postcode in ("*", ""):
        return True
    if "-" in postcode:
        lo, hi = postcode.split("-", 1)
        pc = address.get("postcode") or ""
        return lo <= pc <= hi
    return postcode == address.get("postcode")


def expected_tax_rate(resolved_address, customer_tax_class_id, product_tax_class_id, tax_rules, tax_rates):
    rates_by_id = {r["id"]: r for r in tax_rates}

    candidate_rules = [
        rule for rule in tax_rules
        if customer_tax_class_id in (rule.get("customer_tax_class_ids") or [])
        and product_tax_class_id in (rule.get("product_tax_class_ids") or [])
    ]
    candidate_rules.sort(key=lambda rule: rule.get("priority", 0))

    for rule in candidate_rules:
        matched_rate_total = 0.0
        matched_any = False
        for rate_id in rule.get("tax_rate_ids") or []:
            rate = rates_by_id.get(rate_id)
            if not rate:
                continue
            if _rate_matches_address(rate, resolved_address):
                matched_rate_total += float(rate.get("rate", 0) or 0)
                matched_any = True
        if matched_any:
            return {"expectedRate": matched_rate_total, "matchedRuleId": rule.get("id")}

    return {"expectedRate": 0.0, "matchedRuleId": None}

function rateMatchesAddress(rate, address) {
  if (String(rate.tax_country_id) !== String(address.country_id)) return false;
  const regionId = rate.tax_region_id;
  if (regionId !== null && regionId !== 0 && regionId !== "0" && String(regionId) !== String(address.region_id)) {
    return false;
  }
  const postcode = rate.tax_postcode || "*";
  if (postcode === "*" || postcode === "") return true;
  if (postcode.includes("-")) {
    const [lo, hi] = postcode.split("-");
    const pc = address.postcode || "";
    return lo <= pc && pc <= hi;
  }
  return postcode === address.postcode;
}

export function expectedTaxRate(resolvedAddress, customerTaxClassId, productTaxClassId, taxRules, taxRates) {
  const ratesById = new Map(taxRates.map((r) => [r.id, r]));

  const candidateRules = taxRules
    .filter((rule) =>
      (rule.customer_tax_class_ids || []).includes(customerTaxClassId) &&
      (rule.product_tax_class_ids || []).includes(productTaxClassId)
    )
    .sort((a, b) => (a.priority || 0) - (b.priority || 0));

  for (const rule of candidateRules) {
    let matchedRateTotal = 0;
    let matchedAny = false;
    for (const rateId of rule.tax_rate_ids || []) {
      const rate = ratesById.get(rateId);
      if (!rate) continue;
      if (rateMatchesAddress(rate, resolvedAddress)) {
        matchedRateTotal += Number(rate.rate || 0);
        matchedAny = true;
      }
    }
    if (matchedAny) return { expectedRate: matchedRateTotal, matchedRuleId: rule.id };
  }

  return { expectedRate: 0, matchedRuleId: null };
}

def detect_tax_mismatch(order_actual_rate, expected_result, epsilon=0.05):
    delta = abs(order_actual_rate - expected_result["expectedRate"])
    return {
        "isMismatch": delta > epsilon,
        "expectedRate": expected_result["expectedRate"],
        "actualRate": order_actual_rate,
        "delta": round(delta, 4),
        "matchedRuleId": expected_result["matchedRuleId"],
    }


def is_default_address_leak(shipping_customer_address_id, default_shipping_id, default_billing_id):
    if shipping_customer_address_id is None:
        return False
    return (
        str(shipping_customer_address_id) != str(default_shipping_id)
        and str(shipping_customer_address_id) != str(default_billing_id)
    )

export function detectTaxMismatch(orderActualRate, expectedResult, epsilon = 0.05) {
  const delta = Math.abs(orderActualRate - expectedResult.expectedRate);
  return {
    isMismatch: delta > epsilon,
    expectedRate: expectedResult.expectedRate,
    actualRate: orderActualRate,
    delta: Math.round(delta * 10000) / 10000,
    matchedRuleId: expectedResult.matchedRuleId,
  };
}

export function isDefaultAddressLeak(shippingCustomerAddressId, defaultShippingId, defaultBillingId) {
  if (shippingCustomerAddressId === null || shippingCustomerAddressId === undefined) return false;
  return (
    String(shippingCustomerAddressId) !== String(defaultShippingId) &&
    String(shippingCustomerAddressId) !== String(defaultBillingId)
  );
}

"""Detect Magento 2 or Adobe Commerce orders where the applied tax rate does
not match the address the store's own Tax Calculation Based On setting says
should have been used.

Magento resolves the applicable tax zone using the address selected by the
store-wide Tax Calculation Based On setting (Stores, Configuration, Sales,
Tax, Calculation Settings), which can be Billing Address, Shipping Address,
or Shipping Origin. For a logged-in customer with more than one saved
address, quote and order totals collection can resolve the tax class against
the customer's default address record instead of re-resolving it against the
shipping address actually selected at checkout, especially across
multi-address customers or multi-country carts. This is confirmed in
magento2 issue 38232, where a French address was taxed at 0% because the
customer's default Belgium address was used instead. The tax rule engine
itself is deterministic; the defect is an address resolution problem
upstream of rule matching, not a rule configuration error.

This script never rewrites tax_amount on a placed order, since there is no
supported REST endpoint for that. It independently computes the expected
rate for the address the store's based_on setting points at, compares it to
what the order actually applied, and separately flags any order whose
shipping address customer_address_id differs from the customer's own
default_shipping or default_billing id, the highest risk signature of this
leak. It writes a report row for every order it flags and exits non-zero so
CI or alerting notices. A human reconciles a confirmed mismatch with a
credit memo to refund the wrong tax line, followed by a corrected invoice.
Only with DRY_RUN=false and REPAIR_CONFIRM=true does it post a documentation
comment via /rest/V1/orders/{id}/comments; it never mutates tax or money on
its own. Safe to run again and again.
"""
import os
import csv
import sys
import logging
import requests

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

MAGENTO_URL = os.environ.get("MAGENTO_URL", "https://example.test").rstrip("/")
ADMIN_USERNAME = os.environ.get("MAGENTO_ADMIN_USERNAME", "admin")
ADMIN_PASSWORD = os.environ.get("MAGENTO_ADMIN_PASSWORD", "change-me")
ADMIN_TOKEN = os.environ.get("MAGENTO_ADMIN_TOKEN")
ORDER_IDS = [o.strip() for o in os.environ.get("ORDER_IDS", "").split(",") if o.strip()]
RATE_EPSILON = float(os.environ.get("RATE_EPSILON", "0.05"))
DRY_RUN = os.environ.get("DRY_RUN", "true").lower() == "true"
REPAIR_CONFIRM = os.environ.get("REPAIR_CONFIRM", "false").lower() == "true"
OUTPUT_CSV = os.environ.get("OUTPUT_CSV", "tax_address_mismatches.csv")
PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "100"))


def get_token():
    if ADMIN_TOKEN:
        return ADMIN_TOKEN
    r = requests.post(
        f"{MAGENTO_URL}/rest/V1/integration/admin/token",
        json={"username": ADMIN_USERNAME, "password": ADMIN_PASSWORD},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()


def get_store_tax_based_on(token):
    r = requests.get(
        f"{MAGENTO_URL}/rest/V1/store/storeConfigs",
        headers={"Authorization": f"Bearer {token}"},
        timeout=30,
    )
    r.raise_for_status()
    configs = r.json()
    first = configs[0] if configs else {}
    return (first.get("extension_attributes", {}) or {}).get(
        "tax_calculation_based_on", "shipping"
    )


def get_order(token, order_id):
    r = requests.get(
        f"{MAGENTO_URL}/rest/V1/orders/{order_id}",
        headers={"Authorization": f"Bearer {token}"},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()


def get_customer(token, customer_id):
    r = requests.get(
        f"{MAGENTO_URL}/rest/V1/customers/{customer_id}",
        headers={"Authorization": f"Bearer {token}"},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()


def search_all(token, path, page_size=PAGE_SIZE):
    items = []
    page = 1
    while True:
        params = {
            "searchCriteria[pageSize]": page_size,
            "searchCriteria[currentPage]": page,
        }
        r = requests.get(
            f"{MAGENTO_URL}/rest/V1/{path}",
            params=params,
            headers={"Authorization": f"Bearer {token}"},
            timeout=30,
        )
        r.raise_for_status()
        body = r.json()
        batch = body.get("items", [])
        items.extend(batch)
        if len(batch) < page_size:
            return items
        page += 1


def get_tax_rules(token):
    return search_all(token, "taxRules/search")


def get_tax_rates(token):
    return search_all(token, "taxRates/search")


def post_order_comment(token, order_id, message):
    r = requests.post(
        f"{MAGENTO_URL}/rest/V1/orders/{order_id}/comments",
        json={"statusHistory": {"comment": message, "isVisibleOnFront": 0}},
        headers={"Authorization": f"Bearer {token}"},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()


def _rate_matches_address(rate, address):
    if str(rate.get("tax_country_id")) != str(address.get("country_id")):
        return False
    region_id = rate.get("tax_region_id")
    if region_id not in (None, 0, "0") and str(region_id) != str(address.get("region_id")):
        return False
    postcode = rate.get("tax_postcode") or "*"
    if postcode in ("*", ""):
        return True
    if "-" in postcode:
        lo, hi = postcode.split("-", 1)
        pc = address.get("postcode") or ""
        return lo <= pc <= hi
    return postcode == address.get("postcode")


def expected_tax_rate(resolved_address, customer_tax_class_id, product_tax_class_id, tax_rules, tax_rates):
    rates_by_id = {r["id"]: r for r in tax_rates}

    candidate_rules = [
        rule for rule in tax_rules
        if customer_tax_class_id in (rule.get("customer_tax_class_ids") or [])
        and product_tax_class_id in (rule.get("product_tax_class_ids") or [])
    ]
    candidate_rules.sort(key=lambda rule: rule.get("priority", 0))

    for rule in candidate_rules:
        matched_rate_total = 0.0
        matched_any = False
        for rate_id in rule.get("tax_rate_ids") or []:
            rate = rates_by_id.get(rate_id)
            if not rate:
                continue
            if _rate_matches_address(rate, resolved_address):
                matched_rate_total += float(rate.get("rate", 0) or 0)
                matched_any = True
        if matched_any:
            return {"expectedRate": matched_rate_total, "matchedRuleId": rule.get("id")}

    return {"expectedRate": 0.0, "matchedRuleId": None}


def detect_tax_mismatch(order_actual_rate, expected_result, epsilon=RATE_EPSILON):
    delta = abs(order_actual_rate - expected_result["expectedRate"])
    return {
        "isMismatch": delta > epsilon,
        "expectedRate": expected_result["expectedRate"],
        "actualRate": order_actual_rate,
        "delta": round(delta, 4),
        "matchedRuleId": expected_result["matchedRuleId"],
    }


def is_default_address_leak(shipping_customer_address_id, default_shipping_id, default_billing_id):
    if shipping_customer_address_id is None:
        return False
    return (
        str(shipping_customer_address_id) != str(default_shipping_id)
        and str(shipping_customer_address_id) != str(default_billing_id)
    )


def resolved_address_for_order(order, based_on):
    ext = order.get("extension_attributes", {}) or {}
    assignments = ext.get("shipping_assignments") or []
    shipping_address = {}
    if assignments:
        shipping_address = (assignments[0].get("shipping") or {}).get("address") or {}
    billing_address = order.get("billing_address") or {}
    if based_on == "billing":
        return billing_address
    return shipping_address


def order_actual_rate(order):
    applied = order.get("applied_taxes") or []
    if applied:
        return float(applied[0].get("percent", applied[0].get("rate", 0)) or 0)
    items = order.get("items") or []
    for item in items:
        if item.get("tax_percent") is not None:
            return float(item["tax_percent"])
    return 0.0


def build_report_row(order, mismatch, leak):
    return {
        "order_id": order.get("entity_id"),
        "increment_id": order.get("increment_id"),
        "expected_rate": mismatch["expectedRate"],
        "actual_rate": mismatch["actualRate"],
        "delta": mismatch["delta"],
        "matched_rule_id": mismatch["matchedRuleId"],
        "default_address_leak": leak,
    }


def run():
    token = get_token()
    based_on = get_store_tax_based_on(token)
    tax_rules = get_tax_rules(token)
    tax_rates = get_tax_rates(token)
    flagged = []

    for order_id in ORDER_IDS:
        order = get_order(token, order_id)
        address = resolved_address_for_order(order, based_on)
        if not address:
            continue

        customer_tax_class_id = order.get("customer_tax_class_id")
        items = order.get("items") or []
        product_tax_class_id = items[0].get("tax_class_id") if items else None

        expected = expected_tax_rate(address, customer_tax_class_id, product_tax_class_id, tax_rules, tax_rates)
        actual_rate = order_actual_rate(order)
        mismatch = detect_tax_mismatch(actual_rate, expected)

        customer_id = order.get("customer_id")
        leak = False
        if customer_id:
            customer = get_customer(token, customer_id)
            ext = order.get("extension_attributes", {}) or {}
            assignments = ext.get("shipping_assignments") or []
            shipping_customer_address_id = None
            if assignments:
                shipping_customer_address_id = (assignments[0].get("shipping") or {}).get("address", {}).get("customer_address_id")
            leak = is_default_address_leak(
                shipping_customer_address_id,
                customer.get("default_shipping"),
                customer.get("default_billing"),
            )

        if not mismatch["isMismatch"] and not leak:
            continue

        row = build_report_row(order, mismatch, leak)
        flagged.append(row)
        log.warning(
            "Order %s tax mismatch: expected_rate=%s actual_rate=%s delta=%s default_address_leak=%s",
            row["increment_id"], row["expected_rate"], row["actual_rate"], row["delta"], row["default_address_leak"],
        )

        if not DRY_RUN and REPAIR_CONFIRM:
            post_order_comment(
                token, order_id,
                f"Tax review: expected rate {row['expected_rate']}%, applied rate {row['actual_rate']}%, "
                f"delta {row['delta']}. Possible default-address leak: {row['default_address_leak']}. "
                "Flagged for finance review; no tax or money was changed automatically.",
            )

    if flagged:
        with open(OUTPUT_CSV, "w", newline="") as fh:
            writer = csv.DictWriter(fh, fieldnames=[
                "order_id", "increment_id", "expected_rate", "actual_rate", "delta",
                "matched_rule_id", "default_address_leak",
            ])
            writer.writeheader()
            writer.writerows(flagged)
        log.info("Wrote report to %s%s", OUTPUT_CSV, "" if not DRY_RUN else " (dry run, report only)")

    log.info("Done. %d order(s) flagged with a tax or address mismatch.", len(flagged))
    return flagged


if __name__ == "__main__":
    flagged_orders = run()
    sys.exit(1 if flagged_orders else 0)

/**
 * Detect Magento 2 or Adobe Commerce orders where the applied tax rate does
 * not match the address the store's own Tax Calculation Based On setting
 * says should have been used.
 *
 * Magento resolves the applicable tax zone using the address selected by the
 * store-wide Tax Calculation Based On setting (Stores, Configuration, Sales,
 * Tax, Calculation Settings), which can be Billing Address, Shipping
 * Address, or Shipping Origin. For a logged-in customer with more than one
 * saved address, quote and order totals collection can resolve the tax
 * class against the customer's default address record instead of
 * re-resolving it against the shipping address actually selected at
 * checkout, especially across multi-address customers or multi-country
 * carts. This is confirmed in magento2 issue 38232, where a French address
 * was taxed at 0% because the customer's default Belgium address was used
 * instead. The tax rule engine itself is deterministic; the defect is an
 * address resolution problem upstream of rule matching, not a rule
 * configuration error.
 *
 * This script never rewrites tax_amount on a placed order, since there is
 * no supported REST endpoint for that. It independently computes the
 * expected rate for the address the store's based_on setting points at,
 * compares it to what the order actually applied, and separately flags any
 * order whose shipping address customer_address_id differs from the
 * customer's own default_shipping or default_billing id, the highest risk
 * signature of this leak. It writes a report row for every order it flags
 * and exits non-zero so CI or alerting notices. A human reconciles a
 * confirmed mismatch with a credit memo to refund the wrong tax line,
 * followed by a corrected invoice. Only with DRY_RUN=false and
 * REPAIR_CONFIRM=true does it post a documentation comment via
 * /rest/V1/orders/{id}/comments; it never mutates tax or money on its own.
 * Safe to run again and again.
 *
 * Guide: https://www.allanninal.dev/magento/tax-rate-wrong-for-shipping-address/
 */
import { pathToFileURL } from "node:url";

const MAGENTO_URL = (process.env.MAGENTO_URL || "https://example.test").replace(/\/$/, "");
const ADMIN_USERNAME = process.env.MAGENTO_ADMIN_USERNAME || "admin";
const ADMIN_PASSWORD = process.env.MAGENTO_ADMIN_PASSWORD || "change-me";
const ADMIN_TOKEN = process.env.MAGENTO_ADMIN_TOKEN || "";
const ORDER_IDS = (process.env.ORDER_IDS || "").split(",").map((o) => o.trim()).filter(Boolean);
const RATE_EPSILON = Number(process.env.RATE_EPSILON || 0.05);
const DRY_RUN = (process.env.DRY_RUN || "true").toLowerCase() === "true";
const REPAIR_CONFIRM = (process.env.REPAIR_CONFIRM || "false").toLowerCase() === "true";
const PAGE_SIZE = Number(process.env.PAGE_SIZE || 100);

function rateMatchesAddress(rate, address) {
  if (String(rate.tax_country_id) !== String(address.country_id)) return false;
  const regionId = rate.tax_region_id;
  if (regionId !== null && regionId !== undefined && regionId !== 0 && regionId !== "0" && String(regionId) !== String(address.region_id)) {
    return false;
  }
  const postcode = rate.tax_postcode || "*";
  if (postcode === "*" || postcode === "") return true;
  if (postcode.includes("-")) {
    const [lo, hi] = postcode.split("-");
    const pc = address.postcode || "";
    return lo <= pc && pc <= hi;
  }
  return postcode === address.postcode;
}

export function expectedTaxRate(resolvedAddress, customerTaxClassId, productTaxClassId, taxRules, taxRates) {
  const ratesById = new Map(taxRates.map((r) => [r.id, r]));

  const candidateRules = taxRules
    .filter((rule) =>
      (rule.customer_tax_class_ids || []).includes(customerTaxClassId) &&
      (rule.product_tax_class_ids || []).includes(productTaxClassId)
    )
    .sort((a, b) => (a.priority || 0) - (b.priority || 0));

  for (const rule of candidateRules) {
    let matchedRateTotal = 0;
    let matchedAny = false;
    for (const rateId of rule.tax_rate_ids || []) {
      const rate = ratesById.get(rateId);
      if (!rate) continue;
      if (rateMatchesAddress(rate, resolvedAddress)) {
        matchedRateTotal += Number(rate.rate || 0);
        matchedAny = true;
      }
    }
    if (matchedAny) return { expectedRate: matchedRateTotal, matchedRuleId: rule.id };
  }

  return { expectedRate: 0, matchedRuleId: null };
}

export function detectTaxMismatch(orderActualRate, expectedResult, epsilon = RATE_EPSILON) {
  const delta = Math.abs(orderActualRate - expectedResult.expectedRate);
  return {
    isMismatch: delta > epsilon,
    expectedRate: expectedResult.expectedRate,
    actualRate: orderActualRate,
    delta: Math.round(delta * 10000) / 10000,
    matchedRuleId: expectedResult.matchedRuleId,
  };
}

export function isDefaultAddressLeak(shippingCustomerAddressId, defaultShippingId, defaultBillingId) {
  if (shippingCustomerAddressId === null || shippingCustomerAddressId === undefined) return false;
  return (
    String(shippingCustomerAddressId) !== String(defaultShippingId) &&
    String(shippingCustomerAddressId) !== String(defaultBillingId)
  );
}

export function resolvedAddressForOrder(order, basedOn) {
  const ext = order.extension_attributes || {};
  const assignments = ext.shipping_assignments || [];
  let shippingAddress = {};
  if (assignments.length) {
    shippingAddress = (assignments[0].shipping || {}).address || {};
  }
  const billingAddress = order.billing_address || {};
  return basedOn === "billing" ? billingAddress : shippingAddress;
}

export function orderActualRate(order) {
  const applied = order.applied_taxes || [];
  if (applied.length) return Number(applied[0].percent ?? applied[0].rate ?? 0);
  const items = order.items || [];
  for (const item of items) {
    if (item.tax_percent !== undefined && item.tax_percent !== null) return Number(item.tax_percent);
  }
  return 0;
}

export function buildReportRow(order, mismatch, leak) {
  return {
    order_id: order.entity_id,
    increment_id: order.increment_id,
    expected_rate: mismatch.expectedRate,
    actual_rate: mismatch.actualRate,
    delta: mismatch.delta,
    matched_rule_id: mismatch.matchedRuleId,
    default_address_leak: leak,
  };
}

async function getToken() {
  if (ADMIN_TOKEN) return ADMIN_TOKEN;
  const res = await fetch(`${MAGENTO_URL}/rest/V1/integration/admin/token`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ username: ADMIN_USERNAME, password: ADMIN_PASSWORD }),
  });
  if (!res.ok) throw new Error(`Magento ${res.status}`);
  return res.json();
}

async function getStoreTaxBasedOn(token) {
  const res = await fetch(`${MAGENTO_URL}/rest/V1/store/storeConfigs`, {
    headers: { Authorization: `Bearer ${token}` },
  });
  if (!res.ok) throw new Error(`Magento ${res.status}`);
  const configs = await res.json();
  const first = configs[0] || {};
  return (first.extension_attributes || {}).tax_calculation_based_on || "shipping";
}

async function getOrder(token, orderId) {
  const res = await fetch(`${MAGENTO_URL}/rest/V1/orders/${orderId}`, {
    headers: { Authorization: `Bearer ${token}` },
  });
  if (!res.ok) throw new Error(`Magento ${res.status}`);
  return res.json();
}

async function getCustomer(token, customerId) {
  const res = await fetch(`${MAGENTO_URL}/rest/V1/customers/${customerId}`, {
    headers: { Authorization: `Bearer ${token}` },
  });
  if (!res.ok) throw new Error(`Magento ${res.status}`);
  return res.json();
}

async function searchAll(token, path, pageSize = PAGE_SIZE) {
  const items = [];
  let page = 1;
  while (true) {
    const params = new URLSearchParams({
      "searchCriteria[pageSize]": String(pageSize),
      "searchCriteria[currentPage]": String(page),
    });
    const res = await fetch(`${MAGENTO_URL}/rest/V1/${path}?${params}`, {
      headers: { Authorization: `Bearer ${token}` },
    });
    if (!res.ok) throw new Error(`Magento ${res.status}`);
    const body = await res.json();
    const batch = body.items || [];
    items.push(...batch);
    if (batch.length < pageSize) return items;
    page += 1;
  }
}

async function getTaxRules(token) {
  return searchAll(token, "taxRules/search");
}

async function getTaxRates(token) {
  return searchAll(token, "taxRates/search");
}

async function postOrderComment(token, orderId, message) {
  const res = await fetch(`${MAGENTO_URL}/rest/V1/orders/${orderId}/comments`, {
    method: "POST",
    headers: { Authorization: `Bearer ${token}`, "Content-Type": "application/json" },
    body: JSON.stringify({ statusHistory: { comment: message, isVisibleOnFront: 0 } }),
  });
  if (!res.ok) throw new Error(`Magento ${res.status}`);
  return res.json();
}

export async function run() {
  const token = await getToken();
  const basedOn = await getStoreTaxBasedOn(token);
  const taxRules = await getTaxRules(token);
  const taxRates = await getTaxRates(token);
  const flagged = [];

  for (const orderId of ORDER_IDS) {
    const order = await getOrder(token, orderId);
    const address = resolvedAddressForOrder(order, basedOn);
    if (!address || Object.keys(address).length === 0) continue;

    const customerTaxClassId = order.customer_tax_class_id;
    const items = order.items || [];
    const productTaxClassId = items.length ? items[0].tax_class_id : null;

    const expected = expectedTaxRate(address, customerTaxClassId, productTaxClassId, taxRules, taxRates);
    const actualRate = orderActualRate(order);
    const mismatch = detectTaxMismatch(actualRate, expected);

    let leak = false;
    if (order.customer_id) {
      const customer = await getCustomer(token, order.customer_id);
      const ext = order.extension_attributes || {};
      const assignments = ext.shipping_assignments || [];
      let shippingCustomerAddressId = null;
      if (assignments.length) {
        shippingCustomerAddressId = ((assignments[0].shipping || {}).address || {}).customer_address_id;
      }
      leak = isDefaultAddressLeak(shippingCustomerAddressId, customer.default_shipping, customer.default_billing);
    }

    if (!mismatch.isMismatch && !leak) continue;

    const row = buildReportRow(order, mismatch, leak);
    flagged.push(row);
    console.warn(`Order ${row.increment_id} tax mismatch: expected_rate=${row.expected_rate} actual_rate=${row.actual_rate} delta=${row.delta} default_address_leak=${row.default_address_leak}`);

    if (!DRY_RUN && REPAIR_CONFIRM) {
      await postOrderComment(
        token, orderId,
        `Tax review: expected rate ${row.expected_rate}%, applied rate ${row.actual_rate}%, delta ${row.delta}. ` +
        `Possible default-address leak: ${row.default_address_leak}. Flagged for finance review; no tax or money was changed automatically.`,
      );
    }
  }

  console.log(`Done. ${flagged.length} order(s) flagged with a tax or address mismatch.${DRY_RUN ? " (dry run, report only)" : ""}`);
  return flagged;
}

if (import.meta.url === pathToFileURL(process.argv[1]).href) {
  run()
    .then((flagged) => { if (flagged.length) process.exit(1); })
    .catch((err) => { console.error(err); process.exit(1); });
}

from detect_tax_address_mismatch import (
    expected_tax_rate,
    detect_tax_mismatch,
    is_default_address_leak,
)

TAX_RATES = [
    {"id": 1, "tax_country_id": "BE", "tax_region_id": 0, "tax_postcode": "*", "rate": 0.0},
    {"id": 2, "tax_country_id": "FR", "tax_region_id": 0, "tax_postcode": "*", "rate": 20.0},
    {"id": 3, "tax_country_id": "US", "tax_region_id": 12, "tax_postcode": "90001-90099", "rate": 8.25},
]

TAX_RULES = [
    {"id": 1, "priority": 0, "customer_tax_class_ids": [3], "product_tax_class_ids": [2], "tax_rate_ids": [1, 2]},
    {"id": 2, "priority": 1, "customer_tax_class_ids": [3], "product_tax_class_ids": [2], "tax_rate_ids": [3]},
]


def test_french_shipping_address_expects_french_vat():
    france = {"country_id": "FR", "region_id": None, "postcode": "75001"}
    result = expected_tax_rate(france, 3, 2, TAX_RULES, TAX_RATES)
    assert result["expectedRate"] == 20.0
    assert result["matchedRuleId"] == 1


def test_belgium_default_address_expects_zero():
    belgium = {"country_id": "BE", "region_id": None, "postcode": "1000"}
    result = expected_tax_rate(belgium, 3, 2, TAX_RULES, TAX_RATES)
    assert result["expectedRate"] == 0.0


def test_issue_38232_style_mismatch_is_detected():
    # order shipped to France but was taxed as if the address were Belgium (0%)
    france = {"country_id": "FR", "region_id": None, "postcode": "75001"}
    expected = expected_tax_rate(france, 3, 2, TAX_RULES, TAX_RATES)
    mismatch = detect_tax_mismatch(order_actual_rate=0.0, expected_result=expected)
    assert mismatch["isMismatch"] is True
    assert mismatch["expectedRate"] == 20.0
    assert mismatch["delta"] == 20.0


def test_matching_rate_is_not_a_mismatch():
    france = {"country_id": "FR", "region_id": None, "postcode": "75001"}
    expected = expected_tax_rate(france, 3, 2, TAX_RULES, TAX_RATES)
    mismatch = detect_tax_mismatch(order_actual_rate=20.0, expected_result=expected)
    assert mismatch["isMismatch"] is False


def test_within_epsilon_is_not_a_mismatch():
    france = {"country_id": "FR", "region_id": None, "postcode": "75001"}
    expected = expected_tax_rate(france, 3, 2, TAX_RULES, TAX_RATES)
    mismatch = detect_tax_mismatch(order_actual_rate=19.98, expected_result=expected, epsilon=0.05)
    assert mismatch["isMismatch"] is False


def test_us_postcode_range_rate_matches():
    address = {"country_id": "US", "region_id": 12, "postcode": "90045"}
    result = expected_tax_rate(address, 3, 2, TAX_RULES, TAX_RATES)
    assert result["expectedRate"] == 8.25
    assert result["matchedRuleId"] == 2


def test_default_address_leak_detected_when_shipping_id_differs():
    assert is_default_address_leak(shipping_customer_address_id=42, default_shipping_id=7, default_billing_id=7) is True


def test_no_leak_when_shipping_matches_default():
    assert is_default_address_leak(shipping_customer_address_id=7, default_shipping_id=7, default_billing_id=9) is False


def test_no_leak_when_no_customer_address_id_present():
    assert is_default_address_leak(shipping_customer_address_id=None, default_shipping_id=7, default_billing_id=9) is False

import { test } from "node:test";
import assert from "node:assert/strict";
import {
  expectedTaxRate,
  detectTaxMismatch,
  isDefaultAddressLeak,
} from "./detect-tax-address-mismatch.js";

const TAX_RATES = [
  { id: 1, tax_country_id: "BE", tax_region_id: 0, tax_postcode: "*", rate: 0.0 },
  { id: 2, tax_country_id: "FR", tax_region_id: 0, tax_postcode: "*", rate: 20.0 },
  { id: 3, tax_country_id: "US", tax_region_id: 12, tax_postcode: "90001-90099", rate: 8.25 },
];

const TAX_RULES = [
  { id: 1, priority: 0, customer_tax_class_ids: [3], product_tax_class_ids: [2], tax_rate_ids: [1, 2] },
  { id: 2, priority: 1, customer_tax_class_ids: [3], product_tax_class_ids: [2], tax_rate_ids: [3] },
];

test("French shipping address expects French VAT", () => {
  const france = { country_id: "FR", region_id: null, postcode: "75001" };
  const result = expectedTaxRate(france, 3, 2, TAX_RULES, TAX_RATES);
  assert.equal(result.expectedRate, 20.0);
  assert.equal(result.matchedRuleId, 1);
});

test("Belgium default address expects zero", () => {
  const belgium = { country_id: "BE", region_id: null, postcode: "1000" };
  const result = expectedTaxRate(belgium, 3, 2, TAX_RULES, TAX_RATES);
  assert.equal(result.expectedRate, 0.0);
});

test("issue 38232 style mismatch is detected", () => {
  const france = { country_id: "FR", region_id: null, postcode: "75001" };
  const expected = expectedTaxRate(france, 3, 2, TAX_RULES, TAX_RATES);
  const mismatch = detectTaxMismatch(0.0, expected);
  assert.equal(mismatch.isMismatch, true);
  assert.equal(mismatch.expectedRate, 20.0);
  assert.equal(mismatch.delta, 20.0);
});

test("matching rate is not a mismatch", () => {
  const france = { country_id: "FR", region_id: null, postcode: "75001" };
  const expected = expectedTaxRate(france, 3, 2, TAX_RULES, TAX_RATES);
  const mismatch = detectTaxMismatch(20.0, expected);
  assert.equal(mismatch.isMismatch, false);
});

test("within epsilon is not a mismatch", () => {
  const france = { country_id: "FR", region_id: null, postcode: "75001" };
  const expected = expectedTaxRate(france, 3, 2, TAX_RULES, TAX_RATES);
  const mismatch = detectTaxMismatch(19.98, expected, 0.05);
  assert.equal(mismatch.isMismatch, false);
});

test("US postcode range rate matches", () => {
  const address = { country_id: "US", region_id: 12, postcode: "90045" };
  const result = expectedTaxRate(address, 3, 2, TAX_RULES, TAX_RATES);
  assert.equal(result.expectedRate, 8.25);
  assert.equal(result.matchedRuleId, 2);
});

test("default address leak detected when shipping id differs", () => {
  assert.equal(isDefaultAddressLeak(42, 7, 7), true);
});

test("no leak when shipping matches default", () => {
  assert.equal(isDefaultAddressLeak(7, 7, 9), false);
});

test("no leak when no customer address id present", () => {
  assert.equal(isDefaultAddressLeak(null, 7, 9), false);
});