Marius Mutu
bd41a3406e
- Refactor telegram bot tests (remove old, add new real flow tests) - Add conftest.py for telegram bot test fixtures - Update validate.md command 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
122 lines
3.4 KiB
Python
122 lines
3.4 KiB
Python
# reports-app/telegram-bot/tests/conftest.py
|
|
"""
|
|
Pytest fixtures for telegram bot tests.
|
|
Includes fixtures for both pure tests and integration tests with real backend.
|
|
"""
|
|
import pytest
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
|
|
# Add paths for imports
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
# Get test telegram ID from env or use default
|
|
TEST_TELEGRAM_ID = int(os.getenv("TEST_TELEGRAM_ID", "123456789"))
|
|
BACKEND_URL = os.getenv("BACKEND_URL", "http://localhost:8001")
|
|
|
|
|
|
@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 backend_available():
|
|
"""Check if backend is available, skip integration tests if not"""
|
|
try:
|
|
from app.api.client import get_backend_client
|
|
async with get_backend_client() as client:
|
|
is_healthy = await client.health_check()
|
|
if is_healthy:
|
|
return True
|
|
except Exception as e:
|
|
pytest.skip(f"Backend not available: {e}")
|
|
return False
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def backend_client(backend_available):
|
|
"""Real backend client for integration tests"""
|
|
from app.api.client import get_backend_client
|
|
|
|
async with get_backend_client() as client:
|
|
yield client
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def auth_data(backend_available):
|
|
"""Get auth data for test user from SQLite database"""
|
|
import aiosqlite
|
|
from app.db.database import DB_PATH
|
|
|
|
try:
|
|
async with aiosqlite.connect(DB_PATH) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
cursor = await db.execute("""
|
|
SELECT
|
|
telegram_user_id,
|
|
oracle_username,
|
|
jwt_token,
|
|
jwt_refresh_token,
|
|
linked_at
|
|
FROM telegram_users
|
|
WHERE oracle_username IS NOT NULL
|
|
AND jwt_token IS NOT NULL
|
|
ORDER BY linked_at DESC
|
|
LIMIT 1
|
|
""")
|
|
|
|
row = await cursor.fetchone()
|
|
if row:
|
|
return dict(row)
|
|
|
|
pytest.skip("No linked user found in database")
|
|
|
|
except Exception as e:
|
|
pytest.skip(f"Could not get auth data: {e}")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def jwt_token(auth_data):
|
|
"""Get JWT token from auth data"""
|
|
return auth_data.get("jwt_token")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def test_telegram_id(auth_data):
|
|
"""Get telegram user ID from auth data"""
|
|
return auth_data.get("telegram_user_id")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def test_companies(backend_client, jwt_token):
|
|
"""Get list of companies for test user"""
|
|
companies = await backend_client.get_user_companies(jwt_token=jwt_token)
|
|
|
|
if not companies:
|
|
pytest.skip("No companies available for testing")
|
|
|
|
return companies
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def test_company(test_companies):
|
|
"""Get first company for testing"""
|
|
return test_companies[0]
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def test_company_id(test_company):
|
|
"""Get company ID for testing"""
|
|
return test_company.get("id") or test_company.get("id_firma")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def test_company_name(test_company):
|
|
"""Get company name for testing"""
|
|
return test_company.get("nume_firma") or test_company.get("name")
|