autoChatGPTPlus-Paypal/test_server.py

import json
import threading
import unittest
from http.server import ThreadingHTTPServer
from pathlib import Path
from urllib import request as urlrequest
from unittest.mock import patch

from config import AppConfig


class StartSsoProxyTests(unittest.TestCase):
    def _post_json(self, url, payload):
        data = json.dumps(payload).encode("utf-8")
        req = urlrequest.Request(url, data=data, method="POST", headers={"Content-Type": "application/json"})
        try:
            with urlrequest.urlopen(req, timeout=5) as resp:
                return resp.status, json.loads(resp.read().decode("utf-8"))
        except urlrequest.HTTPError as exc:
            return exc.code, json.loads(exc.read().decode("utf-8"))

    def test_start_sso_does_not_inherit_global_proxy_by_default(self):
        import server

        cfg = AppConfig(
            proxy_url="http://global.proxy:8080",
            sso_mail_domain="aef.claudeai.life",
            cpa_url="http://cpa.example",
            cpa_management_key="token",
        )
        httpd = ThreadingHTTPServer(("127.0.0.1", 0), server.Handler)
        thread = threading.Thread(target=httpd.serve_forever, daemon=True)

        with patch("server.AppConfig.load", return_value=cfg), patch.object(server.JOB, "start_sso", return_value="") as start_sso:
            thread.start()
            self.addCleanup(thread.join, 2)
            self.addCleanup(httpd.server_close)
            self.addCleanup(httpd.shutdown)

            status, body = self._post_json(
                f"http://127.0.0.1:{httpd.server_address[1]}/api/start-sso",
                {"account_count": 1, "cpa_url": "http://cpa.example", "cpa_management_key": "token"},
            )

        self.assertEqual(status, 200)
        self.assertTrue(body["ok"])
        self.assertEqual(start_sso.call_args.kwargs["proxy_url"], "")


class DeprecatedFullAutomationTests(unittest.TestCase):
    def _post_json(self, url, payload=None):
        data = json.dumps(payload or {}).encode("utf-8")
        req = urlrequest.Request(url, data=data, method="POST", headers={"Content-Type": "application/json"})
        try:
            with urlrequest.urlopen(req, timeout=5) as resp:
                return resp.status, json.loads(resp.read().decode("utf-8"))
        except urlrequest.HTTPError as exc:
            return exc.code, json.loads(exc.read().decode("utf-8"))

    def _serve(self):
        import server

        httpd = ThreadingHTTPServer(("127.0.0.1", 0), server.Handler)
        thread = threading.Thread(target=httpd.serve_forever, daemon=True)
        thread.start()
        self.addCleanup(thread.join, 2)
        self.addCleanup(httpd.server_close)
        self.addCleanup(httpd.shutdown)
        return httpd

    def test_start_endpoint_is_disabled_for_full_automation(self):
        import server

        httpd = self._serve()
        status, body = self._post_json(f"http://127.0.0.1:{httpd.server_address[1]}/api/start")

        self.assertEqual(status, 410)
        self.assertIn("已停用", body["error"])
        self.assertFalse(hasattr(server.JOB, "start"))

    def test_tasks_api_rejects_full_mode_without_creating_task(self):
        import server

        httpd = self._serve()
        with patch("server.create_task") as create_task:
            status, body = self._post_json(
                f"http://127.0.0.1:{httpd.server_address[1]}/api/tasks",
                {"mode": "full"},
            )

        self.assertEqual(status, 410)
        self.assertIn("full", body["error"])
        create_task.assert_not_called()

    def test_status_does_not_expose_pool_maintainer(self):
        import server

        self.assertNotIn("pool_maintainer", server.JOB.status())

    def test_server_no_longer_starts_pool_maintainer(self):
        server_py = Path(__file__).resolve().parent.joinpath("server.py").read_text(encoding="utf-8")

        self.assertNotIn("start_pool_maintainer", server_py)


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