File size: 4,722 Bytes
1686de5
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
#!/usr/bin/env python3
"""Core application tests"""

import os
import sys
import asyncio
from pathlib import Path

sys.path.append(os.path.join(os.path.dirname(__file__), '..'))

def test_basic():
    """Basic application health check"""
    print("TESTING: Basic application health...")
    
    try:
        from app.main import app
        print("SUCCESS: FastAPI app imported")
        
        if hasattr(app, 'routes'):
            print("SUCCESS: App has routes")
        else:
            print("WARNING: App missing routes")
            
        return True
    except ImportError as e:
        print(f"ERROR: Could not import app: {e}")
        return False
    except Exception as e:
        print(f"ERROR: Basic test failed: {e}")
        return False

def test_database():
    """Test database connection and basic operations"""
    print("\nTESTING: Database connection...")
    
    try:
        from app.database import engine
        from sqlalchemy import text
        
        with engine.connect() as conn:
            result = conn.execute(text("SELECT 1"))
            print("SUCCESS: Database connection successful")
            
            result = conn.execute(text("SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'"))
            tables = [row[0] for row in result.fetchall()]
            print(f"INFO: Available tables: {tables}")
            
            key_tables = ['models', 'images', 'captions']
            for table in key_tables:
                if table in tables:
                    result = conn.execute(text(f"SELECT COUNT(*) FROM {table}"))
                    count = result.fetchone()[0]
                    print(f"SUCCESS: {table} table has {count} records")
                else:
                    print(f"WARNING: {table} table not found")
            
            return True
            
    except ImportError as e:
        print(f"ERROR: Could not import database: {e}")
        return False
    except Exception as e:
        print(f"ERROR: Database test failed: {e}")
        return False

async def test_api_endpoints():
    """Test basic API endpoint availability"""
    print("\nTESTING: API endpoints...")
    
    try:
        from app.main import app
        from fastapi.testclient import TestClient
        
        client = TestClient(app)
        
        endpoints = [
            ("/", "Root endpoint"),
            ("/docs", "API documentation"),
            ("/api/images", "Images endpoint"),
            ("/api/models", "Models endpoint")
        ]
        
        for endpoint, description in endpoints:
            try:
                response = client.get(endpoint)
                if response.status_code in [200, 404, 405]:
                    print(f"SUCCESS: {description} ({endpoint}) - Status: {response.status_code}")
                else:
                    print(f"WARNING: {description} ({endpoint}) - Status: {response.status_code}")
            except Exception as e:
                print(f"ERROR: {description} ({endpoint}) - Exception: {e}")
        
        return True
        
    except ImportError as e:
        print(f"ERROR: Could not import FastAPI test client: {e}")
        return False
    except Exception as e:
        print(f"ERROR: API endpoint test failed: {e}")
        return False

def test_environment():
    """Test environment variables and configuration"""
    print("\nTESTING: Environment configuration...")
    
    required_vars = [
        'DATABASE_URL',
        'HF_API_KEY',
        'OPENAI_API_KEY',
        'GOOGLE_API_KEY'
    ]
    
    missing_vars = []
    for var in required_vars:
        if os.getenv(var):
            print(f"SUCCESS: {var} is set")
        else:
            print(f"WARNING: {var} is not set")
            missing_vars.append(var)
    
    if missing_vars:
        print(f"INFO: Missing environment variables: {missing_vars}")
        print("INFO: Some tests may fail without these variables")
    
    return len(missing_vars) == 0

async def main():
    """Run all core tests"""
    print("Core Application Tests")
    print("=" * 40)
    
    results = []
    
    results.append(test_basic())
    results.append(test_database())
    results.append(test_environment())
    results.append(await test_api_endpoints())
    
    print("\n" + "=" * 40)
    print("TEST SUMMARY")
    print("=" * 40)
    
    passed = sum(results)
    total = len(results)
    
    print(f"Passed: {passed}/{total}")
    
    if passed == total:
        print("SUCCESS: All core tests passed")
        return 0
    else:
        print(f"WARNING: {total - passed} test(s) failed")
        return 1

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