import asyncio from datetime import datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import AsyncMock, patch from bot.handlers.admin.sync_admin import ( _absorb_duplicate_panel_identity, _coerce_panel_telegram_id, _description_matches, _panel_description_for_user, _panel_identity_matches_user, _panel_identity_needs_full_fetch, _panel_identity_payload_with_expiry, _panel_identity_view_for_comparison, _should_update_lifetime_used_traffic, _subscription_update_delta, ) from db.models import Subscription def test_description_match_ignores_whitespace_shape(): assert _description_matches("email@example.com username", "email@example.com\nusername") def test_description_match_accepts_cp1251_mojibake_from_panel(): desired = "user@example.com\nalice\nАлексей\nЧерников" panel_value = "user@example.com\nalice\nÀëåêñåé\n×åðíèêîâ" assert _description_matches(panel_value, desired) def test_description_match_rejects_different_identity_after_mojibake_repair(): desired = "user@example.com\nalice\nАлексей" panel_value = "other@example.com\nalice\nÀëåêñåé" assert not _description_matches(panel_value, desired) def test_panel_telegram_id_is_coerced_to_int(): assert _coerce_panel_telegram_id("12345") == 12345 assert _coerce_panel_telegram_id("") is None def test_panel_description_for_user_excludes_email(): user = SimpleNamespace( email="linked@example.com", username="alice", first_name="Alice", last_name="Smith", ) assert _panel_description_for_user(user) == "alice\nAlice\nSmith" def test_panel_identity_match_accepts_list_description_without_email(): user = SimpleNamespace( email="linked@example.com", telegram_id=42, username="alice", first_name="Alice", last_name=None, ) panel_user = { "description": "alice\nAlice", "email": "linked@example.com", "telegramId": 42, } assert _panel_identity_matches_user( panel_user, user, _panel_description_for_user(user), ) def test_panel_identity_payload_with_expiry_keeps_email_out_of_description(): expire_at = datetime(2026, 6, 1, 12, 0, tzinfo=timezone.utc) user = SimpleNamespace( email="linked@example.com", telegram_id=42, username="alice", first_name="Alice", last_name=None, ) payload = _panel_identity_payload_with_expiry(user, expire_at=expire_at) assert payload["description"] == "alice\nAlice" assert payload["email"] == "linked@example.com" assert payload["telegramId"] == 42 def test_panel_identity_match_treats_missing_list_email_as_unknown(): user = SimpleNamespace( email="linked@example.com", telegram_id=42, ) panel_user = { "description": "linked@example.com\nalice", "telegramId": 42, } assert _panel_identity_matches_user( panel_user, user, "linked@example.com\nalice", ) def test_panel_identity_match_treats_missing_full_email_as_mismatch(): user = SimpleNamespace( email="linked@example.com", telegram_id=42, ) panel_user = { "description": "linked@example.com\nalice", "telegramId": 42, } assert not _panel_identity_matches_user( panel_user, user, "linked@example.com\nalice", missing_identity_fields_match=False, ) def test_panel_identity_match_rejects_different_returned_email(): user = SimpleNamespace( email="linked@example.com", telegram_id=42, ) panel_user = { "description": "linked@example.com\nalice", "email": "other@example.com", "telegramId": 42, } assert not _panel_identity_matches_user( panel_user, user, "linked@example.com\nalice", ) def test_panel_identity_needs_full_fetch_for_missing_or_blank_identity_fields(): user = SimpleNamespace( email="linked@example.com", telegram_id=42, ) assert _panel_identity_needs_full_fetch({"telegramId": 42}, user) assert _panel_identity_needs_full_fetch({"email": "", "telegramId": 42}, user) assert _panel_identity_needs_full_fetch({"email": "linked@example.com"}, user) assert not _panel_identity_needs_full_fetch( {"email": "linked@example.com", "telegramId": "42"}, user, ) def test_panel_identity_view_fetches_full_user_when_list_email_missing(): panel_service = SimpleNamespace( get_user_by_uuid=AsyncMock( return_value={ "uuid": "panel-1", "description": "linked@example.com\nalice", "email": "linked@example.com", "telegramId": 42, } ) ) user = SimpleNamespace( email="linked@example.com", telegram_id=42, ) panel_user, missing_fields_match = asyncio.run( _panel_identity_view_for_comparison( panel_service, "panel-1", { "uuid": "panel-1", "description": "linked@example.com\nalice", "telegramId": 42, }, user, ) ) assert panel_user["email"] == "linked@example.com" assert not missing_fields_match panel_service.get_user_by_uuid.assert_awaited_once_with("panel-1") def test_subscription_update_delta_skips_unchanged_fields(): end_date = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc) subscription = Subscription( user_id=1, panel_user_uuid="panel-1", panel_subscription_uuid="sub-1", end_date=end_date, is_active=True, status_from_panel="ACTIVE", ) assert ( _subscription_update_delta( subscription, { "user_id": 1, "panel_user_uuid": "panel-1", "end_date": end_date + timedelta(milliseconds=500), "is_active": True, "status_from_panel": "ACTIVE", }, ) == {} ) def test_subscription_update_delta_returns_only_changed_fields(): end_date = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc) subscription = Subscription( user_id=1, panel_user_uuid="panel-1", panel_subscription_uuid="sub-1", end_date=end_date, is_active=True, status_from_panel="ACTIVE", ) assert _subscription_update_delta( subscription, { "user_id": 2, "panel_user_uuid": "panel-1", "end_date": end_date + timedelta(seconds=2), "is_active": False, "status_from_panel": "EXPIRED", }, ) == { "user_id": 2, "end_date": end_date + timedelta(seconds=2), "is_active": False, "status_from_panel": "EXPIRED", } def test_lifetime_traffic_update_waits_for_time_window_for_small_delta(): now = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc) settings = SimpleNamespace( PANEL_SYNC_LIFETIME_TRAFFIC_MIN_INTERVAL_SECONDS=3600, PANEL_SYNC_LIFETIME_TRAFFIC_MIN_DELTA_BYTES=100 * 1024 * 1024, ) user = SimpleNamespace( lifetime_used_traffic_bytes=10 * 1024 * 1024, lifetime_used_traffic_synced_at=now - timedelta(minutes=15), ) assert not _should_update_lifetime_used_traffic( user, 11 * 1024 * 1024, now=now, settings=settings, ) assert _should_update_lifetime_used_traffic( user, 11 * 1024 * 1024, now=now + timedelta(hours=1), settings=settings, ) def test_lifetime_traffic_update_allows_large_delta_and_skips_duplicate_panel_identity(): now = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc) settings = SimpleNamespace( PANEL_SYNC_LIFETIME_TRAFFIC_MIN_INTERVAL_SECONDS=3600, PANEL_SYNC_LIFETIME_TRAFFIC_MIN_DELTA_BYTES=100 * 1024 * 1024, ) user = SimpleNamespace( lifetime_used_traffic_bytes=10 * 1024 * 1024, lifetime_used_traffic_synced_at=now, ) assert _should_update_lifetime_used_traffic( user, 200 * 1024 * 1024, now=now, settings=settings, ) assert not _should_update_lifetime_used_traffic( user, 0, now=now + timedelta(hours=2), settings=settings, is_duplicate_panel_identity=True, ) def test_absorb_duplicate_panel_identity_extends_kept_user_and_deletes_duplicate(): now = datetime.now(timezone.utc) target_sub = SimpleNamespace( subscription_id=10, user_id=42, panel_user_uuid="panel-keep", panel_subscription_uuid="sub-keep", end_date=now - timedelta(days=2), is_active=False, status_from_panel="EXPIRED", ) duplicate_sub = SimpleNamespace( subscription_id=11, user_id=42, panel_user_uuid="panel-duplicate", panel_subscription_uuid="sub-duplicate", end_date=now + timedelta(days=30), is_active=True, skip_notifications=False, status_from_panel="ACTIVE", ) panel_service = SimpleNamespace( update_user_details_on_panel=AsyncMock(return_value={"uuid": "panel-keep"}), delete_user_from_panel=AsyncMock(return_value=True), ) session = SimpleNamespace(execute=AsyncMock()) settings = SimpleNamespace(user_traffic_limit_bytes=0) user = SimpleNamespace( user_id=42, panel_user_uuid="panel-keep", telegram_id=969808056, email="paid@example.com", username="alice", first_name="Alice", last_name=None, ) async def update_subscription(_session, subscription_id, update_data): sub = target_sub if subscription_id == target_sub.subscription_id else duplicate_sub for key, value in update_data.items(): setattr(sub, key, value) return sub with patch( "bot.handlers.admin.sync_admin.subscription_dal.update_subscription", AsyncMock(side_effect=update_subscription), ): result = asyncio.run( _absorb_duplicate_panel_identity( session, panel_service=panel_service, existing_user=user, keep_panel_uuid="panel-keep", keep_panel_user={ "uuid": "panel-keep", "subscriptionUuid": "sub-keep", "status": "EXPIRED", "expireAt": (now - timedelta(days=2)).isoformat(), }, duplicate_panel_user={ "uuid": "panel-duplicate", "subscriptionUuid": "sub-duplicate", "telegramId": 969808056, "status": "ACTIVE", "expireAt": (now + timedelta(days=30)).isoformat(), }, settings=settings, subscriptions_by_panel_uuid={ "sub-keep": target_sub, "sub-duplicate": duplicate_sub, }, active_subscriptions_by_user_panel={}, ) ) assert result["resolved"] assert result["subscriptions_updated"] == 2 assert target_sub.is_active assert target_sub.status_from_panel == "ACTIVE_EXTENDED_BY_PANEL_DUPLICATE_MERGE" assert target_sub.panel_user_uuid == "panel-keep" assert target_sub.end_date > now + timedelta(days=29) assert not duplicate_sub.is_active assert duplicate_sub.skip_notifications assert duplicate_sub.status_from_panel == "MERGED_PANEL_DUPLICATE" panel_service.update_user_details_on_panel.assert_awaited_once() update_uuid, update_payload = panel_service.update_user_details_on_panel.await_args.args[:2] assert update_uuid == "panel-keep" assert update_payload["status"] == "ACTIVE" assert update_payload["telegramId"] == 969808056 panel_service.delete_user_from_panel.assert_awaited_once_with( "panel-duplicate", log_response=False, )