"""Unit tests for US-004: oracle_server_id in telegram_users. Tests cover: - link_user_to_oracle() persists server_id - link_user_to_oracle() works without server_id (backwards compat) - get_user() returns oracle_server_id from stored row - get_user_auth_data() includes server_id in returned dict - Round-trip: link with server_id → auth data returns same server_id """ import asyncio import sys import os import pytest import aiosqlite from datetime import datetime, timedelta from pathlib import Path from unittest.mock import AsyncMock, patch sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../')) # python-telegram-bot is not installed in the test environment; mock it so that # linking.py can be imported (it has `from telegram import User` at top-level). from unittest.mock import MagicMock sys.modules.setdefault('telegram', MagicMock()) import backend.modules.telegram.auth.linking # noqa: F401 — registers module for patch() resolution from backend.modules.telegram.db.operations import ( create_or_update_user, link_user_to_oracle, get_user, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- async def _make_test_db(tmp_path: Path) -> Path: """Create a minimal telegram_users table in a temp SQLite file.""" db_file = tmp_path / "test_app.db" async with aiosqlite.connect(db_file) as db: await db.execute("""CREATE TABLE telegram_users ( telegram_user_id INTEGER PRIMARY KEY, username TEXT, first_name TEXT NOT NULL, last_name TEXT, oracle_username TEXT, oracle_server_id TEXT, jwt_token TEXT, jwt_refresh_token TEXT, token_expires_at TIMESTAMP, linked_at TIMESTAMP, last_active_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, is_active BOOLEAN DEFAULT 1 )""") await db.commit() return db_file async def _insert_user(db_path: Path, telegram_user_id: int): async with aiosqlite.connect(db_path) as db: await db.execute( "INSERT INTO telegram_users (telegram_user_id, first_name) VALUES (?, ?)", (telegram_user_id, "Test") ) await db.commit() # --------------------------------------------------------------------------- # Tests: link_user_to_oracle persists server_id # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_link_user_persists_server_id(tmp_path): db_file = await _make_test_db(tmp_path) await _insert_user(db_file, 12345) with patch("backend.modules.telegram.db.operations.DB_PATH", db_file): result = await link_user_to_oracle( telegram_user_id=12345, oracle_username="TESTUSER", jwt_token="tok", jwt_refresh_token="rtok", token_expires_at=datetime.now() + timedelta(minutes=30), server_id="SERVER1" ) assert result is True async with aiosqlite.connect(db_file) as db: db.row_factory = aiosqlite.Row cursor = await db.execute( "SELECT oracle_server_id FROM telegram_users WHERE telegram_user_id = ?", (12345,) ) row = await cursor.fetchone() assert row["oracle_server_id"] == "SERVER1" @pytest.mark.asyncio async def test_link_user_without_server_id(tmp_path): """Backwards compat: omitting server_id stores NULL.""" db_file = await _make_test_db(tmp_path) await _insert_user(db_file, 22222) with patch("backend.modules.telegram.db.operations.DB_PATH", db_file): result = await link_user_to_oracle( telegram_user_id=22222, oracle_username="TESTUSER2", jwt_token="tok", jwt_refresh_token="rtok", token_expires_at=datetime.now() + timedelta(minutes=30), ) assert result is True async with aiosqlite.connect(db_file) as db: db.row_factory = aiosqlite.Row cursor = await db.execute( "SELECT oracle_server_id FROM telegram_users WHERE telegram_user_id = ?", (22222,) ) row = await cursor.fetchone() assert row["oracle_server_id"] is None # --------------------------------------------------------------------------- # Tests: get_user returns oracle_server_id # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_get_user_returns_oracle_server_id(tmp_path): db_file = await _make_test_db(tmp_path) await _insert_user(db_file, 33333) with patch("backend.modules.telegram.db.operations.DB_PATH", db_file): await link_user_to_oracle( telegram_user_id=33333, oracle_username="TESTUSER3", jwt_token="tok", jwt_refresh_token="rtok", token_expires_at=datetime.now() + timedelta(minutes=30), server_id="PROD_SERVER" ) user = await get_user(33333) assert user is not None assert user["oracle_server_id"] == "PROD_SERVER" # --------------------------------------------------------------------------- # Tests: get_user_auth_data returns server_id (round-trip) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_get_user_auth_data_includes_server_id(tmp_path): """Round-trip: link with server_id → get_user_auth_data returns same server_id.""" db_file = await _make_test_db(tmp_path) await _insert_user(db_file, 44444) expires_at = datetime.now() + timedelta(hours=1) with patch("backend.modules.telegram.db.operations.DB_PATH", db_file): await link_user_to_oracle( telegram_user_id=44444, oracle_username="TESTUSER4", jwt_token="valid_token", jwt_refresh_token="rtoken", token_expires_at=expires_at, server_id="ALPHA_SERVER" ) mock_companies = [{"id": 1, "name": "Test Co"}] mock_backend = AsyncMock() mock_backend.__aenter__ = AsyncMock(return_value=mock_backend) mock_backend.__aexit__ = AsyncMock(return_value=False) mock_backend.get_user_companies = AsyncMock(return_value=mock_companies) with ( patch("backend.modules.telegram.db.operations.DB_PATH", db_file), patch("backend.modules.telegram.auth.linking.get_user") as mock_get_user, patch("backend.modules.telegram.auth.linking.get_backend_client", return_value=mock_backend), ): # Return stored user row (simulating what get_user() would return) mock_get_user.return_value = { "telegram_user_id": 44444, "oracle_username": "TESTUSER4", "oracle_server_id": "ALPHA_SERVER", "jwt_token": "valid_token", "jwt_refresh_token": "rtoken", "token_expires_at": expires_at.isoformat(), } from backend.modules.telegram.auth.linking import get_user_auth_data auth_data = await get_user_auth_data(44444) assert auth_data is not None assert "server_id" in auth_data assert auth_data["server_id"] == "ALPHA_SERVER" assert auth_data["username"] == "TESTUSER4" @pytest.mark.asyncio async def test_get_user_auth_data_server_id_none_when_not_set(tmp_path): """server_id is None in auth data when user was linked without a server_id.""" mock_companies = [] mock_backend = AsyncMock() mock_backend.__aenter__ = AsyncMock(return_value=mock_backend) mock_backend.__aexit__ = AsyncMock(return_value=False) mock_backend.get_user_companies = AsyncMock(return_value=mock_companies) expires_at = datetime.now() + timedelta(hours=1) with ( patch("backend.modules.telegram.auth.linking.get_user") as mock_get_user, patch("backend.modules.telegram.auth.linking.get_backend_client", return_value=mock_backend), ): mock_get_user.return_value = { "telegram_user_id": 55555, "oracle_username": "TESTUSER5", "oracle_server_id": None, "jwt_token": "valid_token", "jwt_refresh_token": "rtoken", "token_expires_at": expires_at.isoformat(), } from backend.modules.telegram.auth.linking import get_user_auth_data auth_data = await get_user_auth_data(55555) assert auth_data is not None assert "server_id" in auth_data assert auth_data["server_id"] is None