""" Integration test for discrete_material_plan.extractor module Tests the complete extraction workflow using real ERP data. Requires valid ERP credentials in .env file. This test: 1. Reads order IDs from tests/id-demo.txt 2. Downloads data from Yonyou BIP ERP system 3. Saves raw Excel files to data/downloads/ 4. Converts and merges to data/output/merged_result.xlsx """ import sys from pathlib import Path from datetime import datetime # Add project root to Python path PROJECT_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(PROJECT_ROOT)) # Load environment variables from dotenv import load_dotenv load_dotenv(PROJECT_ROOT / ".env") # Configure browser path import os os.environ["PLAYWRIGHT_BROWSERS_PATH"] = os.getenv("PLAYWRIGHT_BROWSERS_PATH", "") # Generate timestamp for unique output files TIMESTAMP = datetime.now().strftime("%Y%m%d_%H%M%S") # Output paths DOWNLOAD_DIR = PROJECT_ROOT / "data" / "downloads" OUTPUT_DIR = PROJECT_ROOT / "data" / "output" OUTPUT_FILE = OUTPUT_DIR / f"merged_result_{TIMESTAMP}.xlsx" # Ensure directories exist DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True) OUTPUT_DIR.mkdir(parents=True, exist_ok=True) print("=" * 60) print("Discrete Material Plan Extractor - Integration Test") print("=" * 60) print(f"\nConfiguration:") print(f" ID File: {PROJECT_ROOT / 'tests' / 'id-demo.txt'}") print(f" Download Dir: {DOWNLOAD_DIR}") print(f" Output File: {OUTPUT_FILE}") print(f" ERP URL: {os.getenv('ERP_URL', 'Not set')}") print(f" Headless: {os.getenv('ERP_HEADLESS', 'true')}") print("=" * 60) # Verify environment variables required_vars = ["ERP_USERNAME", "ERP_PASSWORD", "ERP_URL"] missing_vars = [var for var in required_vars if not os.getenv(var)] if missing_vars: print(f"\n[ERROR] Missing required environment variables: {missing_vars}") print("Please ensure .env file contains:") print(" ERP_USERNAME=your_username") print(" ERP_PASSWORD=your_password") print(" ERP_URL=https://your-erp-url.com") sys.exit(1) try: # Import required modules print("\n[1/6] Importing modules...") from playwright.sync_api import sync_playwright from utils.auth import login from utils.discrete_material_plan import ( navigate_to_discrete_material_page, extract_and_post_process, read_order_ids_from_file, get_login_url, ) print("[OK] Modules imported successfully") # Read order IDs from demo file print("\n[2/6] Reading order IDs from data/demo/id-demo.txt...") id_file = PROJECT_ROOT / "data" / "demo" / "id-demo.txt" order_ids = read_order_ids_from_file(str(id_file)) print(f"[OK] Loaded {len(order_ids)} order IDs") test_order_ids = order_ids # Setup browser and extract data print("\n[3/6] Launching browser and logging in...") with sync_playwright() as playwright: # Login browser, context, page, main_frame = login( playwright=playwright, username=os.getenv("ERP_USERNAME"), password=os.getenv("ERP_PASSWORD"), url=get_login_url(os.getenv("ERP_URL")), headless=os.getenv("ERP_HEADLESS", "true").lower() == "true", ignore_https_errors=os.getenv("ERP_IGNORE_HTTPS_ERRORS", "true").lower() == "true", verbose=True, ) try: # Navigate to discrete material page print("\n[4/6] Navigating to discrete material plan page...") work_frame, page1 = navigate_to_discrete_material_page(main_frame, page) print("[OK] Navigation successful") # Extract and post-process print("\n[5/6] Extracting data and post-processing...") print(f" Orders: {len(test_order_ids)}") print(f" Batch size: 100") print(f" Download dir: {DOWNLOAD_DIR}") print(f" Output file: {OUTPUT_FILE}") print(f" Cleanup temp files: True (default)") output_path, df = extract_and_post_process( work_frame=work_frame, page=page1, order_ids=test_order_ids, download_dir=str(DOWNLOAD_DIR), output_file=str(OUTPUT_FILE), batch_size=100, verbose=True, cleanup_temp_files=True, # Default: True, set to False to keep temp files ) print("\n[6/6] Verifying results...") print(f"[OK] Extraction completed successfully") print(f" Output file: {output_path}") print(f" Total records: {len(df)}") print(f" Columns: {list(df.columns)}") # Verify output file exists if not Path(output_path).exists(): raise FileNotFoundError(f"Output file not found: {output_path}") print(f"[OK] Output file verified: {output_path}") # Show sample data print("\n" + "=" * 60) print("Sample Data (first 3 rows, first 5 columns):") print("=" * 60) # Use unicode encoding for display sample = df.head(3).iloc[:, :5] print(sample.to_string()) print("\n[OK] Data extraction verified - Chinese characters displayed correctly in console encoding") finally: # Cleanup browser print("\n\n[Cleanup] Closing browser...") context.close() browser.close() print("[OK] Browser closed") print("\n" + "=" * 60) print("[SUCCESS] Integration test completed successfully!") print("=" * 60) print(f"\nOutputs:") print(f" Raw Excel files: {DOWNLOAD_DIR}/*.xlsx") print(f" Merged result: {OUTPUT_FILE}") print(f" Total records: {len(df)}") print("=" * 60) except Exception as e: print(f"\n[ERROR] Integration test failed!") print(f"Error: {e}") import traceback traceback.print_exc() sys.exit(1)