import json
import os
import stat
import tempfile
import unittest
from contextlib import redirect_stdout
from io import StringIO
from pathlib import Path

from obsidian_wiki.kingdee_api import (
    DEFAULT_SALORDER_API_URL,
    ensure_kingdee_config,
    fetch_kingdee_sales_orders,
    get_kingdee_access_token,
)


class FakeResponse:
    def __init__(self, payload):
        self.payload = payload

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        return False

    def read(self):
        return json.dumps(self.payload).encode("utf-8")


class RecordingUrlopen:
    def __init__(self, responses):
        self.responses = list(responses)
        self.calls = []

    def __call__(self, request, timeout=60):
        self.calls.append({"request": request, "timeout": timeout})
        if not self.responses:
            raise AssertionError("unexpected HTTP request")
        return FakeResponse(self.responses.pop(0))


class KingdeeApiTests(unittest.TestCase):
    def setUp(self):
        self.tempdir = tempfile.TemporaryDirectory()
        self.root = Path(self.tempdir.name)

    def tearDown(self):
        self.tempdir.cleanup()

    def test_config_command_creates_private_template_without_overwriting(self):
        config_path = self.root / ".runtime" / "kingdee.env"

        created = ensure_kingdee_config(config_path)
        config_path.write_text(config_path.read_text(encoding="utf-8") + "\nKINGDEE_TOKEN_URL=https://token.example\n", encoding="utf-8")
        created_again = ensure_kingdee_config(config_path)

        self.assertTrue(created)
        self.assertFalse(created_again)
        self.assertEqual(stat.S_IMODE(config_path.stat().st_mode), 0o600)
        text = config_path.read_text(encoding="utf-8")
        self.assertIn(f"KINGDEE_SALORDER_API_URL={DEFAULT_SALORDER_API_URL}", text)
        self.assertIn("KINGDEE_TOKEN_BODY_JSON=", text)
        self.assertIn("KINGDEE_TOKEN_URL=https://token.example", text)
        self.assertIn("是否必填：自动登录时必填；如果只填 KINGDEE_ACCESS_TOKEN，则不用填。", text)
        self.assertIn("用途：销售订单 API 要求的 x-acgw-identity 请求头。", text)
        self.assertIn("是否必填：可选，手动测试用；长期定时同步建议使用自动登录配置。", text)
        self.assertIn("是否必填：可选，默认从 data.access_token 读取。", text)
        self.assertIn("变量：KINGDEE_CLIENT_ID", text)
        self.assertIn("变量：KINGDEE_USERNAME", text)
        self.assertIn("变量：KINGDEE_ACCOUNT_ID", text)
        self.assertIn("nonce 和 timestamp 会按官方文档自动生成", text)

    def test_default_config_path_accepts_none(self):
        cwd = Path.cwd()
        try:
            os.chdir(self.root)
            created = ensure_kingdee_config(None)
        finally:
            os.chdir(cwd)

        self.assertTrue(created)
        self.assertTrue((self.root / ".runtime" / "kingdee.env").is_file())

    def test_cli_config_kingdee_writes_template(self):
        from obsidian_wiki.cli import main

        config_path = self.root / ".runtime" / "kingdee.env"
        output = StringIO()
        with redirect_stdout(output):
            code = main(["business", "config", "kingdee", "--config", str(config_path)])

        self.assertEqual(code, 0)
        self.assertTrue(config_path.is_file())
        self.assertIn("Kingdee config", output.getvalue())

    def test_access_token_is_fetched_and_reused_from_private_cache(self):
        cache_path = self.root / ".runtime" / "kingdee-token.json"
        config = {
            "KINGDEE_TOKEN_URL": "https://token.example/access",
            "KINGDEE_TOKEN_BODY_JSON": '{"app_key":"abc","app_secret":"secret"}',
            "KINGDEE_TOKEN_HEADERS_JSON": '{"X-App":"demo"}',
            "KINGDEE_TOKEN_FIELD": "data.access_token",
            "KINGDEE_TOKEN_EXPIRES_IN_FIELD": "data.expires_in",
            "KINGDEE_REFRESH_TOKEN_FIELD": "data.refresh_token",
        }
        http = RecordingUrlopen(
            [
                {
                    "data": {
                        "access_token": "access-1",
                        "expires_in": 3600000,
                        "refresh_token": "refresh-1",
                    }
                }
            ]
        )

        token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)
        cached = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1100)

        self.assertEqual(token, "access-1")
        self.assertEqual(cached, "access-1")
        self.assertEqual(len(http.calls), 1)
        request = http.calls[0]["request"]
        self.assertEqual(request.get_method(), "POST")
        self.assertEqual(request.full_url, "https://token.example/access")
        self.assertEqual(json.loads(request.data.decode("utf-8"))["app_key"], "abc")
        self.assertEqual(request.get_header("X-app"), "demo")
        self.assertEqual(stat.S_IMODE(cache_path.stat().st_mode), 0o600)

    def test_access_token_body_defaults_to_official_fields_with_nonce_and_current_time(self):
        cache_path = self.root / ".runtime" / "kingdee-token.json"
        config = {
            "KINGDEE_TOKEN_URL": "https://token.example/access",
            "KINGDEE_CLIENT_ID": "thirdapp",
            "KINGDEE_CLIENT_SECRET": "secret",
            "KINGDEE_USERNAME": "zhangSan",
            "KINGDEE_ACCOUNT_ID": "1355633519610561531",
            "KINGDEE_LANGUAGE": "zh_CN",
            "KINGDEE_TOKEN_FIELD": "data.access_token",
        }
        http = RecordingUrlopen([{"data": {"access_token": "access-1"}}])

        token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)

        self.assertEqual(token, "access-1")
        body = json.loads(http.calls[0]["request"].data.decode("utf-8"))
        self.assertEqual(body["client_id"], "thirdapp")
        self.assertEqual(body["client_secret"], "secret")
        self.assertEqual(body["username"], "zhangSan")
        self.assertEqual(body["accountId"], "1355633519610561531")
        self.assertEqual(body["language"], "zh_CN")
        self.assertRegex(body["nonce"], r"^[0-9a-f]{32}$")
        self.assertEqual(body["timestamp"], "1970-01-01T00:16:40.000Z")

    def test_expired_token_uses_refresh_endpoint(self):
        cache_path = self.root / ".runtime" / "kingdee-token.json"
        cache_path.parent.mkdir(parents=True)
        cache_path.write_text(
            json.dumps(
                {
                    "access_token": "old-token",
                    "refresh_token": "refresh-1",
                    "expires_at": 999,
                    "obtained_at": 100,
                }
            ),
            encoding="utf-8",
        )
        config = {
            "KINGDEE_TOKEN_URL": "https://token.example/access",
            "KINGDEE_TOKEN_BODY_JSON": '{"app_key":"abc"}',
            "KINGDEE_REFRESH_URL": "https://token.example/refresh",
            "KINGDEE_REFRESH_BODY_JSON": '{"refresh_token":"{refresh_token}"}',
            "KINGDEE_TOKEN_FIELD": "data.access_token",
            "KINGDEE_TOKEN_EXPIRES_IN_FIELD": "data.expires_in",
            "KINGDEE_REFRESH_TOKEN_FIELD": "data.refresh_token",
        }
        http = RecordingUrlopen(
            [
                {
                    "data": {
                        "access_token": "new-token",
                        "expires_in": 3600000,
                        "refresh_token": "refresh-2",
                    }
                }
            ]
        )

        token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)

        self.assertEqual(token, "new-token")
        request = http.calls[0]["request"]
        self.assertEqual(request.full_url, "https://token.example/refresh")
        self.assertEqual(json.loads(request.data.decode("utf-8"))["refresh_token"], "refresh-1")

    def test_sales_order_fetch_posts_filters_headers_and_paginates(self):
        cache_path = self.root / ".runtime" / "kingdee-token.json"
        cache_path.parent.mkdir(parents=True)
        cache_path.write_text(
            json.dumps({"access_token": "access-1", "expires_at": 9999, "obtained_at": 1000}),
            encoding="utf-8",
        )
        config = {
            "KINGDEE_SALORDER_API_URL": "https://kingdee.example/querySalOrder",
            "KINGDEE_ACGW_IDENTITY": "identity-1",
            "KINGDEE_PAGE_SIZE": "2",
        }
        page_1 = {
            "status": True,
            "errorCode": "",
            "message": "",
            "data": {
                "pageNo": 1,
                "pageSize": 2,
                "rows": [{"id": "1", "billno": "SO-001"}, {"id": "2", "billno": "SO-002"}],
                "totalCount": 3,
            },
        }
        page_2 = {
            "status": True,
            "errorCode": "",
            "message": "",
            "data": {
                "pageNo": 2,
                "pageSize": 2,
                "rows": [{"id": "3", "billno": "SO-003"}],
                "totalCount": 3,
            },
        }
        http = RecordingUrlopen([page_1, page_2])

        rows, payload = fetch_kingdee_sales_orders(
            config,
            filters={"start_bizdate": "2026-06-01", "billstatus": ["C"]},
            token_cache_path=cache_path,
            urlopen=http,
            now_epoch=1000,
        )

        self.assertEqual([row["billno"] for row in rows], ["SO-001", "SO-002", "SO-003"])
        self.assertEqual(payload["pages_fetched"], 2)
        self.assertEqual(len(http.calls), 2)
        first = http.calls[0]["request"]
        self.assertEqual(first.get_method(), "POST")
        self.assertEqual(first.get_header("Accesstoken"), "access-1")
        self.assertEqual(first.get_header("X-acgw-identity"), "identity-1")
        body = json.loads(first.data.decode("utf-8"))
        self.assertEqual(body["data"]["start_bizdate"], "2026-06-01")
        self.assertEqual(body["data"]["billstatus"], ["C"])
        self.assertEqual(body["pageNo"], 1)
        self.assertEqual(body["pageSize"], 2)

    def test_sales_order_api_failure_reports_code_and_message(self):
        cache_path = self.root / ".runtime" / "kingdee-token.json"
        cache_path.parent.mkdir(parents=True)
        cache_path.write_text(
            json.dumps({"access_token": "access-1", "expires_at": 9999, "obtained_at": 1000}),
            encoding="utf-8",
        )
        config = {
            "KINGDEE_SALORDER_API_URL": "https://kingdee.example/querySalOrder",
            "KINGDEE_ACGW_IDENTITY": "identity-1",
        }
        http = RecordingUrlopen([{"status": False, "errorCode": "E001", "message": "bad filter"}])

        with self.assertRaisesRegex(RuntimeError, "E001.*bad filter"):
            fetch_kingdee_sales_orders(config, filters={}, token_cache_path=cache_path, urlopen=http, now_epoch=1000)


if __name__ == "__main__":
    unittest.main()