feat: Add backend tests with full cache initialization
- Add .env.test with test credentials and company config - Add pytest fixtures for cache initialization (temp SQLite) - Add test_api_real.py (18 tests) - API endpoint tests - Add test_cache_real.py (8 tests) - Cache system tests - Add test_services_real.py (9 tests) - Service layer tests - Use company 110 (MARIUSM_AUTO) - only schema with full data in TEST All 35 backend tests pass. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
186
reports-app/backend/tests/conftest.py
Normal file
186
reports-app/backend/tests/conftest.py
Normal file
@@ -0,0 +1,186 @@
|
||||
# reports-app/backend/tests/conftest.py
|
||||
"""
|
||||
Pytest fixtures for backend tests with real Oracle database.
|
||||
Requires SSH tunnel and valid Oracle credentials.
|
||||
"""
|
||||
import pytest
|
||||
import asyncio
|
||||
import sys
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
# Add paths for imports
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..', 'shared'))
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load .env.test first (test-specific config), then .env as fallback
|
||||
env_test_path = os.path.join(os.path.dirname(__file__), '..', '.env.test')
|
||||
env_path = os.path.join(os.path.dirname(__file__), '..', '.env')
|
||||
|
||||
if os.path.exists(env_test_path):
|
||||
load_dotenv(env_test_path, override=True)
|
||||
load_dotenv(env_path)
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
from httpx import AsyncClient, ASGITransport
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def event_loop():
|
||||
"""Create event loop for async tests"""
|
||||
loop = asyncio.get_event_loop_policy().new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def oracle_available():
|
||||
"""Check if Oracle is available, skip tests if not"""
|
||||
try:
|
||||
from database.oracle_pool import oracle_pool
|
||||
|
||||
# Initialize the pool if not already initialized
|
||||
if oracle_pool._pool is None:
|
||||
await oracle_pool.initialize()
|
||||
|
||||
async with oracle_pool.get_connection() as conn:
|
||||
with conn.cursor() as cursor:
|
||||
cursor.execute("SELECT 1 FROM DUAL")
|
||||
result = cursor.fetchone()
|
||||
if result and result[0] == 1:
|
||||
return True
|
||||
except Exception as e:
|
||||
pytest.skip(f"Oracle not available: {e}")
|
||||
return False
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def cache_initialized():
|
||||
"""
|
||||
Initialize cache system for tests with temp SQLite database.
|
||||
|
||||
This fixture ensures the cache is fully initialized before tests run,
|
||||
using a temporary database file that's cleaned up after the session.
|
||||
"""
|
||||
from app.cache import init_cache, close_cache, get_cache
|
||||
from app.cache.config import CacheConfig
|
||||
import app.cache.cache_manager as cache_module
|
||||
|
||||
# Create temp directory for test cache
|
||||
temp_dir = tempfile.mkdtemp(prefix="roa2web_test_cache_")
|
||||
temp_db_path = os.path.join(temp_dir, "test_cache.db")
|
||||
|
||||
# Override environment for test cache
|
||||
original_sqlite_path = os.environ.get('CACHE_SQLITE_PATH')
|
||||
os.environ['CACHE_SQLITE_PATH'] = temp_db_path
|
||||
os.environ['CACHE_ENABLED'] = 'True'
|
||||
os.environ['CACHE_TYPE'] = 'hybrid'
|
||||
os.environ['CACHE_BENCHMARK_ON_STARTUP'] = 'False' # Skip benchmarks in tests
|
||||
os.environ['CACHE_TRACK_PERFORMANCE'] = 'False' # Disable perf tracking for tests
|
||||
|
||||
try:
|
||||
# Reset global cache manager if already initialized
|
||||
if cache_module._cache_manager is not None:
|
||||
await cache_module._cache_manager.close()
|
||||
cache_module._cache_manager = None
|
||||
|
||||
# Create config and initialize cache
|
||||
config = CacheConfig.from_env()
|
||||
await init_cache(config)
|
||||
|
||||
cache = get_cache()
|
||||
if cache is None:
|
||||
pytest.skip("Cache initialization failed")
|
||||
|
||||
yield cache
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
try:
|
||||
await close_cache()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Reset environment
|
||||
if original_sqlite_path:
|
||||
os.environ['CACHE_SQLITE_PATH'] = original_sqlite_path
|
||||
elif 'CACHE_SQLITE_PATH' in os.environ:
|
||||
del os.environ['CACHE_SQLITE_PATH']
|
||||
|
||||
# Remove temp files
|
||||
try:
|
||||
import shutil
|
||||
shutil.rmtree(temp_dir, ignore_errors=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def app(cache_initialized):
|
||||
"""Get FastAPI app with cache already initialized via cache_initialized fixture"""
|
||||
from app.main import app
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def async_client(app):
|
||||
"""Async HTTP client for testing"""
|
||||
transport = ASGITransport(app=app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def auth_token(async_client, oracle_available):
|
||||
"""Get valid JWT token for tests"""
|
||||
# Use test credentials from environment
|
||||
test_user = os.getenv("TEST_ORACLE_USER", "test_user")
|
||||
test_pass = os.getenv("TEST_ORACLE_PASS", "test_pass")
|
||||
|
||||
response = await async_client.post("/api/auth/login", json={
|
||||
"username": test_user,
|
||||
"password": test_pass
|
||||
})
|
||||
|
||||
if response.status_code != 200:
|
||||
pytest.skip(f"Could not authenticate: {response.text}")
|
||||
|
||||
data = response.json()
|
||||
return data.get("access_token")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def auth_headers(auth_token):
|
||||
"""Headers with authorization"""
|
||||
return {"Authorization": f"Bearer {auth_token}"}
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def test_company_id(oracle_available):
|
||||
"""
|
||||
Returns company ID 110 (MARIUSM_AUTO schema) for tests.
|
||||
|
||||
This is the only company with full data in TEST environment.
|
||||
Schema: MARIUSM_AUTO
|
||||
"""
|
||||
# Use company 110 - MARIUSM AUTO (only schema with full data in TEST)
|
||||
return 110
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def test_company_data(oracle_available):
|
||||
"""
|
||||
Returns company data for company 110 (MARIUSM_AUTO schema).
|
||||
|
||||
This is the only company with full data in TEST environment.
|
||||
"""
|
||||
# Return static company data for MARIUSM AUTO (company 110)
|
||||
return {
|
||||
'id_firma': 110,
|
||||
'name': 'MARIUSM AUTO',
|
||||
'schema_name': 'MARIUSM_AUTO',
|
||||
'fiscal_code': 'RO1879855',
|
||||
'is_active': True
|
||||
}
|
||||
Reference in New Issue
Block a user