antissoft / lvqianmeiye_ERP

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
员工工资服务接口测试脚本
"""
import requests
import json
import sys
import os

# 配置
BASE_URL = "http://localhost:2011"
LOGIN_URL = f"{BASE_URL}/api/oauth/Login"
API_BASE = f"{BASE_URL}/api/Extend/LqEmployeeSalaryStatistics"

# 测试结果
test_results = []

def print_result(test_name, success, message="", data=None):
    """打印测试结果"""
    status = "✓ 通过" if success else "✗ 失败"
    print(f"{status} - {test_name}")
    if message:
        print(f"  说明: {message}")
    if data and success:
        if isinstance(data, dict):
            # 只显示关键信息
            if 'list' in data:
                print(f"  返回数据: 列表数量={len(data.get('list', []))}, 总数={data.get('pagination', {}).get('total', 0)}")
            elif 'SuccessCount' in data:
                print(f"  导入结果: 成功={data.get('SuccessCount', 0)}, 失败={data.get('FailCount', 0)}")
            else:
                print(f"  返回数据: {json.dumps(data, ensure_ascii=False, indent=2)[:200]}...")
        else:
            print(f"  返回数据: {str(data)[:100]}...")
    print()
    test_results.append({
        'name': test_name,
        'success': success,
        'message': message
    })

def get_token():
    """获取登录token"""
    print("=" * 80)
    print("步骤 1: 获取登录Token")
    print("=" * 80)
    
    login_data = {
        "account": "admin",
        "password": "e10adc3949ba59abbe56e057f20f883e"  # md5加密的密码
    }
    
    try:
        response = requests.post(LOGIN_URL, data=login_data, 
                               headers={"Content-Type": "application/x-www-form-urlencoded"})
        
        if response.status_code == 200:
            result = response.json()
            if result.get('code') == 200 and result.get('data') and result.get('data').get('token'):
                token = result['data']['token']
                print(f"✓ Token获取成功: {token[:50]}...")
                print()
                return token
            else:
                print(f"✗ Token获取失败: {result}")
                return None
        else:
            print(f"✗ 登录请求失败: HTTP {response.status_code}")
            print(f"  响应: {response.text[:200]}")
            return None
    except Exception as e:
        print(f"✗ 登录请求异常: {e}")
        return None

def test_1_list(token):
    """测试1: 查看所有员工工资列表"""
    print("=" * 80)
    print("测试 1: 查看所有员工工资列表（分页查询）")
    print("=" * 80)
    
    url = f"{API_BASE}/list"
    headers = {
        "Authorization": token,
        "Content-Type": "application/json"
    }
    
    # 测试数据
    test_cases = [
        {
            "name": "基础查询（无参数）",
            "data": {},
            "params": {"currentPage": 1, "pageSize": 10}
        },
        {
            "name": "按月份查询",
            "data": {"statisticsMonth": "202501"},
            "params": {"currentPage": 1, "pageSize": 10}
        },
        {
            "name": "按关键词搜索",
            "data": {"keyword": "测试"},
            "params": {"currentPage": 1, "pageSize": 10}
        }
    ]
    
    for case in test_cases:
        try:
            response = requests.post(
                url,
                json=case["data"],
                params=case["params"],
                headers=headers,
                timeout=10
            )
            
            if response.status_code == 200:
                result = response.json()
                if isinstance(result, dict) and ('list' in result or 'pagination' in result):
                    print_result(case["name"], True, f"HTTP {response.status_code}", result)
                else:
                    print_result(case["name"], True, f"HTTP {response.status_code}, 返回: {result}")
            else:
                print_result(case["name"], False, f"HTTP {response.status_code}, 响应: {response.text[:200]}")
        except Exception as e:
            print_result(case["name"], False, f"请求异常: {str(e)}")

def test_2_get_by_employee(token):
    """测试2: 根据员工ID或手机号获取工资"""
    print("=" * 80)
    print("测试 2: 根据员工ID或手机号获取员工工资")
    print("=" * 80)
    
    url = f"{API_BASE}/get-by-employee"
    headers = {
        "Authorization": token,
        "Content-Type": "application/json"
    }
    
    test_cases = [
        {
            "name": "缺少月份参数（应失败）",
            "data": {"employeeId": "test123"},
            "should_fail": True
        },
        {
            "name": "缺少员工ID和手机号（应失败）",
            "data": {"statisticsMonth": "202501"},
            "should_fail": True
        },
        {
            "name": "正常查询（使用员工ID）",
            "data": {
                "employeeId": "test123",
                "statisticsMonth": "202501"
            },
            "should_fail": False
        }
    ]
    
    for case in test_cases:
        try:
            response = requests.post(
                url,
                json=case["data"],
                headers=headers,
                timeout=10
            )
            
            if case.get("should_fail"):
                if response.status_code != 200:
                    print_result(case["name"], True, f"正确返回错误: HTTP {response.status_code}")
                else:
                    result = response.json()
                    if result.get('code') != 200:
                        print_result(case["name"], True, f"正确返回错误: {result.get('msg', '')}")
                    else:
                        print_result(case["name"], False, "应该返回错误但没有")
            else:
                if response.status_code == 200:
                    result = response.json()
                    print_result(case["name"], True, f"HTTP {response.status_code}", result)
                else:
                    print_result(case["name"], False, f"HTTP {response.status_code}, 响应: {response.text[:200]}")
        except Exception as e:
            print_result(case["name"], False, f"请求异常: {str(e)}")

def test_3_add(token):
    """测试3: 添加员工工资"""
    print("=" * 80)
    print("测试 3: 添加员工工资")
    print("=" * 80)
    
    url = f"{API_BASE}/add"
    headers = {
        "Authorization": token,
        "Content-Type": "application/json"
    }
    
    test_cases = [
        {
            "name": "缺少必填字段（应失败）",
            "data": {
                "employeeName": "测试员工"
            },
            "should_fail": True
        },
        {
            "name": "正常添加",
            "data": {
                "statisticsMonth": "202501",
                "storeId": "test_store_001",
                "storeName": "测试门店",
                "employeeId": f"test_emp_{int(__import__('time').time())}",
                "employeeName": "测试员工",
                "employeePhone": "13800138000",
                "position": "健康师",
                "baseSalary": 3000.00,
                "totalPerformance": 50000.00,
                "totalCommission": 5000.00,
                "calculatedGrossSalary": 8000.00,
                "finalGrossSalary": 8000.00,
                "actualSalary": 7500.00
            },
            "should_fail": False
        }
    ]
    
    added_id = None
    
    for case in test_cases:
        try:
            response = requests.post(
                url,
                json=case["data"],
                headers=headers,
                timeout=10
            )
            
            if case.get("should_fail"):
                if response.status_code != 200:
                    print_result(case["name"], True, f"正确返回错误: HTTP {response.status_code}")
                else:
                    result = response.json()
                    if result.get('code') != 200:
                        print_result(case["name"], True, f"正确返回错误: {result.get('msg', '')}")
                    else:
                        print_result(case["name"], False, "应该返回错误但没有")
            else:
                if response.status_code == 200:
                    result = response.json()
                    if isinstance(result, str):
                        added_id = result
                        print_result(case["name"], True, f"创建成功，ID: {added_id}", result)
                    else:
                        print_result(case["name"], True, f"HTTP {response.status_code}", result)
                else:
                    print_result(case["name"], False, f"HTTP {response.status_code}, 响应: {response.text[:200]}")
        except Exception as e:
            print_result(case["name"], False, f"请求异常: {str(e)}")
    
    return added_id

def test_4_update(token, salary_id):
    """测试4: 修改员工工资"""
    print("=" * 80)
    print("测试 4: 修改员工工资")
    print("=" * 80)
    
    if not salary_id:
        print("⚠ 跳过测试（需要先成功添加一条记录）")
        print()
        return
    
    url = f"{API_BASE}/update"
    headers = {
        "Authorization": token,
        "Content-Type": "application/json"
    }
    
    test_cases = [
        {
            "name": "缺少ID（应失败）",
            "data": {
                "baseSalary": 3500.00
            },
            "should_fail": True
        },
        {
            "name": "正常修改",
            "data": {
                "id": salary_id,
                "baseSalary": 3500.00,
                "totalPerformance": 60000.00
            },
            "should_fail": False
        }
    ]
    
    for case in test_cases:
        try:
            response = requests.put(
                url,
                json=case["data"],
                headers=headers,
                timeout=10
            )
            
            if case.get("should_fail"):
                if response.status_code != 200:
                    print_result(case["name"], True, f"正确返回错误: HTTP {response.status_code}")
                else:
                    result = response.json()
                    if result.get('code') != 200:
                        print_result(case["name"], True, f"正确返回错误: {result.get('msg', '')}")
                    else:
                        print_result(case["name"], False, "应该返回错误但没有")
            else:
                if response.status_code == 200:
                    result = response.json()
                    print_result(case["name"], True, f"修改成功", result)
                else:
                    print_result(case["name"], False, f"HTTP {response.status_code}, 响应: {response.text[:200]}")
        except Exception as e:
            print_result(case["name"], False, f"请求异常: {str(e)}")

def test_5_confirm(token, salary_id):
    """测试5: 员工工资确认"""
    print("=" * 80)
    print("测试 5: 员工工资确认")
    print("=" * 80)
    
    if not salary_id:
        print("⚠ 跳过测试（需要先成功添加一条记录）")
        print()
        return
    
    url = f"{API_BASE}/confirm"
    headers = {
        "Authorization": token,
        "Content-Type": "application/json"
    }
    
    test_cases = [
        {
            "name": "缺少ID（应失败）",
            "data": {
                "employeeId": "test_emp_123"
            },
            "should_fail": True
        },
        {
            "name": "缺少员工ID（应失败）",
            "data": {
                "id": salary_id
            },
            "should_fail": True
        },
        {
            "name": "正常确认（需要先查询获取正确的employeeId）",
            "data": {
                "id": salary_id,
                "employeeId": "test_emp_123"  # 这个需要从添加的记录中获取
            },
            "should_fail": False,
            "skip": True  # 跳过，因为需要正确的employeeId
        }
    ]
    
    for case in test_cases:
        if case.get("skip"):
            print(f"⚠ 跳过 - {case['name']}（需要先查询获取正确的employeeId）")
            print()
            continue
            
        try:
            response = requests.post(
                url,
                json=case["data"],
                headers=headers,
                timeout=10
            )
            
            if case.get("should_fail"):
                if response.status_code != 200:
                    print_result(case["name"], True, f"正确返回错误: HTTP {response.status_code}")
                else:
                    result = response.json()
                    if result.get('code') != 200:
                        print_result(case["name"], True, f"正确返回错误: {result.get('msg', '')}")
                    else:
                        print_result(case["name"], False, "应该返回错误但没有")
            else:
                if response.status_code == 200:
                    result = response.json()
                    print_result(case["name"], True, f"确认成功", result)
                else:
                    print_result(case["name"], False, f"HTTP {response.status_code}, 响应: {response.text[:200]}")
        except Exception as e:
            print_result(case["name"], False, f"请求异常: {str(e)}")

def test_6_import(token):
    """测试6: 导入员工工资"""
    print("=" * 80)
    print("测试 6: 导入员工工资")
    print("=" * 80)
    
    url = f"{API_BASE}/import"
    headers = {
        "Authorization": token
    }
    
    # 检查模板文件是否存在
    template_path = os.path.join(os.path.dirname(__file__), "excel/员工工资导入模板.xlsx")
    if not os.path.exists(template_path):
        print(f"⚠ 模板文件不存在: {template_path}")
        print("  请先运行 python3 scripts/py/create_salary_template.py 生成模板文件")
        print()
        return
    
    test_cases = [
        {
            "name": "不上传文件（应失败）",
            "files": None,
            "should_fail": True
        },
        {
            "name": "上传模板文件（空数据）",
            "files": {"file": ("员工工资导入模板.xlsx", open(template_path, "rb"), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
            "should_fail": False
        }
    ]
    
    for case in test_cases:
        try:
            if case.get("should_fail"):
                # 测试不上传文件
                response = requests.post(
                    url,
                    headers=headers,
                    timeout=30
                )
                
                if response.status_code != 200:
                    print_result(case["name"], True, f"正确返回错误: HTTP {response.status_code}")
                else:
                    result = response.json()
                    if result.get('code') != 200:
                        print_result(case["name"], True, f"正确返回错误: {result.get('msg', '')}")
                    else:
                        print_result(case["name"], False, "应该返回错误但没有")
            else:
                # 上传文件
                if case["files"]:
                    files = case["files"]
                    response = requests.post(
                        url,
                        files=files,
                        headers=headers,
                        timeout=30
                    )
                    
                    if response.status_code == 200:
                        result = response.json()
                        print_result(case["name"], True, f"导入完成", result)
                    else:
                        print_result(case["name"], False, f"HTTP {response.status_code}, 响应: {response.text[:200]}")
                    
                    # 关闭文件
                    for file_obj in files.values():
                        if hasattr(file_obj, 'close'):
                            file_obj.close()
        except Exception as e:
            print_result(case["name"], False, f"请求异常: {str(e)}")

def print_summary():
    """打印测试总结"""
    print("=" * 80)
    print("测试总结")
    print("=" * 80)
    
    total = len(test_results)
    passed = sum(1 for r in test_results if r['success'])
    failed = total - passed
    
    print(f"总测试数: {total}")
    print(f"通过: {passed} ✓")
    print(f"失败: {failed} ✗")
    print()
    
    if failed > 0:
        print("失败的测试:")
        for r in test_results:
            if not r['success']:
                print(f"  ✗ {r['name']}: {r['message']}")
        print()
    
    if failed == 0:
        print("🎉 所有测试通过！")
    else:
        print(f"⚠ 有 {failed} 个测试失败，请检查")

def main():
    """主函数"""
    print("\n" + "=" * 80)
    print("员工工资服务接口测试")
    print("=" * 80)
    print(f"API地址: {API_BASE}")
    print()
    
    # 获取token
    token = get_token()
    if not token:
        print("✗ 无法获取Token，测试终止")
        sys.exit(1)
    
    # 执行测试
    test_1_list(token)
    test_2_get_by_employee(token)
    added_id = test_3_add(token)
    test_4_update(token, added_id)
    test_5_confirm(token, added_id)
    test_6_import(token)
    
    # 打印总结
    print_summary()

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n\n测试被用户中断")
        sys.exit(1)
    except Exception as e:
        print(f"\n\n测试异常: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)