Create features and tests for internet radio stations of Subsonic API

Implements:
getInternetRadioStations.view
createInternetRadioStation.view
updateInternetRadioStation.view
deleteInternetRadioStation.view

supysonic/api/__init__.py

@@ -116,4 +116,5 @@ from .chat import *
 from .search import *
 from .playlists import *
 from .jukebox import *
+from .radio import *
 from .unsupported import *

supysonic/api/radio.py

@@ -0,0 +1,73 @@
+# coding: utf-8
+#
+# This file is part of Supysonic.
+# Supysonic is a Python implementation of the Subsonic server API.
+#
+# Copyright (C) 2020 Alban 'spl0k' Féron
+#
+# Distributed under terms of the GNU AGPLv3 license.
+
+from flask import request
+
+from ..db import RadioStation
+
+from . import api, get_entity
+from .exceptions import Forbidden, MissingParameter, NotFound
+
+
+@api.route("/getInternetRadioStations.view", methods=["GET", "POST"])
+def get_radio_stations():
+    query = RadioStation.select().sort_by(
+        RadioStation.name
+    )
+    return request.formatter(
+        "internetRadioStations",
+        dict(internetRadioStation=[p.as_subsonic_station() for p in query]),
+    )
+
+
+@api.route("/createInternetRadioStation.view", methods=["GET", "POST"])
+def create_radio_station():
+    if not request.user.admin:
+        raise Forbidden()
+
+    stream_url, name, homepage_url = map(request.values.get, ["streamUrl", "name", "homepageUrl"])
+
+    if stream_url and name:
+        RadioStation(stream_url=stream_url, name=name, homepage_url=homepage_url)
+    else:
+        raise MissingParameter("streamUrl or name")
+
+    return request.formatter.empty
+
+
+@api.route("/updateInternetRadioStation.view", methods=["GET", "POST"])
+def update_radio_station():
+    if not request.user.admin:
+        raise Forbidden()
+
+    res = get_entity(RadioStation)
+
+    stream_url, name, homepage_url = map(request.values.get, ["streamUrl", "name", "homepageUrl"])
+    if stream_url and name:
+        res.stream_url = stream_url
+        res.name = name
+
+        if homepage_url:
+            res.homepage_url = homepage_url
+    else:
+        raise MissingParameter("streamUrl or name")
+
+    return request.formatter.empty
+
+
+@api.route("/deleteInternetRadioStation.view", methods=["GET", "POST"])
+def delete_radio_station():
+    if not request.user.admin:
+        raise Forbidden()
+
+    res = get_entity(RadioStation)
+    res.delete()
+
+    return request.formatter.empty
+

supysonic/db.py

@@ -553,6 +553,25 @@ class Playlist(db.Entity):
         self.tracks = ",".join(t for t in tracks if t)
 
 
+class RadioStation(db.Entity):
+    _table_ = "radio_station"
+
+    id = PrimaryKey(UUID, default=uuid4)
+    stream_url = Required(str)
+    name = Required(str)
+    homepage_url = Optional(str, nullable=True)
+    created = Required(datetime, precision=0, default=now)
+
+    def as_subsonic_station(self):
+        info = dict(
+            id=str(self.id),
+            streamUrl=self.stream_url,
+            name=self.name,
+            homePageUrl=self.homepage_url,
+        )
+        return info
+
+
 def parse_uri(database_uri):
     if not isinstance(database_uri, str):
         raise TypeError("Expecting a string")

supysonic/schema/migration/mysql/20200607.sql

@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS radio_station (
+    id BINARY(16) PRIMARY KEY,
+    stream_url VARCHAR(256) NOT NULL,
+    name VARCHAR(256) NOT NULL,
+    homepage_url VARCHAR(256),
+    created DATETIME NOT NULL
+) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
+

supysonic/schema/migration/postgres/20200607.sql

@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS radio_station (
+    id UUID PRIMARY KEY,
+    stream_url VARCHAR(256) NOT NULL,
+    name VARCHAR(256) NOT NULL,
+    homepage_url VARCHAR(256),
+    created TIMESTAMP NOT NULL
+);
+

supysonic/schema/migration/sqlite/20200607.sql

@@ -0,0 +1,12 @@
+BEGIN TRANSACTION;
+
+CREATE TABLE IF NOT EXISTS radio_station (
+    id CHAR(36) PRIMARY KEY,
+    stream_url VARCHAR(256) NOT NULL,
+    name VARCHAR(256) NOT NULL,
+    homepage_url VARCHAR(256),
+    created DATETIME NOT NULL
+);
+
+COMMIT;
+VACUUM;

supysonic/schema/mysql.sql

@@ -150,3 +150,11 @@ CREATE TABLE meta (
     value VARCHAR(256) NOT NULL
 ) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
 
+CREATE TABLE IF NOT EXISTS radio_station (
+    id BINARY(16) PRIMARY KEY,
+    stream_url VARCHAR(256) NOT NULL,
+    name VARCHAR(256) NOT NULL,
+    homepage_url VARCHAR(256),
+    created DATETIME NOT NULL
+) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
+

supysonic/schema/postgres.sql

@@ -150,3 +150,11 @@ CREATE TABLE meta (
     value VARCHAR(256) NOT NULL
 );
 
+CREATE TABLE IF NOT EXISTS radio_station (
+    id UUID PRIMARY KEY,
+    stream_url VARCHAR(256) NOT NULL,
+    name VARCHAR(256) NOT NULL,
+    homepage_url VARCHAR(256),
+    created TIMESTAMP NOT NULL
+);
+

supysonic/schema/sqlite.sql

@@ -152,3 +152,11 @@ CREATE TABLE meta (
     value CHAR(256) NOT NULL
 );
 
+CREATE TABLE IF NOT EXISTS radio_station (
+    id CHAR(36) PRIMARY KEY,
+    stream_url VARCHAR(256) NOT NULL,
+    name VARCHAR(256) NOT NULL,
+    homepage_url VARCHAR(256),
+    created DATETIME NOT NULL
+);
+

tests/api/__init__.py

@@ -21,6 +21,7 @@ from .test_album_songs import AlbumSongsTestCase
 from .test_annotation import AnnotationTestCase
 from .test_media import MediaTestCase
 from .test_transcoding import TranscodingTestCase
+from .test_radio import RadioStationTestCase
 
 
 def suite():
@@ -38,5 +39,6 @@ def suite():
     suite.addTest(unittest.makeSuite(AnnotationTestCase))
     suite.addTest(unittest.makeSuite(MediaTestCase))
     suite.addTest(unittest.makeSuite(TranscodingTestCase))
+    suite.addTest(unittest.makeSuite(RadioStationTestCase))
 
     return suite

tests/api/test_radio.py

@@ -0,0 +1,202 @@
+#!/usr/bin/env python
+# coding: utf-8
+#
+# This file is part of Supysonic.
+# Supysonic is a Python implementation of the Subsonic server API.
+#
+# Copyright (C) 2020 Alban 'spl0k' Féron
+#
+# Distributed under terms of the GNU AGPLv3 license.
+
+import uuid
+
+from pony.orm import db_session
+
+from supysonic.db import RadioStation
+
+from .apitestbase import ApiTestBase
+
+
+class RadioStationTestCase(ApiTestBase):
+    def setUp(self):
+        super(RadioStationTestCase, self).setUp()
+
+    @db_session
+    def assertRadioStationCountEqual(self, count):
+        self.assertEqual(RadioStation.select().count(), count)
+
+    def assertRadioStationEquals(self, station, stream_url, name, homepage_url=None):
+        self.assertEqual(station.stream_url, stream_url)
+        self.assertEqual(station.name, name)
+        self.assertEqual(station.homepage_url, homepage_url)
+
+    def test_create_radio_station(self):
+        # test for non-admin access
+        self._make_request(
+            "createInternetRadioStation",
+            {"u": "bob", "p": "B0b", "username": "alice"},
+            error=50
+        )
+
+        # check params
+        self._make_request("createInternetRadioStation", error=10)
+        self._make_request("createInternetRadioStation", {"streamUrl": "missingName"}, error=10)
+        self._make_request("createInternetRadioStation", {"name": "missing stream"}, error=10)
+
+        # create w/ required fields
+        stream_url = "http://example.com/radio/create"
+        name = "radio station"
+
+        self._make_request("createInternetRadioStation", {
+            "streamUrl": stream_url,
+            "name": name,
+        })
+
+        # the correct value is 2 because _make_request uses GET then POST
+        self.assertRadioStationCountEqual(2)
+
+        with db_session:
+            for rs in RadioStation.select():
+                self.assertRadioStationEquals(rs, stream_url, name)
+
+            RadioStation.select().delete(bulk=True)
+
+        # create w/ all fields
+        stream_url = "http://example.com/radio/create1"
+        name = "radio station1"
+        homepage_url = "http://example.com/home"
+
+        self._make_request("createInternetRadioStation", {
+            "streamUrl": stream_url,
+            "name": name,
+            "homepageUrl": homepage_url,
+        })
+
+        # the correct value is 2 because _make_request uses GET then POST
+        self.assertRadioStationCountEqual(2)
+
+        with db_session:
+            for rs in RadioStation.select():
+                self.assertRadioStationEquals(rs, stream_url, name, homepage_url)
+
+    def test_update_radio_station(self):
+        self._make_request(
+            "updateInternetRadioStation",
+            {"u": "bob", "p": "B0b", "username": "alice"},
+            error=50
+        )
+
+        # test data
+        test = {
+            "stream_url": "http://example.com/radio/update",
+            "name": "Radio Update",
+            "homepage_url": "http://example.com/update",
+        }
+        update = {
+            "stream_url": test["stream_url"] + "-1",
+            "name": test["name"] + "-1",
+            "homepage_url": test["homepage_url"] + "-1",
+        }
+
+        # load a test record
+        with db_session:
+            station = RadioStation(
+                stream_url=test["stream_url"],
+                name=test["name"],
+                homepage_url=test["homepage_url"],
+            )
+
+        # check params
+        self._make_request("updateInternetRadioStation", {
+            "id": station.id, "homepageUrl": "missing required params",
+        }, error=10)
+        self._make_request("updateInternetRadioStation", {
+            "id": station.id, "name": "missing streamUrl",
+        }, error=10)
+        self._make_request("updateInternetRadioStation", {
+            "id": station.id, "streamUrl": "missing name",
+        }, error=10)
+
+        # update the record w/ required fields
+        self._make_request("updateInternetRadioStation", {
+            "id": station.id,
+            "streamUrl": update["stream_url"],
+            "name": update["name"],
+        })
+
+        with db_session:
+            rs_update = RadioStation[station.id]
+
+        self.assertRadioStationEquals(rs_update, update["stream_url"], update["name"], test["homepage_url"])
+
+        # update the record w/ all fields
+        self._make_request("updateInternetRadioStation", {
+            "id": station.id,
+            "streamUrl": update["stream_url"],
+            "name": update["name"],
+            "homepageUrl": update["homepage_url"],
+        })
+
+        with db_session:
+            rs_update = RadioStation[station.id]
+
+        self.assertRadioStationEquals(rs_update, update["stream_url"], update["name"], update["homepage_url"])
+
+    def test_delete_radio_station(self):
+        # test for non-admin access
+        self._make_request(
+            "deleteInternetRadioStation",
+            {"u": "bob", "p": "B0b", "username": "alice"},
+            error=50
+        )
+
+        # check params
+        self._make_request("deleteInternetRadioStation", error=10)
+        self._make_request("deleteInternetRadioStation", {"id": 1}, error=0)
+        self._make_request("deleteInternetRadioStation", {"id": str(uuid.uuid4())}, error=70)
+
+        # delete
+        with db_session:
+            station = RadioStation(
+                stream_url="http://example.com/radio/delete",
+                name="Radio Delete",
+                homepage_url="http://example.com/update"
+            )
+
+        self._make_request("deleteInternetRadioStation", {"id": station.id}, skip_post=True)
+
+        self.assertRadioStationCountEqual(0)
+
+
+    def test_get_radio_stations(self):
+        test_range = 3
+        with db_session:
+            for x in range(0, test_range):
+                RadioStation(
+                    stream_url="http://example.com/radio-{}".format(x),
+                    name="Radio {}".format(x),
+                    homepage_url="http://example.com/update-{}".format(x),
+                )
+
+        # verify happy path is clean
+        self.assertRadioStationCountEqual(test_range)
+        rv, child = self._make_request("getInternetRadioStations", tag="internetRadioStations")
+        self.assertEqual(len(child), test_range)
+        # This order is guaranteed to work because the api returns in order by name.
+        # Test data is sequential by design.
+        for x in range(0, test_range):
+            station = child[x]
+            self.assertTrue(station.get("streamUrl").endswith("radio-{}".format(x)))
+            self.assertTrue(station.get("name").endswith("Radio {}".format(x)))
+            self.assertTrue(station.get("homePageUrl").endswith("update-{}".format(x)))
+
+
+        # test for non-admin access
+        rv, child = self._make_request(
+            "getInternetRadioStations",
+            {"u": "bob", "p": "B0b", "username": "alice"},
+            tag="internetRadioStations",
+        )
+
+        self.assertEqual(len(child), test_range)
+