invalidate-new-follower

tests/test_follower_counters.py

@@ -0,0 +1,162 @@
+"""
+Тест обновления счетчиков подписчиков в статистике авторов
+"""
+from __future__ import annotations
+
+import asyncio
+import time
+import pytest
+
+from resolvers.author import get_authors_with_stats, invalidate_authors_cache
+from resolvers.stat import get_followers_count
+from orm.author import Author, AuthorFollower
+from storage.db import local_session
+from storage.redis import redis
+
+
+@pytest.mark.asyncio
+async def test_follower_counters_update():
+    """
+    Тест обновления счетчиков подписчиков после инвалидации кеша
+    """
+    # Создаем тестовых пользователей
+    with local_session() as session:
+        # Создаем автора, у которого будут подписчики
+        target_author = Author(
+            name="Popular Author", 
+            slug=f"popular-author-{int(time.time())}",
+            email=f"popular-{int(time.time())}@test.com"
+        )
+        session.add(target_author)
+        
+        # Создаем подписчиков
+        follower1 = Author(
+            name="Follower 1",
+            slug=f"follower-1-{int(time.time())}",
+            email=f"follower-1-{int(time.time())}@test.com"
+        )
+        session.add(follower1)
+        
+        follower2 = Author(
+            name="Follower 2",
+            slug=f"follower-2-{int(time.time())}",
+            email=f"follower-2-{int(time.time())}@test.com"
+        )
+        session.add(follower2)
+        
+        session.commit()
+        
+        target_author_id = target_author.id
+        follower1_id = follower1.id
+        follower2_id = follower2.id
+
+    try:
+        # 1. Очищаем кеш авторов
+        await invalidate_authors_cache()
+        
+        # 2. Получаем начальную статистику (должно быть 0 подписчиков)
+        initial_stats = await get_authors_with_stats(limit=100, offset=0)
+        target_stats = next((author for author in initial_stats if author["id"] == target_author_id), None)
+        
+        assert target_stats is not None, "Автор должен быть найден в статистике"
+        assert target_stats["stat"]["followers"] == 0, "Изначально подписчиков быть не должно"
+        
+        # 3. Добавляем подписчика в БД
+        with local_session() as session:
+            subscription1 = AuthorFollower(
+                follower=follower1_id,
+                following=target_author_id
+            )
+            session.add(subscription1)
+            session.commit()
+        
+        # 4. Проверяем, что прямой запрос к БД возвращает 1 подписчика
+        direct_count = get_followers_count("author", target_author_id)
+        assert direct_count == 1, "Прямой запрос должен вернуть 1 подписчика"
+        
+        # 5. Инвалидируем кеш (имитируя логику follow)
+        await invalidate_authors_cache(target_author_id)
+        
+        # 6. Получаем обновленную статистику
+        updated_stats = await get_authors_with_stats(limit=100, offset=0)
+        updated_target_stats = next((author for author in updated_stats if author["id"] == target_author_id), None)
+        
+        assert updated_target_stats is not None, "Автор должен быть найден после обновления"
+        assert updated_target_stats["stat"]["followers"] == 1, "После инвалидации кеша должен быть 1 подписчик"
+        
+        # 7. Добавляем второго подписчика
+        with local_session() as session:
+            subscription2 = AuthorFollower(
+                follower=follower2_id,
+                following=target_author_id
+            )
+            session.add(subscription2)
+            session.commit()
+        
+        # 8. Инвалидируем кеш снова
+        await invalidate_authors_cache(target_author_id)
+        
+        # 9. Проверяем финальную статистику
+        final_stats = await get_authors_with_stats(limit=100, offset=0)
+        final_target_stats = next((author for author in final_stats if author["id"] == target_author_id), None)
+        
+        assert final_target_stats is not None, "Автор должен быть найден в финальной статистике"
+        assert final_target_stats["stat"]["followers"] == 2, "В финальной статистике должно быть 2 подписчика"
+        
+        print("✅ Тест обновления счетчиков подписчиков прошел успешно!")
+        
+    finally:
+        # Очистка тестовых данных
+        with local_session() as session:
+            session.query(AuthorFollower).filter(
+                AuthorFollower.following == target_author_id
+            ).delete()
+            session.query(Author).filter(Author.id.in_([target_author_id, follower1_id, follower2_id])).delete()
+            session.commit()
+        
+        # Очищаем кеш
+        await invalidate_authors_cache()
+
+
+@pytest.mark.asyncio
+async def test_follower_counter_edge_cases():
+    """
+    Тест крайних случаев для счетчиков подписчиков
+    """
+    with local_session() as session:
+        author = Author(
+            name="Edge Case Author", 
+            slug=f"edge-case-{int(time.time())}",
+            email=f"edge-case-{int(time.time())}@test.com"
+        )
+        session.add(author)
+        session.commit()
+        author_id = author.id
+
+    try:
+        # Тест 1: Автор без подписчиков
+        await invalidate_authors_cache()
+        stats = await get_authors_with_stats(limit=100, offset=0)
+        author_stats = next((a for a in stats if a["id"] == author_id), None)
+        
+        if author_stats:  # Автор может не попасть в топ-100, это нормально
+            assert author_stats["stat"]["followers"] == 0, "Автор без подписчиков должен иметь 0 в счетчике"
+        
+        # Тест 2: Проверяем прямой запрос для несуществующего автора
+        nonexistent_count = get_followers_count("author", 999999)
+        assert nonexistent_count == 0, "Несуществующий автор должен иметь 0 подписчиков"
+        
+        print("✅ Тест крайних случаев прошел успешно!")
+        
+    finally:
+        with local_session() as session:
+            session.query(Author).filter(Author.id == author_id).delete()
+            session.commit()
+        
+        await invalidate_authors_cache()
+
+
+if __name__ == "__main__":
+    asyncio.run(test_follower_counters_update())
+    asyncio.run(test_follower_counter_edge_cases())
+    print("🎯 Все тесты счетчиков подписчиков прошли успешно!")