Moved non-bot setup specific methods to Utils class

minimike86 · Oct 29, 2024 · 4d6e794 · 4d6e794
1 parent 159b378
commit 4d6e794
Show file tree

Hide file tree

Showing 7 changed files with 159 additions and 227 deletions.
diff --git a/tests/test_custom_bot.py b/tests/test_custom_bot.py
@@ -7,7 +7,7 @@
 @mock_aws
 @pytest.mark.filterwarnings("ignore::DeprecationWarning")
 def test_get_app_token(mocker):
-    from twitchrce.main import get_app_token
+    from twitchrce.utils.utils import Utils
 
     # Mock the response from client_credentials_grant_flow
     client_credentials_grant_flow_response = {
@@ -25,6 +25,6 @@ def test_get_app_token(mocker):
     )
 
     event_loop = asyncio.get_event_loop()
-    access_token = event_loop.run_until_complete(get_app_token())
+    access_token = event_loop.run_until_complete(Utils.get_app_token())
     assert access_token == "1234567890abcdef1234567890abcdef"
     mock_client_credentials_grant_flow.assert_awaited_once()
diff --git a/tests/test_main.py b/tests/test_main.py
@@ -3,7 +3,6 @@
 
 import boto3
 import pytest
-from botocore.exceptions import NoCredentialsError, PartialCredentialsError
 from moto import mock_aws
 
 from twitchrce.config import bot_config
@@ -43,142 +42,6 @@ def set_environment_variables(monkeypatch):
     monkeypatch.setenv("VIRUS_TOTAL_API_KEY", "xyz")
 
 
-@pytest.mark.asyncio
-@pytest.mark.filterwarnings("ignore::DeprecationWarning")
-async def test_get_app_token(mocker):
-    from twitchrce.main import get_app_token
-
-    mock_client_credentials_grant_flow_response = {
-        "access_token": "access_token_xyz789",
-        "expires_in": 12345,
-        "token_type": "bearer",
-    }
-
-    mock_client_credentials_grant_flow = mocker.patch(
-        "twitchrce.api.twitch.twitch_api_auth.TwitchApiAuth.client_credentials_grant_flow"
-    )
-    mock_client_credentials_grant_flow.return_value = (
-        mock_client_credentials_grant_flow_response
-    )
-
-    access_token = await get_app_token()
-    assert access_token == "access_token_xyz789"
-    mock_client_credentials_grant_flow.assert_awaited_once()
-
-
-@pytest.mark.asyncio
-async def test_check_valid_token_is_valid(mocker):
-    from twitchrce.main import check_valid_token
-
-    mock_validate_token = mocker.patch(
-        "twitchrce.api.twitch.twitch_api_auth.TwitchApiAuth.validate_token"
-    )
-    mock_validate_token.return_value = True
-
-    mock_user = {}
-
-    is_valid_token = await check_valid_token(user=mock_user)
-    assert bool(is_valid_token)
-    mock_validate_token.assert_awaited_once()
-
-
-@mock_aws
-def test_check_valid_token_is_invalid(mocker):
-    from twitchrce.main import check_valid_token
-
-    BOT_CONFIG = bot_config.BotConfig().get_bot_config()
-
-    mock_validate_token = mocker.patch(
-        "twitchrce.api.twitch.twitch_api_auth.TwitchApiAuth.validate_token"
-    )
-    mock_validate_token.side_effect = [False, True]
-
-    mock_refresh_access_token = mocker.patch(
-        "twitchrce.api.twitch.twitch_api_auth.TwitchApiAuth.refresh_access_token"
-    )
-    mock_refresh_access_token.return_value = MOCK_REFRESH_ACCESS_TOKEN_RESPONSE_SUCCESS
-
-    mock_dynamodb = boto3.resource(
-        "dynamodb", region_name=BOT_CONFIG.get("aws").get("region_name")
-    )
-    mock_table_name = "user"
-    mock_dynamodb.create_table(
-        TableName=mock_table_name,
-        KeySchema=[
-            {"AttributeName": "id", "KeyType": "HASH"},  # Partition key
-        ],
-        AttributeDefinitions=[
-            {"AttributeName": "id", "AttributeType": "S"},  # String
-        ],
-        ProvisionedThroughput={
-            "ReadCapacityUnits": 5,
-            "WriteCapacityUnits": 5,
-        },
-    )
-    mock_table = mock_dynamodb.Table(mock_table_name)
-    mock_table.put_item(
-        Item={
-            "id": "123456",
-            "access_token": "access_token_abc123",
-            "client_id": BOT_CONFIG.get("twitch").get("client_id"),
-            "expires_in": 12345,
-            "login": "username_bot",
-            "refresh_token": "refresh_token_abc123",
-        }
-    )
-    mock_user_table = mocker.patch("twitchrce.main.user_table")
-    mock_user_table.return_value = mock_table
-
-    mock_user = {"id": "123", "login": "username_bot"}
-
-    event_loop = asyncio.get_event_loop()
-    is_valid_token = event_loop.run_until_complete(check_valid_token(user=mock_user))
-    assert bool(is_valid_token)
-    assert mock_validate_token.await_count == 2
-    mock_validate_token.assert_awaited()
-    mock_refresh_access_token.assert_awaited_once()
-
-
-@pytest.mark.asyncio
-async def test_refresh_user_token_no_credentials(mocker):
-    from twitchrce.main import refresh_user_token
-
-    mock_refresh_access_token = mocker.patch(
-        "twitchrce.api.twitch.twitch_api_auth.TwitchApiAuth.refresh_access_token"
-    )
-    mock_refresh_access_token.return_value = MOCK_REFRESH_ACCESS_TOKEN_RESPONSE_SUCCESS
-
-    mock_user_table_update_item = mocker.patch("twitchrce.main.user_table.update_item")
-    mock_user_table_update_item.side_effect = (
-        NoCredentialsError()
-    )  # simulate missing AWS credentials
-
-    mock_user = {"id": "123", "login": "username_bot"}
-
-    with pytest.raises(NoCredentialsError):
-        await refresh_user_token(mock_user)
-
-
-@pytest.mark.asyncio
-async def test_refresh_user_token_partial_credentials(mocker):
-    from twitchrce.main import refresh_user_token
-
-    mock_refresh_access_token = mocker.patch(
-        "twitchrce.api.twitch.twitch_api_auth.TwitchApiAuth.refresh_access_token"
-    )
-    mock_refresh_access_token.return_value = MOCK_REFRESH_ACCESS_TOKEN_RESPONSE_SUCCESS
-
-    mock_user_table_update_item = mocker.patch("twitchrce.main.user_table.update_item")
-    mock_user_table_update_item.side_effect = PartialCredentialsError(
-        provider="mock_provider", cred_var="mock_cred_var"
-    )  # simulate missing AWS credentials
-
-    mock_user = {"id": "123", "login": "username_bot"}
-
-    with pytest.raises(PartialCredentialsError):
-        await refresh_user_token(mock_user)
-
-
 @mock_aws
 def test_main(mocker, capfd):
     from twitchrce.main import setup_bot
@@ -277,7 +140,7 @@ def test_setup_bot_if_bot_user_has_no_access_token_should_raise_value_error(mock
 
     BOT_CONFIG = bot_config.BotConfig().get_bot_config()
 
-    mock_get_app_token = mocker.patch("twitchrce.main.get_app_token")
+    mock_get_app_token = mocker.patch("twitchrce.utils.utils.Utils.get_app_token")
     mock_get_app_token.return_value = "access_token_abc123"
 
     mock_dynamodb = boto3.resource(
@@ -325,7 +188,7 @@ def test_setup_bot_if_db_has_no_bot_user_should_raise_value_error(mocker):
 
     BOT_CONFIG = bot_config.BotConfig().get_bot_config()
 
-    mock_get_app_token = mocker.patch("twitchrce.main.get_app_token")
+    mock_get_app_token = mocker.patch("twitchrce.utils.utils.Utils.get_app_token")
     mock_get_app_token.return_value = "access_token_abc123"
 
     mock_dynamodb = boto3.resource(
@@ -364,7 +227,7 @@ def test_setup_bot_if_bot_user_has_access_token_but_describe_instances_has_no_re
 
     BOT_CONFIG = bot_config.BotConfig().get_bot_config()
 
-    mock_get_app_token = mocker.patch("twitchrce.main.get_app_token")
+    mock_get_app_token = mocker.patch("twitchrce.utils.utils.Utils.get_app_token")
     mock_get_app_token.return_value = "access_token_abc123"
 
     mock_dynamodb = boto3.resource(
@@ -408,7 +271,7 @@ def test_setup_bot_if_bot_user_has_access_token_but_describe_instances_has_no_re
         ),
     ]
 
-    mock_check_valid_token = mocker.patch("twitchrce.main.check_valid_token")
+    mock_check_valid_token = mocker.patch("twitchrce.utils.utils.Utils.check_valid_token")
     mock_check_valid_token.return_value = True
 
     mock_describe_instances_response = {}

diff --git a/tests/utils/test_utils.py b/tests/utils/test_utils.py
@@ -3,7 +3,41 @@
 from twitchrce.utils.utils import Utils
 
 
-# Test cases for the redact_secret_string method
+@pytest.mark.asyncio
+@pytest.mark.filterwarnings("ignore::DeprecationWarning")
+async def test_get_app_token(mocker):
+    mock_client_credentials_grant_flow_response = {
+        "access_token": "access_token_xyz789",
+        "expires_in": 12345,
+        "token_type": "bearer",
+    }
+
+    mock_client_credentials_grant_flow = mocker.patch(
+        "twitchrce.api.twitch.twitch_api_auth.TwitchApiAuth.client_credentials_grant_flow"
+    )
+    mock_client_credentials_grant_flow.return_value = (
+        mock_client_credentials_grant_flow_response
+    )
+
+    access_token = await Utils.get_app_token()
+    assert access_token == "access_token_xyz789"
+    mock_client_credentials_grant_flow.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_check_valid_token_is_valid(mocker):
+    mock_validate_token = mocker.patch(
+        "twitchrce.api.twitch.twitch_api_auth.TwitchApiAuth.validate_token"
+    )
+    mock_validate_token.return_value = True
+
+    mock_user = {}
+
+    is_valid_token = await Utils().check_valid_token(user=mock_user)
+    assert bool(is_valid_token)
+    mock_validate_token.assert_awaited_once()
+
+
 @pytest.mark.parametrize(
     "secret_string, visible_chars, expected",
     [

diff --git a/twitchrce/__init__.py b/twitchrce/__init__.py
diff --git a/twitchrce/api/twitch/twitch_api_auth.py b/twitchrce/api/twitch/twitch_api_auth.py
@@ -69,7 +69,7 @@ async def obtain_access_token(self, code: str, redirect_uri: str) -> dict:
         Parameter	    Required?	Type	Description
         client_id	    Yes	        String	Your app’s registered client ID.
         client_secret	Yes	        String	Your app’s registered client secret.
-        code	        Yes	        String	The code that the /authorize response returned in the code query parameter.
+        code	        Yes	        String	The code that the /authorize response returned to the code query parameter.
         grant_type	    Yes	        String	Must be set to authorization_code.
         redirect_uri	Yes	        URI	    Your app’s registered redirect URI.
         """
@@ -128,6 +128,7 @@ async def refresh_access_token(self, refresh_token: str):
                 status = resp.status
                 data = await resp.json()
         if status == 400:
+            from twitchrce.utils.utils import Utils
             logger.error(
                 f"{Fore.RED}Refresh of user oauth access_token using refresh_token [{Fore.MAGENTA}"
                 f"{Utils.redact_secret_string(refresh_token)}{Fore.RED}] has FAILED!.{Style.RESET_ALL}"