diff --git a/tests/README.md b/tests/README.md new file mode 100644 index 0000000..f1b83d5 --- /dev/null +++ b/tests/README.md @@ -0,0 +1,29 @@ +# Tests for pyTwitchAPI + +This directory contains unit tests for the pyTwitchAPI library. + +## Running Tests + +Install test dependencies: +```bash +pip install pytest pytest-asyncio +``` + +Run all tests: +```bash +pytest tests/ +``` + +Run with verbose output: +```bash +pytest tests/ -v +``` + +Run specific test file: +```bash +pytest tests/test_chat_vip_status.py -v +``` + +## Test Coverage + +- `test_chat_vip_status.py` - VIP status detection and caching in Chat class diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_chat_vip_status.py b/tests/test_chat_vip_status.py new file mode 100644 index 0000000..8a0ec16 --- /dev/null +++ b/tests/test_chat_vip_status.py @@ -0,0 +1,205 @@ +""" +Unit tests for VIP status detection in Chat class. + +Tests the new is_vip() method and _handle_user_state() VIP caching logic. +VIP status is detected from IRC USERSTATE badges (not legacy tags). +""" + +import pytest +from unittest.mock import Mock +from twitchAPI.chat import Chat +from twitchAPI.type import ChatRoom + + +class TestVIPStatusDetection: + """Test VIP status detection and caching.""" + + @pytest.fixture + def chat_instance(self): + """Create a Chat instance for testing (bypass async init).""" + # Mock the Twitch instance + mock_twitch = Mock() + mock_twitch.has_required_auth = Mock(return_value=True) + + # Create Chat instance (calls __init__ directly) + chat = Chat.__new__(Chat) + chat.twitch = mock_twitch + chat.username = "testbot" + chat._mod_status_cache = {} + chat._subscriber_status_cache = {} + chat._vip_status_cache = {} + chat.logger = Mock() + + return chat + + def test_vip_status_detected_from_badges(self, chat_instance): + """Test that VIP status is correctly detected from badges dict.""" + parsed = { + 'tags': { + 'badges': {'vip': '1'}, # VIP badge present + 'mod': '0', + 'subscriber': '0', + 'vip': '0' # Legacy tag (always 0, ignored) + }, + 'command': {'channel': '#testchannel'} + } + + # Synchronously test (no await needed for internal method) + import asyncio + asyncio.run(chat_instance._handle_user_state(parsed)) + + assert chat_instance.is_vip('testchannel') is True + assert chat_instance.is_mod('testchannel') is False + + def test_non_vip_user(self, chat_instance): + """Test that non-VIP users are correctly identified.""" + parsed = { + 'tags': { + 'badges': {}, # No badges + 'mod': '0', + 'subscriber': '0', + 'vip': '0' + }, + 'command': {'channel': '#testchannel'} + } + + import asyncio + asyncio.run(chat_instance._handle_user_state(parsed)) + + assert chat_instance.is_vip('testchannel') is False + + def test_broadcaster_is_not_vip(self, chat_instance): + """Test that broadcaster is mod but NOT VIP.""" + parsed = { + 'tags': { + 'badges': {'broadcaster': '1'}, # Broadcaster badge + 'mod': '0', + 'subscriber': '0', + 'vip': '0' + }, + 'command': {'channel': '#testchannel'} + } + + import asyncio + asyncio.run(chat_instance._handle_user_state(parsed)) + + assert chat_instance.is_vip('testchannel') is False + assert chat_instance.is_mod('testchannel') is True + + def test_moderator_is_not_vip(self, chat_instance): + """Test that moderator is mod but NOT VIP.""" + parsed = { + 'tags': { + 'badges': {'moderator': '1'}, # Moderator badge + 'mod': '1', + 'subscriber': '0', + 'vip': '0' + }, + 'command': {'channel': '#testchannel'} + } + + import asyncio + asyncio.run(chat_instance._handle_user_state(parsed)) + + assert chat_instance.is_vip('testchannel') is False + assert chat_instance.is_mod('testchannel') is True + + def test_vip_and_subscriber(self, chat_instance): + """Test user who is both VIP and subscriber.""" + parsed = { + 'tags': { + 'badges': {'vip': '1', 'subscriber': '12'}, + 'mod': '0', + 'subscriber': '1', + 'vip': '0' + }, + 'command': {'channel': '#testchannel'} + } + + import asyncio + asyncio.run(chat_instance._handle_user_state(parsed)) + + assert chat_instance.is_vip('testchannel') is True + assert chat_instance.is_subscriber('testchannel') is True + assert chat_instance.is_mod('testchannel') is False + + def test_vip_cache_per_channel(self, chat_instance): + """Test that VIP status is cached separately per channel.""" + # User is VIP in channel1 + parsed_vip = { + 'tags': { + 'badges': {'vip': '1'}, + 'mod': '0', + 'subscriber': '0', + 'vip': '0' + }, + 'command': {'channel': '#channel1'} + } + + # User is NOT VIP in channel2 + parsed_non_vip = { + 'tags': { + 'badges': {}, + 'mod': '0', + 'subscriber': '0', + 'vip': '0' + }, + 'command': {'channel': '#channel2'} + } + + import asyncio + asyncio.run(chat_instance._handle_user_state(parsed_vip)) + asyncio.run(chat_instance._handle_user_state(parsed_non_vip)) + + assert chat_instance.is_vip('channel1') is True + assert chat_instance.is_vip('channel2') is False + + def test_is_vip_with_chatroom_object(self, chat_instance): + """Test is_vip() accepts ChatRoom object as parameter.""" + parsed = { + 'tags': { + 'badges': {'vip': '1'}, + 'mod': '0', + 'subscriber': '0', + 'vip': '0' + }, + 'command': {'channel': '#testchannel'} + } + + import asyncio + asyncio.run(chat_instance._handle_user_state(parsed)) + + # Test with ChatRoom object + room = ChatRoom(name='testchannel', is_emote_only=False, is_subs_only=False, + is_followers_only=False, is_unique_only=False, + follower_only_delay=-1, room_id='12345', slow=0) + + assert chat_instance.is_vip(room) is True + + def test_is_vip_raises_on_empty_room(self, chat_instance): + """Test that is_vip() raises ValueError on empty room name.""" + with pytest.raises(ValueError, match='please specify a room'): + chat_instance.is_vip('') + + def test_legacy_vip_tag_ignored(self, chat_instance): + """Test that legacy vip='0' tag is ignored (only badges matter).""" + # Even though vip='0' in tags, badges has VIP + parsed = { + 'tags': { + 'badges': {'vip': '1'}, + 'mod': '0', + 'subscriber': '0', + 'vip': '0' # Legacy tag says NOT VIP (ignored) + }, + 'command': {'channel': '#testchannel'} + } + + import asyncio + asyncio.run(chat_instance._handle_user_state(parsed)) + + # Should detect VIP from badges, not legacy tag + assert chat_instance.is_vip('testchannel') is True + + +if __name__ == '__main__': + pytest.main([__file__, '-v']) diff --git a/twitchAPI/chat/__init__.py b/twitchAPI/chat/__init__.py index 9dbb69b..bd47867 100644 --- a/twitchAPI/chat/__init__.py +++ b/twitchAPI/chat/__init__.py @@ -627,6 +627,7 @@ def __init__(self, """Time in seconds till a channel join attempt times out""" self._mod_status_cache = {} self._subscriber_status_cache = {} + self._vip_status_cache = {} self._channel_command_prefix = {} self._command_middleware: List['BaseCommandMiddleware'] = [] self._command_specific_middleware: Dict[str, List['BaseCommandMiddleware']] = {} @@ -1077,15 +1078,27 @@ async def _handle_room_state(self, parsed: dict): async def _handle_user_state(self, parsed: dict): self.logger.debug('got user state event') - is_broadcaster = False - if parsed['tags'].get('badges') is not None: - is_broadcaster = parsed['tags']['badges'].get('broadcaster') is not None - self._mod_status_cache[parsed['command']['channel'][1:]] = 'mod' if parsed['tags']['mod'] == '1' or is_broadcaster else 'user' - self._subscriber_status_cache[parsed['command']['channel'][1:]] = 'sub' if parsed['tags']['subscriber'] == '1' else 'non-sub' - - async def _handle_ping(self, parsed: dict): - self.logger.debug('got PING') - await self._send_message('PONG ' + parsed['parameters']) + + # Parse user badges (source of truth for user status) + # Note: Legacy tags (mod='1', subscriber='1') exist for backwards compatibility, + # but badges dict is the authoritative source. VIP status is only available in badges. + badges = parsed['tags'].get('badges', {}) + channel = parsed['command']['channel'][1:] + + # MOD STATUS: broadcaster OR moderator badge + is_mod = badges.get('broadcaster') is not None or badges.get('moderator') is not None + self._mod_status_cache[channel] = 'mod' if is_mod else 'user' + + # VIP STATUS: vip badge (not available in legacy tags) + is_vip = badges.get('vip') is not None + self._vip_status_cache[channel] = 'vip' if is_vip else 'non-vip' + + # SUB STATUS: subscriber badge + is_sub = badges.get('subscriber') is not None + self._subscriber_status_cache[channel] = 'sub' if is_sub else 'non-sub' + async def _handle_ping(self, parsed: dict): + self.logger.debug('got PING') + await self._send_message('PONG ' + parsed['parameters']) # noinspection PyUnusedLocal async def _handle_ready(self, parsed: dict): @@ -1294,6 +1307,20 @@ def is_subscriber(self, room: CHATROOM_TYPE) -> bool: room = room[1:] return self._subscriber_status_cache.get(room.lower(), 'user') == 'sub' + def is_vip(self, room: CHATROOM_TYPE) -> bool: + """Check if chat bot is a VIP in a channel + + :param room: The chat room you want to check if bot is a VIP in. + This can either be a instance of :const:`~twitchAPI.type.ChatRoom` or a string with the room name (either with leading # or without) + :return: Returns True if chat bot is a VIP """ + if isinstance(room, ChatRoom): + room = room.name + if room is None or len(room) == 0: + raise ValueError('please specify a room') + if room[0] == '#': + room = room[1:] + return self._vip_status_cache.get(room.lower(), 'non-vip') == 'vip' + def is_in_room(self, room: CHATROOM_TYPE) -> bool: """Check if the bot is currently in the given chat room