#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ API測試腳本 """ import requests import json import sys import time from multiprocessing import Process def start_flask_app(): """在子進程中啟動Flask應用""" try: # 簡化的Flask應用啟動 from flask import Flask, jsonify, request, session import pymysql app = Flask(__name__) app.config['SECRET_KEY'] = 'test-secret-key' @app.route('/health') def health(): """健康檢查API""" return jsonify({ 'status': 'ok', 'timestamp': time.time() }) @app.route('/api/v1/auth/login', methods=['POST']) def login(): """簡化的登入API""" try: data = request.get_json() username = data.get('username') password = data.get('password') if not username or not password: return jsonify({ 'success': False, 'error': 'MISSING_CREDENTIALS', 'message': '缺少帳號或密碼' }), 400 # 測試LDAP認證 import ldap3 from ldap3 import Server, Connection, ALL server = Server('panjit.com.tw', port=389, get_info=ALL) bind_dn = "CN=LdapBind,CN=Users,DC=PANJIT,DC=COM,DC=TW" bind_password = "panjit2481" service_conn = Connection(server, user=bind_dn, password=bind_password, auto_bind=True) # 搜尋使用者 search_base = "OU=PANJIT,DC=panjit,DC=com,DC=tw" search_filter = f"(userPrincipalName={username})" result = service_conn.search(search_base, search_filter, attributes=['displayName', 'mail', 'department', 'distinguishedName']) if not result or not service_conn.entries: service_conn.unbind() return jsonify({ 'success': False, 'error': 'USER_NOT_FOUND', 'message': '使用者不存在' }), 404 user_entry = service_conn.entries[0] user_dn = str(user_entry.distinguishedName) # 驗證使用者密碼 user_conn = Connection(server, user=user_dn, password=password) if not user_conn.bind(): service_conn.unbind() return jsonify({ 'success': False, 'error': 'INVALID_PASSWORD', 'message': '密碼錯誤' }), 401 user_conn.unbind() service_conn.unbind() # 模擬成功登入 user_info = { 'id': 1, 'username': username.split('@')[0], 'display_name': str(user_entry.displayName) if user_entry.displayName else username, 'email': str(user_entry.mail) if user_entry.mail else username, 'department': str(user_entry.department) if user_entry.department else 'Unknown', 'is_admin': username.lower() == 'ymirliu@panjit.com.tw' } # 設定session session['user_id'] = user_info['id'] session['username'] = user_info['username'] session['is_admin'] = user_info['is_admin'] return jsonify({ 'success': True, 'data': { 'user': user_info }, 'message': '登入成功' }) except Exception as e: print(f"Login error: {e}") return jsonify({ 'success': False, 'error': 'INTERNAL_ERROR', 'message': f'系統錯誤: {str(e)}' }), 500 @app.route('/api/v1/auth/me') def get_current_user(): """取得當前使用者""" user_id = session.get('user_id') if not user_id: return jsonify({ 'success': False, 'error': 'NOT_AUTHENTICATED', 'message': '未登入' }), 401 return jsonify({ 'success': True, 'data': { 'user': { 'id': user_id, 'username': session.get('username'), 'is_admin': session.get('is_admin', False) } } }) print("Starting test Flask server on port 5000...") app.run(host='127.0.0.1', port=5000, debug=False) except Exception as e: print(f"Flask app failed to start: {e}") def test_apis(): """測試API端點""" base_url = 'http://127.0.0.1:5000' # 等待Flask應用啟動 print("Waiting for Flask server to start...") time.sleep(3) test_results = [] # 1. 測試健康檢查 try: response = requests.get(f'{base_url}/health', timeout=5) if response.status_code == 200: test_results.append(('Health Check', 'PASS')) print("✓ Health check API works") else: test_results.append(('Health Check', 'FAIL')) print(f"✗ Health check failed: {response.status_code}") except Exception as e: test_results.append(('Health Check', 'FAIL')) print(f"✗ Health check failed: {e}") # 2. 測試登入API（無效憑證） try: login_data = { 'username': 'invalid@panjit.com.tw', 'password': 'wrongpassword' } response = requests.post(f'{base_url}/api/v1/auth/login', json=login_data, timeout=10) if response.status_code == 404: test_results.append(('Invalid Login', 'PASS')) print("✓ Invalid login properly rejected") else: test_results.append(('Invalid Login', 'FAIL')) print(f"✗ Invalid login test failed: {response.status_code}") except Exception as e: test_results.append(('Invalid Login', 'FAIL')) print(f"✗ Invalid login test failed: {e}") # 3. 測試登入API（有效憑證） try: login_data = { 'username': 'ymirliu@panjit.com.tw', 'password': 'ˇ3EDC4rfv5tgb' # 使用提供的測試密碼 } response = requests.post(f'{base_url}/api/v1/auth/login', json=login_data, timeout=15) if response.status_code == 200: result = response.json() if result.get('success'): test_results.append(('Valid Login', 'PASS')) print("✓ Valid login successful") # 保存session cookies cookies = response.cookies # 4. 測試取得當前使用者 try: me_response = requests.get(f'{base_url}/api/v1/auth/me', cookies=cookies, timeout=5) if me_response.status_code == 200: me_result = me_response.json() if me_result.get('success'): test_results.append(('Get Current User', 'PASS')) print("✓ Get current user API works") else: test_results.append(('Get Current User', 'FAIL')) else: test_results.append(('Get Current User', 'FAIL')) except Exception as e: test_results.append(('Get Current User', 'FAIL')) print(f"✗ Get current user failed: {e}") else: test_results.append(('Valid Login', 'FAIL')) print(f"✗ Login failed: {result.get('message', 'Unknown error')}") else: test_results.append(('Valid Login', 'FAIL')) print(f"✗ Valid login failed: {response.status_code}") if response.headers.get('content-type', '').startswith('application/json'): print(f"Response: {response.json()}") except Exception as e: test_results.append(('Valid Login', 'FAIL')) print(f"✗ Valid login test failed: {e}") # 輸出測試結果 print("\n=== API Test Results ===") for test_name, result in test_results: print(f"{test_name}: {result}") passed = sum(1 for _, result in test_results if result == 'PASS') total = len(test_results) print(f"\nPassed: {passed}/{total}") return test_results if __name__ == '__main__': if len(sys.argv) > 1 and sys.argv[1] == 'server': # 只啟動服務器 start_flask_app() else: # 在子進程中啟動Flask應用 flask_process = Process(target=start_flask_app) flask_process.start() try: # 運行測試 test_results = test_apis() finally: # 關閉Flask進程 flask_process.terminate() flask_process.join(timeout=3) if flask_process.is_alive(): flask_process.kill()