Fix tests for crash recovery
This commit is contained in:
20
app/main.py
20
app/main.py
@@ -1,6 +1,8 @@
|
|||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import time
|
import time
|
||||||
|
from dataclasses import asdict
|
||||||
|
from json import JSONDecodeError
|
||||||
|
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
from requests import HTTPError
|
from requests import HTTPError
|
||||||
@@ -12,17 +14,20 @@ from app.twitch_client import StreamInformation, TwitchClient
|
|||||||
class Main:
|
class Main:
|
||||||
is_live: bool = False
|
is_live: bool = False
|
||||||
current_stream_id: str = ""
|
current_stream_id: str = ""
|
||||||
streams: dict[str, StreamInformation]
|
streams: dict[str, StreamInformation] = dict()
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self.twitch_client = TwitchClient(streamer=os.environ["STREAMER_NAME"])
|
self.twitch_client = TwitchClient(streamer=os.environ["STREAMER_NAME"])
|
||||||
self.twitch_client.update_access_token()
|
self.twitch_client.update_access_token()
|
||||||
self.profile_image = self.twitch_client.get_streamer_profile_picture()
|
self.profile_image = self.twitch_client.get_streamer_profile_picture()
|
||||||
self.discord_client = DiscordClient()
|
self.discord_client = DiscordClient()
|
||||||
with open("streams.json", "r") as file:
|
try:
|
||||||
saved_streams = json.load(file)
|
with open("streams.json", "r") as file:
|
||||||
for stream_id, saved_stream in saved_streams:
|
saved_streams = json.load(file)
|
||||||
self.streams[stream_id] = StreamInformation(**saved_stream)
|
for stream_id, saved_stream in saved_streams:
|
||||||
|
self.streams[stream_id] = StreamInformation(**saved_stream)
|
||||||
|
except (FileNotFoundError, JSONDecodeError):
|
||||||
|
pass
|
||||||
|
|
||||||
def update_status(self):
|
def update_status(self):
|
||||||
try:
|
try:
|
||||||
@@ -101,7 +106,10 @@ def entry() -> None:
|
|||||||
while True:
|
while True:
|
||||||
main.update_status()
|
main.update_status()
|
||||||
with open("streams.json", "w") as file:
|
with open("streams.json", "w") as file:
|
||||||
json.dump(main.streams, file)
|
serializable_streams = {
|
||||||
|
k: asdict(v) for k, v in main.streams.items()
|
||||||
|
}
|
||||||
|
json.dump(serializable_streams, file)
|
||||||
time.sleep(
|
time.sleep(
|
||||||
DELAY_SECONDS - ((time.time() - start_time) % DELAY_SECONDS)
|
DELAY_SECONDS - ((time.time() - start_time) % DELAY_SECONDS)
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ def test_require_webhook_url():
|
|||||||
|
|
||||||
def test_send_information_to_discord(mock_loggers):
|
def test_send_information_to_discord(mock_loggers):
|
||||||
stream = StreamInformation(
|
stream = StreamInformation(
|
||||||
|
id="0",
|
||||||
user_id="0",
|
user_id="0",
|
||||||
user_name="Test",
|
user_name="Test",
|
||||||
user_login="test",
|
user_login="test",
|
||||||
@@ -56,6 +57,7 @@ def test_send_information_to_discord(mock_loggers):
|
|||||||
|
|
||||||
def test_send_information_to_discord_fails(mock_loggers):
|
def test_send_information_to_discord_fails(mock_loggers):
|
||||||
stream = StreamInformation(
|
stream = StreamInformation(
|
||||||
|
id="",
|
||||||
user_id="",
|
user_id="",
|
||||||
user_name="",
|
user_name="",
|
||||||
user_login="",
|
user_login="",
|
||||||
|
|||||||
@@ -196,3 +196,5 @@ def test_run_main_streamer_already_live(mock_loggers):
|
|||||||
finalize_notification_request = requests_mocker.request_history[-1]
|
finalize_notification_request = requests_mocker.request_history[-1]
|
||||||
assert finalize_notification_request.path == "/webhook/messages/123456"
|
assert finalize_notification_request.path == "/webhook/messages/123456"
|
||||||
assert finalize_notification_request.json()
|
assert finalize_notification_request.json()
|
||||||
|
|
||||||
|
os.remove("streams.json")
|
||||||
|
|||||||
Reference in New Issue
Block a user