"""Tests for dynamic master data (companies, bank accounts, aliases).

Unit tests exercise normalization, masking, effective-window boundaries and
alias-match priority against an in-memory database; the HTTP integration
class drives a live server to cover the submission -> review -> disable
workflow, uniqueness under concurrency, masking boundaries and RBAC.
"""

from __future__ import annotations

from pathlib import Path
import tempfile
import threading
import unittest

from bank_importer import auth, master_data
from bank_importer.db import connect, migrate

import server
from test_server_auth import Client, as_json


BOOTSTRAP_PASSWORD = "BootAdmin123"
ADMIN_PASSWORD = "AdminPass123"
CASHIER_PASSWORD = "Cashier123"


class MasterDataUnitTests(unittest.TestCase):
    def setUp(self) -> None:
        self.connection = connect(":memory:")
        migrate(self.connection)
        now = master_data.utc_now()
        cursor = self.connection.execute(
            "INSERT INTO companies (name, created_at, updated_at) VALUES ('甲公司', ?, ?)",
            (now, now),
        )
        self.company_id = int(cursor.lastrowid)
        self.connection.commit()

    def tearDown(self) -> None:
        self.connection.close()

    def _active_account(self, number: str = "1234567890123") -> object:
        return master_data.submit_bank_account(
            self.connection,
            company_id=self.company_id,
            bank_name="中信银行",
            account_type="基本户",
            account_number=number,
            start_date="2026-01-01",
            actor=None,
        )

    def test_normalize_account_number(self) -> None:
        self.assertEqual("1234567890", master_data.normalize_account_number("1234 5678-90"))
        self.assertEqual("1234567890", master_data.normalize_account_number("１２３４５６７８９０"))
        for invalid in ("", "123", "abc123456", "12345678901234567890123456789012345"):
            with self.assertRaises(ValueError):
                master_data.normalize_account_number(invalid)

    def test_mask_account_number(self) -> None:
        self.assertEqual("****9012", master_data.mask_account_number("123456789012"))
        self.assertEqual("****123", master_data.mask_account_number("123"))

    def test_is_usable_window_boundaries(self) -> None:
        account = self._active_account()
        account = master_data.review_bank_account(
            self.connection, account["id"], "approve", None, self._admin(),
            effective_from="2026-02-01",
        )
        self.assertFalse(master_data.is_usable(account, "2026-01-31"))
        self.assertTrue(master_data.is_usable(account, "2026-02-01"))
        # A scheduled effective_to bounds the usable window (inclusive).
        with self.connection:
            self.connection.execute(
                "UPDATE bank_accounts SET effective_to = '2026-06-30' WHERE id = ?",
                (account["id"],),
            )
        account = master_data.get_account(self.connection, account["id"])
        self.assertTrue(master_data.is_usable(account, "2026-06-30"))
        self.assertFalse(master_data.is_usable(account, "2026-07-01"))
        # Disabled accounts never accept new uploads, even inside the window.
        account = master_data.review_bank_account(
            self.connection, account["id"], "disable", "账户销户", self._admin(),
            effective_to="2026-06-30",
        )
        self.assertFalse(master_data.is_usable(account, "2026-06-30"))
        # …but they still identify historical rows inside the window.
        self.assertTrue(master_data.is_identifiable(account, "2026-06-30"))
        self.assertFalse(master_data.is_identifiable(account, "2026-07-01"))

    def _admin(self):
        user = self.connection.execute(
            "SELECT * FROM users WHERE username = 'admin-u'"
        ).fetchone()
        if user is None:
            auth.create_user(self.connection, "admin-u", "AdminPass123", "admin")
            user = self.connection.execute(
                "SELECT * FROM users WHERE username = 'admin-u'"
            ).fetchone()
        return user

    def test_pending_and_returned_accounts_never_usable(self) -> None:
        account = master_data.submit_bank_account(
            self.connection,
            company_id=self.company_id,
            bank_name="中信银行",
            account_type="一般户",
            account_number="9988776655",
            start_date="2026-01-01",
            actor=None,
        )
        self.assertFalse(master_data.is_usable(account, "2026-06-01"))
        account = master_data.review_bank_account(
            self.connection, account["id"], "return", "资料不全", self._admin()
        )
        self.assertEqual("returned", account["status"])
        self.assertFalse(master_data.is_usable(account, "2026-06-01"))

    def test_alias_match_priority_and_window(self) -> None:
        exact_account = self._active_account("1111222233334")
        exact_account = master_data.review_bank_account(
            self.connection, exact_account["id"], "approve", None, self._admin(),
            effective_from="2026-01-01",
        )
        alias_account = master_data.submit_bank_account(
            self.connection,
            company_id=self.company_id,
            bank_name="建设银行",
            account_type="一般户",
            account_number="5555666677778",
            start_date="2026-01-01",
            actor=None,
        )
        alias_account = master_data.review_bank_account(
            self.connection, alias_account["id"], "approve", None, self._admin(),
            effective_from="2026-01-01",
        )
        master_data.add_alias(
            self.connection, alias_account["id"], "account", "9999000011112",
            priority=5, actor=None,
        )
        master_data.add_alias(
            self.connection, alias_account["id"], "name", "甲公司郑州分部",
            effective_from="2026-03-01", effective_to="2026-03-31", actor=None,
        )

        # Exact account number beats the account alias tier.
        master_data.add_alias(
            self.connection, alias_account["id"], "account", "1111222233334",
            priority=1, actor=None,
        )
        hits = master_data.match_account(
            self.connection, account_number="1111-2222 3333 4", on_date="2026-06-01"
        )
        self.assertEqual(exact_account["id"], hits[0]["bank_account_id"])
        self.assertEqual("account_exact", hits[0]["via"])
        self.assertEqual(alias_account["id"], hits[1]["bank_account_id"])
        self.assertEqual("account_alias", hits[1]["via"])

        # Name alias only matches inside its effective window.
        hits = master_data.match_account(
            self.connection, name="甲公司郑州分部 ", on_date="2026-03-15"
        )
        self.assertEqual(alias_account["id"], hits[0]["bank_account_id"])
        self.assertEqual("name_alias", hits[0]["via"])
        self.assertEqual(
            [],
            master_data.match_account(
                self.connection, name="甲公司郑州分部", on_date="2026-04-01"
            ),
        )

    def test_match_never_returns_pending_or_disabled(self) -> None:
        pending = master_data.submit_bank_account(
            self.connection,
            company_id=self.company_id,
            bank_name="郑州银行",
            account_type="一般户",
            account_number="7777888899990",
            actor=None,
        )
        self.assertEqual(
            [],
            master_data.match_account(self.connection, account_number="7777888899990"),
        )
        account = master_data.review_bank_account(
            self.connection, pending["id"], "approve", None, self._admin(),
            effective_from="2026-01-01",
        )
        self.assertTrue(
            master_data.match_account(self.connection, account_number="7777888899990")
        )
        master_data.review_bank_account(
            self.connection, account["id"], "disable", "销户", self._admin(),
            effective_to="2026-06-30",
        )
        # Disabled accounts still resolve dates inside their effective window…
        self.assertTrue(
            master_data.match_account(
                self.connection, account_number="7777888899990", on_date="2026-06-30"
            )
        )
        # …and never resolve dates after it.
        self.assertEqual(
            [],
            master_data.match_account(
                self.connection, account_number="7777888899990", on_date="2026-07-01"
            ),
        )


class MasterDataApiTests(unittest.TestCase):
    """Live-server workflow tests for account registration and review."""

    @classmethod
    def setUpClass(cls) -> None:
        import os

        cls.temp_dir = tempfile.TemporaryDirectory()
        root = Path(cls.temp_dir.name)
        cls.db_path = root / "app.db"
        cls.storage = root / "files"

        cls._old_db_path = server.DB_PATH
        cls._old_storage = server.STORAGE_DIR
        server.DB_PATH = cls.db_path
        server.STORAGE_DIR = cls.storage

        os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD
        connection = connect(cls.db_path)
        migrate(connection)
        server.ensure_bootstrap_admin(connection)
        connection.close()

        class QuietHandler(server.AppHandler):
            def log_message(self, *args) -> None:
                pass

        cls.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
        cls.port = cls.httpd.server_address[1]
        cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True)
        cls.thread.start()

        cls.admin = Client("127.0.0.1", cls.port)
        status, _, data = cls.admin.post_json(
            "/api/login",
            {"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"},
        )
        assert status == 200, data
        status, _, data = cls.admin.post_json(
            "/api/password/change",
            {"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD},
        )
        assert status == 200, data

        # Company A created together with its cashier login in one call.
        status, _, data = cls.admin.post_json(
            "/api/admin/companies",
            {"name": "甲公司", "credit_code": "91410100TEST", "cashier_name": "牛女士",
             "username": "cashier-a"},
        )
        assert status == 200, data
        payload = as_json(data)
        cls.company_a = payload["company_id"]
        cls.initial_a = payload["initial_password"]
        assert cls.initial_a != "cashier-a"
        assert len(cls.initial_a) >= 12

        status, _, data = cls.admin.post_json("/api/admin/companies", {"name": "乙公司"})
        assert status == 200, data
        cls.company_b = as_json(data)["company_id"]
        status, _, data = cls.admin.post_json(
            "/api/admin/users", {"username": "cashier-b", "company_id": cls.company_b}
        )
        assert status == 200, data
        cls.initial_b = as_json(data)["initial_password"]
        assert cls.initial_b != "cashier-b"

        cls.cashier_a = cls._login_company_user("cashier-a", cls.initial_a)
        cls.cashier_b = cls._login_company_user("cashier-b", cls.initial_b)

    @classmethod
    def _login_company_user(cls, username: str, initial: str) -> Client:
        client = Client("127.0.0.1", cls.port)
        status, _, data = client.post_json(
            "/api/login", {"username": username, "password": initial, "portal": "company"}
        )
        assert status == 200, data
        status, _, data = client.post_json(
            "/api/password/change",
            {"old_password": initial, "new_password": CASHIER_PASSWORD},
        )
        assert status == 200, data
        return client

    @classmethod
    def tearDownClass(cls) -> None:
        import os

        cls.httpd.shutdown()
        cls.httpd.server_close()
        server.DB_PATH = cls._old_db_path
        server.STORAGE_DIR = cls._old_storage
        os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None)
        cls.temp_dir.cleanup()

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    def submit_account(self, client: Client, number: str, bank: str = "郑州银行"):
        return client.post_json(
            "/api/company/accounts",
            {"bank_name": bank, "account_type": "一般户",
             "account_number": number, "start_date": "2026-08-01"},
        )

    def review(self, account_id: int, decision: str, reason: str = ""):
        return self.admin.post_json(
            f"/api/admin/accounts/{account_id}/review",
            {"decision": decision, "reason": reason},
        )

    # ------------------------------------------------------------------
    # Company management
    # ------------------------------------------------------------------

    def test_new_company_appears_in_admin_list_with_user(self) -> None:
        status, _, data = self.admin.get("/api/admin/companies")
        self.assertEqual(200, status)
        companies = {c["name"]: c for c in as_json(data)["companies"]}
        self.assertIn("甲公司", companies)
        self.assertEqual("91410100TEST", companies["甲公司"]["credit_code"])
        self.assertEqual("牛女士", companies["甲公司"]["cashier_name"])
        self.assertIn("cashier-a", companies["甲公司"]["usernames"])
        self.assertEqual(0, companies["乙公司"]["account_count"])

    def test_company_user_cannot_change_own_company_binding(self) -> None:
        # A company_id in the submission body is ignored: the account is
        # always bound to the session company.
        status, _, data = self.cashier_a.post_json(
            "/api/company/accounts",
            {"bank_name": "民生银行", "account_type": "一般户",
             "account_number": "6001000100010001", "company_id": self.company_b},
        )
        self.assertEqual(200, status, data)
        self.assertEqual(self.company_a, as_json(data)["account"]["company_id"])

    # ------------------------------------------------------------------
    # Account registration -> review workflow
    # ------------------------------------------------------------------

    def test_full_registration_workflow(self) -> None:
        status, _, data = self.submit_account(self.cashier_a, "6222 0210-0100 8888")
        self.assertEqual(200, status, data)
        account = as_json(data)["account"]
        account_id = account["id"]
        self.assertEqual("pending", account["status"])
        # Company view is masked only.
        self.assertEqual("****8888", account["account_number_masked"])
        self.assertNotIn("account_number", account)
        self.assertFalse(account.get("usable", True))

        # Admin audit view shows the full normalized number.
        status, _, data = self.admin.get("/api/admin/accounts")
        self.assertEqual(200, status)
        full = [a for a in as_json(data)["accounts"] if a["id"] == account_id][0]
        self.assertEqual("62220210010088 88".replace(" ", ""), full["account_number"])
        self.assertEqual("甲公司", full["company_name"])

        # Pending account cannot upload.
        sample = Path("流水模板/中信银行账户流水.xlsx")
        status, _, data = self.cashier_a.post_multipart(
            "/api/parse", {"bank_account_id": str(account_id)},
            sample.name, sample.read_bytes(),
        )
        self.assertEqual(409, status, data)

        # Approve -> active and usable.
        status, _, data = self.review(account_id, "approve")
        self.assertEqual(200, status, data)
        self.assertEqual("active", as_json(data)["account"]["status"])
        self.assertEqual("2026-08-01", as_json(data)["account"]["effective_from"])

        status, _, data = self.cashier_a.get("/api/company/accounts")
        mine = [a for a in as_json(data)["accounts"] if a["id"] == account_id][0]
        self.assertTrue(mine["usable"])

        # Approved account accepts uploads.
        status, _, data = self.cashier_a.post_multipart(
            "/api/parse", {"bank_account_id": str(account_id)},
            sample.name, sample.read_bytes(),
        )
        self.assertEqual(200, status, data)

        # Disable -> leaves the usable window.
        status, _, data = self.review(account_id, "disable", "账户销户")
        self.assertEqual(200, status, data)
        disabled = as_json(data)["account"]
        self.assertEqual("disabled", disabled["status"])
        self.assertIsNotNone(disabled["effective_to"])

        status, _, data = self.cashier_a.get("/api/company/accounts")
        mine = [a for a in as_json(data)["accounts"] if a["id"] == account_id][0]
        self.assertFalse(mine["usable"])

    def test_return_and_resubmit_reuses_same_row(self) -> None:
        status, _, data = self.submit_account(self.cashier_a, "3100998877665")
        self.assertEqual(200, status, data)
        account_id = as_json(data)["account"]["id"]

        status, _, data = self.review(account_id, "return", "开户许可证模糊")
        self.assertEqual(200, status, data)
        self.assertEqual("returned", as_json(data)["account"]["status"])

        # Return requires a reason.
        status, _, data = self.submit_account(self.cashier_a, "3100998877666")
        other_id = as_json(data)["account"]["id"]
        status, _, _ = self.review(other_id, "return")
        self.assertEqual(400, status)

        # Resubmission reopens the same row as pending.
        status, _, data = self.submit_account(self.cashier_a, "3100 9988-7766 5")
        self.assertEqual(200, status, data)
        resubmitted = as_json(data)["account"]
        self.assertEqual(account_id, resubmitted["id"])
        self.assertEqual("pending", resubmitted["status"])

        # The whole trail is auditable with actor, time and reason.
        status, _, data = self.admin.get(
            f"/api/admin/master-changes?entity_type=bank_account&entity_id={account_id}"
        )
        self.assertEqual(200, status)
        changes = as_json(data)["changes"]
        actions = [change["action"] for change in changes]
        self.assertEqual(["resubmit", "return", "submit"], actions)
        returned = [c for c in changes if c["action"] == "return"][0]
        self.assertEqual("开户许可证模糊", returned["reason"])
        self.assertEqual("group-admin", returned["actor_username"])
        self.assertIn('"pending"', returned["before_json"])
        self.assertIn('"returned"', returned["after_json"])

    def test_duplicate_normalized_number_conflicts(self) -> None:
        status, _, _ = self.submit_account(self.cashier_a, "4501111222233")
        self.assertEqual(200, status)
        # Same company, same number with different separators.
        status, _, data = self.submit_account(self.cashier_a, "4501-1112 2223-3")
        self.assertEqual(409, status, data)
        # Another company registering the same number conflicts too.
        status, _, data = self.submit_account(self.cashier_b, "4501111222233")
        self.assertEqual(409, status, data)
        self.assertIn("其他公司", as_json(data)["message"])

    def test_concurrent_submissions_create_only_one_account(self) -> None:
        number = "8800123456789"
        results: list[int] = []

        def submit() -> None:
            client = Client("127.0.0.1", self.port)
            client.cookies.update(self.cashier_a.cookies)
            status, _, _ = client.post_json(
                "/api/company/accounts",
                {"bank_name": "中信银行", "account_type": "基本户",
                 "account_number": number},
            )
            results.append(status)

        threads = [threading.Thread(target=submit) for _ in range(2)]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()
        self.assertEqual(sorted(results), [200, 409])

        connection = connect(self.db_path)
        try:
            count = connection.execute(
                "SELECT COUNT(*) AS n FROM bank_accounts WHERE account_number = ?",
                (number,),
            ).fetchone()["n"]
        finally:
            connection.close()
        self.assertEqual(1, count)

    # ------------------------------------------------------------------
    # RBAC and masking
    # ------------------------------------------------------------------

    def test_company_user_forbidden_on_admin_account_endpoints(self) -> None:
        for call in (
            lambda: self.cashier_a.get("/api/admin/accounts"),
            lambda: self.cashier_a.post_json(
                "/api/admin/accounts/1/review", {"decision": "approve"}
            ),
            lambda: self.cashier_a.get("/api/admin/accounts/1/aliases"),
            lambda: self.cashier_a.post_json(
                "/api/admin/accounts/1/aliases",
                {"alias_kind": "name", "alias_value": "x"},
            ),
            lambda: self.cashier_a.get("/api/admin/master-changes"),
        ):
            status, _, data = call()
            self.assertEqual(403, status, data)

    def test_admin_cannot_use_company_account_endpoint(self) -> None:
        status, _, _ = self.admin.get("/api/company/accounts")
        self.assertEqual(403, status)
        status, _, _ = self.submit_account(self.admin, "1000200030004")
        self.assertEqual(403, status)

    def test_unauthenticated_master_data_calls_return_401(self) -> None:
        anon = Client("127.0.0.1", self.port)
        for call in (
            lambda: anon.get("/api/company/accounts"),
            lambda: anon.get("/api/admin/accounts"),
            lambda: anon.post_json("/api/company/accounts", {}),
            lambda: anon.get("/api/admin/master-changes"),
        ):
            status, _, _ = call()
            self.assertEqual(401, status)

    def test_company_account_list_scoped_and_masked(self) -> None:
        status, _, data = self.cashier_b.get("/api/company/accounts")
        self.assertEqual(200, status)
        for account in as_json(data)["accounts"]:
            self.assertEqual(self.company_b, account["company_id"])
            self.assertNotIn("account_number", account)
            self.assertTrue(account["account_number_masked"].startswith("****"))

    def test_upload_with_other_companys_account_returns_404(self) -> None:
        status, _, data = self.submit_account(self.cashier_a, "5550001112223")
        account_id = as_json(data)["account"]["id"]
        status, _, _ = self.review(account_id, "approve")
        self.assertEqual(200, status)
        sample = Path("流水模板/中信银行账户流水.xlsx")
        status, _, data = self.cashier_b.post_multipart(
            "/api/parse", {"bank_account_id": str(account_id)},
            sample.name, sample.read_bytes(),
        )
        self.assertEqual(404, status, data)

    # ------------------------------------------------------------------
    # Aliases via API
    # ------------------------------------------------------------------

    def test_alias_crud_and_audit(self) -> None:
        status, _, data = self.submit_account(self.cashier_a, "6601234509876")
        account_id = as_json(data)["account"]["id"]
        status, _, _ = self.review(account_id, "approve")
        self.assertEqual(200, status)

        status, _, data = self.admin.post_json(
            f"/api/admin/accounts/{account_id}/aliases",
            {"alias_kind": "name", "alias_value": "甲公司工会",
             "effective_from": "2026-01-01", "effective_to": "2026-12-31",
             "priority": 10},
        )
        self.assertEqual(200, status, data)
        alias_id = as_json(data)["alias_id"]

        # Duplicate alias on the same account conflicts.
        status, _, _ = self.admin.post_json(
            f"/api/admin/accounts/{account_id}/aliases",
            {"alias_kind": "name", "alias_value": "甲公司工会"},
        )
        self.assertEqual(409, status)

        status, _, data = self.admin.get(f"/api/admin/accounts/{account_id}/aliases")
        self.assertEqual(200, status)
        aliases = as_json(data)["aliases"]
        self.assertEqual(1, len(aliases))
        self.assertEqual("甲公司工会", aliases[0]["alias_value"])
        self.assertEqual(10, aliases[0]["priority"])

        status, _, data = self.admin.get(
            f"/api/admin/master-changes?entity_type=account_alias&entity_id={alias_id}"
        )
        changes = as_json(data)["changes"]
        self.assertEqual("create", changes[0]["action"])
        self.assertIn("2026-12-31", changes[0]["after_json"])

        # Invalid alias input is rejected, unknown account is 404.
        status, _, _ = self.admin.post_json(
            f"/api/admin/accounts/{account_id}/aliases",
            {"alias_kind": "account", "alias_value": "abc"},
        )
        self.assertEqual(400, status)
        status, _, _ = self.admin.post_json(
            "/api/admin/accounts/99999/aliases",
            {"alias_kind": "name", "alias_value": "x"},
        )
        self.assertEqual(404, status)


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