Spaces:
Sleeping
Sleeping
File size: 5,047 Bytes
127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda 37ee895 127bfda | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | import asyncio
import os
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from bot.bootstrap_runtime import (
build_irc_token_manager,
get_secret,
require_env,
resolve_client_secret_for_irc_refresh,
resolve_irc_channel_logins,
run_eventsub_mode,
run_irc_mode,
)
class TestBootstrapRuntimeV4:
def test_get_secret(self):
with patch.dict(os.environ, {"TWITCH_CLIENT_SECRET": "test_secret"}):
assert get_secret("twitch-client-secret") == "test_secret"
with patch.dict(os.environ, clear=True):
with pytest.raises(RuntimeError):
get_secret("missing")
def test_require_env(self):
with patch.dict(os.environ, {"EXISTING": "val"}):
assert require_env("EXISTING") == "val"
with patch.dict(os.environ, clear=True):
with pytest.raises(RuntimeError):
require_env("MISSING")
@pytest.mark.asyncio
async def test_resolve_irc_channel_logins(self):
with patch("bot.bootstrap_runtime.TWITCH_CHANNEL_LOGINS_RAW", "chan1,chan2"):
assert await resolve_irc_channel_logins() == ["chan1", "chan2"]
with patch("bot.bootstrap_runtime.TWITCH_CHANNEL_LOGINS_RAW", ""):
with patch("bot.bootstrap_runtime.TWITCH_CHANNEL_LOGIN", "chan3"):
assert await resolve_irc_channel_logins() == ["chan3"]
with patch("bot.bootstrap_runtime.TWITCH_CHANNEL_LOGINS_RAW", ""):
with patch("bot.bootstrap_runtime.TWITCH_CHANNEL_LOGIN", ""):
with patch.dict(os.environ, {"TWITCH_CHANNEL_LOGIN": "chan4"}):
assert await resolve_irc_channel_logins() == ["chan4"]
def test_resolve_client_secret_for_irc_refresh(self):
with patch("bot.bootstrap_runtime.TWITCH_CLIENT_SECRET_INLINE", "inline"):
assert resolve_client_secret_for_irc_refresh() == "inline"
with patch("bot.bootstrap_runtime.TWITCH_CLIENT_SECRET_INLINE", ""):
with patch("bot.bootstrap_runtime.TWITCH_CLIENT_SECRET_NAME", "my-secret"):
with patch("bot.bootstrap_runtime.get_secret", return_value="env_secret"):
assert resolve_client_secret_for_irc_refresh() == "env_secret"
@patch("bot.bootstrap_runtime.require_env")
def test_build_irc_token_manager(self, mock_require):
mock_require.side_effect = lambda x: {
"TWITCH_USER_TOKEN": "utoken",
"TWITCH_CLIENT_ID": "cid",
}.get(x, "")
with patch("os.environ.get") as mock_env:
mock_env.return_value = ""
mgr = build_irc_token_manager()
assert mgr.access_token == "utoken"
assert not hasattr(mgr, "refresh_token") or not mgr.refresh_token
mock_env.return_value = "rtoken"
with patch(
"bot.bootstrap_runtime.resolve_client_secret_for_irc_refresh",
return_value="csec",
):
mgr = build_irc_token_manager()
assert mgr.refresh_token == "rtoken"
assert mgr.client_secret == "csec"
@patch("bot.bootstrap_runtime.IrcByteBot")
@patch("bot.bootstrap_runtime.build_irc_token_manager")
@patch("bot.bootstrap_runtime.resolve_irc_channel_logins")
@patch("bot.bootstrap_runtime.asyncio.run")
def test_run_irc_mode(self, mock_run, mock_resolve, mock_build, mock_bot_cls):
def run_side_effect(coroutine):
coroutine.close()
mock_run.side_effect = run_side_effect
mock_resolve.return_value = ["test"]
mock_mgr = MagicMock()
mock_build.return_value = mock_mgr
mock_bot = MagicMock()
mock_bot_cls.return_value = mock_bot
run_irc_mode()
mock_run.assert_called_once()
@pytest.mark.skip(reason="Teste de implementacao interna - loop infinito")
@patch("bot.bootstrap_runtime.IrcByteBot")
@patch("bot.bootstrap_runtime.build_irc_token_manager")
@patch("bot.bootstrap_runtime.resolve_irc_channel_logins")
@patch("bot.bootstrap_runtime.asyncio.run")
def test_run_irc_mode_exception(self, mock_run, mock_resolve, mock_build, mock_bot_cls):
mock_resolve.return_value = ["test"]
mock_build.return_value = MagicMock()
mock_bot_cls.side_effect = Exception("init failed")
# Should catch and log, not crash
with patch("bot.bootstrap_runtime.logger.error") as mock_log:
run_irc_mode()
mock_log.assert_called()
@patch("bot.bootstrap_runtime.ByteBot")
def test_run_eventsub_mode(self, mock_bot_cls):
with patch.dict(
os.environ,
{
"TWITCH_CLIENT_ID": "cid",
"TWITCH_BOT_ID": "bid",
"TWITCH_CHANNEL_ID": "chid",
"TWITCH_CLIENT_SECRET": "sec",
},
):
mock_bot = MagicMock()
mock_bot_cls.return_value = mock_bot
run_eventsub_mode()
mock_bot.run.assert_called_once()
|