From c866c22f45efd60398dd43892cfa50b558f31967 Mon Sep 17 00:00:00 2001 From: CodeBeaverAI Date: Thu, 20 Feb 2025 18:13:04 +0100 Subject: [PATCH 1/4] test: Add coverage improvement test for tests/test_streamtypes.py --- tests/test_streamtypes.py | 479 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 479 insertions(+) create mode 100644 tests/test_streamtypes.py diff --git a/tests/test_streamtypes.py b/tests/test_streamtypes.py new file mode 100644 index 00000000000..c58e1154b5b --- /dev/null +++ b/tests/test_streamtypes.py @@ -0,0 +1,479 @@ +import aiohttp +import asyncio +import contextlib +import discord +import json +import logging +import pytest +import time +import xml.etree.ElementTree as ET +from datetime import ( + datetime, + timedelta, + timezone, +) +from random import ( + choice, +) +from redbot.cogs.streams.streamtypes import ( + APIError, + InvalidTwitchCredentials, + InvalidYoutubeCredentials, + OfflineStream, + PicartoStream, + StreamNotFound, + TWITCH_BASE_URL, + TWITCH_FOLLOWS_ENDPOINT, + TWITCH_ID_ENDPOINT, + TWITCH_STREAMS_ENDPOINT, + TwitchStream, + YOUTUBE_CHANNELS_ENDPOINT, + YOUTUBE_CHANNEL_RSS, + YOUTUBE_SEARCH_ENDPOINT, + YOUTUBE_VIDEOS_ENDPOINT, + YoutubeQuotaExceeded, + YoutubeStream, +) +from redbot.core.utils.chat_formatting import ( + humanize_number, +) +from string import ( + ascii_letters, +) +from typing import ( + ClassVar, + List, + Optional, + Tuple, +) +from unittest.mock import ( + MagicMock, +) + + +@pytest.mark.asyncio +async def test_youtube_no_api_key(): + """ + Test that YoutubeStream.is_online() correctly raises InvalidYoutubeCredentials + when no YouTube API key has been provided. + """ + dummy_bot = MagicMock() + dummy_config = MagicMock() + yt_stream = YoutubeStream( + _bot=dummy_bot, + name="test_channel", + channels=[], + messages=[], + config=dummy_config, + token=None, + ) + with pytest.raises(InvalidYoutubeCredentials): + await yt_stream.is_online() + + +@pytest.mark.asyncio +async def test_youtube_offline_stream(monkeypatch): + """ + Test that YoutubeStream.is_online() correctly raises OfflineStream when the RSS feed + returns no video entries (i.e. the channel is offline). + """ + + class DummyResponseLocal: + + def __init__(self, url): + self.url = url + self.status = 200 + + async def text(self): + return "" + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + pass + + class DummySessionLocal: + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + pass + + def get(self, url, **kwargs): + return DummyResponseLocal(url) + + monkeypatch.setattr( + aiohttp, "ClientSession", lambda *args, **kwargs: DummySessionLocal() + ) + dummy_bot = MagicMock() + dummy_config = MagicMock() + yt_stream = YoutubeStream( + _bot=dummy_bot, + name="dummy_channel", + channels=[], + messages=[], + config=dummy_config, + token={"api_key": "dummy_api_key"}, + id="dummy_channel_id", + ) + with pytest.raises(OfflineStream): + await yt_stream.is_online() + + +class DummyResponse: + """Dummy implementation for aiohttp.ClientResponse.""" + + def __init__(self, url, status, text_data=None, json_data=None, headers=None): + self.url = url + self.status = status + self._text = text_data + self._json = json_data + self.headers = headers or {} + + async def text(self, encoding="utf-8"): + return self._text + + async def json(self, encoding="utf-8"): + return self._json + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + pass + + +class DummySession: + """Dummy implementation for aiohttp.ClientSession for testing purposes.""" + + responses = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + pass + + def get(self, url, **kwargs): + return DummySession.responses.pop(0) + + +@pytest.mark.asyncio +async def test_youtube_live_stream(monkeypatch): + """ + Test that YoutubeStream.is_online() returns a valid embed when the channel is live. + This simulates the scenario where the RSS feed returns one video id, + and the API calls to YOUTUBE_VIDEOS_ENDPOINT return live streaming data. + """ + xml_feed = 'dummy_vid' + live_start = datetime.now(timezone.utc).isoformat() + video_response_data_initial = { + "items": [ + { + "id": "dummy_vid", + "snippet": { + "title": "Test Live Stream", + "channelTitle": "Test Channel", + "thumbnails": {"medium": {"url": "http://example.com/thumb.jpg"}}, + }, + "liveStreamingDetails": {"actualStartTime": live_start}, + } + ] + } + video_response_data_final = { + "items": [ + { + "id": "dummy_vid", + "snippet": { + "title": "Test Live Stream", + "channelTitle": "Test Channel", + "thumbnails": {"medium": {"url": "http://example.com/thumb.jpg"}}, + }, + "liveStreamingDetails": {"actualStartTime": live_start}, + } + ] + } + DummySession.responses = [ + DummyResponse( + url=YOUTUBE_CHANNEL_RSS.format(channel_id="dummy_channel_id"), + status=200, + text_data=xml_feed, + ), + DummyResponse( + url=YOUTUBE_VIDEOS_ENDPOINT, + status=200, + json_data=video_response_data_initial, + ), + DummyResponse( + url=YOUTUBE_VIDEOS_ENDPOINT, status=200, json_data=video_response_data_final + ), + ] + monkeypatch.setattr( + aiohttp, "ClientSession", lambda *args, **kwargs: DummySession() + ) + dummy_bot = MagicMock() + dummy_config = MagicMock() + yt_stream = YoutubeStream( + _bot=dummy_bot, + name="dummy_channel", + channels=[], + messages=[], + config=dummy_config, + token={"api_key": "dummy_api_key"}, + id="dummy_channel_id", + ) + result = await yt_stream.is_online() + embed, is_schedule = result + assert isinstance(embed, discord.Embed) + assert embed.title == "Test Live Stream" + assert embed.url == "https://youtube.com/watch?v=dummy_vid" + assert is_schedule is False + assert embed.author.name == "Test Channel" + + +@pytest.mark.asyncio +async def test_youtube_api_error_in_video_endpoint(monkeypatch): + """ + Test that YoutubeStream.is_online() raises OfflineStream when the video details API call + returns an API error. This simulates a case where the API call returns an unexpected error, + and although the error is caught internally, no valid live stream data is obtained. + """ + xml_feed = 'error_vid' + api_error_data = { + "error": { + "code": 500, + "errors": [{"reason": "internalError", "message": "Internal error"}], + "message": "Internal error encountered.", + } + } + DummySession.responses = [ + DummyResponse( + url=YOUTUBE_CHANNEL_RSS.format(channel_id="dummy_channel_id"), + status=200, + text_data=xml_feed, + ), + DummyResponse( + url=YOUTUBE_VIDEOS_ENDPOINT, status=200, json_data=api_error_data + ), + ] + monkeypatch.setattr( + aiohttp, "ClientSession", lambda *args, **kwargs: DummySession() + ) + dummy_bot = MagicMock() + dummy_config = MagicMock() + yt_stream = YoutubeStream( + _bot=dummy_bot, + name="dummy_channel", + channels=[], + messages=[], + config=dummy_config, + token={"api_key": "dummy_api_key"}, + id="dummy_channel_id", + ) + with pytest.raises(OfflineStream): + await yt_stream.is_online() + + +@pytest.mark.asyncio +async def test_picarto_live_stream(monkeypatch): + """ + Test that PicartoStream.is_online() returns a valid embed when the Picarto API reports a live channel. + The test simulates a live Picarto stream response and verifies that the returned embed includes the correct + title, URL, author, and footer information (including category and tags). + """ + picarto_data = { + "online": True, + "avatar": "http://example.com/avatar.png", + "name": "picarto_channel", + "title": "Picarto Live!", + "thumbnails": {"web": "http://example.com/thumb.png"}, + "followers": 1234, + "viewers_total": 4321, + "tags": ["tag1", "tag2"], + "adult": False, + "category": "Art", + } + dummy_text = json.dumps(picarto_data) + dummy_response = DummyResponse( + url="https://api.picarto.tv/api/v1/channel/name/picarto_channel", + status=200, + text_data=dummy_text, + ) + DummySession.responses = [dummy_response] + monkeypatch.setattr( + aiohttp, "ClientSession", lambda *args, **kwargs: DummySession() + ) + dummy_bot = MagicMock() + picarto_stream = PicartoStream( + _bot=dummy_bot, name="picarto_channel", channels=[], messages=[] + ) + embed = await picarto_stream.is_online() + assert isinstance(embed, discord.Embed) + assert embed.title == "Picarto Live!" + assert embed.url == "https://picarto.tv/picarto_channel" + assert embed.author.name == "picarto_channel" + footer_text = embed.footer.text + assert "Category: Art" in footer_text + assert "Tags: tag1, tag2" in footer_text + + +@pytest.mark.asyncio +async def test_twitch_live_stream(monkeypatch): + """ + Test that TwitchStream.is_online() returns a valid embed when the Twitch API indicates a live stream. + This test simulates responses for: + - The user profile endpoint (TWITCH_ID_ENDPOINT) to retrieve channel ID and profile image. + - The streams endpoint (TWITCH_STREAMS_ENDPOINT) to indicate that the stream is live. + - The followers endpoint (TWITCH_FOLLOWS_ENDPOINT) to return a follower count. + It then verifies that the returned embed has the expected title, author, fields, and images. + """ + profile_data = { + "data": [ + { + "id": "dummy_id", + "login": "dummy_login", + "profile_image_url": "http://example.com/profile.png", + } + ] + } + stream_data = { + "data": [ + { + "user_name": "dummy_stream", + "game_name": "dummy_game", + "thumbnail_url": "http://example.com/thumbnail.jpg", + "title": "Dummy Twitch Live Stream", + "type": "live", + "viewer_count": 100, + } + ] + } + follows_data = {"total": 2000} + DummySession.responses = [ + DummyResponse(url=TWITCH_ID_ENDPOINT, status=200, json_data=profile_data), + DummyResponse(url=TWITCH_STREAMS_ENDPOINT, status=200, json_data=stream_data), + DummyResponse(url=TWITCH_FOLLOWS_ENDPOINT, status=200, json_data=follows_data), + ] + monkeypatch.setattr( + aiohttp, "ClientSession", lambda *args, **kwargs: DummySession() + ) + dummy_bot = MagicMock() + twitch_stream = TwitchStream( + _bot=dummy_bot, + name="dummy_login", + channels=[], + messages=[], + token="dummy_client_id", + ) + result = await twitch_stream.is_online() + embed, is_rerun = result + assert isinstance(embed, discord.Embed) + assert embed.title == "Dummy Twitch Live Stream" + assert embed.url == "https://www.twitch.tv/dummy_login" + assert embed.author.name == "dummy_stream" + field_names = [field.name for field in embed.fields] + assert "Followers" in field_names + assert "Total views" in field_names + fields_dict = {field.name: field.value for field in embed.fields} + assert fields_dict["Followers"] == humanize_number(2000) + assert fields_dict["Total views"] == humanize_number(100) + assert embed.thumbnail.url == "http://example.com/profile.png" + assert embed.image.url.startswith("http://example.com/thumbnail.jpg") + assert is_rerun is False + + +@pytest.mark.asyncio +async def test_youtube_scheduled_stream(monkeypatch): + """ + Test that YoutubeStream.is_online() returns a valid embed for a scheduled stream. + This simulates a YouTube channel with a scheduled stream (scheduledStartTime set in the future) + and verifies that the returned embed correctly displays that the stream will start in the future, + sets the embed timestamp accordingly, and flags the stream as scheduled. + """ + scheduled_time = datetime.now(timezone.utc) + timedelta(hours=1) + scheduled_time_iso = scheduled_time.isoformat() + xml_feed = 'dummy_scheduled_vid' + video_response_data_initial = { + "items": [ + { + "id": "dummy_scheduled_vid", + "snippet": { + "title": "Test Scheduled Stream", + "channelTitle": "Scheduled Channel", + "thumbnails": { + "medium": {"url": "http://example.com/sched_thumb.jpg"} + }, + }, + "liveStreamingDetails": {"scheduledStartTime": scheduled_time_iso}, + } + ] + } + video_response_data_final = video_response_data_initial.copy() + DummySession.responses = [ + DummyResponse( + url=YOUTUBE_CHANNEL_RSS.format(channel_id="dummy_channel_id"), + status=200, + text_data=xml_feed, + ), + DummyResponse( + url=YOUTUBE_VIDEOS_ENDPOINT, + status=200, + json_data=video_response_data_initial, + ), + DummyResponse( + url=YOUTUBE_VIDEOS_ENDPOINT, status=200, json_data=video_response_data_final + ), + ] + monkeypatch.setattr( + aiohttp, "ClientSession", lambda *args, **kwargs: DummySession() + ) + dummy_bot = MagicMock() + dummy_config = MagicMock() + yt_stream = YoutubeStream( + _bot=dummy_bot, + name="dummy_channel", + channels=[], + messages=[], + config=dummy_config, + token={"api_key": "dummy_api_key"}, + id="dummy_channel_id", + ) + embed, is_schedule = await yt_stream.is_online() + assert isinstance(embed, discord.Embed) + assert embed.title == "Test Scheduled Stream" + assert embed.url == "https://youtube.com/watch?v=dummy_scheduled_vid" + assert "will start" in embed.description + delta = abs((embed.timestamp - scheduled_time).total_seconds()) + assert delta < 1 + assert embed.author.name == "Scheduled Channel" + assert is_schedule is True + + +@pytest.mark.asyncio +async def test_twitch_invalid_credentials(monkeypatch): + """ + Test that TwitchStream.is_online() raises InvalidTwitchCredentials when the Twitch streams endpoint + returns a 400 error code. This simulates invalid credentials for Twitch. + """ + DummySession.responses = [ + DummyResponse(url=TWITCH_STREAMS_ENDPOINT, status=400, json_data={}) + ] + monkeypatch.setattr( + aiohttp, "ClientSession", lambda *args, **kwargs: DummySession() + ) + from unittest.mock import MagicMock + + dummy_bot = MagicMock() + twitch_stream = TwitchStream( + _bot=dummy_bot, + name="dummy_login", + channels=[], + messages=[], + token="dummy_client_id", + id="dummy_id", + bearer="dummy_bearer", + ) + with pytest.raises(InvalidTwitchCredentials): + await twitch_stream.is_online() From 18d76162393245ae18daf90da3971a6c63a37499 Mon Sep 17 00:00:00 2001 From: CodeBeaverAI Date: Thu, 20 Feb 2025 18:13:06 +0100 Subject: [PATCH 2/4] test: Add coverage improvement test for tests/test__cli.py --- tests/test__cli.py | 256 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 256 insertions(+) create mode 100644 tests/test__cli.py diff --git a/tests/test__cli.py b/tests/test__cli.py new file mode 100644 index 00000000000..bcab2d47b6a --- /dev/null +++ b/tests/test__cli.py @@ -0,0 +1,256 @@ +import argparse +import asyncio +import discord +import logging +import pytest +import sys +from discord import ( + __version__, +) +from enum import ( + IntEnum, +) +from redbot.core._cli import ( + ExitCodes, + confirm, + interactive_config, + message_cache_size_int, + non_negative_int, + parse_cli_flags, +) +from redbot.core.utils._internal_utils import ( + cli_level_to_log_level, +) +from typing import ( + Optional, +) + + +def test_non_negative_and_message_cache_int(): + """ + Test non_negative_int and message_cache_size_int functions. + + This test verifies that: + - non_negative_int properly converts valid string input to integer. + - non_negative_int raises an error if a negative number is provided. + - message_cache_size_int returns the proper integer if the provided string is 1000 or above. + - message_cache_size_int raises an error if the provided number is less than 1000. + """ + assert non_negative_int("10") == 10 + with pytest.raises(argparse.ArgumentTypeError) as exc_neg: + non_negative_int("-5") + assert "non-negative" in str(exc_neg.value).lower() + assert message_cache_size_int("1500") == 1500 + with pytest.raises(argparse.ArgumentTypeError) as exc_cache: + message_cache_size_int("500") + assert "greater than or equal to 1000" in str(exc_cache.value) + + +def test_confirm(monkeypatch): + """ + Test the confirm function for various scenarios: + - Returning True when user enters "yes". + - Returning False when user enters "no". + - Using the default value when input is empty. + - Re-prompting when input is invalid. + - Handling KeyboardInterrupt and EOFError by exiting with appropriate codes. + """ + monkeypatch.setattr("builtins.input", lambda prompt: "yes") + assert confirm("Test?") is True + monkeypatch.setattr("builtins.input", lambda prompt: "no") + assert confirm("Test?") is False + monkeypatch.setattr("builtins.input", lambda prompt: "") + assert confirm("Test?", default=True) is True + monkeypatch.setattr("builtins.input", lambda prompt: "") + assert confirm("Test?", default=False) is False + responses = iter(["maybe", "y"]) + monkeypatch.setattr("builtins.input", lambda prompt: next(responses)) + assert confirm("Test?") is True + + def raise_keyboard(prompt): + raise KeyboardInterrupt + + monkeypatch.setattr("builtins.input", raise_keyboard) + with pytest.raises(SystemExit) as excinfo: + confirm("Test?") + assert excinfo.value.code == ExitCodes.SHUTDOWN + + def raise_eof(prompt): + raise EOFError + + monkeypatch.setattr("builtins.input", raise_eof) + with pytest.raises(SystemExit) as excinfo: + confirm("Test?") + assert excinfo.value.code == ExitCodes.INVALID_CLI_USAGE + + +def test_parse_cli_flags_parses_args(): + """ + Test parse_cli_flags function for correct parsing of CLI flags, including: + - Setting the positional instance name. + - Accumulation of verbosity flags and correct logging level conversion. + - Sorting of provided prefixes in reverse order. + """ + test_args = ["testinstance", "--verbose", "-v", "--prefix", "!", "--prefix", "?"] + args = parse_cli_flags(test_args) + assert args.instance_name == "testinstance" + expected_log_level = cli_level_to_log_level(2) + assert args.logging_level == expected_log_level + expected_prefixes = sorted(["!", "?"], reverse=True) + assert args.prefix == expected_prefixes + + +class DummyValue: + + def __init__(self): + self.value = None + + async def set(self, value): + self.value = value + + +class DummyConfig: + + def __init__(self): + self.token = DummyValue() + self.prefix = DummyValue() + + +class DummyRed: + + def __init__(self): + self._config = DummyConfig() + + +@pytest.mark.asyncio +async def test_interactive_config_valid(monkeypatch): + """ + Test interactive_config by simulating user inputs for both token and prefix. + + The test creates a dummy red instance with a fake _config that records the token + and prefix values. It then simulates inputs for a valid token (a 50-character string), + a valid prefix, and confirmation that accepts the chosen prefix. The test ensures that: + - The returned token is the valid token. + - The dummy red instance has its config values set appropriately. + """ + red = DummyRed() + inputs = iter(["a" * 50, "!", "yes"]) + monkeypatch.setattr("builtins.input", lambda prompt: next(inputs)) + returned_token = await interactive_config( + red, token_set=False, prefix_set=False, print_header=False + ) + assert returned_token == "a" * 50 + assert red._config.token.value == "a" * 50 + assert red._config.prefix.value == ["!"] + + +@pytest.mark.asyncio +async def test_interactive_config_token_set(monkeypatch, capsys): + """ + Test interactive_config when token is already set. + This simulation only needs to configure the prefix because token_set is True. + The test simulates an overly long prefix that gets rejected and then a valid one, + and also asserts that the configuration header is printed. + """ + red = DummyRed() + inputs = iter(["abcdefghijk", "no", "!", "yes"]) + monkeypatch.setattr("builtins.input", lambda prompt: next(inputs)) + token_returned = await interactive_config( + red, token_set=True, prefix_set=False, print_header=True + ) + captured = capsys.readouterr() + assert "Red - Discord Bot | Configuration process" in captured.out + assert token_returned is None + assert red._config.prefix.value == ["!"] + + +@pytest.mark.asyncio +async def test_interactive_config_both_token_and_prefix_set(monkeypatch, capsys): + """ + Test interactive_config when both token_set and prefix_set are True. + In this scenario, no user input should be requested, and the function should + simply print the header (if print_header is True) and return None without + modifying the configuration values. + """ + + class DummyValue: + + def __init__(self): + self.value = None + + async def set(self, value): + self.value = value + + class DummyConfig: + + def __init__(self): + self.token = DummyValue() + self.prefix = DummyValue() + + class DummyRed: + + def __init__(self): + self._config = DummyConfig() + + red = DummyRed() + ret_token = await interactive_config( + red, token_set=True, prefix_set=True, print_header=True + ) + captured = capsys.readouterr() + assert "Red - Discord Bot | Configuration process" in captured.out + assert ret_token is None + assert red._config.token.value is None + assert red._config.prefix.value is None + + +@pytest.mark.asyncio +async def test_interactive_config_invalid_then_valid(monkeypatch): + """ + Test interactive_config with an initially invalid token (too short) followed by a valid token, + and a scenario where an overly long prefix is rejected before a valid prefix is provided. + This test simulates user input through monkeypatch to trigger the re-prompting logic. + """ + + class DummyValue: + + def __init__(self): + self.value = None + + async def set(self, value): + self.value = value + + class DummyConfig: + + def __init__(self): + self.token = DummyValue() + self.prefix = DummyValue() + + class DummyRed: + + def __init__(self): + self._config = DummyConfig() + + red = DummyRed() + responses = iter(["short", "a" * 50, "verylongprefix", "no", "!", "yes"]) + monkeypatch.setattr("builtins.input", lambda prompt: next(responses)) + ret_token = await interactive_config( + red, token_set=False, prefix_set=False, print_header=False + ) + assert ret_token == "a" * 50 + assert red._config.token.value == "a" * 50 + assert red._config.prefix.value == ["!"] + + +def test_parse_cli_flags_rpc_options(): + """ + Test parse_cli_flags to ensure that RPC related flags are correctly parsed. + This test verifies that: + - The --rpc flag is correctly interpreted as True. + - The --rpc-port flag sets the RPC port to the specified value. + - Other unrelated values (like the instance name) are properly set. + """ + test_args = ["myinstance", "--rpc", "--rpc-port", "7000"] + args = parse_cli_flags(test_args) + assert args.instance_name == "myinstance" + assert args.rpc is True + assert args.rpc_port == 7000 From 066b0123983a92d9c1ba322cfd970a73f48c6b5a Mon Sep 17 00:00:00 2001 From: CodeBeaverAI Date: Thu, 20 Feb 2025 18:13:07 +0100 Subject: [PATCH 3/4] test: Add coverage improvement test for tests/test__i18n.py --- tests/test__i18n.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) create mode 100644 tests/test__i18n.py diff --git a/tests/test__i18n.py b/tests/test__i18n.py new file mode 100644 index 00000000000..a33867a32cb --- /dev/null +++ b/tests/test__i18n.py @@ -0,0 +1,15 @@ +import pytest +from redbot.core._i18n import ( + set_contextual_locale, +) + + +def test_invalid_contextual_locale_verification(): + """ + Test that set_contextual_locale raises a ValueError when verify_language_code is True + and the language code is invalid (missing the country/territory part). + """ + with pytest.raises( + ValueError, match="Invalid format - language code has to include country code" + ): + set_contextual_locale("en", verify_language_code=True) From 3a5df443eddbaad0fc7d4ef66fc1c79795aed1b3 Mon Sep 17 00:00:00 2001 From: CodeBeaverAI Date: Thu, 20 Feb 2025 18:13:09 +0100 Subject: [PATCH 4/4] --- codebeaver.yml | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 codebeaver.yml diff --git a/codebeaver.yml b/codebeaver.yml new file mode 100644 index 00000000000..ec00f397d18 --- /dev/null +++ b/codebeaver.yml @@ -0,0 +1,2 @@ +from:pytest +# This file was generated automatically by CodeBeaver based on your repository. Learn how to customize it here: https://docs.codebeaver.ai/configuration/ \ No newline at end of file