diff --git a/app/main.py b/app/main.py index bbab36c..ea9ee75 100644 --- a/app/main.py +++ b/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() diff --git a/app/tests/test_main.py b/app/tests/test_main.py index 05f1953..e9c8f6c 100644 --- a/app/tests/test_main.py +++ b/app/tests/test_main.py @@ -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()