from __future__ import annotations

from collections import deque
from urllib.parse import unquote

import httpx

from pingcap_datapipelines.netsuite_restlet import (
    NetSuiteRestletClient,
    OAuth1TbaCredentials,
    OAuth1TbaSigner,
)


def _oauth_header_values(header: str) -> dict[str, str]:
    return {
        pair.split("=", 1)[0]: unquote(pair.split("=", 1)[1].strip('"'))
        for pair in header.removeprefix("OAuth ").split(", ")
    }


def test_oauth1_tba_signs_full_query_and_retries_with_a_new_nonce() -> None:
    nonces = deque(["nonce-one", "nonce-two"])
    signer = OAuth1TbaSigner(
        OAuth1TbaCredentials(
            realm="7429932",
            consumer_key="consumer",
            consumer_secret="consumer-secret",
            token_key="token",
            token_secret="token-secret",
        ),
        nonce_factory=nonces.popleft,
        timestamp_factory=lambda: 1700000000,
    )
    requests: list[httpx.Request] = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        return httpx.Response(500 if len(requests) == 1 else 200, json={"ok": True})

    with NetSuiteRestletClient(
        url="https://7429932.restlets.api.netsuite.com/app/site/hosting/restlet.nl?script=616&deploy=1",
        signer=signer,
        max_attempts=2,
        transport=httpx.MockTransport(handler),
        sleeper=lambda _delay: None,
    ) as client:
        assert client.fetch_page(posting_period_id="245", cursor="1", page_size=500) == {"ok": True}

    assert len(requests) == 2
    for part in ("script=616", "deploy=1", "postingPeriodId=245", "pageSize=500", "cursor=1"):
        assert part in str(requests[0].url)
    first = _oauth_header_values(requests[0].headers["Authorization"])
    second = _oauth_header_values(requests[1].headers["Authorization"])
    assert first["realm"] == "7429932"
    assert first["oauth_nonce"] == "nonce-one"
    assert second["oauth_nonce"] == "nonce-two"
    assert first["oauth_signature"] != second["oauth_signature"]


def test_oauth1_tba_signature_changes_when_only_a_query_value_changes() -> None:
    credentials = OAuth1TbaCredentials(
        realm="7429932",
        consumer_key="consumer",
        consumer_secret="consumer-secret",
        token_key="token",
        token_secret="token-secret",
    )
    signer_one = OAuth1TbaSigner(credentials, nonce_factory=lambda: "nonce", timestamp_factory=lambda: 1700000000)
    signer_two = OAuth1TbaSigner(credentials, nonce_factory=lambda: "nonce", timestamp_factory=lambda: 1700000000)
    one = _oauth_header_values(
        signer_one.authorization_header("GET", "https://example.test/restlet?script=616&cursor=1")
    )
    two = _oauth_header_values(
        signer_two.authorization_header("GET", "https://example.test/restlet?script=616&cursor=2")
    )
    assert one["oauth_signature"] != two["oauth_signature"]