#!/usr/bin/env python3 """ Debug script for the RAG web interface using Playwright """ import asyncio import sys import os import subprocess import time import signal from pathlib import Path from playwright.async_api import async_playwright class WebDebugger: def __init__(self): self.flask_process = None self.browser = None self.context = None self.page = None async def start_flask_app(self): """Start the Flask app in the background""" print("Starting Flask app...") # Activate virtual environment and start Flask env = os.environ.copy() env['PATH'] = f"{os.path.dirname(sys.executable)}:{env['PATH']}" self.flask_process = subprocess.Popen( [sys.executable, 'app.py'], cwd='/home/dmtnaga/Documents/work/airagagent', env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, preexec_fn=os.setsid ) # Wait for Flask to start and check output time.sleep(5) # Read output to see if Flask started if self.flask_process.poll() is None: # Process is still running, try to read some output try: stdout_data = self.flask_process.stdout.read(1024) stderr_data = self.flask_process.stderr.read(1024) stdout_str = stdout_data.decode() if stdout_data else "" stderr_str = stderr_data.decode() if stderr_data else "" print(f"Flask stdout: {stdout_str}") print(f"Flask stderr: {stderr_str}") # Check if Flask is actually running import socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) result = sock.connect_ex(('127.0.0.1', 5000)) sock.close() if result == 0: print("Flask app is responding on port 5000") return True else: print("Flask app started but not responding on port 5000") return False except Exception as e: print(f"Error checking Flask status: {e}") return False else: # Process has terminated stdout, stderr = self.flask_process.communicate() print(f"Flask failed to start!") print(f"STDOUT: {stdout.decode()}") print(f"STDERR: {stderr.decode()}") return False async def setup_browser(self): """Setup Playwright browser""" print("Setting up browser...") playwright = await async_playwright().start() self.browser = await playwright.chromium.launch( headless=False, # Set to False to see the browser args=['--no-sandbox', '--disable-setuid-sandbox'] ) self.context = await self.browser.new_context() self.page = await self.context.new_page() # Capture console logs self.page.on("console", lambda msg: print(f"CONSOLE: {msg.type}: {msg.text}")) self.page.on("pageerror", lambda err: print(f"PAGE ERROR: {err}")) # Capture network requests self.page.on("request", lambda request: print(f"REQUEST: {request.method} {request.url}")) self.page.on("response", lambda response: print(f"RESPONSE: {response.status} {response.url}")) print("Browser setup complete") async def test_upload_functionality(self): """Test the file upload functionality""" try: print("Navigating to web interface...") await self.page.goto("http://localhost:5000") await self.page.wait_for_load_state('networkidle') # Wait for page to be ready await self.page.wait_for_selector('#upload-area', timeout=10000) print("Page loaded successfully") # Check if upload area is visible upload_area = self.page.locator('#upload-area') is_visible = await upload_area.is_visible() print(f"Upload area visible: {is_visible}") # Get the file input element file_input = self.page.locator('#file-input') file_input_visible = await file_input.is_visible() print(f"File input visible: {file_input_visible}") # Create a test file test_file_path = "/home/dmtnaga/Documents/work/airagagent/test_upload.txt" with open(test_file_path, 'w') as f: f.write("This is a test file for upload debugging.") print(f"Attempting to upload file: {test_file_path}") # Make the file input visible for testing await self.page.evaluate('document.getElementById("file-input").style.display = "block"') # Now try to upload the file await file_input.set_input_files(test_file_path) # Wait for upload to complete or error await self.page.wait_for_timeout(5000) # Check for any error messages error_alerts = self.page.locator('.alert-danger') error_count = await error_alerts.count() if error_count > 0: error_text = await error_alerts.first.inner_text() print(f"UPLOAD ERROR: {error_text}") else: print("No error alerts found") # Check for success messages success_alerts = self.page.locator('.alert-success') success_count = await success_alerts.count() if success_count > 0: success_text = await success_alerts.first.inner_text() print(f"UPLOAD SUCCESS: {success_text}") else: print("No success alerts found") # Check status update status_div = self.page.locator('#system-status') if await status_div.is_visible(): status_text = await status_div.inner_text() print(f"System status: {status_text}") # Clean up test file os.remove(test_file_path) except Exception as e: print(f"Test failed with error: {e}") import traceback traceback.print_exc() async def test_ask_functionality(self): """Test the Q&A functionality""" try: print("\nTesting Q&A functionality...") # Find the question input question_input = self.page.locator('#question-input') ask_button = self.page.locator('#ask-button') if await question_input.is_visible() and await ask_button.is_visible(): print("Q&A interface elements found") # Type a test question await question_input.fill("What is AI?") print("Entered test question") # Click ask button await ask_button.click() print("Clicked ask button") # Wait for response await self.page.wait_for_timeout(10000) # Check for response chat_messages = self.page.locator('.chat-message') message_count = await chat_messages.count() print(f"Found {message_count} chat messages") if message_count > 0: last_message = await chat_messages.last.inner_text() print(f"Last message preview: {last_message[:200]}...") else: print("No chat messages found") else: print("Q&A interface elements not found") except Exception as e: print(f"Q&A test failed: {e}") async def run_tests(self): """Run all tests""" try: # Check if Flask is already running import socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) result = sock.connect_ex(('127.0.0.1', 5000)) sock.close() if result != 0: print("Flask app not running on port 5000. Starting it...") if not await self.start_flask_app(): print("Failed to start Flask app") return else: print("Flask app already running on port 5000") # Setup browser await self.setup_browser() # Run upload test await self.test_upload_functionality() # Run Q&A test await self.test_ask_functionality() print("\nAll tests completed!") except Exception as e: print(f"Test suite failed: {e}") import traceback traceback.print_exc() finally: # Cleanup if self.browser: await self.browser.close() if self.flask_process: os.killpg(os.getpgid(self.flask_process.pid), signal.SIGTERM) self.flask_process.wait() print("Flask app stopped") async def main(): debugger = WebDebugger() await debugger.run_tests() if __name__ == "__main__": asyncio.run(main())