Python脚本如何适配测试环境调试运行

wen python案例 2026-06-26 241

本文目录导读：

环境配置管理
连接测试环境
调试工具和日志
测试环境切换脚本
实用调试技巧
最佳实践建议

我来详细介绍Python脚本适配测试环境调试运行的几种方法。

环境配置管理

使用环境变量

import os
import json
class Config:
    def __init__(self):
        # 从环境变量获取当前环境
        self.env = os.getenv('APP_ENV', 'development')
        # 基础配置
        self.base_config = {
            'debug': True,
            'log_level': 'DEBUG'
        }
        # 环境特定配置
        self.env_configs = {
            'development': {
                'database': {
                    'host': 'localhost',
                    'port': 3306,
                    'user': 'dev_user',
                    'password': 'dev_pass',
                    'database': 'test_db'
                },
                'api_base_url': 'http://localhost:8080/api',
                'timeout': 30
            },
            'testing': {
                'database': {
                    'host': 'test-db.example.com',
                    'port': 3306,
                    'user': 'test_user',
                    'password': 'test_pass',
                    'database': 'test_db'
                },
                'api_base_url': 'http://test-api.example.com/api',
                'timeout': 60
            },
            'production': {
                'database': {
                    'host': 'prod-db.example.com',
                    'port': 3306,
                    'user': 'prod_user',
                    'password': 'prod_pass',
                    'database': 'prod_db'
                },
                'api_base_url': 'https://api.example.com/api',
                'timeout': 120
            }
        }
    def get(self, key, default=None):
        """获取配置值"""
        env_config = self.env_configs.get(self.env, {})
        config = {**self.base_config, **env_config}
        return config.get(key, default)
# 使用示例
config = Config()
db_config = config.get('database')
print(f"当前环境: {config.env}")
print(f"数据库配置: {db_config}")

使用配置文件

# config.py
import yaml
import json
import os
class ConfigManager:
    def __init__(self, config_dir='config'):
        self.config_dir = config_dir
        self.env = os.getenv('APP_ENV', 'development')
        self.config = self.load_config()
    def load_config(self):
        """加载配置文件"""
        config_file = os.path.join(self.config_dir, f'{self.env}.yaml')
        # 尝试加载YAML配置
        if os.path.exists(config_file):
            with open(config_file, 'r') as f:
                return yaml.safe_load(f)
        # 尝试加载JSON配置
        config_file = config_file.replace('.yaml', '.json')
        if os.path.exists(config_file):
            with open(config_file, 'r') as f:
                return json.load(f)
        # 默认配置
        return {
            'debug': True,
            'log_level': 'DEBUG',
            'database': {
                'host': 'localhost',
                'port': 3306
            }
        }
    def get_db_config(self):
        return self.config.get('database', {})
    def get_api_config(self):
        return self.config.get('api', {})
# 配置文件示例 (config/development.yaml)
"""
debug: true
log_level: DEBUG
database:
  host: localhost
  port: 3306
  user: dev_user
  password: dev_pass
  database: test_db
api:
  base_url: http://localhost:8080/api
  timeout: 30
  retry_count: 3
"""

连接测试环境

数据库连接管理

import pymysql
from dbutils.pooled_db import PooledDB
class DatabaseManager:
    def __init__(self, config):
        self.config = config
        self.pool = None
        self.init_pool()
    def init_pool(self):
        """初始化数据库连接池"""
        db_config = self.config.get_db_config()
        # 测试环境使用测试数据库
        if self.config.env == 'testing':
            db_config['database'] = 'test_db'
            db_config['host'] = 'test-db.example.com'
        self.pool = PooledDB(
            creator=pymysql,
            maxconnections=10,
            mincached=2,
            **db_config
        )
    def get_connection(self):
        """获取数据库连接"""
        return self.pool.connection()
    def execute_query(self, sql, params=None):
        """执行查询"""
        conn = self.get_connection()
        try:
            with conn.cursor() as cursor:
                cursor.execute(sql, params)
                return cursor.fetchall()
        finally:
            conn.close()
# 使用示例
db_manager = DatabaseManager(config)

API接口测试

import requests
import logging
from functools import wraps
class APITester:
    def __init__(self, config):
        self.config = config
        self.base_url = config.get('api_base_url', 'http://localhost:8000')
        self.session = requests.Session()
        self.session.headers.update({
            'Content-Type': 'application/json',
            'X-Environment': config.env
        })
        # 测试环境特殊配置
        if config.env == 'testing':
            self.session.verify = False  # 忽略SSL证书验证
            self.session.timeout = 60   # 超时时间加长
    def get(self, endpoint, params=None):
        """发送GET请求"""
        url = f"{self.base_url}{endpoint}"
        response = self.session.get(url, params=params)
        return self.handle_response(response)
    def post(self, endpoint, data=None, json_data=None):
        """发送POST请求"""
        url = f"{self.base_url}{endpoint}"
        response = self.session.post(url, data=data, json=json_data)
        return self.handle_response(response)
    def handle_response(self, response):
        """处理响应"""
        logging.debug(f"Response status: {response.status_code}")
        logging.debug(f"Response body: {response.text}")
        if response.status_code >= 400:
            raise requests.RequestException(f"API Error: {response.status_code}")
        return response.json() if response.headers.get('content-type') == 'application/json' else response.text
# 使用示例
api_tester = APITester(config)
result = api_tester.get('/users', params={'page': 1, 'limit': 10})

调试工具和日志

配置日志系统

import logging
import logging.handlers
import sys
class DebugLogger:
    def __init__(self, config):
        self.config = config
        self.logger = logging.getLogger('test_debug')
        self.setup_logger()
    def setup_logger(self):
        """配置日志系统"""
        log_level = self.config.get('log_level', 'DEBUG')
        self.logger.setLevel(getattr(logging, log_level))
        # 控制台输出
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setLevel(logging.DEBUG)
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        console_handler.setFormatter(formatter)
        self.logger.addHandler(console_handler)
        # 文件输出（测试环境切换到文件）
        if self.config.env == 'testing':
            file_handler = logging.handlers.RotatingFileHandler(
                'test_debug.log',
                maxBytes=1024*1024,  # 1MB
                backupCount=5
            )
            file_handler.setLevel(logging.DEBUG)
            file_handler.setFormatter(formatter)
            self.logger.addHandler(file_handler)
    def debug(self, message, *args, **kwargs):
        self.logger.debug(message, *args, **kwargs)
    def info(self, message, *args, **kwargs):
        self.logger.info(message, *args, **kwargs)
    def error(self, message, *args, **kwargs):
        self.logger.error(message, *args, **kwargs)
# 使用装饰器记录函数调用
def debug_call(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger = logging.getLogger('debug')
        logger.debug(f"Calling {func.__name__} with args={args}, kwargs={kwargs}")
        try:
            result = func(*args, **kwargs)
            logger.debug(f"{func.__name__} returned: {result}")
            return result
        except Exception as e:
            logger.error(f"{func.__name__} raised: {e}")
            raise
    return wrapper
# 使用示例
@debug_call
def test_function(param1, param2):
    return param1 + param2

测试环境切换脚本

#!/usr/bin/env python3
# switch_env.py
import os
import sys
import json
import argparse
class EnvironmentSwitcher:
    def __init__(self):
        self.env_config_file = '.env-config.json'
        self.current_env = self.get_current_env()
    def get_current_env(self):
        """获取当前环境"""
        if os.path.exists(self.env_config_file):
            with open(self.env_config_file, 'r') as f:
                config = json.load(f)
                return config.get('environment', 'development')
        return 'development'
    def save_env(self, env_name):
        """保存环境配置"""
        config = {'environment': env_name}
        with open(self.env_config_file, 'w') as f:
            json.dump(config, f)
        # 设置环境变量
        os.environ['APP_ENV'] = env_name
        print(f"环境已切换到: {env_name}")
    def list_environments(self):
        """列出可用环境"""
        environments = ['development', 'testing', 'staging', 'production']
        print("可用环境:")
        for env in environments:
            indicator = "*" if env == self.current_env else " "
            print(f"  [{indicator}] {env}")
    def run_script(self, script_name):
        """运行指定脚本"""
        current_env = self.get_current_env()
        print(f"当前环境: {current_env}")
        # 根据环境加载不同配置
        if current_env == 'testing':
            self.load_test_config()
        # 执行脚本
        exec(open(script_name).read())
    def load_test_config(self):
        """加载测试配置"""
        # 模拟测试数据库
        self.test_db = {
            'host': 'localhost:3307',  # 测试数据库端口
            'user': 'test_user',
            'password': 'test_pass',
            'database': 'test_db'
        }
        # 设置测试标志
        os.environ['IS_TEST_ENV'] = 'true'
        print("已加载测试环境配置")
def main():
    parser = argparse.ArgumentParser(description='环境切换工具')
    parser.add_argument('action', choices=['switch', 'list', 'run'],
                       help='操作类型')
    parser.add_argument('--env', '-e', help='目标环境')
    parser.add_argument('--script', '-s', help='要运行的脚本')
    args = parser.parse_args()
    switcher = EnvironmentSwitcher()
    if args.action == 'switch':
        if not args.env:
            print("请指定环境名称")
            sys.exit(1)
        switcher.save_env(args.env)
    elif args.action == 'list':
        switcher.list_environments()
    elif args.action == 'run':
        if not args.script:
            print("请指定要运行的脚本")
            sys.exit(1)
        switcher.run_script(args.script)
if __name__ == '__main__':
    main()

实用调试技巧

断点调试

import ipdb
import pdb
def debug_function():
    # 设置断点
    import pdb; pdb.set_trace()  # Python内置调试器
    # 或使用 ipdb (需要安装: pip install ipdb)
    # import ipdb; ipdb.set_trace()
    x = 10
    y = 20
    result = x + y
    return result
# 条件断点
def conditional_debug():
    for i in range(100):
        if i == 50:  # 只在i=50时中断
            import pdb; pdb.set_trace()
        print(f"Processing {i}")

使用unittest进行测试

import unittest
from unittest.mock import patch, MagicMock
class TestDatabaseManager(unittest.TestCase):
    def setUp(self):
        """测试前准备"""
        self.config = MagicMock()
        self.config.env = 'testing'
        self.config.get_db_config.return_value = {
            'host': 'localhost',
            'port': 3306
        }
    def test_connection(self):
        """测试数据库连接"""
        with patch('pymysql.connect') as mock_connect:
            mock_connect.return_value = MagicMock()
            db_manager = DatabaseManager(self.config)
            self.assertIsNotNone(db_manager.pool)
    def test_query_execution(self):
        """测试查询执行"""
        with patch('pymysql.connect') as mock_connect:
            mock_cursor = MagicMock()
            mock_cursor.fetchall.return_value = [{'id': 1, 'name': 'test'}]
            mock_connection = MagicMock()
            mock_connection.cursor.return_value = mock_cursor
            mock_connect.return_value = mock_connection
            db_manager = DatabaseManager(self.config)
            result = db_manager.execute_query("SELECT * FROM users")
            self.assertEqual(len(result), 1)
            self.assertEqual(result[0]['name'], 'test')
if __name__ == '__main__':
    unittest.main()