"""Phase 3a 端到端冒烟:异步 submit + WS 阶段 + JWT + tears(容器)。""" import sys import os import asyncio import tempfile import shutil # Set up paths for both local and container environments if os.path.exists("/app"): sys.path.insert(0, "/app") sys.path.insert(0, "/app/vnpy_v4.4.0") print("=== Using container paths ===") else: # Local development environment repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, repo_root) vnpy_path = os.path.join(repo_root, "vnpy_v4.4.0") if vnpy_path not in sys.path: sys.path.insert(0, vnpy_path) print(f"=== Using local paths: {repo_root} ===") def test_sys_path(): """Test sys.path configuration""" print("=== SYS.PATH CONFIG ===") for i, p in enumerate(sys.path[:3]): print(f" path[{i}]: {p}") print("=== PASS: sys.path configured ===") print() def test_jwt_login(): """Test JWT login endpoint""" print("=== JWT LOGIN TEST ===") try: from fastapi.testclient import TestClient from sanguo_api.app import create_app from sanguo_api.auth import hash_password # Create temporary directory for test tmpdir = tempfile.mkdtemp() db_path = os.path.join(tmpdir, "test.db") file_dir = tmpdir try: # Generate real password hash using bcrypt directly test_password = "password123" password_hash = hash_password(test_password) # Create app with auth config app = create_app( db_path=db_path, file_dir=file_dir, auth_config={ "username": "admin", "password_hash": password_hash, "jwt_secret": "test-secret", "expire_minutes": 60 }, max_workers=1 ) # Test login with TestClient - REAL password verification client = TestClient(app) response = client.post("/api/v1/auth/login", json={ "username": "admin", "password": test_password }) assert response.status_code == 200, f"Expected 200, got {response.status_code}" data = response.json() assert "token" in data, "Token not in response" print(f" Login successful, token: {data['token'][:20]}...") print("=== PASS: JWT login ===") return data['token'] finally: shutil.rmtree(tmpdir, ignore_errors=True) except Exception as e: print(f" === FAIL: JWT login test failed: {type(e).__name__}: {e} ===") import traceback traceback.print_exc() print("=== FAIL: JWT login ===") return None def test_protected_route_auth(): """Test protected route authentication""" print("=== PROTECTED ROUTE AUTH TEST ===") try: from fastapi.testclient import TestClient from sanguo_api.app import create_app from sanguo_api.auth import hash_password tmpdir = tempfile.mkdtemp() db_path = os.path.join(tmpdir, "test.db") file_dir = tmpdir try: # Generate real password hash using bcrypt directly test_password = "password123" password_hash = hash_password(test_password) app = create_app( db_path=db_path, file_dir=file_dir, auth_config={ "username": "admin", "password_hash": password_hash, "jwt_secret": "test-secret", "expire_minutes": 60 }, max_workers=1 ) client = TestClient(app) # Test without token - should get 401 response = client.get("/api/v1/task/foo") assert response.status_code == 401, f"Expected 401 without token, got {response.status_code}" print(" No token: 401 Unauthorized ✓") # Get valid token using REAL password verification login_response = client.post("/api/v1/auth/login", json={ "username": "admin", "password": test_password }) token = login_response.json()["token"] # Test with token but non-existent task - should get 404 response = client.get( "/api/v1/task/foo", headers={"Authorization": f"Bearer {token}"} ) assert response.status_code == 404, f"Expected 404 with valid token, got {response.status_code}" print(" Valid token, non-existent task: 404 Not Found ✓") print("=== PASS: protected route auth ===") finally: shutil.rmtree(tmpdir, ignore_errors=True) except Exception as e: print(f" === FAIL: Protected route auth test failed: {type(e).__name__}: {e} ===") import traceback traceback.print_exc() print("=== FAIL: protected route auth ===") async def test_orchestrator_async(): """Test orchestrator async submission""" print("=== ORCHESTRATOR ASYNC TEST ===") from sanguo_orchestrator.runner import Orchestrator tmpdir = tempfile.mkdtemp() db_path = os.path.join(tmpdir, "test.db") try: orch = Orchestrator(db_path=db_path, file_dir=tmpdir, max_workers=1) # Check executor type from concurrent.futures import ProcessPoolExecutor executor_type = type(orch.pool.executor).__name__ print(f" Orchestrator pool executor: {executor_type}") assert executor_type == "ProcessPoolExecutor", f"Expected ProcessPoolExecutor, got {executor_type}" # Submit a tiny factor task task_id = await orch.submit_factor( symbols=["600000.SSE"], factor_names=["ma5"], start="2024-01-01", end="2024-06-30", cfg=None, # Use default config for smoke test output_dir=tmpdir ) print(f" Task submitted: {task_id}") assert task_id.startswith("factor_"), f"Expected task_id to start with 'factor_', got {task_id}" # Check task status status = orch.get_status(task_id) print(f" Task status: {status}") print("=== PASS: orchestrator async ===") return task_id finally: shutil.rmtree(tmpdir, ignore_errors=True) async def test_ws_stage_wiring(): """Test WebSocket stage callback wiring""" print("=== WS STAGE WIRING TEST ===") from sanguo_api.app import create_app from sanguo_api.ws import manager from sanguo_api.routes import get_orchestrator from sanguo_api.auth import hash_password tmpdir = tempfile.mkdtemp() db_path = os.path.join(tmpdir, "test.db") try: # Generate real password hash using bcrypt directly test_password = "password123" password_hash = hash_password(test_password) # Create app to trigger wiring app = create_app( db_path=db_path, file_dir=tmpdir, auth_config={ "username": "admin", "password_hash": password_hash, "jwt_secret": "test-secret", "expire_minutes": 60 }, max_workers=1 ) # Check orchestrator has on_stage callback set orch = get_orchestrator() assert orch._on_stage is not None, "on_stage callback not set" print(" WS manager exists ✓") print(" on_stage callback wired ✓") print("=== PASS: WS stage wiring ===") finally: shutil.rmtree(tmpdir, ignore_errors=True) async def test_real_factor_tears_pipeline(): """Test real factor tears pipeline with multiprocessing guard""" print("=== REAL FACTOR TEARS PIPELINE TEST ===") # Check if we're in container with full dependencies try: import polars import alphalens print(" Container environment detected (polars + alphalens available)") except ImportError as e: print(f" === SKIP: polars/alphalens not available ({e}) ===") return tmpdir = tempfile.mkdtemp() output_dir = os.path.join(tmpdir, "factor_output") try: from sanguo_factor.analyzer import run_factor_analysis from sanguo_data.config import load_config # Load config for database access if os.path.exists("/app/config/data_platform.yaml"): cfg = load_config("/app/config/data_platform.yaml") print(" Loaded database config") else: print(" === SKIP: database config not found ===") return # ≥2 symbols: alphalens IC is cross-sectional (1 symbol → empty bins → fails) symbols = ["600000", "000001", "300750"] # BARE symbols (no .EXCHANGE suffix) for DB lookup factor_names = ["ma5"] # Simple factor start = "2024-01-01" end = "2024-06-30" # Use longer range for reliable IC (Phase 1 confirmed 541 bars) print(f" Running factor analysis: {symbols}, {factor_names}, {start} to {end}") print(" This will test the multiprocessing pipeline...") # Run the analysis result = run_factor_analysis( symbols=symbols, factor_names=factor_names, start=start, end=end, cfg=cfg, output_dir=output_dir ) print(f" Analysis complete: {len(result.factor_names)} factors processed") print(f" IC summary: {result.ic_summary}") print(f" Report paths: {result.report_paths}") # Verify we got a result assert result is not None, "Result is None" assert len(result.factor_names) > 0, "No factors processed" # CRITICAL: Verify IC is non-empty (Root cause C fix) assert result.ic_summary, "ic_summary empty - no IC computed" assert "ma5" in result.ic_summary, "ma5 not in ic_summary" assert result.ic_summary["ma5"].get("ic"), "no IC data for ma5" print(f" ✓ IC computed: {result.ic_summary['ma5']['ic']}") # Check if any reports were generated if result.report_paths: print(f" Tears report generated: {result.report_paths}") for factor_name, path in result.report_paths.items(): if os.path.exists(path.replace('.html', '.png')): # Check for actual PNG file print(f" ✓ {factor_name}: {path}") else: print(f" ⚠ {factor_name}: {path} (file not found)") print("=== PASS: real factor tears pipeline ===") except Exception as e: print(f" === FAIL: real tears pipeline failed with {type(e).__name__}: {e} ===") import traceback traceback.print_exc() raise finally: shutil.rmtree(tmpdir, ignore_errors=True) async def main(): """Run all Phase 3a smoke tests""" print("\n") print("=" * 60) print("PHASE 3A SMOKE TEST SUITE") print("=" * 60) print() # Test 1: sys.path configuration test_sys_path() # Test 2: JWT login test_jwt_login() # Test 3: Protected route authentication test_protected_route_auth() # Test 4: Orchestrator async submission await test_orchestrator_async() # Test 5: WebSocket stage wiring await test_ws_stage_wiring() # Test 6: Real factor tears pipeline (container only) await test_real_factor_tears_pipeline() print() print("=" * 60) print("PHASE 3A SMOKE TEST SUITE: ALL TESTS PASSED") print("=" * 60) print() if __name__ == "__main__": # CRITICAL: multiprocessing guard for vnpy.alpha # This is required because vnpy.alpha's AlphaDataset.prepare_data() spawns # a multiprocessing.Pool, which fails without this guard on spawn platforms print("=== Running with multiprocessing guard ===") asyncio.run(main())