matrix-openclaw-bot/tests/test_token_manager.py

# test_token_manager.py - Token 管理器測試
"""
Tests for TokenManager class.
Uses sandbox testing to avoid modifying real files.
"""

import json
import pytest
import tempfile
from pathlib import Path
from datetime import datetime

# 需要確保可以導入
import sys
sys.path.insert(0, str(Path(__file__).parent.parent))

from src.services.token_manager import TokenManager


class TestTokenManager:
    """TokenManager 測試套件"""
    
    @pytest.fixture
    def temp_json_file(self, tmp_path):
        """創建臨時 JSON 文件用於沙箱測試"""
        json_file = tmp_path / "openclaw.json"
        test_data = {
            "channels": {
                "matrix": {
                    "accessToken": "old_token_12345",
                    "homeserver": "https://matrix.org"
                },
                "telegram": {
                    "botToken": "telegram_token"
                }
            },
            "settings": {
                "debug": True
            }
        }
        json_file.write_text(json.dumps(test_data, indent=2))
        return json_file
    
    @pytest.fixture
    def token_manager(self, temp_json_file):
        """創建 TokenManager 實例"""
        return TokenManager(
            json_path=temp_json_file,
            token_key="channels.matrix.accessToken"
        )
    
    def test_create_backup(self, token_manager, temp_json_file):
        """測試備份創建"""
        backup_path = token_manager.create_backup()
        
        assert backup_path is not None
        assert backup_path.exists()
        assert temp_json_file.name in backup_path.name
        assert ".bak" in backup_path.name
        
        # 驗證備份內容與原文件相同
        original = json.loads(temp_json_file.read_text())
        backup = json.loads(backup_path.read_text())
        assert original == backup
    
    def test_verify_backup_exists(self, token_manager):
        """測試備份驗證"""
        # 未創建備份時應該返回 False
        assert token_manager.verify_backup_exists() is False
        
        # 創建備份後應該返回 True
        token_manager.create_backup()
        assert token_manager.verify_backup_exists() is True
    
    def test_get_nested_value(self, token_manager):
        """測試嵌套值讀取"""
        data = {
            "level1": {
                "level2": {
                    "level3": "value"
                }
            }
        }
        
        assert token_manager._get_nested_value(data, "level1.level2.level3") == "value"
        assert token_manager._get_nested_value(data, "level1.level2") is None  # 不是字符串
        assert token_manager._get_nested_value(data, "nonexistent") is None
    
    def test_set_nested_value(self, token_manager):
        """測試嵌套值設置"""
        data = {}
        
        result = token_manager._set_nested_value(data, "a.b.c", "new_value")
        
        assert result["a"]["b"]["c"] == "new_value"
    
    @pytest.mark.asyncio
    async def test_read_current_token(self, token_manager):
        """測試讀取當前 token"""
        token = await token_manager.read_current_token()
        
        assert token == "old_token_12345"
    
    @pytest.mark.asyncio
    async def test_update_access_token(self, token_manager, temp_json_file):
        """測試更新 access token（核心功能）"""
        new_token = "new_token_67890"
        
        # 執行更新
        success = await token_manager.update_access_token(new_token)
        
        # 驗證成功
        assert success is True
        
        # 驗證備份存在
        assert token_manager.verify_backup_exists() is True
        
        # 驗證新 token 已寫入
        updated_data = json.loads(temp_json_file.read_text())
        assert updated_data["channels"]["matrix"]["accessToken"] == new_token
        
        # 驗證其他數據未被破壞
        assert updated_data["channels"]["telegram"]["botToken"] == "telegram_token"
        assert updated_data["settings"]["debug"] is True
    
    def test_restore_from_backup(self, token_manager, temp_json_file):
        """測試從備份恢復"""
        # 創建備份
        backup_path = token_manager.create_backup()
        
        # 修改原文件
        temp_json_file.write_text('{"corrupted": true}')
        
        # 恢復
        success = token_manager.restore_from_backup()
        
        assert success is True
        
        # 驗證恢復結果
        restored_data = json.loads(temp_json_file.read_text())
        assert restored_data["channels"]["matrix"]["accessToken"] == "old_token_12345"


class TestTokenManagerEdgeCases:
    """TokenManager 邊緣情況測試"""
    
    def test_backup_nonexistent_file(self, tmp_path):
        """測試備份不存在的文件"""
        manager = TokenManager(
            json_path=tmp_path / "nonexistent.json",
            token_key="test"
        )
        
        backup_path = manager.create_backup()
        assert backup_path is None
    
    @pytest.mark.asyncio
    async def test_update_nonexistent_file(self, tmp_path):
        """測試更新不存在的文件"""
        manager = TokenManager(
            json_path=tmp_path / "nonexistent.json",
            token_key="test"
        )
        
        success = await manager.update_access_token("new_token")
        assert success is False


class TestBackupRotation:
    """備份輪轉功能測試（沙箱測試）"""
    
    @pytest.fixture
    def temp_json_file(self, tmp_path):
        """創建臨時 JSON 文件"""
        json_file = tmp_path / "openclaw.json"
        test_data = {"channels": {"matrix": {"accessToken": "test_token"}}}
        json_file.write_text(json.dumps(test_data))
        return json_file
    
    @pytest.fixture
    def token_manager(self, temp_json_file):
        """創建 TokenManager 實例"""
        return TokenManager(
            json_path=temp_json_file,
            token_key="channels.matrix.accessToken"
        )
    
    def test_max_backups_constant(self):
        """測試最大備份數量常量"""
        assert TokenManager.MAX_BACKUPS == 5
    
    def test_get_backup_files_empty(self, token_manager):
        """測試沒有備份時返回空列表"""
        backups = token_manager._get_backup_files()
        assert backups == []
    
    def test_get_backup_files_with_backups(self, token_manager, tmp_path):
        """測試獲取備份文件列表"""
        # 創建幾個備份
        for i in range(3):
            token_manager.create_backup()
            import time
            time.sleep(0.01)  # 確保時間戳不同
        
        backups = token_manager._get_backup_files()
        assert len(backups) == 3
        
        # 確認按文件名排序（時間戳在名稱中，字典序=時間序，最新在前）
        for i in range(len(backups) - 1):
            assert backups[i].name > backups[i + 1].name
    
    def test_rotation_under_limit(self, token_manager):
        """測試備份數量未超過限制時不刪除"""
        # 創建 3 個備份（低於限制）
        for i in range(3):
            token_manager.create_backup()
            import time
            time.sleep(0.01)
        
        assert token_manager.get_backup_count() == 3
    
    def test_rotation_at_limit(self, token_manager):
        """測試備份數量剛好達到限制"""
        # 創建 5 個備份（剛好等於限制）
        for i in range(5):
            token_manager.create_backup()
            import time
            time.sleep(0.01)
        
        assert token_manager.get_backup_count() == 5
    
    def test_rotation_over_limit(self, token_manager):
        """測試備份數量超過限制時自動刪除最舊的"""
        # 創建 7 個備份（超過限制）
        for i in range(7):
            token_manager.create_backup()
            import time
            time.sleep(0.01)
        
        # 應該只保留 5 個
        assert token_manager.get_backup_count() == 5
    
    def test_rotation_preserves_newest(self, token_manager, temp_json_file):
        """測試輪轉保留最新的備份"""
        backup_paths = []
        
        # 創建 7 個備份
        for i in range(7):
            path = token_manager.create_backup()
            if path:
                backup_paths.append(path)
            import time
            time.sleep(0.01)
        
        # 檢查最新的 5 個備份仍然存在
        remaining_backups = token_manager._get_backup_files()
        assert len(remaining_backups) == 5
        
        # 最新的 5 個應該是最後創建的 5 個
        newest_5 = backup_paths[-5:]
        for path in newest_5:
            assert path.exists(), f"{path} should exist"
        
        # 最舊的 2 個應該被刪除
        oldest_2 = backup_paths[:2]
        for path in oldest_2:
            assert not path.exists(), f"{path} should be deleted"
    
    def test_get_backup_count(self, token_manager):
        """測試獲取備份數量"""
        assert token_manager.get_backup_count() == 0
        
        token_manager.create_backup()
        assert token_manager.get_backup_count() == 1
        
        token_manager.create_backup()
        assert token_manager.get_backup_count() == 2