def get_fresh_token():
    """每次請(qǐng)求前獲取新Token"""
    login_data = {"username": "testuser", "password": "testpass"}
    resp = requests.post(f"{BASE_URL}/login", json=login_data)
    return resp.json()["data"]["access_token"]
def test_api_with_fresh_token():
    token = get_fresh_token()
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.get(f"{BASE_URL}/api/data", headers=headers)
    return response.json()

import pytest
class TestBase:
    token = None
    @classmethod
    def setup_class(cls):
        """測(cè)試類執(zhí)行前獲取Token"""
        cls.token = cls.get_token_once()
    @classmethod
    def get_token_once(cls):
        login_data = {"username": "testuser", "password": "testpass"}
        resp = requests.post(f"{BASE_URL}/login", json=login_data)
        return resp.json()["data"]["access_token"]
    def test_user_info(self):
        headers = {"Authorization": f"Bearer {self.token}"}
        response = requests.get(f"{BASE_URL}/user/info", headers=headers)
        assert response.status_code == 200

class SmartTokenClient:
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.username = username
        self.password = password
        self.token = None
    def login(self):
        """登錄獲取Token"""
        login_data = {
            "username": self.username,
            "password": self.password
        }
        resp = requests.post(f"{self.base_url}/login", json=login_data)
        self.token = resp.json()["data"]["access_token"]
        return self.token
    def request_with_retry(self, method, endpoint, **kwargs):
        """帶重試機(jī)制的請(qǐng)求"""
        if not self.token:
            self.login()
        headers = kwargs.get('headers', {})
        headers['Authorization'] = f'Bearer {self.token}'
        kwargs['headers'] = headers
        response = requests.request(method, f"{self.base_url}{endpoint}", **kwargs)
        # 檢測(cè)到Token過(guò)期，重新登錄并重試
        if response.status_code == 401:
            print("檢測(cè)到Token過(guò)期，重新登錄...")
            self.login()
            kwargs['headers']['Authorization'] = f'Bearer {self.token}'
            response = requests.request(method, f"{self.base_url}{endpoint}", **kwargs)
        return response
# 使用示例
client = SmartTokenClient(BASE_URL, "testuser", "testpass")
response = client.request_with_retry("GET", "/api/users")

import time
class ProactiveTokenClient:
    def __init__(self, base_url, username, password, token_ttl=1800):
        self.base_url = base_url
        self.username = username
        self.password = password
        self.token_ttl = token_ttl  # Token有效期，默認(rèn)30分鐘
        self.token = None
        self.token_time = 0
    def get_token(self):
        """獲取Token，如果快過(guò)期則主動(dòng)刷新"""
        current_time = time.time()
        # 沒(méi)有Token或Token即將過(guò)期（提前5分鐘刷新）
        if not self.token or (current_time - self.token_time) > (self.token_ttl - 300):
            self.refresh_token()
        return self.token
    def refresh_token(self):
        """刷新Token"""
        login_data = {
            "username": self.username, 
            "password": self.password
        }
        resp = requests.post(f"{self.base_url}/login", json=login_data)
        self.token = resp.json()["data"]["access_token"]
        self.token_time = time.time()
        print(f"Token已刷新: {self.token[:20]}...")
    def get(self, endpoint, **kwargs):
        """GET請(qǐng)求"""
        token = self.get_token()
        headers = kwargs.get('headers', {})
        headers['Authorization'] = f'Bearer {token}'
        kwargs['headers'] = headers
        return requests.get(f"{self.base_url}{endpoint}", **kwargs)

class DualTokenClient:
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.username = username
        self.password = password
        self.access_token = None
        self.refresh_token = None
        self.setup_tokens()
    def setup_tokens(self):
        """初始獲取雙Token"""
        login_data = {
            "username": self.username,
            "password": self.password
        }
        resp = requests.post(f"{self.base_url}/login", json=login_data)
        token_data = resp.json()["data"]
        self.access_token = token_data["access_token"]
        self.refresh_token = token_data["refresh_token"]
    def refresh_access_token(self):
        """使用Refresh Token刷新Access Token"""
        refresh_data = {
            "refresh_token": self.refresh_token
        }
        try:
            resp = requests.post(f"{self.base_url}/refresh", json=refresh_data)
            if resp.status_code == 200:
                token_data = resp.json()["data"]
                self.access_token = token_data["access_token"]
                # 有些系統(tǒng)會(huì)返回新的refresh_token
                if "refresh_token" in token_data:
                    self.refresh_token = token_data["refresh_token"]
                return True
        except Exception as e:
            print(f"刷新Token失敗: {e}")
        # 刷新失敗，重新登錄
        self.setup_tokens()
        return False
    def request(self, method, endpoint, max_retry=1, **kwargs):
        """支持Token自動(dòng)刷新的請(qǐng)求"""
        for attempt in range(max_retry + 1):
            headers = kwargs.get('headers', {})
            headers['Authorization'] = f'Bearer {self.access_token}'
            kwargs['headers'] = headers
            response = requests.request(method, f"{self.base_url}{endpoint}", **kwargs)
            if response.status_code != 401:
                return response
            if attempt < max_retry:
                print("Access Token過(guò)期，嘗試刷新...")
                if self.refresh_access_token():
                    continue
        return response

def auto_token_refresh(func):
    """自動(dòng)Token刷新裝飾器"""
    def wrapper(self, *args, **kwargs):
        # 確保有有效Token
        self.ensure_valid_token()
        # 執(zhí)行原函數(shù)
        result = func(self, *args, **kwargs)
        # 如果返回401，刷新Token后重試
        if hasattr(result, 'status_code') and result.status_code == 401:
            print("檢測(cè)到401錯(cuò)誤，自動(dòng)刷新Token并重試...")
            self.refresh_token()
            result = func(self, *args, **kwargs)
        return result
    return wrapper
class DecoratorTokenClient:
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.username = username
        self.password = password
        self.token = None
    def login(self):
        """登錄獲取Token"""
        login_data = {
            "username": self.username,
            "password": self.password
        }
        resp = requests.post(f"{self.base_url}/login", json=login_data)
        self.token = resp.json()["data"]["access_token"]
        return self.token
    def refresh_token(self):
        """刷新Token"""
        return self.login()
    def ensure_valid_token(self):
        """確保有有效的Token"""
        if not self.token:
            self.login()
    @auto_token_refresh
    def get_users(self):
        """獲取用戶列表"""
        headers = {"Authorization": f"Bearer {self.token}"}
        return requests.get(f"{self.base_url}/users", headers=headers)
    @auto_token_refresh  
    def create_order(self, order_data):
        """創(chuàng)建訂單"""
        headers = {"Authorization": f"Bearer {self.token}"}
        return requests.post(f"{self.base_url}/orders", json=order_data, headers=headers)
# 使用示例
client = DecoratorTokenClient(BASE_URL, "testuser", "testpass")
response = client.get_users()  # 自動(dòng)處理Token

class SessionTokenClient:
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.username = username
        self.password = password
        self.session = requests.Session()
        self.is_logged_in = False
    def login(self):
        """登錄并保持會(huì)話"""
        login_data = {
            "username": self.username,
            "password": self.password
        }
        response = self.session.post(f"{self.base_url}/login", json=login_data)
        if response.status_code == 200:
            self.is_logged_in = True
            # 如果是Token認(rèn)證，可能需要手動(dòng)設(shè)置Header
            token = response.json()["data"]["access_token"]
            self.session.headers.update({"Authorization": f"Bearer {token}"})
        return response
    def ensure_login(self):
        """確保處于登錄狀態(tài)"""
        if not self.is_logged_in:
            self.login()
    def get(self, endpoint, **kwargs):
        """GET請(qǐng)求"""
        self.ensure_login()
        response = self.session.get(f"{self.base_url}{endpoint}", **kwargs)
        # 檢查是否因會(huì)話過(guò)期而失敗
        if response.status_code == 401:
            print("會(huì)話過(guò)期，重新登錄...")
            self.login()
            response = self.session.get(f"{self.base_url}{endpoint}", **kwargs)
        return response
    def post(self, endpoint, **kwargs):
        """POST請(qǐng)求"""
        self.ensure_login()
        response = self.session.post(f"{self.base_url}{endpoint}", **kwargs)
        if response.status_code == 401:
            print("會(huì)話過(guò)期，重新登錄...")
            self.login()
            response = self.session.post(f"{self.base_url}{endpoint}", **kwargs)
        return response
# 使用示例
client = SessionTokenClient(BASE_URL, "testuser", "testpass")
response = client.get("/api/users")  # 自動(dòng)處理會(huì)話

from abc import ABC, abstractmethod
class TokenManager(ABC):
    """Token管理器抽象類"""
    @abstractmethod
    def get_token(self) -> str:
        pass
    @abstractmethod
    def refresh_token(self) -> bool:
        pass
class DevTokenManager(TokenManager):
    """開(kāi)發(fā)環(huán)境Token管理器"""
    def get_token(self) -> str:
        # 開(kāi)發(fā)環(huán)境可能返回固定Token或mock Token
        return "dev_mock_token"
    def refresh_token(self) -> bool:
        return True
class TestTokenManager(TokenManager):
    """測(cè)試環(huán)境Token管理器"""
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.username = username
        self.password = password
        self.token = None
    def get_token(self) -> str:
        if not self.token:
            self.refresh_token()
        return self.token
    def refresh_token(self) -> bool:
        login_data = {
            "username": self.username,
            "password": self.password
        }
        resp = requests.post(f"{self.base_url}/login", json=login_data)
        self.token = resp.json()["data"]["access_token"]
        return True
class ProdTokenManager(TokenManager):
    """生產(chǎn)環(huán)境Token管理器 - 更嚴(yán)格的安全控制"""
    def get_token(self) -> str:
        # 生產(chǎn)環(huán)境可能需要更復(fù)雜的認(rèn)證流程
        raise Exception("生產(chǎn)環(huán)境請(qǐng)使用正式的認(rèn)證流程")
    def refresh_token(self) -> bool:
        raise Exception("生產(chǎn)環(huán)境請(qǐng)使用正式的認(rèn)證流程")
class TokenManagerFactory:
    """Token管理器工廠"""
    @staticmethod
    def create_manager(env: str, **kwargs) -> TokenManager:
        if env == "dev":
            return DevTokenManager()
        elif env == "test":
            return TestTokenManager(kwargs['base_url'], kwargs['username'], kwargs['password'])
        elif env == "prod":
            return ProdTokenManager()
        else:
            raise ValueError(f"不支持的環(huán)境: {env}")
# 使用示例
token_manager = TokenManagerFactory.create_manager(
    "test", 
    base_url="https://test.api.com",
    username="testuser", 
    password="testpass"
)
token = token_manager.get_token()
headers = {"Authorization": f"Bearer {token}"}
response = requests.get("https://test.api.com/api/data", headers=headers)

import requests
import json
import os
import time
from typing import Optional, Dict, Any
# =======================
# ?? 認(rèn)證客戶端（支持自動(dòng)刷新）
# =======================
class AuthHttpClient:
    def __init__(
        self,
        base_url: str,
        login_url: str = "/auth/login",
        refresh_url: str = "/auth/refresh",
        username: str = "",
        password: str = "",
        token_file: str = "cache/token.json"
    ):
        self.base_url = base_url.rstrip("/")
        self.login_url = login_url
        self.refresh_url = refresh_url
        self.username = username
        self.password = password
        self.token_file = token_file
        self.session = requests.Session()
        # 確保緩存目錄存在
        os.makedirs(os.path.dirname(token_file), exist_ok=True)
    def _save_token(self, token: str, refresh_token: str, expires_in: int):
        """保存 Token 到本地文件"""
        data = {
            "access_token": token,
            "refresh_token": refresh_token,
            "expires_at": time.time() + expires_in - 60,  # 提前60秒過(guò)期
            "username": self.username
        }
        with open(self.token_file, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2)
    def _load_token(self) -> Optional[Dict[str, Any]]:
        """從文件加載 Token"""
        if not os.path.exists(self.token_file):
            return None
        try:
            with open(self.token_file, "r", encoding="utf-8") as f:
                return json.load(f)
        except:
            return None
    def _is_token_expired(self) -> bool:
        """檢查 Token 是否過(guò)期"""
        token_data = self._load_token()
        if not token_data:
            return True
        return time.time() >= token_data.get("expires_at", 0)
    def login(self) -> bool:
        """登錄并獲取 Token"""
        try:
            resp = self.session.post(
                f"{self.base_url}{self.login_url}",
                json={"username": self.username, "password": self.password}
            )
            if resp.status_code == 200:
                data = resp.json()
                token = data["data"]["access_token"]
                refresh_token = data["data"]["refresh_token"]
                expires_in = data["data"].get("expires_in", 1800)  # 默認(rèn)30分鐘
                self._save_token(token, refresh_token, expires_in)
                self.session.headers.update({"Authorization": f"Bearer {token}"})
                print(f"? 登錄成功，Token 已保存")
                return True
            else:
                print(f"? 登錄失敗: {resp.text}")
                return False
        except Exception as e:
            print(f"? 登錄異常: {e}")
            return False
    def refresh_token(self) -> bool:
        """刷新 Token"""
        token_data = self._load_token()
        if not token_data:
            return False
        try:
            resp = self.session.post(
                f"{self.base_url}{self.refresh_url}",
                json={"refresh_token": token_data["refresh_token"]}
            )
            if resp.status_code == 200:
                data = resp.json()
                new_token = data["data"]["access_token"]
                new_refresh = data["data"]["refresh_token"]
                expires_in = data["data"].get("expires_in", 1800)
                self._save_token(new_token, new_refresh, expires_in)
                self.session.headers.update({"Authorization": f"Bearer {new_token}"})
                print(f"?? Token 刷新成功")
                return True
            else:
                print(f"? Token 刷新失敗: {resp.text}")
                return False
        except Exception as e:
            print(f"? 刷新異常: {e}")
            return False
    def ensure_auth(self) -> bool:
        """確保當(dāng)前有有效 Token（自動(dòng)登錄或刷新）"""
        if self._is_token_expired():
            print("?? Token 已過(guò)期，嘗試刷新...")
            if not self.refresh_token():
                print("?? 刷新失敗，正在重新登錄...")
                return self.login()
        else:
            # 加載有效 Token
            token_data = self._load_token()
            self.session.headers.update({"Authorization": f"Bearer {token_data['access_token']}"})
            print("?? 使用現(xiàn)有 Token")
        return True
    def request(self, method: str, url: str, **kwargs) -> requests.Response:
        """重寫(xiě) request，自動(dòng)處理認(rèn)證"""
        full_url = self.base_url + ("" if url.startswith("/") else "/") + url
        # 確保認(rèn)證有效
        if not self.ensure_auth():
            raise Exception("認(rèn)證失敗，無(wú)法繼續(xù)請(qǐng)求")
        try:
            resp = self.session.request(method, full_url, **kwargs)
            # 如果返回 401，可能是刷新失敗，嘗試重新登錄
            if resp.status_code == 401:
                print("?? 收到 401，嘗試重新登錄...")
                self.login()
                # 重新發(fā)送請(qǐng)求
                resp = self.session.request(method, full_url, **kwargs)
            return resp
        except Exception as e:
            print(f"? 請(qǐng)求異常: {e}")
            raise
    # 便捷方法
    def get(self, url, **kwargs):
        return self.request("GET", url, **kwargs)
    def post(self, url, **kwargs):
        return self.request("POST", url, **kwargs)
    def put(self, url, **kwargs):
        return self.request("PUT", url, **kwargs)
    def delete(self, url, **kwargs):
        return self.request("DELETE", url, **kwargs)