autoChatGPTPlus-Paypal/test_cpa_oauth.py

import unittest
from unittest.mock import patch


class CallbackUrlTests(unittest.TestCase):
    def test_accepts_localhost_auth_and_codex_callbacks_with_code_and_state(self):
        import cpa_oauth

        self.assertTrue(cpa_oauth.is_localhost_oauth_callback_url("http://localhost:1455/auth/callback?code=abc&state=xyz"))
        self.assertTrue(cpa_oauth.is_localhost_oauth_callback_url("http://127.0.0.1:1455/codex/callback?code=abc&state=xyz"))

    def test_rejects_non_localhost_or_missing_oauth_params(self):
        import cpa_oauth

        self.assertFalse(cpa_oauth.is_localhost_oauth_callback_url("https://example.com/auth/callback?code=abc&state=xyz"))
        self.assertFalse(cpa_oauth.is_localhost_oauth_callback_url("http://localhost:1455/auth/callback?code=abc"))
        self.assertFalse(cpa_oauth.is_localhost_oauth_callback_url("http://localhost:1455/other?code=abc&state=xyz"))


class FetchOauthUrlTests(unittest.TestCase):
    def test_stops_retrying_when_management_login_stays_on_login_page(self):
        import cpa_oauth

        class _FakeLocator:
            def __init__(self, page, selector):
                self.page = page
                self.selector = selector

            def count(self):
                if "password" in self.selector:
                    return 1
                if 'has-text("Login")' in self.selector:
                    return 1
                return 0

            def nth(self, _idx):
                return self

            def is_visible(self):
                return True

            def is_enabled(self):
                return True

            def fill(self, value):
                self.page.filled.append((self.selector, value))

            def click(self, **_kwargs):
                self.page.clicked.append(self.selector)

        class _FakePage:
            def __init__(self):
                self.url = ""
                self.filled = []
                self.clicked = []

            def goto(self, url, **_kwargs):
                self.url = url.rsplit("#", 1)[0] + "#/login"

            def wait_for_timeout(self, *_args, **_kwargs):
                pass

            def locator(self, selector):
                return _FakeLocator(self, selector)

        page = _FakePage()

        with self.assertRaisesRegex(RuntimeError, "CPA 管理登录失败"):
            cpa_oauth.fetch_cpa_oauth_url(
                page,
                cpa_url="https://cpa.example/management.html#/oauth",
                management_key="bad-key",
                log=lambda *_: None,
                timeout_sec=5,
            )

        self.assertLessEqual(len(page.clicked), 3)

    def test_waits_for_oauth_panel_after_management_login(self):
        import cpa_oauth

        class _FakeLocator:
            def __init__(self, page, selector):
                self.page = page
                self.selector = selector

            def count(self):
                if "authUrlValue" in self.selector:
                    if self.page.state == "oauth_generating":
                        self.page.auth_url_polls += 1
                        if self.page.auth_url_polls >= 2:
                            self.page.state = "auth_url"
                    return 1 if self.page.state == "auth_url" else 0
                if "password" in self.selector:
                    return 1 if self.page.state == "login" else 0
                if 'has-text("Login")' in self.selector or 'has-text("登录")' in self.selector:
                    return 1 if self.page.state in ("login", "oauth_card") else 0
                return 0

            def nth(self, _idx):
                return self

            def is_visible(self):
                return True

            def is_enabled(self):
                return True

            def fill(self, value):
                self.page.filled.append((self.selector, value))

            def click(self, **_kwargs):
                self.page.clicked.append((self.page.state, self.selector))
                if self.page.state == "login":
                    self.page.state = "loading_after_login"
                elif self.page.state == "oauth_card":
                    self.page.state = "oauth_generating"

            def inner_text(self, **_kwargs):
                if self.page.state == "auth_url":
                    return "https://auth.openai.com/oauth/authorize?client_id=codex"
                return ""

        class _FakePage:
            def __init__(self):
                self.url = ""
                self.state = "new"
                self.wait_calls = 0
                self.auth_url_polls = 0
                self.clicked = []
                self.filled = []

            def goto(self, url, **_kwargs):
                self.url = url.rsplit("#", 1)[0] + "#/login"
                self.state = "login"

            def wait_for_timeout(self, *_args, **_kwargs):
                self.wait_calls += 1
                if self.state == "loading_after_login" and self.wait_calls >= 3:
                    self.url = self.url.rsplit("#", 1)[0] + "#/oauth"
                    self.state = "oauth_card"

            def locator(self, selector):
                return _FakeLocator(self, selector)

        page = _FakePage()

        oauth_url = cpa_oauth.fetch_cpa_oauth_url(
            page,
            cpa_url="https://cpa.example/management.html#/oauth",
            management_key="secret",
            log=lambda *_: None,
            timeout_sec=2,
        )

        self.assertTrue(oauth_url.startswith("https://auth.openai.com/oauth/authorize"))
        self.assertTrue(any(state == "login" for state, _selector in page.clicked))
        self.assertTrue(any(state == "oauth_card" for state, _selector in page.clicked))

    def test_reopens_oauth_route_when_management_login_lands_on_home(self):
        import cpa_oauth

        class _FakeLocator:
            def __init__(self, page, selector):
                self.page = page
                self.selector = selector

            def count(self):
                if "authUrlValue" in self.selector:
                    if self.page.state == "oauth_generating":
                        self.page.state = "auth_url"
                    return 1 if self.page.state == "auth_url" else 0
                if "password" in self.selector:
                    return 1 if self.page.state == "login" else 0
                if 'has-text("Login")' in self.selector or 'has-text("登录")' in self.selector:
                    return 1 if self.page.state in ("login", "oauth_card") else 0
                return 0

            def nth(self, _idx):
                return self

            def is_visible(self):
                return True

            def is_enabled(self):
                return True

            def fill(self, value):
                self.page.filled.append((self.selector, value))

            def click(self, **_kwargs):
                self.page.clicked.append((self.page.state, self.selector))
                if self.page.state == "login":
                    self.page.url = "https://cpa.example/management.html#/"
                    self.page.state = "home"
                elif self.page.state == "oauth_card":
                    self.page.state = "oauth_generating"

            def inner_text(self, **_kwargs):
                if self.page.state == "auth_url":
                    return "https://auth.openai.com/oauth/authorize?client_id=codex"
                return ""

        class _FakePage:
            def __init__(self):
                self.url = ""
                self.state = "new"
                self.goto_calls = []
                self.clicked = []
                self.filled = []

            def goto(self, url, **_kwargs):
                self.goto_calls.append(url)
                self.url = url
                if len(self.goto_calls) == 1:
                    self.url = "https://cpa.example/management.html#/login"
                    self.state = "login"
                elif url.endswith("#/oauth"):
                    self.state = "oauth_card"

            def wait_for_timeout(self, *_args, **_kwargs):
                pass

            def locator(self, selector):
                return _FakeLocator(self, selector)

        page = _FakePage()

        oauth_url = cpa_oauth.fetch_cpa_oauth_url(
            page,
            cpa_url="https://cpa.example/management.html#/oauth",
            management_key="secret",
            log=lambda *_: None,
            timeout_sec=2,
        )

        self.assertTrue(oauth_url.startswith("https://auth.openai.com/oauth/authorize"))
        self.assertGreaterEqual(page.goto_calls.count("https://cpa.example/management.html#/oauth"), 2)
        self.assertTrue(any(state == "home" for state, _selector in page.clicked) is False)


class SubmitCallbackTests(unittest.TestCase):
    def test_submit_callback_fills_url_and_clicks_submit(self):
        import cpa_oauth

        class _FakeLocator:
            def __init__(self, page, selector):
                self.page = page
                self.selector = selector

            def count(self):
                if self.selector == 'input[placeholder*="localhost"]':
                    return 1
                if "Submit Callback URL" in self.selector:
                    return 1
                return 0

            def nth(self, _idx):
                return self

            def is_visible(self):
                return True

            def is_enabled(self):
                return True

            @property
            def first(self):
                return self

            def click(self, **_kwargs):
                self.page.clicked.append(self.selector)

            def fill(self, value):
                self.page.filled.append((self.selector, value))

        class _FakePage:
            def __init__(self):
                self.url = ""
                self.clicked = []
                self.filled = []
                self.goto_calls = []

            def goto(self, url, **_kwargs):
                self.url = url
                self.goto_calls.append(url)

            def locator(self, selector):
                return _FakeLocator(self, selector)

            def wait_for_timeout(self, *_args, **_kwargs):
                pass

        page = _FakePage()
        callback_url = "http://localhost:1455/auth/callback?code=abc&state=xyz"

        cpa_oauth.submit_oauth_callback_to_cpa(
            page,
            cpa_url="https://cpa.example/management.html#/oauth",
            callback_url=callback_url,
            log=lambda *_: None,
            wait_for_success=False,
        )

        self.assertEqual(page.goto_calls, ["https://cpa.example/management.html#/oauth"])
        self.assertEqual(page.filled, [('input[placeholder*="localhost"]', callback_url)])
        self.assertTrue(any("Submit Callback URL" in selector for selector in page.clicked))


class CaptureCallbackTests(unittest.TestCase):
    def test_fills_email_hint_when_openai_oauth_shows_email_prompt(self):
        import cpa_oauth

        callback_url = "http://localhost:1455/auth/callback?code=abc&state=xyz"

        class _FakeLocator:
            def __init__(self, page, selector):
                self.page = page
                self.selector = selector

            def count(self):
                if 'input[type="email"]' in self.selector:
                    return 1 if self.page.state == "email_prompt" else 0
                if 'button[type="submit"]' in self.selector:
                    return 1 if self.page.state == "email_prompt" else 0
                return 0

            def nth(self, _idx):
                return self

            def is_visible(self):
                return True

            def is_enabled(self):
                return True

            def fill(self, value):
                self.page.filled.append((self.selector, value))

            def click(self, **_kwargs):
                self.page.clicked.append(self.selector)
                self.page.url = callback_url
                self.page.state = "callback"

        class _FakePage:
            def __init__(self):
                self.url = ""
                self.state = "new"
                self.filled = []
                self.clicked = []

            def goto(self, url, **_kwargs):
                self.url = url
                self.state = "email_prompt"

            def wait_for_timeout(self, *_args, **_kwargs):
                pass

            def locator(self, selector):
                return _FakeLocator(self, selector)

        page = _FakePage()

        captured = cpa_oauth.approve_openai_oauth_and_capture_callback(
            page,
            "https://auth.openai.com/oauth/authorize?client_id=codex",
            email_hint="n@example.com",
            log=lambda *_: None,
        )

        self.assertEqual(captured, callback_url)
        self.assertEqual(page.filled, [('input[type="email"]', "n@example.com")])
        self.assertIn('button[type="submit"]', page.clicked)

    def test_clicks_delayed_codex_consent_continue_after_email_continue(self):
        import cpa_oauth

        callback_url = "http://localhost:1455/auth/callback?code=abc&state=xyz"

        class _FakeLocator:
            def __init__(self, page, selector):
                self.page = page
                self.selector = selector

            def count(self):
                if 'input[type="email"]' in self.selector:
                    return 1 if self.page.state == "email_prompt" else 0
                if 'button[type="submit"]' in self.selector:
                    if self.page.state == "email_prompt":
                        return 1
                    if self.page.state == "loading_consent":
                        self.page.button_polls += 1
                        if self.page.button_polls >= 2:
                            self.page.state = "consent"
                            self.page.url = "https://auth.openai.com/sign-in-with-chatgpt/codex/consent"
                            return 1
                    return 1 if self.page.state == "consent" else 0
                return 0

            def nth(self, _idx):
                return self

            def is_visible(self):
                return True

            def is_enabled(self):
                return True

            def fill(self, value):
                self.page.filled.append((self.selector, value))

            def click(self, **_kwargs):
                self.page.clicked.append((self.page.state, self.selector))
                if self.page.state == "email_prompt":
                    self.page.url = "https://auth.openai.com/log-in"
                    self.page.state = "loading_consent"
                elif self.page.state == "consent":
                    self.page.url = callback_url
                    self.page.state = "callback"

        class _FakePage:
            def __init__(self):
                self.url = ""
                self.state = "new"
                self.button_polls = 0
                self.filled = []
                self.clicked = []

            def goto(self, url, **_kwargs):
                self.url = url
                self.state = "email_prompt"

            def wait_for_timeout(self, *_args, **_kwargs):
                pass

            def locator(self, selector):
                return _FakeLocator(self, selector)

        page = _FakePage()

        captured = cpa_oauth.approve_openai_oauth_and_capture_callback(
            page,
            "https://auth.openai.com/oauth/authorize?client_id=codex",
            email_hint="n@example.com",
            log=lambda *_: None,
            timeout_sec=1,
        )

        self.assertEqual(captured, callback_url)
        self.assertIn(("email_prompt", 'button[type="submit"]'), page.clicked)
        self.assertIn(("consent", 'button[type="submit"]'), page.clicked)

    def test_submits_email_only_once_when_prompt_remains_visible_before_consent(self):
        import cpa_oauth

        callback_url = "http://localhost:1455/auth/callback?code=abc&state=xyz"

        class _FakeLocator:
            def __init__(self, page, selector):
                self.page = page
                self.selector = selector

            def count(self):
                if 'input[type="email"]' in self.selector:
                    return 1 if self.page.state in ("email_prompt", "loading_consent") else 0
                if 'button[type="submit"]' in self.selector:
                    return 1 if self.page.state in ("email_prompt", "loading_consent", "consent") else 0
                return 0

            def nth(self, _idx):
                return self

            def is_visible(self):
                return True

            def is_enabled(self):
                return True

            def fill(self, value):
                self.page.filled.append((self.page.state, self.selector, value))

            def click(self, **_kwargs):
                self.page.clicked.append((self.page.state, self.selector))
                if self.page.state == "email_prompt":
                    self.page.state = "loading_consent"
                    self.page.url = "https://auth.openai.com/log-in"
                elif self.page.state == "loading_consent":
                    self.page.state = "consent"
                    self.page.url = "https://auth.openai.com/sign-in-with-chatgpt/codex/consent"
                elif self.page.state == "consent":
                    self.page.url = callback_url
                    self.page.state = "callback"

        class _FakePage:
            def __init__(self):
                self.url = ""
                self.state = "new"
                self.filled = []
                self.clicked = []

            def goto(self, url, **_kwargs):
                self.url = url
                self.state = "email_prompt"

            def wait_for_timeout(self, *_args, **_kwargs):
                pass

            def locator(self, selector):
                return _FakeLocator(self, selector)

        page = _FakePage()

        captured = cpa_oauth.approve_openai_oauth_and_capture_callback(
            page,
            "https://auth.openai.com/oauth/authorize?client_id=codex",
            email_hint="n@example.com",
            log=lambda *_: None,
            timeout_sec=1,
        )

        self.assertEqual(captured, callback_url)
        self.assertEqual(page.filled, [("email_prompt", 'input[type="email"]', "n@example.com")])
        self.assertIn(("consent", 'button[type="submit"]'), page.clicked)

    def test_detects_callback_opened_in_another_context_page(self):
        import cpa_oauth

        callback_url = "http://localhost:1455/auth/callback?code=abc&state=xyz"

        class _FakeContext:
            def __init__(self):
                self.pages = []

        class _FakeCallbackPage:
            def __init__(self, url):
                self.url = url

        class _FakeLocator:
            def __init__(self, page, selector):
                self.page = page
                self.selector = selector

            def count(self):
                if 'button[type="submit"]' in self.selector:
                    return 1
                return 0

            def nth(self, _idx):
                return self

            def is_visible(self):
                return True

            def is_enabled(self):
                return True

            def click(self, **_kwargs):
                self.page.clicked.append(self.selector)
                self.page.context.pages.append(_FakeCallbackPage(callback_url))

        class _FakePage:
            def __init__(self):
                self.url = ""
                self.clicked = []
                self.context = _FakeContext()
                self.context.pages.append(self)

            def goto(self, url, **_kwargs):
                self.url = "https://auth.openai.com/sign-in-with-chatgpt/codex/consent"

            def wait_for_timeout(self, *_args, **_kwargs):
                pass

            def locator(self, selector):
                return _FakeLocator(self, selector)

        page = _FakePage()

        captured = cpa_oauth.approve_openai_oauth_and_capture_callback(
            page,
            "https://auth.openai.com/oauth/authorize?client_id=codex",
            log=lambda *_: None,
            timeout_sec=1,
        )

        self.assertEqual(captured, callback_url)
        self.assertIn('button[type="submit"]', page.clicked)

    def test_capture_callback_even_when_localhost_navigation_raises(self):
        import cpa_oauth

        callback_url = "http://localhost:1455/auth/callback?code=abc&state=xyz"

        class _FakePage:
            def __init__(self):
                self.url = ""

            def goto(self, *_args, **_kwargs):
                self.url = callback_url
                raise RuntimeError("net::ERR_CONNECTION_REFUSED")

            def wait_for_timeout(self, *_args, **_kwargs):
                pass

            def locator(self, _selector):
                raise AssertionError("callback 已捕获时不应继续查找按钮")

        captured = cpa_oauth.approve_openai_oauth_and_capture_callback(
            _FakePage(),
            "https://auth.openai.com/oauth",
            log=lambda *_: None,
        )

        self.assertEqual(captured, callback_url)


class AuthorizeFlowTests(unittest.TestCase):
    def test_authorize_keeps_cpa_panel_page_and_uses_separate_oauth_page(self):
        import cpa_oauth

        class _FakeContext:
            def __init__(self, oauth_page):
                self.oauth_page = oauth_page
                self.new_page_calls = 0

            def new_page(self):
                self.new_page_calls += 1
                return self.oauth_page

        class _FakePage:
            def __init__(self, name):
                self.name = name
                self.closed = False
                self.context = None

            def close(self):
                self.closed = True

        panel_page = _FakePage("panel")
        oauth_page = _FakePage("oauth")
        panel_page.context = _FakeContext(oauth_page)

        def fake_approve(page, oauth_url, **kwargs):
            self.assertIs(page, oauth_page)
            self.assertEqual(oauth_url, "https://auth.openai.com/oauth/authorize")
            self.assertEqual(kwargs["email_hint"], "n@example.com")
            return "http://localhost:1455/auth/callback?code=abc&state=xyz"

        def fake_submit(page, **kwargs):
            self.assertIs(page, panel_page)
            self.assertEqual(kwargs["callback_url"], "http://localhost:1455/auth/callback?code=abc&state=xyz")
            return "Authentication successful!"

        with patch.object(cpa_oauth, "fetch_cpa_oauth_url", return_value="https://auth.openai.com/oauth/authorize") as fetch:
            with patch.object(cpa_oauth, "approve_openai_oauth_and_capture_callback", side_effect=fake_approve) as approve:
                with patch.object(cpa_oauth, "submit_oauth_callback_to_cpa", side_effect=fake_submit) as submit:
                    result = cpa_oauth.authorize_codex_oauth_to_cpa(
                        panel_page,
                        cpa_url="https://cpa.example/management.html#/oauth",
                        management_key="secret",
                        email_hint="n@example.com",
                        log=lambda *_: None,
                    )

        fetch.assert_called_once()
        approve.assert_called_once()
        submit.assert_called_once()
        self.assertEqual(panel_page.context.new_page_calls, 1)
        self.assertTrue(oauth_page.closed)
        self.assertFalse(panel_page.closed)
        self.assertEqual(result["callbackUrl"], "http://localhost:1455/auth/callback?code=abc&state=xyz")


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