"""NetSuite runner + TBA client behavior.

Two layers:

* **Client (OAuth 1.0a TBA signing)** — host derivation, account-id
  validation, and an independent re-computation of the HMAC-SHA256
  signature for a fixed nonce/timestamp to prove the base-string +
  signing-key construction matches the OAuth spec.
* **Runner decision tree** — wire calls are stubbed at
  ``NetSuiteClient.request`` so the tests pin the RUNNER's logic
  (eid: upsert, missing-key guards, default-subsidiary injection,
  deactivate idempotency), not httpx plumbing.

Vendor response shapes (NetSuite REST record API):

* Successful upsert → ``204`` + ``Location: …/employee/12345``.
* Error → ``{"o:errorDetails": [{"detail": "...", "o:errorCode": "..."}]}``.
"""
from __future__ import annotations

import base64
import hashlib
import hmac
from typing import Any
from urllib.parse import quote

import httpx
import pytest

from app.domains.identity_sync.runners.netsuite import (
    NetSuiteClient,
    NetSuiteDepartmentRunner,
    NetSuiteError,
    NetSuiteRunner,
    _internal_id_from_location,
    _pencode,
)

# ``asyncio_mode = auto`` (pyproject) collects the async tests here
# without a module-level marker — which would otherwise warn on the
# sync client/signature tests in this same file.


def _client(**kw: Any) -> NetSuiteClient:
    base = dict(
        account_id="7429932",
        consumer_key="ck",
        consumer_secret="cs",
        token_id="ti",
        token_secret="ts",
    )
    base.update(kw)
    return NetSuiteClient(**base)


# ── Scripted request stub ─────────────────────────────────────────────
class _ScriptedCalls:
    """Replays (status, body, raw, location) per (method, path-prefix)."""

    def __init__(self, script: list[tuple[str, str, tuple[int, Any, str, str | None]]]):
        self.script = list(script)
        self.calls: list[tuple[str, str, dict | None]] = []

    async def __call__(
        self, method: str, path: str, *,
        json_body: dict | None = None, params: dict | None = None,
        retry_on_429: int = 2,
    ):
        self.calls.append((method, path, json_body))
        for i, (m, prefix, resp) in enumerate(self.script):
            if m == method and path.startswith(prefix):
                self.script.pop(i)
                return resp
        raise AssertionError(f"unexpected call {method} {path}")


def _runner(script, **runner_kw) -> tuple[NetSuiteRunner, _ScriptedCalls]:
    runner = NetSuiteRunner(client=_client(), **runner_kw)
    calls = _ScriptedCalls(script)
    runner._client.request = calls  # type: ignore[method-assign]
    return runner, calls


_PAYLOAD = {
    "externalId": "P01431",
    "firstName": "Shuai",
    "lastName": "Feng",
    "email": "shuai.feng@pingcap.com",
    "subsidiary": {"id": "3"},
}


# ── Client: host derivation + account-id validation ────────────────────
class TestClientHost:
    def test_prod_numeric_host(self):
        c = _client(account_id="7429932")
        assert c.host == "7429932.suitetalk.api.netsuite.com"

    def test_sandbox_underscore_becomes_dash(self):
        c = _client(account_id="1234567_SB1")
        assert c.host == "1234567-sb1.suitetalk.api.netsuite.com"

    def test_bad_account_id_rejected(self):
        # A host-injection attempt must be refused at construction.
        with pytest.raises(ValueError):
            _client(account_id="evil.com/x")

    def test_missing_credential_rejected(self):
        with pytest.raises(ValueError):
            _client(token_secret="")


# ── Client: OAuth 1.0a signature correctness ───────────────────────────
class TestOAuthSignature:
    def test_signature_matches_independent_recompute(self, monkeypatch):
        c = _client()
        # Pin nonce + timestamp so we can recompute independently.
        import app.domains.identity_sync.runners.netsuite as mod
        monkeypatch.setattr(mod.time, "time", lambda: 1_700_000_000.0)
        monkeypatch.setattr(mod.time, "time_ns", lambda: 1_700_000_000_000_000_000)

        method = "PUT"
        url = (
            "https://7429932.suitetalk.api.netsuite.com"
            "/services/rest/record/v1/employee/eid:P01431"
        )
        header = c._auth_header(method, url, None)

        # Parse the header back into a param dict.
        assert header.startswith("OAuth ")
        parts = header[len("OAuth "):].split(", ")
        params: dict[str, str] = {}
        for p in parts:
            k, v = p.split("=", 1)
            params[k] = v.strip('"')

        assert params["realm"] == "7429932"
        assert params["oauth_signature_method"] == "HMAC-SHA256"
        assert params["oauth_consumer_key"] == quote("ck", safe="")
        assert params["oauth_token"] == quote("ti", safe="")
        assert params["oauth_version"] == "1.0"

        # Independent recompute of the signature from the SIGNED params
        # (everything except oauth_signature), URL-decoded.
        from urllib.parse import unquote
        signed = {
            unquote(k): unquote(v)
            for k, v in params.items()
            if k not in ("realm", "oauth_signature")
        }
        param_pairs = sorted(
            (quote(k, safe=""), quote(v, safe="")) for k, v in signed.items()
        )
        param_string = "&".join(f"{k}={v}" for k, v in param_pairs)
        base_string = "&".join([
            method.upper(),
            quote(url, safe=""),
            quote(param_string, safe=""),
        ])
        signing_key = f"{quote('cs', safe='')}&{quote('ts', safe='')}"
        expected = base64.b64encode(
            hmac.new(
                signing_key.encode(), base_string.encode(), hashlib.sha256,
            ).digest()
        ).decode()
        # The header value is percent-encoded; decode before compare.
        assert unquote(params["oauth_signature"]) == expected

    def test_pencode_rfc3986(self):
        # Reserved chars must be encoded; unreserved must not.
        assert _pencode("a b/c:d") == "a%20b%2Fc%3Ad"
        assert _pencode("Aa0-._~") == "Aa0-._~"


# ── Runner: upsert (UPDATE-ONLY via PATCH; create gated) ───────────────
# Default mode addresses by ``eid:{工号}`` and UPDATES via PATCH (which
# never creates and merges, preserving omitted SWC/Authing-era fields). A
# 404 is ``excluded`` ("not adopted"), NOT a blind create — the guard
# against duplicating the un-adopted existing population. ``create_missing``
# opts into PUT-creating genuine new hires once the population is adopted.
class TestUpsert:
    async def test_update_via_patch_success(self):
        runner, calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (204, None, "", None)),
        ])
        out = await runner.upsert(
            person_email="shuai.feng@pingcap.com",
            payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "success"
        assert out.external_id == "eid:P01431"
        assert out.detail["op"] == "update"
        method, path, body = calls.calls[0]
        assert method == "PATCH"  # UPDATE, never a blind PUT-create
        assert path.endswith("/employee/eid:P01431")
        assert "externalId" not in body  # stripped → eid: URL key
        assert body["email"] == "shuai.feng@pingcap.com"

    async def test_missing_external_id_fails_without_call(self):
        runner, calls = _runner([])
        payload = {k: v for k, v in _PAYLOAD.items() if k != "externalId"}
        out = await runner.upsert(
            person_email="x@pingcap.com", payload=payload,
            prior_external_id=None,
        )
        assert out.status == "failed"
        assert "externalId" in (out.error or "")
        assert calls.calls == []  # never hit the wire

    async def test_not_adopted_404_is_excluded_not_created(self):
        # THE crux safety: a 404 in update-only mode is NOT a create. An
        # un-adopted existing employee (empty externalId in NetSuite) must
        # never be duplicated — exactly ONE call (the PATCH), no PUT.
        runner, calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (404, {"o:errorDetails": [
                 {"o:errorCode": "NONEXISTENT_EXTERNAL_ID", "detail": "x"}]},
              "", None)),
        ])
        out = await runner.upsert(
            person_email="x@pingcap.com", payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "excluded"
        assert out.detail["reason"] == "netsuite_not_adopted"
        assert out.external_id == "eid:P01431"
        assert len(calls.calls) == 1  # NO create attempted
        assert calls.calls[0][0] == "PATCH"

    async def test_update_does_not_inject_default_subsidiary(self):
        # On UPDATE we must NOT inject a default subsidiary — the existing
        # record already has the right one; overwriting it from a half-
        # filled map is the damage update-only mode exists to prevent.
        runner, calls = _runner(
            [("PATCH", "/services/rest/record/v1/employee/eid:P01431",
              (204, None, "", None))],
            default_subsidiary_id="9",
        )
        payload = {k: v for k, v in _PAYLOAD.items() if k != "subsidiary"}
        out = await runner.upsert(
            person_email="x@pingcap.com", payload=payload,
            prior_external_id=None,
        )
        assert out.status == "success"
        _m, _p, body = calls.calls[0]
        assert "subsidiary" not in body  # NOT injected on update

    async def test_create_missing_creates_on_404(self):
        # Opt-in: after adoption a 404 means new hire → PATCH then PUT.
        runner, calls = _runner(
            [("PATCH", "/services/rest/record/v1/employee/eid:P01431",
              (404, {"o:errorDetails": [
                  {"o:errorCode": "NONEXISTENT_EXTERNAL_ID"}]}, "", None)),
             ("PUT", "/services/rest/record/v1/employee/eid:P01431",
              (204, None, "",
               "https://7429932.suitetalk.api.netsuite.com"
               "/services/rest/record/v1/employee/999"))],
            create_missing=True,
        )
        out = await runner.upsert(
            person_email="x@pingcap.com", payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "success"
        assert out.external_id == "999"
        assert out.detail["op"] == "create"
        assert [c[0] for c in calls.calls] == ["PATCH", "PUT"]

    async def test_create_missing_injects_default_subsidiary(self):
        runner, calls = _runner(
            [("PATCH", "/services/rest/record/v1/employee/eid:P01431",
              (404, {"o:errorDetails": [
                  {"o:errorCode": "NONEXISTENT_EXTERNAL_ID"}]}, "", None)),
             ("PUT", "/services/rest/record/v1/employee/eid:P01431",
              (204, None, "", None))],
            create_missing=True, default_subsidiary_id="9",
        )
        payload = {k: v for k, v in _PAYLOAD.items() if k != "subsidiary"}
        out = await runner.upsert(
            person_email="x@pingcap.com", payload=payload,
            prior_external_id=None,
        )
        assert out.status == "success"
        put_body = calls.calls[1][2]
        assert put_body["subsidiary"] == {"id": "9"}  # injected on CREATE

    async def test_create_missing_requires_email(self):
        runner, calls = _runner(
            [("PATCH", "/services/rest/record/v1/employee/eid:P01431",
              (404, {"o:errorDetails": [
                  {"o:errorCode": "NONEXISTENT_EXTERNAL_ID"}]}, "", None))],
            create_missing=True,
        )
        payload = {k: v for k, v in _PAYLOAD.items() if k != "email"}
        out = await runner.upsert(
            person_email="x@pingcap.com", payload=payload,
            prior_external_id=None,
        )
        assert out.status == "failed"
        assert "email" in (out.error or "")
        assert len(calls.calls) == 1  # PATCH only; no PUT without email

    async def test_create_missing_skips_listed_work_num(self):
        # Even with create_missing on, a 404 for a 工号 on the create
        # skip-list is EXCLUDED, never PUT-created (admins would get a
        # duplicate; finance-multi-subsidiary / test rows must not be built).
        runner, calls = _runner(
            [("PATCH", "/services/rest/record/v1/employee/eid:P01431",
              (404, {"o:errorDetails": [
                  {"o:errorCode": "NONEXISTENT_EXTERNAL_ID"}]}, "", None))],
            create_missing=True,
            skip_create_work_nums=frozenset({"P01431"}),
        )
        out = await runner.upsert(
            person_email="admin@pingcap.com", payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "excluded"
        assert out.detail["reason"] == "netsuite_skip_provision"
        assert [c[0] for c in calls.calls] == ["PATCH"]  # no PUT-create

    async def test_create_missing_creates_when_not_on_skip_list(self):
        # A skip-list that doesn't contain this 工号 must not block a
        # genuine new-hire create.
        runner, calls = _runner(
            [("PATCH", "/services/rest/record/v1/employee/eid:P01431",
              (404, {"o:errorDetails": [
                  {"o:errorCode": "NONEXISTENT_EXTERNAL_ID"}]}, "", None)),
             ("PUT", "/services/rest/record/v1/employee/eid:P01431",
              (204, None, "", None))],
            create_missing=True,
            skip_create_work_nums=frozenset({"P9999"}),
        )
        out = await runner.upsert(
            person_email="x@pingcap.com", payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "success"
        assert out.detail["op"] == "create"
        assert [c[0] for c in calls.calls] == ["PATCH", "PUT"]

    async def test_vendor_error_surfaces_message(self):
        err = {
            "o:errorDetails": [
                {"detail": "Invalid subsidiary reference.",
                 "o:errorCode": "INVALID_KEY_OR_REF"},
            ],
        }
        runner, _calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (400, err, "", None)),
        ])
        out = await runner.upsert(
            person_email="x@pingcap.com", payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "failed"
        assert "Invalid subsidiary reference" in (out.error or "")
        assert out.detail["error_code"] == "INVALID_KEY_OR_REF"
        # A failed update must still report the eid: form (re-addressable).
        assert out.external_id == "eid:P01431"

    async def test_admin_record_lock_is_excluded_not_failed_zh(self):
        # NetSuite refuses a non-admin role any access to an Administrator
        # employee record (verified live 2026-06-24 against id 131). The
        # integration's TBA role is deliberately non-admin, so these IT
        # admins can never be synced — by policy they are EXCLUDED, not a
        # red failure (a full sweep must not choke on the admins).
        err = {
            "o:errorDetails": [
                {"detail": "出于安全方面的原因，只有管理员才能够编辑管理员记录。",
                 "o:errorCode": "USER_ERROR"},
            ],
        }
        runner, _calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (400, err, "", None)),
        ])
        out = await runner.upsert(
            person_email="admin@pingcap.com", payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "excluded"
        assert out.error is None
        assert out.external_id == "eid:P01431"
        assert out.detail["reason"] == "netsuite_admin_record_locked"

    async def test_admin_record_lock_is_excluded_not_failed_en(self):
        # Locale tolerance: same lock in an English account.
        err = {
            "o:errorDetails": [
                {"detail": "For security reasons, only an administrator can "
                           "edit an administrator record.",
                 "o:errorCode": "USER_ERROR"},
            ],
        }
        runner, _calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (400, err, "", None)),
        ])
        out = await runner.upsert(
            person_email="admin@pingcap.com", payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "excluded"
        assert out.detail["reason"] == "netsuite_admin_record_locked"

    async def test_ordinary_user_error_still_fails(self):
        # A generic USER_ERROR (NOT the admin-record lock) must remain a
        # hard failure — the exclusion is narrow, message-discriminated.
        err = {
            "o:errorDetails": [
                {"detail": "您的当前角色没有执行此操作的权限。",
                 "o:errorCode": "USER_ERROR"},
            ],
        }
        runner, _calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (400, err, "", None)),
        ])
        out = await runner.upsert(
            person_email="x@pingcap.com", payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "failed"

    async def test_subsidiary_locked_drops_subsidiary_and_retries(self):
        # NetSuite locks an employee's subsidiary once they carry posted
        # transactions. Since subsidiary rides in the same PATCH as the
        # identity fields, this 400 would fail the WHOLE update. The runner
        # must drop `subsidiary` and retry once: the vendor keeps its
        # (locked, financially-correct) subsidiary and the rest still syncs.
        err = {
            "o:errorDetails": [
                {"detail": "您不能更改此实体的子公司，原因是该实体有一到多个事务处理。",
                 "o:errorCode": "USER_ERROR"},
            ],
        }
        runner, calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (400, err, "", None)),
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (204, None, "", "https://x/services/rest/record/v1/employee/999")),
        ])
        out = await runner.upsert(
            person_email="shuai.feng@pingcap.com", payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "success"
        assert out.detail.get("subsidiary_preserved") is True
        assert out.external_id == "999"
        # Two PATCHes: first carried subsidiary, the retry dropped it while
        # keeping the identity fields.
        assert len(calls.calls) == 2
        assert "subsidiary" in calls.calls[0][2]
        assert "subsidiary" not in calls.calls[1][2]
        assert calls.calls[1][2]["email"] == "shuai.feng@pingcap.com"

    async def test_subsidiary_locked_en_locale(self):
        err = {
            "o:errorDetails": [
                {"detail": "You can't change the subsidiary of this entity "
                           "because it has one or more transactions.",
                 "o:errorCode": "USER_ERROR"},
            ],
        }
        runner, calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (400, err, "", None)),
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (204, None, "", None)),
        ])
        out = await runner.upsert(
            person_email="x@pingcap.com", payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "success"
        assert out.detail.get("subsidiary_preserved") is True
        assert len(calls.calls) == 2

    async def test_subsidiary_locked_retry_still_fails_is_failed(self):
        # If the retry (without subsidiary) still errors for another reason,
        # it's a real failure — the narrow retry doesn't mask other errors.
        locked = {
            "o:errorDetails": [
                {"detail": "您不能更改此实体的子公司，原因是该实体有一到多个事务处理。",
                 "o:errorCode": "USER_ERROR"},
            ],
        }
        other = {
            "o:errorDetails": [
                {"detail": "something else broke", "o:errorCode": "USER_ERROR"},
            ],
        }
        runner, calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (400, locked, "", None)),
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (400, other, "", None)),
        ])
        out = await runner.upsert(
            person_email="x@pingcap.com", payload=dict(_PAYLOAD),
            prior_external_id=None,
        )
        assert out.status == "failed"
        assert len(calls.calls) == 2


# ── Runner: deactivate ─────────────────────────────────────────────────
class TestDeactivate:
    async def test_patch_sets_inactive(self):
        runner, calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/4567",
             (204, None, "", None)),
        ])
        out = await runner.deactivate(
            person_email="x@pingcap.com", prior_external_id="4567",
        )
        assert out.status == "deactivated"
        _m, _p, body = calls.calls[0]
        assert body == {"isInactive": True}

    async def test_eid_form_prior_id(self):
        runner, calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/eid:P01431",
             (204, None, "", None)),
        ])
        out = await runner.deactivate(
            person_email="x@pingcap.com", prior_external_id="eid:P01431",
        )
        assert out.status == "deactivated"

    async def test_404_is_idempotent_success(self):
        runner, _calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/4567",
             (404, None, "", None)),
        ])
        out = await runner.deactivate(
            person_email="x@pingcap.com", prior_external_id="4567",
        )
        assert out.status == "deactivated"

    async def test_no_prior_id_is_noop(self):
        runner, calls = _runner([])
        out = await runner.deactivate(
            person_email="x@pingcap.com", prior_external_id=None,
        )
        assert out.status == "deactivated"
        assert calls.calls == []  # nothing to PATCH

    async def test_admin_record_lock_is_excluded_not_failed(self):
        # An NS Administrator leaver: deactivate hits the same lock as
        # upsert. Must be excluded (manual upkeep), NOT failed — else the
        # offboarding sweep reddens + retries forever (lock is permanent).
        err = {
            "o:errorDetails": [
                {"detail": "出于安全方面的原因，只有管理员才能够编辑管理员记录。",
                 "o:errorCode": "USER_ERROR"},
            ],
        }
        runner, _calls = _runner([
            ("PATCH", "/services/rest/record/v1/employee/4567",
             (400, err, "", None)),
        ])
        out = await runner.deactivate(
            person_email="admin@pingcap.com", prior_external_id="4567",
        )
        assert out.status == "excluded"
        assert out.error is None
        assert out.detail["reason"] == "netsuite_admin_record_locked"

    async def test_runner_does_not_support_delete(self):
        assert NetSuiteRunner.supports_delete is False
        assert NetSuiteRunner.supports_deactivate is True


# ── Driver test_connection probe ───────────────────────────────────────
# Regression for the false-negative health check: the connector is
# upsert-only (addresses every person by ``employee/eid:{工号}`` and never
# lists), so the probe must mirror that — a targeted ``eid:`` GET, where a
# ``404`` (sentinel id absent) proves auth + REST + employee-record access.
# A collection ``GET /employee?limit=1`` instead asks for *list/search*
# rights a minimal upsert role legitimately lacks, which NetSuite rejects
# with ``400 / o:errorCode USER_ERROR`` — unrelated to whether the sync works.
from app.integrations.drivers.netsuite import NetSuiteDriver  # noqa: E402


def _driver_with_response(resp: tuple[int, Any, str, str | None]):
    drv = NetSuiteDriver(
        integration_id="ns-1",
        config=dict(
            account_id="7429932", consumer_key="ck", consumer_secret="cs",
            token_id="ti", token_secret="ts",
        ),
    )
    calls = _ScriptedCalls([("GET", "/services/rest/record/v1/employee", resp)])
    real_client = drv._client

    def _patched_client():
        c = real_client()
        c.request = calls  # type: ignore[method-assign]
        return c

    drv._client = _patched_client  # type: ignore[method-assign]
    return drv, calls


class TestDriverTestConnection:
    async def test_404_sentinel_is_healthy(self):
        drv, calls = _driver_with_response(
            (404, {"o:errorDetails": [{"o:errorCode": "NONEXISTENT", "detail": "x"}]}, "", None)
        )
        res = await drv.test_connection()
        assert res.ok is True
        assert res.detail["status_code"] == 404
        # Probe must hit the targeted eid: path, NOT a collection list.
        method, path, body = calls.calls[0]
        assert method == "GET"
        assert path.endswith("/employee/eid:__helix_connection_probe__")
        assert body is None

    async def test_2xx_is_healthy(self):
        drv, _ = _driver_with_response((200, {"id": "1"}, "", None))
        res = await drv.test_connection()
        assert res.ok is True

    async def test_400_user_error_fails_with_targeted_hint(self):
        drv, _ = _driver_with_response(
            (400, {"o:errorDetails": [
                {"o:errorCode": "USER_ERROR", "detail": "您的当前角色没有执行此操作的权限。"}
            ]}, "", None)
        )
        res = await drv.test_connection()
        assert res.ok is False
        assert res.detail["status_code"] == 400
        assert "USER_ERROR" in res.message
        assert "员工" in res.message  # actionable hint names the permission

    async def test_403_fails(self):
        drv, _ = _driver_with_response((403, {"o:errorDetails": [{"detail": "no"}]}, "", None))
        res = await drv.test_connection()
        assert res.ok is False
        assert res.detail["status_code"] == 403


# ── Dept runner: upsert_dept / soft_delete ─────────────────────────────
def _dept_runner(script, **runner_kw) -> tuple[NetSuiteDepartmentRunner, _ScriptedCalls]:
    runner = NetSuiteDepartmentRunner(client=_client(), **runner_kw)
    calls = _ScriptedCalls(script)
    runner._client.request = calls  # type: ignore[method-assign]
    return runner, calls


class TestNetSuiteDepartmentRunner:
    def test_supports_soft_delete(self):
        assert NetSuiteDepartmentRunner.supports_soft_delete is True

    async def test_upsert_dept_update_via_patch(self):
        # Update-only: PATCH the existing dept (never blind PUT-create).
        runner, calls = _dept_runner([
            ("PATCH", "/services/rest/record/v1/department/eid:ENG",
             (204, None, "", None)),
        ])
        out = await runner.upsert_dept(
            dept_code="ENG", dept_name="Engineering",
            parent_dept_code=None, leader_email=None, dept_order=None,
            prior_external_id=None,
        )
        assert out.status == "success"
        assert out.external_id == "eid:ENG"
        assert out.detail["op"] == "update"
        method, path, body = calls.calls[0]
        assert method == "PATCH"
        assert path.endswith("/department/eid:ENG")
        # Update PATCH OMITS name → PATCH-merge preserves the vendor's
        # existing code-prefixed dept name (SWC "{编码} {名称}"). camelCase
        # isInactive matches the record schema.
        assert "name" not in body
        assert body["isInactive"] is False
        assert body["externalId"] == "ENG"
        assert "parent" not in body  # no parent → no parent ref

    async def test_upsert_dept_with_parent_sends_external_id_ref(self):
        runner, calls = _dept_runner([
            ("PATCH", "/services/rest/record/v1/department/eid:ENG_BE",
             (204, None, "", None)),
        ])
        out = await runner.upsert_dept(
            dept_code="ENG_BE", dept_name="Backend",
            parent_dept_code="ENG", leader_email=None, dept_order=None,
            prior_external_id=None,
        )
        assert out.status == "success"
        method, path, body = calls.calls[0]
        assert method == "PATCH"
        assert path.endswith("/department/eid:ENG_BE")
        # Parent addressed by External ID, not internal id.
        assert body["parent"] == {"externalId": "ENG"}

    async def test_upsert_dept_not_adopted_404_is_skipped(self):
        # Update-only: a 404 (no dept carries this external id) is SKIPPED,
        # not created — refusing to build a parallel tree beside the
        # un-adopted Authing-era depts. Exactly one call (the PATCH).
        runner, calls = _dept_runner([
            ("PATCH", "/services/rest/record/v1/department/eid:ENG",
             (404, {"o:errorDetails": [
                 {"o:errorCode": "NONEXISTENT_EXTERNAL_ID"}]}, "", None)),
        ])
        out = await runner.upsert_dept(
            dept_code="ENG", dept_name="Engineering",
            parent_dept_code=None, leader_email=None, dept_order=None,
            prior_external_id=None,
        )
        assert out.status == "skipped"
        assert out.detail["reason"] == "netsuite_dept_not_adopted"
        assert len(calls.calls) == 1
        assert calls.calls[0][0] == "PATCH"

    async def test_upsert_dept_create_missing_on_404(self):
        # Opt-in: tree adopted → 404 means a new dept → PATCH then PUT.
        runner, calls = _dept_runner(
            [("PATCH", "/services/rest/record/v1/department/eid:ENG",
              (404, {"o:errorDetails": [
                  {"o:errorCode": "NONEXISTENT_EXTERNAL_ID"}]}, "", None)),
             ("PUT", "/services/rest/record/v1/department/eid:ENG",
              (204, None, "",
               "https://7429932.suitetalk.api.netsuite.com"
               "/services/rest/record/v1/department/88"))],
            create_missing=True,
        )
        out = await runner.upsert_dept(
            dept_code="ENG", dept_name="Engineering",
            parent_dept_code=None, leader_email=None, dept_order=None,
            prior_external_id=None,
        )
        assert out.status == "success"
        assert out.external_id == "88"
        assert out.detail["op"] == "create"
        assert [c[0] for c in calls.calls] == ["PATCH", "PUT"]
        # A NEW dept DOES carry the clean name on the create PUT (only the
        # update path omits it to preserve the vendor's code-prefixed name).
        put_body = calls.calls[1][2]
        assert put_body["name"] == "Engineering"
        assert put_body["isInactive"] is False
        assert put_body["externalId"] == "ENG"

    async def test_upsert_dept_percent_encodes_code_in_path(self):
        # Free-form org-node codes (NOT ^P\d+$-gated) must be percent-
        # encoded in the eid: path segment so a code with '/' can't escape
        # the record path. Body externalId keeps the raw code.
        runner, calls = _dept_runner([
            ("PATCH", "/services/rest/record/v1/department/eid:A%2FB%3Fx",
             (204, None, "", None)),
        ])
        out = await runner.upsert_dept(
            dept_code="A/B?x", dept_name="Weird",
            parent_dept_code=None, leader_email=None, dept_order=None,
            prior_external_id=None,
        )
        assert out.status == "success"
        _m, path, body = calls.calls[0]
        assert path.endswith("/department/eid:A%2FB%3Fx")
        assert "/department/eid:A/B?x" not in path  # not raw
        assert body["externalId"] == "A/B?x"  # body keeps raw code

    async def test_upsert_dept_missing_code_fails_without_call(self):
        runner, calls = _dept_runner([])
        out = await runner.upsert_dept(
            dept_code="", dept_name="x",
            parent_dept_code=None, leader_email=None, dept_order=None,
            prior_external_id=None,
        )
        assert out.status == "failed"
        assert "dept_code" in (out.error or "")
        assert calls.calls == []  # never hit the wire

    async def test_upsert_dept_vendor_error_surfaces_message(self):
        err = {
            "o:errorDetails": [
                {"detail": "Invalid parent reference.",
                 "o:errorCode": "INVALID_KEY_OR_REF"},
            ],
        }
        runner, _calls = _dept_runner([
            ("PATCH", "/services/rest/record/v1/department/eid:ENG",
             (400, err, "", None)),
        ])
        out = await runner.upsert_dept(
            dept_code="ENG", dept_name="Engineering",
            parent_dept_code=None, leader_email=None, dept_order=None,
            prior_external_id=None,
        )
        assert out.status == "failed"
        assert "Invalid parent reference" in (out.error or "")
        assert out.detail["error_code"] == "INVALID_KEY_OR_REF"
        # A failed upsert must report the eid: form (re-addressable).
        assert out.external_id == "eid:ENG"

    async def test_soft_delete_patches_inactive(self):
        runner, calls = _dept_runner([
            ("PATCH", "/services/rest/record/v1/department/88",
             (204, None, "", None)),
        ])
        out = await runner.soft_delete(
            dept_code="ENG", prior_external_id="88",
        )
        assert out.status == "soft_deleted"
        method, path, body = calls.calls[0]
        assert method == "PATCH"
        assert path.endswith("/department/88")
        assert body == {"isInactive": True}

    async def test_soft_delete_404_is_idempotent_success(self):
        runner, _calls = _dept_runner([
            ("PATCH", "/services/rest/record/v1/department/eid:ENG",
             (404, None, "", None)),
        ])
        out = await runner.soft_delete(
            dept_code="ENG", prior_external_id="eid:ENG",
        )
        assert out.status == "soft_deleted"

    async def test_soft_delete_falls_back_to_eid_without_prior_id(self):
        runner, calls = _dept_runner([
            ("PATCH", "/services/rest/record/v1/department/eid:ENG",
             (204, None, "", None)),
        ])
        out = await runner.soft_delete(
            dept_code="ENG", prior_external_id=None,
        )
        assert out.status == "soft_deleted"
        _m, path, _b = calls.calls[0]
        assert path.endswith("/department/eid:ENG")

    async def test_soft_delete_vendor_error_surfaces(self):
        err = {"o:errorDetails": [{"detail": "still in use", "o:errorCode": "X"}]}
        runner, _calls = _dept_runner([
            ("PATCH", "/services/rest/record/v1/department/88",
             (400, err, "", None)),
        ])
        out = await runner.soft_delete(
            dept_code="ENG", prior_external_id="88",
        )
        assert out.status == "failed"
        assert "still in use" in (out.error or "")


# ── Circuit breaker vs expected business outcomes (PR #1373 update-only) ─
#
# The _ScriptedCalls stub above replaces NetSuiteClient.request wholesale,
# so the REAL request() breaker logic is never exercised by the runner
# tests. These drive a real client over a mocked httpx transport to pin
# the breaker behaviour directly.
def _client_with_transport(handler, **kw) -> NetSuiteClient:
    """Real NetSuiteClient whose httpx transport is mocked → the genuine
    ``request()`` (rate-limit + breaker + status bookkeeping) runs."""
    c = _client(rate_limit_per_minute=10_000_000, **kw)
    c._client = httpx.AsyncClient(
        transport=httpx.MockTransport(handler),
        follow_redirects=False,
        trust_env=False,
    )
    return c


async def test_breaker_not_armed_by_expected_404s():
    """First adoption sweep: every person 404s (empty standard externalId,
    by design). The dead-host breaker must NOT trip — otherwise person #6
    onward is bricked as ``failed`` with a misleading 'host unreachable /
    token revoked' message, defeating the whole update-only design."""

    def handler(_req: httpx.Request) -> httpx.Response:
        return httpx.Response(
            404,
            json={"o:errorDetails": [
                {"detail": "Record does not exist", "o:errorCode": "RCRD_DSNT_EXIST"},
            ]},
        )

    c = _client_with_transport(handler)
    try:
        for _ in range(10):  # well past _breaker_threshold (5)
            status, _body, _text, _loc = await c.request(
                "PATCH", "/services/rest/record/v1/employee/eid:P01431"
            )
            assert status == 404
        assert c._consecutive_failures == 0  # never armed by 404s
    finally:
        await c._client.aclose()


async def test_record_requests_pin_company_content_language_but_suiteql_does_not():
    """Records require the company language; SuiteQL rejects that header."""
    seen: list[httpx.Request] = []

    def handler(req: httpx.Request) -> httpx.Response:
        seen.append(req)
        if req.url.path.endswith("/suiteql"):
            return httpx.Response(200, json={"items": [], "hasMore": False})
        return httpx.Response(204)

    c = _client_with_transport(handler)
    try:
        await c.request(
            "PATCH", "/services/rest/record/v1/employee/eid:P01431",
            json_body={"isInactive": True},
        )
        await c.suiteql("SELECT id FROM employee", max_rows=1)
        assert len(seen) == 2
        assert seen[0].headers["Content-Language"] == "zh"
        assert "Content-Language" not in seen[1].headers
    finally:
        await c._client.aclose()


async def test_breaker_not_armed_by_admin_record_lock():
    """The admin-record lock (USER_ERROR 400) is a *policy* skip → the row
    settles as ``excluded``; it is not a transport/auth failure and must
    not arm the breaker either (a run of consecutive IT-admins is normal)."""

    def handler(_req: httpx.Request) -> httpx.Response:
        return httpx.Response(
            400,
            json={"o:errorDetails": [{
                "detail": (
                    "For security reasons, only an administrator can edit "
                    "an administrator record."
                ),
                "o:errorCode": "USER_ERROR",
            }]},
        )

    c = _client_with_transport(handler)
    try:
        for _ in range(10):
            status, _body, _text, _loc = await c.request(
                "PATCH", "/services/rest/record/v1/employee/eid:ADMIN"
            )
            assert status == 400
        assert c._consecutive_failures == 0
    finally:
        await c._client.aclose()


async def test_breaker_still_opens_on_real_failures():
    """The fix exempts only expected business outcomes — a genuine dead
    host / 5xx still accumulates and opens the breaker, so a truly broken
    connector still fails fast instead of burning the whole retry budget."""

    def handler(_req: httpx.Request) -> httpx.Response:
        return httpx.Response(503, text="upstream unavailable")

    c = _client_with_transport(handler)
    try:
        with pytest.raises(NetSuiteError, match="circuit open"):
            for _ in range(10):
                await c.request(
                    "PATCH", "/services/rest/record/v1/employee/eid:P1",
                    retry_on_429=0,
                )
        assert c._consecutive_failures >= c._breaker_threshold
    finally:
        await c._client.aclose()


# ── Helpers ────────────────────────────────────────────────────────────
def test_internal_id_from_location():
    assert _internal_id_from_location(
        "https://x.suitetalk.api.netsuite.com/services/rest/record/v1/employee/12345"
    ) == "12345"
    assert _internal_id_from_location(".../employee/eid:P01") is None
    assert _internal_id_from_location(None) is None