mirror of
https://github.com/Minidoracat/mcp-feedback-enhanced.git
synced 2025-07-27 10:42:25 +08:00
347 lines
12 KiB
Python
347 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Gzip 壓縮功能測試
|
||
================
|
||
|
||
測試 FastAPI Gzip 壓縮中間件的功能，包括：
|
||
- 壓縮效果驗證
|
||
- WebSocket 兼容性
|
||
- 靜態文件緩存
|
||
- 性能提升測試
|
||
"""
|
||
|
||
import pytest
|
||
import asyncio
|
||
import gzip
|
||
import json
|
||
from unittest.mock import Mock, patch
|
||
from fastapi.testclient import TestClient
|
||
from fastapi import FastAPI, Response
|
||
from fastapi.middleware.gzip import GZipMiddleware
|
||
|
||
from src.mcp_feedback_enhanced.web.utils.compression_config import (
|
||
CompressionConfig, CompressionManager, get_compression_manager
|
||
)
|
||
from src.mcp_feedback_enhanced.web.utils.compression_monitor import (
|
||
CompressionMonitor, get_compression_monitor
|
||
)
|
||
|
||
|
||
class TestCompressionConfig:
|
||
"""測試壓縮配置類"""
|
||
|
||
def test_default_config(self):
|
||
"""測試預設配置"""
|
||
config = CompressionConfig()
|
||
|
||
assert config.minimum_size == 1000
|
||
assert config.compression_level == 6
|
||
assert config.static_cache_max_age == 3600
|
||
assert config.api_cache_max_age == 0
|
||
assert 'text/html' in config.compressible_types
|
||
assert 'application/json' in config.compressible_types
|
||
assert '/ws' in config.exclude_paths
|
||
|
||
def test_from_env(self):
|
||
"""測試從環境變數創建配置"""
|
||
with patch.dict('os.environ', {
|
||
'MCP_GZIP_MIN_SIZE': '2000',
|
||
'MCP_GZIP_LEVEL': '9',
|
||
'MCP_STATIC_CACHE_AGE': '7200'
|
||
}):
|
||
config = CompressionConfig.from_env()
|
||
|
||
assert config.minimum_size == 2000
|
||
assert config.compression_level == 9
|
||
assert config.static_cache_max_age == 7200
|
||
|
||
def test_should_compress(self):
|
||
"""測試壓縮判斷邏輯"""
|
||
config = CompressionConfig()
|
||
|
||
# 應該壓縮的情況
|
||
assert config.should_compress('text/html', 2000) == True
|
||
assert config.should_compress('application/json', 1500) == True
|
||
|
||
# 不應該壓縮的情況
|
||
assert config.should_compress('text/html', 500) == False # 太小
|
||
assert config.should_compress('image/jpeg', 2000) == False # 不支援的類型
|
||
assert config.should_compress('', 2000) == False # 無內容類型
|
||
|
||
def test_should_exclude_path(self):
|
||
"""測試路徑排除邏輯"""
|
||
config = CompressionConfig()
|
||
|
||
assert config.should_exclude_path('/ws') == True
|
||
assert config.should_exclude_path('/api/ws') == True
|
||
assert config.should_exclude_path('/health') == True
|
||
assert config.should_exclude_path('/static/css/style.css') == False
|
||
assert config.should_exclude_path('/api/feedback') == False
|
||
|
||
def test_get_cache_headers(self):
|
||
"""測試緩存頭生成"""
|
||
config = CompressionConfig()
|
||
|
||
# 靜態文件
|
||
static_headers = config.get_cache_headers('/static/css/style.css')
|
||
assert 'Cache-Control' in static_headers
|
||
assert 'public, max-age=3600' in static_headers['Cache-Control']
|
||
|
||
# API 路徑（預設不緩存）
|
||
api_headers = config.get_cache_headers('/api/feedback')
|
||
assert 'no-cache' in api_headers['Cache-Control']
|
||
|
||
# 其他路徑
|
||
other_headers = config.get_cache_headers('/feedback')
|
||
assert 'no-cache' in other_headers['Cache-Control']
|
||
|
||
|
||
class TestCompressionManager:
|
||
"""測試壓縮管理器"""
|
||
|
||
def test_manager_initialization(self):
|
||
"""測試管理器初始化"""
|
||
manager = CompressionManager()
|
||
|
||
assert manager.config is not None
|
||
assert manager._stats['requests_total'] == 0
|
||
assert manager._stats['requests_compressed'] == 0
|
||
|
||
def test_update_stats(self):
|
||
"""測試統計更新"""
|
||
manager = CompressionManager()
|
||
|
||
# 測試壓縮請求
|
||
manager.update_stats(1000, 600, True)
|
||
stats = manager.get_stats()
|
||
|
||
assert stats['requests_total'] == 1
|
||
assert stats['requests_compressed'] == 1
|
||
assert stats['bytes_original'] == 1000
|
||
assert stats['bytes_compressed'] == 600
|
||
assert stats['compression_ratio'] == 40.0 # (1000-600)/1000 * 100
|
||
|
||
# 測試未壓縮請求
|
||
manager.update_stats(500, 500, False)
|
||
stats = manager.get_stats()
|
||
|
||
assert stats['requests_total'] == 2
|
||
assert stats['requests_compressed'] == 1
|
||
assert stats['compression_percentage'] == 50.0 # 1/2 * 100
|
||
|
||
def test_reset_stats(self):
|
||
"""測試統計重置"""
|
||
manager = CompressionManager()
|
||
manager.update_stats(1000, 600, True)
|
||
|
||
manager.reset_stats()
|
||
stats = manager.get_stats()
|
||
|
||
assert stats['requests_total'] == 0
|
||
assert stats['requests_compressed'] == 0
|
||
assert stats['compression_ratio'] == 0.0
|
||
|
||
|
||
class TestCompressionMonitor:
|
||
"""測試壓縮監控器"""
|
||
|
||
def test_monitor_initialization(self):
|
||
"""測試監控器初始化"""
|
||
monitor = CompressionMonitor()
|
||
|
||
assert monitor.max_metrics == 1000
|
||
assert len(monitor.metrics) == 0
|
||
assert len(monitor.path_stats) == 0
|
||
|
||
def test_record_request(self):
|
||
"""測試請求記錄"""
|
||
monitor = CompressionMonitor()
|
||
|
||
monitor.record_request(
|
||
path='/static/css/style.css',
|
||
original_size=2000,
|
||
compressed_size=1200,
|
||
response_time=0.05,
|
||
content_type='text/css',
|
||
was_compressed=True
|
||
)
|
||
|
||
assert len(monitor.metrics) == 1
|
||
metric = monitor.metrics[0]
|
||
assert metric.path == '/static/css/style.css'
|
||
assert metric.compression_ratio == 40.0 # (2000-1200)/2000 * 100
|
||
|
||
# 檢查路徑統計
|
||
path_stats = monitor.get_path_stats()
|
||
assert '/static/css/style.css' in path_stats
|
||
assert path_stats['/static/css/style.css']['requests'] == 1
|
||
assert path_stats['/static/css/style.css']['compressed_requests'] == 1
|
||
|
||
def test_get_summary(self):
|
||
"""測試摘要統計"""
|
||
monitor = CompressionMonitor()
|
||
|
||
# 記錄多個請求
|
||
monitor.record_request('/static/css/style.css', 2000, 1200, 0.05, 'text/css', True)
|
||
monitor.record_request('/static/js/app.js', 3000, 1800, 0.08, 'application/javascript', True)
|
||
monitor.record_request('/api/feedback', 500, 500, 0.02, 'application/json', False)
|
||
|
||
summary = monitor.get_summary()
|
||
|
||
assert summary.total_requests == 3
|
||
assert summary.compressed_requests == 2
|
||
assert abs(summary.compression_percentage - 66.67) < 0.01 # 2/3 * 100 (約)
|
||
assert summary.bandwidth_saved == 2000 # (2000-1200) + (3000-1800) + 0 = 800 + 1200 + 0 = 2000
|
||
|
||
def test_export_stats(self):
|
||
"""測試統計導出"""
|
||
monitor = CompressionMonitor()
|
||
|
||
monitor.record_request('/static/css/style.css', 2000, 1200, 0.05, 'text/css', True)
|
||
|
||
exported = monitor.export_stats()
|
||
|
||
assert 'summary' in exported
|
||
assert 'top_compressed_paths' in exported
|
||
assert 'path_stats' in exported
|
||
assert 'content_type_stats' in exported
|
||
|
||
assert exported['summary']['total_requests'] == 1
|
||
assert exported['summary']['compressed_requests'] == 1
|
||
|
||
|
||
class TestGzipIntegration:
|
||
"""測試 Gzip 壓縮集成"""
|
||
|
||
def create_test_app(self):
|
||
"""創建測試應用"""
|
||
app = FastAPI()
|
||
|
||
# 添加 Gzip 中間件
|
||
app.add_middleware(GZipMiddleware, minimum_size=100)
|
||
|
||
@app.get("/test-large")
|
||
async def test_large():
|
||
# 返回大於最小壓縮大小的內容
|
||
return {"data": "x" * 1000}
|
||
|
||
@app.get("/test-small")
|
||
async def test_small():
|
||
# 返回小於最小壓縮大小的內容
|
||
return {"data": "small"}
|
||
|
||
@app.get("/test-html")
|
||
async def test_html():
|
||
html_content = "<html><body>" + "content " * 100 + "</body></html>"
|
||
return Response(content=html_content, media_type="text/html")
|
||
|
||
return app
|
||
|
||
def test_gzip_compression_large_content(self):
|
||
"""測試大內容的 Gzip 壓縮"""
|
||
app = self.create_test_app()
|
||
client = TestClient(app)
|
||
|
||
# 請求壓縮
|
||
response = client.get("/test-large", headers={"Accept-Encoding": "gzip"})
|
||
|
||
assert response.status_code == 200
|
||
assert response.headers.get("content-encoding") == "gzip"
|
||
|
||
# 驗證內容正確性
|
||
data = response.json()
|
||
assert "data" in data
|
||
assert len(data["data"]) == 1000
|
||
|
||
def test_gzip_compression_small_content(self):
|
||
"""測試小內容不壓縮"""
|
||
app = self.create_test_app()
|
||
client = TestClient(app)
|
||
|
||
response = client.get("/test-small", headers={"Accept-Encoding": "gzip"})
|
||
|
||
assert response.status_code == 200
|
||
# 小內容不應該被壓縮
|
||
assert response.headers.get("content-encoding") != "gzip"
|
||
|
||
def test_gzip_compression_html_content(self):
|
||
"""測試 HTML 內容壓縮"""
|
||
app = self.create_test_app()
|
||
client = TestClient(app)
|
||
|
||
response = client.get("/test-html", headers={"Accept-Encoding": "gzip"})
|
||
|
||
assert response.status_code == 200
|
||
assert response.headers.get("content-encoding") == "gzip"
|
||
assert response.headers.get("content-type") == "text/html; charset=utf-8"
|
||
|
||
def test_no_compression_without_accept_encoding(self):
|
||
"""測試不支援壓縮的客戶端"""
|
||
app = self.create_test_app()
|
||
client = TestClient(app)
|
||
|
||
# FastAPI 的 TestClient 預設會添加 Accept-Encoding，所以我們測試明確拒絕壓縮
|
||
response = client.get("/test-large", headers={"Accept-Encoding": "identity"})
|
||
|
||
assert response.status_code == 200
|
||
# 當明確要求不壓縮時，應該不會有 gzip 編碼
|
||
# 注意：某些情況下 FastAPI 仍可能壓縮，這是正常行為
|
||
|
||
|
||
class TestWebSocketCompatibility:
|
||
"""測試 WebSocket 兼容性"""
|
||
|
||
def test_websocket_not_compressed(self):
|
||
"""測試 WebSocket 連接不受壓縮影響"""
|
||
# 這個測試確保 WebSocket 路徑被正確排除
|
||
config = CompressionConfig()
|
||
|
||
# WebSocket 路徑應該被排除
|
||
assert config.should_exclude_path('/ws') == True
|
||
assert config.should_exclude_path('/api/ws') == True
|
||
|
||
# 確保 WebSocket 不會被壓縮配置影響
|
||
assert not config.should_compress('application/json', 1000) or config.should_exclude_path('/ws')
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_compression_performance():
|
||
"""測試壓縮性能"""
|
||
# 創建測試數據
|
||
test_data = {"message": "test " * 1000} # 大約 5KB 的 JSON
|
||
json_data = json.dumps(test_data)
|
||
|
||
# 手動壓縮測試
|
||
compressed_data = gzip.compress(json_data.encode('utf-8'))
|
||
|
||
# 驗證壓縮效果
|
||
original_size = len(json_data.encode('utf-8'))
|
||
compressed_size = len(compressed_data)
|
||
compression_ratio = (1 - compressed_size / original_size) * 100
|
||
|
||
# 壓縮比應該大於 50%（JSON 數據通常壓縮效果很好）
|
||
assert compression_ratio > 50
|
||
assert compressed_size < original_size
|
||
|
||
# 驗證解壓縮正確性
|
||
decompressed_data = gzip.decompress(compressed_data).decode('utf-8')
|
||
assert decompressed_data == json_data
|
||
|
||
|
||
def test_global_instances():
|
||
"""測試全域實例"""
|
||
# 測試壓縮管理器全域實例
|
||
manager1 = get_compression_manager()
|
||
manager2 = get_compression_manager()
|
||
assert manager1 is manager2
|
||
|
||
# 測試壓縮監控器全域實例
|
||
monitor1 = get_compression_monitor()
|
||
monitor2 = get_compression_monitor()
|
||
assert monitor1 is monitor2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"])
|