write tests for sources

aircox_streamer/tests/test_controllers_sources.py

@@ -1,6 +1,13 @@
+import os
+
 import pytest
 
-from aircox_streamer.controllers import Source
+from aircox_streamer.controllers import (
+    Source,
+    PlaylistSource,
+    QueueSource,
+    Request,
+)
 
 
 @pytest.fixture
@@ -8,6 +15,16 @@ def source(controller):
     return Source(controller, 13)
 
 
+@pytest.fixture
+def playlist_source(controller, program):
+    return PlaylistSource(controller, 14, program)
+
+
+@pytest.fixture
+def queue_source(controller):
+    return QueueSource(controller, 15)
+
+
 class TestSource:
     @pytest.mark.django_db
     def test_station(self, source, station):
@@ -22,60 +39,108 @@ class TestSource:
         ]
 
         source.fetch()
-        assert f"{source.id}.remaining" in socket.sent_data
-        assert f"{source.id}.get" in socket.sent_data
+        assert socket.is_sent(f"{source.id}.remaining")
+        assert socket.is_sent(f"{source.id}.get")
 
         assert source.remaining == remaining
-        assert source["request_uri"]
+        assert source.request_status
 
     @pytest.mark.django_db
     def test_skip(self, socket, source):
+        socket.recv_data = "\nEND"
         source.skip()
-        assert f"{source.id}.skip" in socket.sent_data
+        assert socket.is_sent(f"{source.id}.skip\n")
 
     @pytest.mark.django_db
     def test_restart(self, socket, source):
-        source.skip()
+        source.restart()
         prefix = f"{source.id}.seek"
         assert any(r for r in socket.sent_data if r.startswith(prefix))
 
     @pytest.mark.django_db
     def test_seek(self, socket, source):
         source.seek(10)
-        assert f"{source.id}.skip 10" in socket.sent_data
+        assert socket.is_sent(f"{source.id}.seek 10")
 
 
 class TestPlaylistSource:
     @pytest.mark.django_db
-    def test_get_sound_queryset(self):
-        pass
+    def test_get_sound_queryset(self, playlist_source, sounds):
+        query = playlist_source.get_sound_queryset()
+        assert all(
+            r.program_id == playlist_source.program.pk
+            and r.type == r.TYPE_ARCHIVE
+            for r in query
+        )
 
     @pytest.mark.django_db
-    def test_get_playlist(self):
-        pass
+    def test_get_playlist(self, playlist_source, sounds):
+        expected = {r.file.path for r in sounds}
+        query = playlist_source.get_playlist()
+        assert all(r in expected for r in query)
 
     @pytest.mark.django_db
-    def test_write_playlist(self):
-        pass
+    def test_write_playlist(self, playlist_source):
+        playlist = ["/tmp/a", "/tmp/b"]
+        playlist_source.write_playlist(playlist)
+        with open(playlist_source.path, "r") as file:
+            result = file.read()
+        os.remove(playlist_source.path)
+
+        assert result == "\n".join(playlist)
 
     @pytest.mark.django_db
-    def test_stream(self):
-        pass
+    def test_stream(self, playlist_source, stream):
+        result = playlist_source.stream()
+        assert result == {
+            "begin": stream.begin.strftime("%Hh%M"),
+            "end": stream.end.strftime("%Hh%M"),
+            "delay": 0,
+        }
 
     @pytest.mark.django_db
-    def test_sync(self):
-        pass
+    def test_sync(self, playlist_source):
+        # spoof method
+        playlist = ["/tmp/a", "/tmp/b"]
+        written_playlist = []
+        playlist_source.get_playlist = lambda: playlist
+        playlist_source.write_playlist = lambda p: written_playlist.extend(p)
+
+        playlist_source.sync()
+        assert written_playlist == playlist
 
 
 class TestQueueSource:
     @pytest.mark.django_db
-    def test_push(self):
-        pass
+    def test_requests(self, queue_source, socket, metadata_string):
+        queue_source.queue = [13, 14, 15]
+        socket.recv_data = [
+            f"{metadata_string}\nEND" for _ in queue_source.queue
+        ]
+
+        requests = queue_source.requests
+
+        assert all(isinstance(r, Request) for r in requests)
+        assert all(r.uri for r in requests)
 
     @pytest.mark.django_db
-    def test_fetch(self):
-        pass
+    def test_push(self, queue_source, socket):
+        paths = ["/tmp/a", "/tmp/b"]
+        queue_source.push(*paths)
+        assert all(
+            socket.is_sent(f"{queue_source.id}_queue.push {path}")
+            for path in paths
+        )
 
     @pytest.mark.django_db
-    def test_requests(self):
-        pass
+    def test_fetch(self, queue_source, socket, metadata_string):
+        queue = ["13", "14", "15"]
+        socket.recv_data = [
+            # Source fetch remaining & metadata
+            "13 END",
+            metadata_string,
+            " ".join(queue) + "\nEND",
+        ]
+        queue_source.fetch()
+        assert queue_source.uri
+        assert queue_source.queue == queue