cmshoppe/tests/test_accounts.py

import os
import unittest
from unittest import mock

from _helpers import TempDirMixin

from app import accounts
from app import config as account_config


class AccountsTests(TempDirMixin, unittest.TestCase):
    def make_config(self, temp_dir):
        return {
            "chrome_path": "chrome.exe",
            "user_data_root": os.path.join(temp_dir, "chrome_user_data_dir"),
            "db_path": os.path.join(temp_dir, "cmshopee.db"),
            "debug_port_range": [9222, 9224],
        }

    def test_create_account_stores_record_and_creates_user_data_dir(self):
        with self.make_temp_dir() as temp_dir:
            cfg = self.make_config(temp_dir)
            account = accounts.create_account(
                "主店",
                "alias",
                "https://seller.shopee.tw/portal/",
                9222,
                password="secret",
                note="备注",
                config=cfg,
            )

            self.assertEqual("主店", account.account_name)
            self.assertEqual("alias", account.alias)
            self.assertEqual("seller.shopee.tw", account.region_host)
            self.assertEqual("alias_cdb6fdbe", account.slug)
            self.assertTrue(os.path.isdir(account.user_data_dir))
            self.assertEqual("******", accounts.mask_password(account.password))
            self.assertEqual([account], accounts.list_accounts(config=cfg))

        self.assert_removed(temp_dir)

    def test_update_account_can_change_alias_and_rebuild_directory(self):
        with self.make_temp_dir() as temp_dir:
            cfg = self.make_config(temp_dir)
            accounts.create_account("主店", "alias", debug_port=9222, config=cfg)

            updated = accounts.update_account(
                "alias",
                "新主店",
                "new alias",
                "seller.shopee.tw",
                9333,
                password="new-secret",
                note="新备注",
                config=cfg,
            )

            self.assertEqual("new alias", updated.alias)
            self.assertEqual(account_config.make_slug("new alias"), updated.slug)
            self.assertEqual(9333, updated.debug_port)
            self.assertTrue(os.path.isdir(updated.user_data_dir))
            with self.assertRaises(accounts.AccountError):
                accounts.get_account("alias", config=cfg)
            self.assertEqual("new-secret", accounts.get_account("new alias", config=cfg).password)

        self.assert_removed(temp_dir)

    def test_next_debug_port_skips_existing_accounts(self):
        with self.make_temp_dir() as temp_dir:
            cfg = self.make_config(temp_dir)
            accounts.create_account("店1", "a1", debug_port=9222, config=cfg)
            accounts.create_account("店2", "a2", debug_port=9223, config=cfg)

            self.assertEqual(9224, accounts.next_debug_port(config=cfg))

        self.assert_removed(temp_dir)

    def test_launch_for_login_starts_chrome_when_port_not_running(self):
        with self.make_temp_dir() as temp_dir:
            cfg = self.make_config(temp_dir)
            account = accounts.create_account("主店", "alias", debug_port=9222, config=cfg)

            class FakeProcess:
                pid = 1234

            with mock.patch("app.accounts.chrome.is_running", return_value=False) as is_running, \
                    mock.patch("app.accounts.chrome.launch_chrome", return_value=FakeProcess()) as launch, \
                    mock.patch("app.accounts.chrome.wait_debug_ready", return_value=True) as wait_ready, \
                    mock.patch("app.accounts.cdp.http_get", return_value=[]), \
                    mock.patch(
                        "app.accounts.cdp.create_tab_info",
                        return_value={
                            "id": "target-new",
                            "url": "https://seller.shopee.tw/portal/",
                            "webSocketDebuggerUrl": "ws-new",
                        },
                    ) as create_tab_info, \
                    mock.patch("app.accounts.cdp.activate_tab") as activate_tab:
                result = accounts.launch_for_login("alias", config=cfg)

            self.assertEqual("launched", result["action"])
            self.assertTrue(result["launched"])
            self.assertFalse(result["reused"])
            self.assertEqual(1234, result["pid"])
            self.assertEqual("target-new", result["target_id"])
            is_running.assert_called_once_with(9222)
            launch.assert_called_once_with(account, config=cfg)
            wait_ready.assert_called_once_with(9222, timeout=60)
            create_tab_info.assert_called_once_with(
                "https://seller.shopee.tw/portal/",
                host="127.0.0.1:9222",
            )
            activate_tab.assert_called_once_with("target-new", host="127.0.0.1:9222")

        self.assert_removed(temp_dir)

    def test_launch_for_login_reuses_running_chrome_without_starting_process(self):
        with self.make_temp_dir() as temp_dir:
            cfg = self.make_config(temp_dir)
            accounts.create_account("主店", "alias", debug_port=9222, config=cfg)
            existing_tab = {
                "type": "page",
                "id": "target-existing",
                "url": "https://seller.shopee.tw/portal/",
                "webSocketDebuggerUrl": "ws-existing",
            }

            with mock.patch("app.accounts.chrome.is_running", return_value=True) as is_running, \
                    mock.patch("app.accounts.chrome.launch_chrome") as launch, \
                    mock.patch("app.accounts.chrome.wait_debug_ready") as wait_ready, \
                    mock.patch("app.accounts.cdp.http_get", return_value=[existing_tab]) as http_get, \
                    mock.patch("app.accounts.cdp.create_tab_info") as create_tab_info, \
                    mock.patch("app.accounts.cdp.activate_tab") as activate_tab:
                result = accounts.launch_for_login("alias", config=cfg)

            self.assertEqual("reused", result["action"])
            self.assertTrue(result["reused"])
            self.assertFalse(result["launched"])
            self.assertIsNone(result["pid"])
            self.assertFalse(result["created_tab"])
            self.assertEqual("target-existing", result["target_id"])
            is_running.assert_called_once_with(9222)
            http_get.assert_called_once_with("/json", host="127.0.0.1:9222")
            activate_tab.assert_called_once_with("target-existing", host="127.0.0.1:9222")
            launch.assert_not_called()
            wait_ready.assert_not_called()
            create_tab_info.assert_not_called()

        self.assert_removed(temp_dir)

    def test_create_shortcut_uses_account_record(self):
        with self.make_temp_dir() as temp_dir:
            cfg = self.make_config(temp_dir)
            account = accounts.create_account("主店", "alias", debug_port=9222, config=cfg)
            shortcut_path = os.path.join(temp_dir, "Desktop", "主店.lnk")

            with mock.patch(
                "app.accounts.chrome.create_shortcut",
                return_value=shortcut_path,
            ) as create_shortcut:
                result = accounts.create_shortcut(
                    "alias",
                    shortcut_path=shortcut_path,
                    config=cfg,
                )

            self.assertEqual(shortcut_path, result)
            create_shortcut.assert_called_once_with(
                account,
                shortcut_path=shortcut_path,
                desktop_dir=None,
                config=cfg,
            )

        self.assert_removed(temp_dir)

    def test_detect_login_updates_last_login_at_when_logged_in(self):
        with self.make_temp_dir() as temp_dir:
            cfg = self.make_config(temp_dir)
            accounts.create_account("主店", "alias", debug_port=9222, config=cfg)

            with mock.patch(
                "app.accounts.editor.login_status",
                return_value={"logged_in": True, "reason": None},
            ) as login_status:
                status = accounts.detect_login("alias", timeout=1, config=cfg)

            self.assertTrue(status["logged_in"])
            login_status.assert_called_once()
            self.assertIsNotNone(accounts.get_account("alias", config=cfg).last_login_at)

        self.assert_removed(temp_dir)

    def test_duplicate_alias_raises_clear_error(self):
        with self.make_temp_dir() as temp_dir:
            cfg = self.make_config(temp_dir)
            accounts.create_account("店1", "alias", debug_port=9222, config=cfg)

            with self.assertRaises(accounts.AccountError):
                accounts.create_account("店2", "alias", debug_port=9223, config=cfg)

        self.assert_removed(temp_dir)

    def test_duplicate_debug_port_raises_clear_error(self):
        with self.make_temp_dir() as temp_dir:
            cfg = self.make_config(temp_dir)
            accounts.create_account("店1", "a1", debug_port=9222, config=cfg)
            accounts.create_account("店2", "a2", debug_port=9223, config=cfg)

            with self.assertRaises(accounts.AccountError):
                accounts.create_account("店3", "a3", debug_port=9222, config=cfg)
            with self.assertRaises(accounts.AccountError):
                accounts.update_account(
                    "a2",
                    "店2",
                    "a2",
                    "seller.shopee.tw",
                    9222,
                    config=cfg,
                )

        self.assert_removed(temp_dir)


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