chore: Update telegram bot tests and validate command
- 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>
This commit is contained in:
121
reports-app/telegram-bot/tests/conftest.py
Normal file
121
reports-app/telegram-bot/tests/conftest.py
Normal file
@@ -0,0 +1,121 @@
|
||||
# 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")
|
||||
Reference in New Issue
Block a user