Add a test for running the entire main loop once
This commit is contained in:
13
app/main.py
13
app/main.py
@@ -63,7 +63,10 @@ class Main:
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
DELAY_SECONDS = 30.0
|
||||
|
||||
|
||||
def entry():
|
||||
logger.info("Initiating main...")
|
||||
start_time = time.time()
|
||||
main = Main()
|
||||
@@ -72,8 +75,14 @@ if __name__ == "__main__":
|
||||
try:
|
||||
while True:
|
||||
main.update_status()
|
||||
time.sleep(30.0 - ((time.time() - start_time) % 30.0))
|
||||
time.sleep(
|
||||
DELAY_SECONDS - ((time.time() - start_time) % DELAY_SECONDS)
|
||||
)
|
||||
except (SystemExit, KeyboardInterrupt):
|
||||
logger.info("Caught wish to exit, interrupting and re-raising.")
|
||||
main.interrupt()
|
||||
raise
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
entry()
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
import os
|
||||
import threading
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
import requests_mock
|
||||
|
||||
from app.main import Main
|
||||
from app.main import Main, entry
|
||||
|
||||
|
||||
def test_create_main(mock_loggers):
|
||||
@@ -34,3 +36,44 @@ def test_create_main(mock_loggers):
|
||||
assert main.discord_client
|
||||
assert main.twitch_client
|
||||
assert main.profile_image == "image"
|
||||
|
||||
|
||||
def test_run_main_one_iteration(mock_loggers):
|
||||
with (
|
||||
mock.patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"STREAMER_NAME": "streamer_name",
|
||||
"DISCORD_WEBHOOK_URL": "URL",
|
||||
"TWITCH_CLIENT_ID": "id",
|
||||
"TWITCH_CLIENT_SECRET": "secret",
|
||||
},
|
||||
),
|
||||
mock.patch(
|
||||
"app.main.Main.update_status",
|
||||
side_effect=Main.update_status,
|
||||
autospec=True,
|
||||
) as mock_update_status,
|
||||
mock.patch("app.main.DELAY_SECONDS", 2),
|
||||
requests_mock.Mocker() as requests_mocker,
|
||||
):
|
||||
requests_mocker.post(
|
||||
"https://id.twitch.tv/oauth2/token", json={"access_token": "token"}
|
||||
)
|
||||
requests_mocker.get(
|
||||
"https://api.twitch.tv/helix/users?login=streamer_name",
|
||||
json={"data": [{"profile_image_url": "image"}]},
|
||||
)
|
||||
requests_mocker.get(
|
||||
"https://api.twitch.tv/helix/streams?user_login=streamer_name",
|
||||
json={"data": []},
|
||||
)
|
||||
|
||||
def delayed_raise():
|
||||
mock_update_status.side_effect = KeyboardInterrupt()
|
||||
|
||||
thread = threading.Timer(1, delayed_raise)
|
||||
thread.start()
|
||||
|
||||
with pytest.raises(KeyboardInterrupt):
|
||||
entry()
|
||||
|
||||
Reference in New Issue
Block a user