test: extend core mock, add tests

tests/core_client_mock.py

@@ -19,11 +19,12 @@
 
 import logging
 from time import sleep
+from urllib.parse import urlparse
 
 from aura_engine.base.config import AuraConfig
 from aura_engine.base.lang import synchronized
 from aura_engine.base.utils import SimpleUtil as SU
-from aura_engine.core.channels import ChannelName
+from aura_engine.core.channels import ChannelName, PlayoutStatusResponse
 from aura_engine.core.client import CoreClient
 from aura_engine.core.mixer import Mixer
 from aura_engine.events import EngineEventDispatcher
@@ -40,6 +41,8 @@ class CoreClientMock(CoreClient):
     event_dispatcher = None
     conn = None
     stream_url = None
+    volumes = [0, 0, 0, 0]
+    selections = [0, 0, 0, 0]
 
     def __init__(self, event_dispatcher: EngineEventDispatcher):
         """
@@ -68,15 +71,33 @@ class CoreClientMock(CoreClient):
 
     @synchronized
     def exec(self, namespace: str, action: str, args: str = "") -> str:
+        self.logger.debug(
+            f"Core mock namespace: '{namespace}', action: '{action}', args: '{args}'"
+        )
         if namespace == "mixer":
             if action == "inputs":
                 return f"{ChannelName.QUEUE_A} {ChannelName.QUEUE_B}"
             elif action == "status":
-                return "ready=true selected=true single=false volume=100% remaining=1"
+                chn = int(args)
+                vol = self.volumes[chn]
+                return f"ready=true selected=true single=false volume={vol}% remaining=0"
             elif action == "volume":
-                return "volume=100% remaining=1"
+                argv = args.split(" ")
+                chn = int(argv[0])
+                vol = self.volumes[chn]
+                self.volumes[chn] = int(float(argv[1]) / 100.0)
+                resp = f"volume={vol}% remaining=0"
+                # print(resp)
+                return resp
             elif action == "select":
-                return "volume=100% remaining=1 selected=1"
+                argv = args.split(" ")
+                chn = int(argv[0])
+                vol = self.volumes[chn]
+                sel = 1 if argv[1] == "true" else 0
+                self.selections[chn] = sel
+                return f"volume={vol}% remaining=0 selected={sel}"
+            elif action == "activate":
+                return "OK"
         elif namespace in ChannelName._value2member_map_.values():
             if action == "push":
                 sleep(5)  # simulate some loading time
@@ -85,15 +106,26 @@ class CoreClientMock(CoreClient):
                 sleep(3)  # simulate some loading time
                 return 2  # res id
             elif action == "url":
-                self.stream_url = args
-                return "OK"
+                parse = urlparse(args)
+                if all([parse.scheme, parse.netloc]):
+                    self.stream_url = args
+                    return "OK"
+                else:
+                    return "Invalid URL"
             elif action == "status":
-                return "connected " + self.stream_url
+                return PlayoutStatusResponse.STREAM_STATUS_CONNECTED.value + " " + self.stream_url
             elif action == "start":
                 sleep(1)  # simulate small start delay
                 return "OK"
             elif action == "stop":
                 return "OK"
+            elif action == "clear":
+                return "OK"
+            elif action == "roll":
+                return "OK"
+            elif action == "set_track_metadata":
+                return "OK"
+        raise Exception(f"Unhandled namespace: '{namespace}', action: '{action}', args: '{args}'")
 
 
 class PlayoutClientMock(CoreClientMock):

tests/test_core_channels.py

@@ -90,10 +90,18 @@ class TestChannelFactory(unittest.TestCase):
     def test_load(self):
         print(self._testMethodName)
         queue, stream, live = self.create_channels()
-        queue.load(metadata={"foo": "bar"})
-        stream.load("http://foo")
+        queue.load("./file.flac")
+        stream.load("http://stream.local")
         live.load()
 
+    def test_load_metadata(self):
+        print(self._testMethodName)
+        queue, stream, live = self.create_channels()
+        metadata = {"title": "A Title", "artist": "An Artist", "album": "An Album"}
+        queue.load("./file.flac", metadata=metadata)
+        stream.load("http://stream.local", metadata=metadata)
+        live.load(metadata=metadata)
+
     def test_fade_in_out(self):
         print(self._testMethodName)
         queue, stream, live = self.create_channels()
@@ -108,12 +116,13 @@ class TestChannelFactory(unittest.TestCase):
 
         live.fade_in(volume=100)
         stream.fade_out()
-        live.fade_out()
+        live.fade_out(instant=True)
 
     def test_roll(self):
         print(self._testMethodName)
-        queue, _, _ = self.create_channels()
-        queue.roll(300)
+        queue = self.factory.create_channel(0, ChannelName.QUEUE_A, self.mixer)
+        res = queue.roll(300)
+        self.assertTrue(res)
 
 
 class TestPlayoutStatusResponse(unittest.TestCase):

tests/test_core_mixer.py

@@ -64,8 +64,8 @@ class TestMixer(unittest.TestCase):
             "ready": True,
             "selected": True,
             "single": True,
-            "volume": 100,
-            "remaining": 1.0,
+            "volume": 0,
+            "remaining": 0.0,
         }
         self.assertEqual(len(inputs), 2)
         self.assertEqual(inputs["in_queue_0"], input_state)