diff --git a/tests/test_scheduling_timetable.py b/tests/test_scheduling_timetable.py
index bf8add961baaca8126c10ae2ad16b5928b21c7b1..7248920ead4896b5bdaf46f607e4bbc6376b0443 100644
--- a/tests/test_scheduling_timetable.py
+++ b/tests/test_scheduling_timetable.py
@@ -26,6 +26,7 @@ from unittest import mock
from aura_engine.base.api import SimpleCachedRestApi
from aura_engine.base.config import AuraConfig
+from aura_engine.base.lang import DotDict
from aura_engine.base.utils import SimpleUtil as SU
from aura_engine.scheduling.api import ApiFetcher
from aura_engine.scheduling.domain import Playlist, PlaylistItem, PlaylistType, Timeslot
@@ -47,57 +48,26 @@ class TestSchedulingTimetable(unittest.TestCase):
# Mock
#
- # Mock `requests.get` in `SimpleRestApi`
- def mocked_requests_get(*args, **kwargs):
- class MockResponse:
- def __init__(self, json_data, status_code):
- self.json_data = json_data
- self.status_code = status_code
-
- def json(self):
- return self.json_data
-
- print(f"Calling mocked 'requests.get' with '{args[0]}'")
- if "/api/v1/playout" in args[0]:
- return MockResponse(TestSchedulingTimetable.mocked_steering_json, 200)
- elif "/api/v1/playlists/1" in args[0]:
- return MockResponse(TestSchedulingTimetable.mocked_tank_json1, 200)
- elif "/api/v1/playlists/2" in args[0]:
- return MockResponse(TestSchedulingTimetable.mocked_tank_json2, 200)
-
- return MockResponse(None, 404)
-
- # Mock `requests.get` in `SimpleRestApi`
- def mocked_requests_no_data(*args, **kwargs):
- class MockResponse:
- def __init__(self, json_data, status_code):
- self.json_data = json_data
- self.status_code = status_code
-
- def json(self):
- return self.json_data
-
- print(f"Calling mocked 'requests.get' with '{args[0]}'")
- if "/api/v1/playout" in args[0]:
- return MockResponse(None, 200)
- elif "/api/v1/playlists/1" in args[0]:
- return MockResponse(None, 200)
-
- return MockResponse(None, 404)
-
- # Mock `requests.get` in `SimpleRestApi`
- def mocked_requests_get_exception(*args, **kwargs):
- class MockResponse:
- def __init__(self, json_data, status_code):
- self.json_data = json_data
- self.status_code = status_code
-
- def json(self):
- invalid_timeslots = [{"id": 18, "start": "2023-05-16T13:00:00"}]
- return invalid_timeslots
-
- print(f"Calling mocked 'requests.get' with '{args[0]}'")
- return MockResponse(None, 200)
+ class MockedApiFetcher:
+ def __init__(self, code, message, timeslots, exception):
+ self.code = code
+ self.message = message
+ self.timeslots = timeslots
+ self.exception = exception
+
+ def start(self):
+ print("Called mocked 'start()'")
+
+ def fetch(self):
+ print("Called mocked 'fetch()'")
+ return DotDict(
+ {
+ "code": self.code,
+ "message": self.message,
+ "timeslots": self.timeslots,
+ "exception": self.exception,
+ }
+ )
#
# Setup and teardown
@@ -107,30 +77,9 @@ class TestSchedulingTimetable(unittest.TestCase):
self.config = AuraConfig()
cache_location = self.config.get("cache_dir")
self.timetable = TimetableService(cache_location)
- self.api_fetcher = ApiFetcher()
-
- with open("./tests/json/steering-api-v1-playout.json", "r") as file:
- TestSchedulingTimetable.mocked_steering_json = json.load(file)
- with open("./tests/json/tank-api-v1-playlists-1.json", "r") as file:
- TestSchedulingTimetable.mocked_tank_json1 = json.load(file)
- with open("./tests/json/tank-api-v1-playlists-2.json", "r") as file:
- TestSchedulingTimetable.mocked_tank_json2 = json.load(file)
-
- # Update dates that they are in the future
- # Format e.g. "start":"2023-05-16T10:06:00"
- now = datetime.now()
- hour: int = -1
-
- for timeslot in TestSchedulingTimetable.mocked_steering_json:
- start = (now + timedelta(hours=hour)).strftime("%Y-%m-%dT%H:%M:%S")
- end = (now + timedelta(hours=hour + 1)).strftime("%Y-%m-%dT%H:%M:%S")
- timeslot["start"] = start
- timeslot["end"] = end
- hour += 1
def tearDown(self):
# Clean up cache when finished
- self.api_fetcher.api.prune_cache_dir()
self.timetable.prune_cache_dir()
#
@@ -275,29 +224,58 @@ class TestSchedulingTimetable(unittest.TestCase):
def test_timetable_merge(self):
print(self._testMethodName)
- # @mock.patch("aura_engine.scheduling.programme.api_fetcher", side_effect=mocked_fetch)
- @mock.patch("aura_engine.base.api.requests.get", side_effect=mocked_requests_get)
- def test_fetch_data(self, mock_get):
+ # TODO
+
+ def test_fetch_data(self):
print(self._testMethodName)
+ now = SU.timestamp()
+ ts1 = Timeslot(
+ id=1, repetition_id=None, start=now - 180, end=now - 120, show=None, episode=None
+ )
+ ts2 = Timeslot(
+ id=1, repetition_id=None, start=now - 120, end=now + 120, show=None, episode=None
+ )
+ ts3 = Timeslot(
+ id=1, repetition_id=None, start=now + 120, end=now + 180, show=None, episode=None
+ )
+ ts4 = Timeslot(
+ id=1, repetition_id=None, start=now + 180, end=now + 360, show=None, episode=None
+ )
+
+ pl = Playlist("999", "Playlist X", False)
+ alpha = PlaylistItem("alpha.flac", 20, 100, None)
+ beta = PlaylistItem("beta.flac", 100, 100, None)
+ gamma = PlaylistItem("gamma.flac", 20, 100, None)
+ pl.add(alpha)
+ pl.add(beta)
+ pl.add(gamma)
+
+ ts2.set_playlists(pl, None, None)
+
+ timeslots = [ts1, ts2, ts3, ts4]
+ fetcher = self.MockedApiFetcher(0, "Success", timeslots, None)
+ self.timetable.refresh(fetcher)
+
# Clear cache to keep test cases isolated
- self.api_fetcher.api.prune_cache_dir()
self.timetable.prune_cache_dir()
- @mock.patch("aura_engine.base.api.requests.get", side_effect=mocked_requests_no_data)
- def test_fetch_no_data(self, mock_get):
+ def test_fetch_no_data(self):
print(self._testMethodName)
+ fetcher = self.MockedApiFetcher(1, "No timeslots", None, None)
+ self.timetable.refresh(fetcher)
+
# Clear cache to keep test cases isolated
- self.api_fetcher.api.prune_cache_dir()
self.timetable.prune_cache_dir()
- @mock.patch("aura_engine.base.api.requests.get", side_effect=mocked_requests_get_exception)
- def test_fetch_exception(self, mock_get):
+ def test_fetch_exception(self):
print(self._testMethodName)
+ fetcher = self.MockedApiFetcher(1, "Some weird exception", None, "Quack")
+ self.timetable.refresh(fetcher)
+
# Clear cache to keep test cases isolated
- self.api_fetcher.api.prune_cache_dir()
self.timetable.prune_cache_dir()