taxbaik/scripts/run_local_tests.py

import subprocess
import time
import socket
import sys
import os
import urllib.request

def is_port_in_use(port):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        return s.connect_ex(('localhost', port)) == 0

def wait_for_server(url, timeout=30):
    start_time = time.time()
    print(f"Waiting for local server at {url}...")
    while time.time() - start_time < timeout:
        try:
            with urllib.request.urlopen(url, timeout=2) as response:
                if response.status == 200:
                    print("Local server is up and healthy!")
                    return True
        except Exception:
            pass
        time.sleep(1)
    return False

def is_server_healthy(url):
    try:
        with urllib.request.urlopen(url, timeout=3) as response:
            return response.status == 200
    except Exception:
        return False

def main():
    port = 5001
    base_url = f"http://localhost:{port}/taxbaik"
    health_url = f"{base_url}/healthz"
    
    server_process = None
    started_server = False
    tunnel_process = None

    # Check and establish SSH DB tunnel if needed
    if not is_port_in_use(5432):
        print("Database port 5432 is not active. Starting SSH tunnel to 178.104.200.7...")
        try:
            tunnel_process = subprocess.Popen(
                ["ssh", "-N", "-o", "StrictHostKeyChecking=no", "-o", "BatchMode=yes", "-L", "5432:127.0.0.1:5432", "kjh2064@178.104.200.7"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL
            )
            time.sleep(3)
            if is_port_in_use(5432):
                print("SSH tunnel established successfully.")
            else:
                print("Warning: SSH tunnel started but port 5432 is still not responding.")
        except Exception as e:
            print(f"Failed to start SSH tunnel: {e}")
    else:
        print("Database port 5432 is already active.")

    server_healthy = is_server_healthy(health_url)
    if is_port_in_use(port) and server_healthy:
        print(f"Warning: Port {port} is already in use and the local server is healthy.")
    else:
        if is_port_in_use(port):
            print(f"Warning: Port {port} is already in use but the local server is not healthy. Starting a new server attempt may fail if another process is bound.")
        print(f"Starting dotnet local server on port {port}...")
        # Start dotnet run in background
        env = os.environ.copy()
        env["ASPNETCORE_ENVIRONMENT"] = "Development"
        
        # We start dotnet run --project src/TaxBaik.Web with encoding set to utf-8 and errors ignored/replaced
        try:
            log_file = open("local_server.log", "w", encoding="utf-8", buffering=1)
            server_process = subprocess.Popen(
                ["dotnet", "run", "--project", "src/TaxBaik.Web", "--launch-profile", "http"],
                env=env,
                stdout=log_file,
                stderr=subprocess.STDOUT
            )
            started_server = True
        except Exception as e:
            print(f"Failed to launch dotnet server: {e}")
            if tunnel_process:
                tunnel_process.terminate()
            sys.exit(1)
        
        # Wait for the server to become healthy
        if not wait_for_server(health_url, timeout=45):
            print("Timeout: Local server did not start within 45 seconds.")
            # Let's print out what happened to stdout/stderr of dotnet if any
            try:
                with open("local_server.log", "r", encoding="utf-8") as lf:
                    print("--- Dotnet Server Output ---")
                    print(lf.read())
            except Exception:
                pass
            
            # Cleanup
            server_process.terminate()
            try:
                server_process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                server_process.kill()
            if tunnel_process:
                tunnel_process.terminate()
            sys.exit(1)

    # Run Playwright tests
    print("\n==========================================")
    print("Running Playwright E2E Tests Locally...")
    print("==========================================\n")
    
    test_env = os.environ.copy()
    test_env["E2E_BASE_URL"] = base_url
    test_env["EVIDENCE_ROOT"] = "evidence"
    # Set default test credentials
    test_env["E2E_ADMIN_USERNAME"] = "admin"
    test_env["E2E_ADMIN_PASSWORD"] = "Admin123!@#456"
    
    # Run playwright test command
    # Local 1차 검증은 Desktop Chrome만 대상으로 잡아 불필요한 브라우저 설치 의존을 줄인다.
    cmd = ["npx", "playwright", "test"]
    has_project_arg = any(arg.startswith("--project=") or arg == "--project" for arg in sys.argv[1:])
    if not has_project_arg:
        cmd.extend(["--project", "Desktop Chrome"])
    if len(sys.argv) > 1:
        cmd.extend(sys.argv[1:])
        
    print(f"Executing: {' '.join(cmd)}")
    
    try:
        # Use shell=True on Windows to resolve npm/npx paths properly
        use_shell = sys.platform == 'win32'
        result = subprocess.run(cmd, env=test_env, shell=use_shell, timeout=300)
        exit_code = result.returncode
    except Exception as e:
        print(f"Error running tests: {e}")
        exit_code = 1
    finally:
        if started_server and server_process:
            print("\nShutting down local dotnet server...")
            server_process.terminate()
            try:
                server_process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                server_process.kill()
            print("Server stopped.")
            
        if tunnel_process:
            print("Closing SSH tunnel...")
            tunnel_process.terminate()
            try:
                tunnel_process.wait(timeout=5)
            except Exception:
                tunnel_process.kill()
            print("SSH tunnel closed.")
            
    sys.exit(exit_code)

if __name__ == "__main__":
    main()