test_system.py

"""
Test script to verify system components
Author: Martin Bacigal, 01/2025
"""

import asyncio
import logging
from pathlib import Path
import tempfile
import hashlib

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

async def test_database_connection():
    """Test database connectivity and schema"""
    logging.info("Testing database connection...")

    try:
        from database_manager import DatabaseManager, DatabaseConfig

        # Create test config
        config = DatabaseConfig(
            db_type="sqlite",
            database="test_contracts"
        )

        # Initialize database
        db_manager = DatabaseManager(config)
        db_manager.initialize()

        logging.info("✓ Database initialized successfully")

        # Test hash calculation
        test_file = Path(__file__)
        file_hash = db_manager.calculate_file_hash(test_file)
        logging.info(f"✓ File hash calculation works: {file_hash[:16]}...")

        # Test document creation
        doc, is_new = await db_manager.create_or_update_document(
            file_path=test_file,
            file_hash=file_hash,
            cw_number="TEST-001",
            created_by="test_script"
        )

        logging.info(f"✓ Document {'created' if is_new else 'updated'}: ID={doc.id}")

        # Test processing update
        await db_manager.update_document_processing(
            doc.id,
            'completed',
            message="Test completed",
            duration_seconds=1
        )

        logging.info("✓ Processing status updated")

        # Test ontology tree
        tree = await db_manager.get_ontology_tree()
        logging.info(f"✓ Ontology tree loaded: {len(tree)} root categories")

        # Test progress stats
        stats = await db_manager.get_processing_progress()
        logging.info(f"✓ Progress stats retrieved: {stats}")

        db_manager.close()
        return True

    except Exception as e:
        logging.error(f"✗ Database test failed: {e}")
        return False

def test_settings_manager():
    """Test settings management"""
    logging.info("\nTesting settings manager...")

    try:
        from settings_manager import SettingsManager

        manager = SettingsManager()
        logging.info(f"✓ Settings loaded from: {manager.settings_file}")
        logging.info(f"✓ Output directory: {manager.settings.output_directory}")
        logging.info(f"✓ Batch size: {manager.settings.batch_size}")

        # Test saving
        manager.save_settings()
        logging.info("✓ Settings saved successfully")

        return True

    except Exception as e:
        logging.error(f"✗ Settings test failed: {e}")
        return False

def test_memory_manager():
    """Test memory management"""
    logging.info("\nTesting memory manager...")

    try:
        from memory_manager import MemoryConfig, MemoryMonitor, get_system_memory_info

        # Get system info
        mem_info = get_system_memory_info()
        logging.info(f"✓ System memory: {mem_info['total_mb']:.0f}MB total, {mem_info['available_mb']:.0f}MB available")

        # Create monitor
        config = MemoryConfig(memory_limit_mb=1000, max_batch_size=10)
        monitor = MemoryMonitor(config)

        # Test monitoring
        monitor.start_monitoring()
        info = monitor.get_memory_info()
        logging.info(f"✓ Process memory: {info['process_rss_mb']:.1f}MB")

        monitor.stop_monitoring()
        logging.info("✓ Memory monitoring works")

        return True

    except Exception as e:
        logging.error(f"✗ Memory test failed: {e}")
        return False

def test_imports():
    """Test all required imports"""
    logging.info("\nTesting imports...")

    required_modules = [
        'pandas',
        'pypdf',
        'docx',
        'mammoth',
        'openai',
        'sqlalchemy',
        'networkx',
        'matplotlib',
        'psutil',
        'redis',
        'py2neo'
    ]

    failed = []
    for module in required_modules:
        try:
            __import__(module)
            logging.info(f"✓ {module}")
        except ImportError:
            logging.warning(f"✗ {module} (optional)")
            failed.append(module)

    if failed:
        logging.info(f"\nOptional modules not installed: {', '.join(failed)}")
        logging.info("These are only needed for specific features")

    return True

async def main():
    """Run all tests"""
    print("=" * 60)
    print("Contract Processing System - Component Test")
    print("=" * 60)

    # Run tests
    tests = [
        ("Imports", test_imports()),
        ("Settings Manager", test_settings_manager()),
        ("Memory Manager", test_memory_manager()),
        ("Database Connection", await test_database_connection())
    ]

    # Summary
    print("\n" + "=" * 60)
    print("Test Summary:")
    print("=" * 60)

    passed = sum(1 for _, result in tests if result)
    total = len(tests)

    for name, result in tests:
        status = "PASS" if result else "FAIL"
        print(f"{name:<30} {status}")

    print(f"\nTotal: {passed}/{total} tests passed")

    if passed == total:
        print("\n✅ All systems operational! The application is ready to use.")
    else:
        print("\n⚠️ Some components failed. Check the logs above for details.")

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