hive/test_complete_flow.py

#!/usr/bin/env python3
"""
Complete Frontend Flow Testing with Login and Screenshots
Tests the entire application flow with authentication
"""

import os
import time
from datetime import datetime
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

class HiveFullFlowTester:
    def __init__(self, base_url="http://localhost:3000", headless=True):
        self.base_url = base_url
        self.headless = headless
        self.driver = None
        self.screenshots_dir = "/tmp/hive_flow_screenshots"
        self.setup_driver()
        os.makedirs(self.screenshots_dir, exist_ok=True)
        
    def setup_driver(self):
        """Setup Chrome WebDriver"""
        chrome_options = Options()
        if self.headless:
            chrome_options.add_argument("--headless")
        chrome_options.add_argument("--no-sandbox")
        chrome_options.add_argument("--disable-dev-shm-usage")
        chrome_options.add_argument("--disable-gpu")
        chrome_options.add_argument("--window-size=1920,1080")
        
        self.driver = webdriver.Chrome(options=chrome_options)
        self.driver.implicitly_wait(10)
        
    def take_screenshot(self, name, description=""):
        """Take a screenshot with timestamp"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"{timestamp}_{name}.png"
        filepath = os.path.join(self.screenshots_dir, filename)
        
        self.driver.save_screenshot(filepath)
        print(f"📸 {filename}: {description}")
        return filepath
    
    def login(self, username="admin", password="hiveadmin"):
        """Login to the application"""
        print("🔐 Testing Login Flow...")
        
        # Go to login page
        self.driver.get(f"{self.base_url}/login")
        self.take_screenshot("01_login_page", "Login page loaded")
        
        # Wait for login form - use more specific approach
        username_field = WebDriverWait(self.driver, 10).until(
            EC.presence_of_element_located((By.CSS_SELECTOR, "input[type='text']"))
        )
        
        password_field = self.driver.find_element(By.CSS_SELECTOR, "input[type='password']")
        login_button = self.driver.find_element(By.CSS_SELECTOR, "button[type='submit']")
        
        # Fill in credentials
        username_field.clear()
        username_field.send_keys(username)
        password_field.clear()
        password_field.send_keys(password)
        
        self.take_screenshot("02_credentials_entered", "Login credentials entered")
        
        # Click login
        login_button.click()
        
        # Wait for redirect (either to dashboard or another page)
        time.sleep(3)
        
        current_url = self.driver.current_url
        print(f"After login - Current URL: {current_url}")
        
        if "/login" not in current_url:
            print("✅ Login successful - redirected away from login page")
            self.take_screenshot("03_login_success", "Successfully logged in")
            return True
        else:
            print("❌ Login failed - still on login page")
            self.take_screenshot("03_login_failed", "Login failed")
            return False
    
    def test_pages_after_login(self):
        """Test all main pages after successful login"""
        pages = [
            ("/", "dashboard", "Main Dashboard"),
            ("/agents", "agents", "Agents Management"),
            ("/tasks", "tasks", "Task Management"), 
            ("/workflows", "workflows", "Workflow Management"),
            ("/analytics", "analytics", "Analytics Dashboard"),
            ("/executions", "executions", "Execution History")
        ]
        
        results = {}
        
        for path, page_name, description in pages:
            try:
                print(f"🔍 Testing {description}...")
                
                self.driver.get(f"{self.base_url}{path}")
                time.sleep(3)  # Wait for page load
                
                # Check if we got redirected back to login (session expired)
                if "/login" in self.driver.current_url:
                    print(f"❌ {page_name}: Redirected to login (authentication issue)")
                    results[page_name] = "auth_failed"
                    continue
                
                # Take screenshot
                self.take_screenshot(f"04_{page_name}_page", f"{description} loaded")
                
                # Check for loading states or errors
                page_text = self.driver.find_element(By.TAG_NAME, "body").text.lower()
                
                if "error" in page_text and "error rate" not in page_text:
                    print(f"⚠️  {page_name}: Contains error messages")
                    results[page_name] = "has_errors"
                elif "loading" in page_text:
                    print(f"⏳ {page_name}: Still loading, waiting...")
                    time.sleep(3)
                    self.take_screenshot(f"04_{page_name}_after_load", f"{description} after additional wait")
                    results[page_name] = "loaded_slow"
                else:
                    print(f"✅ {page_name}: Loaded successfully")
                    results[page_name] = "success"
                
                # Check for specific content based on page
                if page_name == "agents":
                    agent_elements = self.driver.find_elements(By.CSS_SELECTOR, "[data-testid*='agent'], .agent-card, .agent-item")
                    print(f"   Found {len(agent_elements)} agent elements")
                
                elif page_name == "analytics":
                    chart_elements = self.driver.find_elements(By.CSS_SELECTOR, "svg, canvas, .recharts-wrapper, .chart")
                    print(f"   Found {len(chart_elements)} chart elements")
                
                elif page_name == "tasks":
                    task_elements = self.driver.find_elements(By.CSS_SELECTOR, "[data-testid*='task'], .task-card, .task-item")
                    print(f"   Found {len(task_elements)} task elements")
                
            except Exception as e:
                print(f"❌ {page_name}: Error testing page - {e}")
                self.take_screenshot(f"04_{page_name}_error", f"Error on {description}")
                results[page_name] = "error"
        
        return results
    
    def test_data_loading(self):
        """Test if data is loading from APIs"""
        print("🔍 Testing API Data Loading...")
        
        # Go to agents page to test API calls
        self.driver.get(f"{self.base_url}/agents")
        time.sleep(5)  # Wait for API calls
        
        # Check network activity (basic check)
        page_source = self.driver.page_source
        
        # Look for signs of real data vs mock data
        data_indicators = {
            "has_agent_data": "agent" in page_source.lower(),
            "has_loading_states": "loading" in page_source.lower(),
            "has_error_states": "error" in page_source.lower() and "no error" not in page_source.lower(),
            "has_empty_states": "no agents" in page_source.lower() or "empty" in page_source.lower()
        }
        
        print("📊 Data Loading Analysis:")
        for indicator, present in data_indicators.items():
            status = "✅" if present else "❌"
            print(f"   {status} {indicator}: {present}")
        
        self.take_screenshot("05_data_analysis", "Data loading analysis")
        
        return data_indicators
    
    def run_complete_test(self):
        """Run complete application test"""
        print("🚀 Starting Complete Hive Frontend Test")
        print("=" * 50)
        
        try:
            # Test 1: Login
            login_success = self.login()
            if not login_success:
                print("❌ Cannot continue - login failed")
                return False
            
            # Test 2: Page navigation
            print("\n🧭 Testing Page Navigation...")
            page_results = self.test_pages_after_login()
            
            # Test 3: Data loading
            print("\n📡 Testing Data Loading...")
            data_results = self.test_data_loading()
            
            # Summary
            print("\n" + "=" * 50)
            print("📊 Complete Test Results:")
            print(f"🔐 Login: {'✅ SUCCESS' if login_success else '❌ FAILED'}")
            
            print("🧭 Page Navigation:")
            for page, result in page_results.items():
                status_emoji = {"success": "✅", "loaded_slow": "⏳", "has_errors": "⚠️", "auth_failed": "🔐", "error": "❌"}
                emoji = status_emoji.get(result, "❓")
                print(f"   {emoji} {page}: {result}")
            
            print("📡 Data Loading:")
            for indicator, present in data_results.items():
                emoji = "✅" if present else "❌"
                print(f"   {emoji} {indicator}")
            
            print(f"\n📁 Screenshots saved to: {self.screenshots_dir}")
            
            # Overall success
            failed_pages = [p for p, r in page_results.items() if r in ["error", "auth_failed"]]
            overall_success = login_success and len(failed_pages) == 0
            
            print(f"\n🎯 Overall Result: {'✅ SUCCESS' if overall_success else '❌ NEEDS WORK'}")
            if failed_pages:
                print(f"   Failed pages: {failed_pages}")
                
            return overall_success
            
        except Exception as e:
            print(f"❌ Test suite failed with error: {e}")
            self.take_screenshot("error_test_suite", "Test suite error")
            return False
        finally:
            if self.driver:
                self.driver.quit()

if __name__ == "__main__":
    tester = HiveFullFlowTester()
    success = tester.run_complete_test()
    exit(0 if success else 1)