test_requirements_simple.py

"""
Simple standalone test script to validate backend requirements
Can be run directly: python test_requirements_simple.py

This script performs basic validation checks without requiring pytest.
"""

import asyncio
import sys
from sqlalchemy import inspect
from sqlalchemy.ext.asyncio import create_async_engine
from app.database import Base
from app.models import (
    Company, User, Country, Currency, Airport, Charge, ChargeAlias,
    Quote, QuoteCharge, Invoice, InvoiceCharge, Anomaly, TrackingEvent
)


class Colors:
    """ANSI color codes for terminal output"""
    GREEN = '\033[92m'
    RED = '\033[91m'
    YELLOW = '\033[93m'
    BLUE = '\033[94m'
    RESET = '\033[0m'
    BOLD = '\033[1m'


def print_header(text):
    """Print a formatted header"""
    print(f"\n{Colors.BOLD}{Colors.BLUE}{'=' * 70}{Colors.RESET}")
    print(f"{Colors.BOLD}{Colors.BLUE}{text}{Colors.RESET}")
    print(f"{Colors.BOLD}{Colors.BLUE}{'=' * 70}{Colors.RESET}\n")


def print_test(name, passed, details=""):
    """Print test result"""
    status = f"{Colors.GREEN}✓ PASS{Colors.RESET}" if passed else f"{Colors.RED}✗ FAIL{Colors.RESET}"
    print(f"{status} - {name}")
    if details:
        print(f"       {Colors.YELLOW}{details}{Colors.RESET}")


async def test_database_models():
    """Test Phase 1: Database Models & Migrations"""
    print_header("PHASE 1: DATABASE MODELS & MIGRATIONS")
    
    results = []
    
    # Test 1.1: Check all models are defined
    required_models = [
        ("Company", Company),
        ("User", User),
        ("Country", Country),
        ("Currency", Currency),
        ("Airport", Airport),
        ("Charge", Charge),
        ("ChargeAlias", ChargeAlias),
        ("Quote", Quote),
        ("QuoteCharge", QuoteCharge),
        ("Invoice", Invoice),
        ("InvoiceCharge", InvoiceCharge),
        ("Anomaly", Anomaly),
        ("TrackingEvent", TrackingEvent)
    ]
    
    for model_name, model_class in required_models:
        has_tablename = hasattr(model_class, "__tablename__")
        print_test(f"Model {model_name} exists", has_tablename)
        results.append(has_tablename)
    
    # Test 1.2: Check Company model fields
    company_fields = ["id", "name", "short_name", "type", "address", "city", "country", "is_active"]
    company_has_fields = all(hasattr(Company, field) for field in company_fields)
    print_test("Company model has all required fields", company_has_fields, 
               f"Fields: {', '.join(company_fields)}")
    results.append(company_has_fields)
    
    # Test 1.3: Check User model fields
    user_fields = ["id", "company_id", "name", "email", "password_hash", "is_admin", "is_active"]
    user_has_fields = all(hasattr(User, field) for field in user_fields)
    print_test("User model has all required fields", user_has_fields,
               f"Fields: {', '.join(user_fields)}")
    results.append(user_has_fields)
    
    # Test 1.4: Check Charge and ChargeAlias relationship
    charge_has_aliases = hasattr(Charge, "aliases")
    print_test("Charge model has aliases relationship", charge_has_aliases)
    results.append(charge_has_aliases)
    
    # Test 1.5: Check Quote model fields
    quote_fields = ["id", "forwarder_id", "buyer_id", "tracking_number", "status", "created_at"]
    quote_has_fields = all(hasattr(Quote, field) for field in quote_fields)
    print_test("Quote model has all required fields", quote_has_fields,
               f"Key fields: {', '.join(quote_fields)}")
    results.append(quote_has_fields)
    
    # Test 1.6: Check QuoteCharge mapping fields
    quote_charge_fields = ["mapping_tier", "similarity_score", "low_confidence"]
    quote_charge_has_fields = all(hasattr(QuoteCharge, field) for field in quote_charge_fields)
    print_test("QuoteCharge has mapping intelligence fields", quote_charge_has_fields,
               f"Fields: {', '.join(quote_charge_fields)}")
    results.append(quote_charge_has_fields)
    
    return all(results)


async def test_database_tables():
    """Test that tables can be created in database"""
    print_header("DATABASE TABLE CREATION TEST")
    
    try:
        # Use in-memory SQLite for testing
        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
        
        # Create all tables
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        
        # Inspect tables
        async with engine.connect() as conn:
            def get_tables(connection):
                inspector = inspect(connection)
                return inspector.get_table_names()
            
            tables = await conn.run_sync(get_tables)
        
        await engine.dispose()
        
        required_tables = [
            "companies", "users", "countries", "currencies", "airports",
            "charges", "charge_aliases", "quotes", "quote_charges",
            "invoices", "invoice_charges", "anomalies", "tracking_events"
        ]
        
        results = []
        for table in required_tables:
            exists = table in tables
            print_test(f"Table '{table}' created", exists)
            results.append(exists)
        
        return all(results)
        
    except Exception as e:
        print_test("Database table creation", False, f"Error: {str(e)}")
        return False


def test_schemas():
    """Test Phase 1.3: Pydantic Schemas"""
    print_header("PHASE 1.3: PYDANTIC SCHEMAS")
    
    try:
        from app.schemas import (
            CompanyCreate, CompanyRead,
            UserCreate, UserRead,
            ChargeCreate, ChargeRead,
            QuoteCreate, QuoteRead,
            QuoteChargeRead,
            InvoiceRead, InvoiceChargeRead,
            AnomalyRead,
            TrackingEventRead,
            TokenResponse
        )
        
        schemas = [
            ("CompanyCreate", CompanyCreate),
            ("CompanyRead", CompanyRead),
            ("UserCreate", UserCreate),
            ("UserRead", UserRead),
            ("ChargeCreate", ChargeCreate),
            ("ChargeRead", ChargeRead),
            ("QuoteCreate", QuoteCreate),
            ("QuoteRead", QuoteRead),
            ("QuoteChargeRead", QuoteChargeRead),
            ("InvoiceRead", InvoiceRead),
            ("InvoiceChargeRead", InvoiceChargeRead),
            ("AnomalyRead", AnomalyRead),
            ("TrackingEventRead", TrackingEventRead),
            ("TokenResponse", TokenResponse)
        ]
        
        results = []
        for schema_name, schema_class in schemas:
            exists = schema_class is not None
            print_test(f"Schema {schema_name} exists", exists)
            results.append(exists)
        
        # Check ChargeRead has aliases field
        has_aliases = hasattr(ChargeRead, "__annotations__") and "aliases" in ChargeRead.__annotations__
        print_test("ChargeRead includes aliases field", has_aliases)
        results.append(has_aliases)
        
        # Check QuoteChargeRead has mapping fields (check base class too)
        mapping_fields = ["mapping_tier", "low_confidence", "similarity_score"]
        all_annotations = {}
        # Collect annotations from base classes
        for base in QuoteChargeRead.__mro__:
            if hasattr(base, "__annotations__"):
                all_annotations.update(base.__annotations__)
        has_mapping = all(field in all_annotations for field in mapping_fields)
        print_test("QuoteChargeRead has mapping intelligence fields", has_mapping)
        results.append(has_mapping)
        
        return all(results)
        
    except ImportError as e:
        print_test("Import schemas", False, f"Error: {str(e)}")
        return False


def test_auth_dependencies():
    """Test Phase 2.1: Auth Dependencies"""
    print_header("PHASE 2: AUTH & DEPENDENCIES")
    
    try:
        from app.auth.dependencies import get_db, get_current_user
        
        print_test("get_db dependency exists", get_db is not None)
        print_test("get_current_user dependency exists", get_current_user is not None)
        
        return True
        
    except ImportError as e:
        print_test("Import auth dependencies", False, f"Error: {str(e)}")
        return False


def test_routers():
    """Test that all routers are defined"""
    print_header("PHASE 2-4: API ROUTERS")
    
    results = []
    
    routers = [
        ("auth", "app.api.auth"),
        ("companies", "app.api.companies"),
        ("masters", "app.api.masters"),
        ("tracking", "app.api.tracking"),
        ("quotes", "app.api.quotes"),
        ("invoices", "app.api.invoices")
    ]
    
    for router_name, module_path in routers:
        try:
            module = __import__(module_path, fromlist=["router"])
            has_router = hasattr(module, "router")
            print_test(f"Router '{router_name}' exists", has_router)
            results.append(has_router)
        except ImportError as e:
            print_test(f"Router '{router_name}' exists", False, f"Import error: {str(e)}")
            results.append(False)
    
    return all(results)


def test_app_bootstrap():
    """Test Phase 5: App Bootstrap"""
    print_header("PHASE 5: APP BOOTSTRAP")
    
    try:
        from app.main import app
        
        print_test("FastAPI app exists", app is not None)
        
        # Check if routers are registered
        routes = [route.path for route in app.routes]
        
        required_routes = [
            "/health",
            "/auth/login",
            "/companies",
            "/masters/countries",
            "/masters/charges",
            "/tracking"
        ]
        
        results = []
        for route in required_routes:
            # Check if route or a pattern matching it exists
            exists = any(route in r for r in routes)
            print_test(f"Route '{route}' registered", exists)
            results.append(exists)
        
        return all(results)
        
    except ImportError as e:
        print_test("Import FastAPI app", False, f"Error: {str(e)}")
        return False


def test_security():
    """Test security utilities"""
    print_header("SECURITY & PASSWORD HASHING")
    
    try:
        from app.auth.security import get_password_hash, verify_password
        
        # Test password hashing
        password = "test123"
        hashed = get_password_hash(password)
        
        print_test("Password hashing works", hashed != password and len(hashed) > 20)
        print_test("Password verification works", verify_password(password, hashed))
        print_test("Wrong password rejected", not verify_password("wrong", hashed))
        
        return True
        
    except ImportError as e:
        print_test("Import security utilities", False, f"Error: {str(e)}")
        return False
    except Exception as e:
        print_test("Security functions", False, f"Error: {str(e)}")
        return False


async def main():
    """Run all tests"""
    print(f"\n{Colors.BOLD}{Colors.BLUE}")
    print("=" * 70)
    print("  FreightAuditPro Backend Requirements Validation")
    print("  Testing implementation against README_tanishk.md")
    print("=" * 70)
    print(Colors.RESET)
    
    results = []
    
    # Run all test suites
    results.append(await test_database_models())
    results.append(await test_database_tables())
    results.append(test_schemas())
    results.append(test_auth_dependencies())
    results.append(test_routers())
    results.append(test_app_bootstrap())
    results.append(test_security())
    
    # Print summary
    print_header("TEST SUMMARY")
    
    total_passed = sum(results)
    total_tests = len(results)
    
    if all(results):
        print(f"{Colors.GREEN}{Colors.BOLD}✓ ALL TESTS PASSED ({total_passed}/{total_tests}){Colors.RESET}")
        print(f"\n{Colors.GREEN}Your backend implementation meets all requirements from README_tanishk.md!{Colors.RESET}\n")
        return 0
    else:
        print(f"{Colors.RED}{Colors.BOLD}✗ SOME TESTS FAILED ({total_passed}/{total_tests} passed){Colors.RESET}")
        print(f"\n{Colors.YELLOW}Please review the failed tests above and fix the issues.{Colors.RESET}\n")
        return 1


if __name__ == "__main__":
    exit_code = asyncio.run(main())
    sys.exit(exit_code)