test: add play active item scenarios

tests/test_scheduling_scheduler.py

@@ -29,6 +29,7 @@ from aura_engine.engine import Engine
 from aura_engine.scheduling.domain import Playlist, PlaylistItem, Timeslot
 from aura_engine.scheduling.scheduler import AuraScheduler
 from aura_engine.scheduling.timetable import TimetableService
+from aura_engine.scheduling.utils import TimetableRenderer
 
 
 class CommandRegistry:
@@ -44,8 +45,9 @@ class CommandRegistry:
 
     def reset(f):
         c: EngineExecutor
-        for c in CommandRegistry.commands.get(f):
-            c.cancel()
+        if CommandRegistry.commands.get(f):
+            for c in CommandRegistry.commands.get(f):
+                c.cancel()
         CommandRegistry.commands[f] = None
 
     def prune():
@@ -126,6 +128,12 @@ class MockedTimeslotCommand:
         caller = inspect.stack()[2].function
         print(f"Called do_play of MockedTimeslotCommand from {caller}")
 
+    def cancel(self):
+        """
+        Cancel the command.
+        """
+        pass
+
 
 class MockedPlayCommand:
     """
@@ -174,6 +182,12 @@ class MockedPlayCommand:
         caller = inspect.stack()[2].function
         print(f"Called do_play of MockedPlayCommand from {caller}")
 
+    def cancel(self):
+        """
+        Cancel the command.
+        """
+        pass
+
 
 class TestSchedulingScheduler(unittest.TestCase):
     """
@@ -200,7 +214,9 @@ class TestSchedulingScheduler(unittest.TestCase):
     # Utilities
     #
 
-    def build_timetable(self, scheduler: AuraScheduler) -> [Timeslot]:
+    def build_timetable(
+        self, scheduler: AuraScheduler, current_start, current_end, source_type="file://"
+    ) -> [Timeslot]:
         # Delete the timetable file to isolate test case
         cache_location = self.config.get("cache_dir")
         f = scheduler.timetable.timetable_file
@@ -213,22 +229,39 @@ class TestSchedulingScheduler(unittest.TestCase):
         # Build some timetable
         now = SU.timestamp()
         ts1 = Timeslot(
-            id=1, repetition_id=None, start=now - 500, end=now - 400, show=None, episode=None
+            id=1,
+            repetition_id=None,
+            start=now - 500,
+            end=now + current_start,
+            show=None,
+            episode=None,
         )
         ts2 = Timeslot(
-            id=2, repetition_id=None, start=now - 400, end=now + 400, show=None, episode=None
+            id=2,
+            repetition_id=None,
+            start=now + current_start,
+            end=now + current_end,
+            show=None,
+            episode=None,
         )
         ts3 = Timeslot(
-            id=3, repetition_id=None, start=now + 400, end=now + 540, show=None, episode=None
+            id=3,
+            repetition_id=None,
+            start=now + current_end,
+            end=now + 540,
+            show=None,
+            episode=None,
         )
         ts4 = Timeslot(
             id=4, repetition_id=None, start=now + 540, end=now + 720, show=None, episode=None
         )
 
         pl = Playlist("9999", "Playlist XYZ", False)
-        alpha = PlaylistItem("file://alpha.flac", 100, 100, None)
-        beta = PlaylistItem("file://beta.flac", 600, 100, None)
-        gamma = PlaylistItem("file://gamma.flac", 100, 100, None)
+        alpha = PlaylistItem(source_type + "alpha.flac", 100, 100, None)
+        beta = PlaylistItem(
+            source_type + "beta.flac", current_end - current_start - 200, 100, None
+        )
+        gamma = PlaylistItem(source_type + "gamma.flac", 100, 100, None)
         pl.add(alpha)
         pl.add(beta)
         pl.add(gamma)
@@ -263,13 +296,13 @@ class TestSchedulingScheduler(unittest.TestCase):
         scheduler.timetable.delete_timetable_file()
 
         # Build some timetable
-        scheduler.timetable.timetable = self.build_timetable(scheduler)
+        scheduler.timetable.timetable = self.build_timetable(scheduler, -400, +400)
 
         tt = scheduler.get_timetable()
         self.assertIsNotNone(tt)
         self.assertEqual(4, len(tt.timetable))
 
-    def test_play_active_item(self):
+    def test_play_active_item_with_roll(self):
         print(self._testMethodName)
 
         # Construct the scheduler
@@ -277,11 +310,25 @@ class TestSchedulingScheduler(unittest.TestCase):
         scheduler.timetable.delete_timetable_file()
 
         # Build some timetable
-        scheduler.timetable.timetable = self.build_timetable(scheduler)
+        scheduler.timetable.timetable = self.build_timetable(scheduler, -400, +400)
 
-        is_playing = scheduler.play_active_item()
-        cmds = CommandRegistry.get("test_play_active_item")
+        # Display timetable for debugging
+        tr = TimetableRenderer(scheduler)
+        report = tr.get_ascii_timeslots()
+        print(report)
+
+        # Before we proceed, ensure there is really an active item
+        active_item = scheduler.timetable.get_current_item()
+        self.assertIsNotNone(active_item)
+
+        scheduler.play_active_item()
+        cmds = CommandRegistry.get(self._testMethodName)
+
+        # Expecting at a timeslot and a play command
         self.assertEqual(2, len(cmds))
+        self.assertTrue(isinstance(cmds[0], MockedTimeslotCommand))
+        self.assertTrue(isinstance(cmds[1], MockedPlayCommand))
+
         timeslot_cmd: MockedTimeslotCommand = cmds[0]
         play_cmd: MockedPlayCommand = cmds[1]
         self.assertEqual(2, timeslot_cmd.timeslot.get_id())
@@ -289,6 +336,40 @@ class TestSchedulingScheduler(unittest.TestCase):
         channel: GenericChannel = play_cmd.items[0].play_channel
         self.assertEqual("main channel", channel.name)
 
+    def test_play_active_item_no_roll(self):
+        print(self._testMethodName)
+
+        # Construct the scheduler
+        scheduler = AuraScheduler(MockedEngine())
+        scheduler.timetable.delete_timetable_file()
+
+        # Build some timetable
+        scheduler.timetable.timetable = self.build_timetable(scheduler, -400, +5)
+
+        scheduler.play_active_item()
+        cmds = CommandRegistry.get(self._testMethodName)
+
+        # Expecting at least one timeslot command
+        self.assertEqual(1, len(cmds))
+        self.assertTrue(isinstance(cmds[0], MockedTimeslotCommand))
+
+    def test_play_active_invalid_source(self):
+        print(self._testMethodName)
+
+        # Construct the scheduler
+        scheduler = AuraScheduler(MockedEngine())
+        scheduler.timetable.delete_timetable_file()
+
+        # Build some timetable
+        scheduler.timetable.timetable = self.build_timetable(scheduler, -400, +5, "3d_video://")
+
+        scheduler.play_active_item()
+        cmds = CommandRegistry.get(self._testMethodName)
+
+        # Expecting at least one timeslot command
+        self.assertEqual(1, len(cmds))
+        self.assertTrue(isinstance(cmds[0], MockedTimeslotCommand))
+
 
 if __name__ == "__main__":
     unittest.main()