Porting supysonic.frontend

supysonic/frontend/__init__.py

@@ -1,7 +1,7 @@
 # This file is part of Supysonic.
 # Supysonic is a Python implementation of the Subsonic server API.
 #
-# Copyright (C) 2013-2021 Alban 'spl0k' Féron
+# Copyright (C) 2013-2022 Alban 'spl0k' Féron
 #                    2017 Óscar García Amor
 #
 # Distributed under terms of the GNU AGPLv3 license.
@@ -17,7 +17,6 @@ from flask import (
 )
 from flask import Blueprint
 from functools import wraps
-from pony.orm import ObjectNotFound
 
 from .. import VERSION, DOWNLOAD_URL
 from ..daemon.client import DaemonClient
@@ -42,7 +41,7 @@ def login_check():
             user = UserManager.get(session.get("userid"))
             request.user = user
             should_login = False
-        except (ValueError, ObjectNotFound):
+        except (ValueError, User.DoesNotExist):
             session.clear()
 
     if should_login and request.endpoint != "frontend.login":

supysonic/frontend/folder.py

@@ -1,12 +1,11 @@
 # This file is part of Supysonic.
 # Supysonic is a Python implementation of the Subsonic server API.
 #
-# Copyright (C) 2013-2019 Alban 'spl0k' Féron
+# Copyright (C) 2013-2022 Alban 'spl0k' Féron
 #
 # Distributed under terms of the GNU AGPLv3 license.
 
 from flask import current_app, flash, redirect, render_template, request, url_for
-from pony.orm import ObjectNotFound
 
 from ..daemon.client import DaemonClient
 from ..daemon.exceptions import DaemonUnavailableError
@@ -29,7 +28,9 @@ def folder_index():
             "warning",
         )
     return render_template(
-        "folders.html", folders=Folder.select(lambda f: f.root), allow_scan=allow_scan
+        "folders.html",
+        folders=Folder.select().where(Folder.root),
+        allow_scan=allow_scan,
     )
 
 
@@ -71,7 +72,7 @@ def del_folder(id):
         flash("Deleted folder")
     except ValueError as e:
         flash(str(e), "error")
-    except ObjectNotFound:
+    except Folder.DoesNotExist:
         flash("No such folder", "error")
 
     return redirect(url_for("frontend.folder_index"))
@@ -90,7 +91,7 @@ def scan_folder(id=None):
         flash("Scanning started")
     except ValueError as e:
         flash(str(e), "error")
-    except ObjectNotFound:
+    except Folder.DoesNotExist:
         flash("No such folder", "error")
     except DaemonUnavailableError:
         flash("Can't start scan", "error")

supysonic/frontend/playlist.py

@@ -1,14 +1,14 @@
 # This file is part of Supysonic.
 # Supysonic is a Python implementation of the Subsonic server API.
 #
-# Copyright (C) 2013-2018 Alban 'spl0k' Féron
+# Copyright (C) 2013-2022 Alban 'spl0k' Féron
 #
 # Distributed under terms of the GNU AGPLv3 license.
 
 import uuid
 
 from flask import Response, flash, redirect, render_template, request, url_for
-from pony.orm import ObjectNotFound
+from functools import wraps
 
 from ..db import Playlist
 
@@ -19,42 +19,40 @@ from . import frontend
 def playlist_index():
     return render_template(
         "playlists.html",
-        mine=Playlist.select(lambda p: p.user == request.user),
+        mine=Playlist.select().where(Playlist.user == request.user),
-        others=Playlist.select(lambda p: p.user != request.user and p.public),
+        others=Playlist.select().where(Playlist.user != request.user, Playlist.public),
     )
 
 
+def resolve_and_inject_playlist(func):
+    @wraps(func)
+    def decorated(uid):
+        try:
+            uid = uuid.UUID(uid)
+        except ValueError:
+            flash("Invalid playlist id")
+            return redirect(url_for("frontend.playlist_index"))
+
+        try:
+            playlist = Playlist[uid]
+        except Playlist.DoesNotExist:
+            flash("Unknown playlist")
+            return redirect(url_for("frontend.playlist_index"))
+
+        return func(uid, playlist)
+
+    return decorated
+
+
 @frontend.route("/playlist/<uid>")
-def playlist_details(uid):
+@resolve_and_inject_playlist
-    try:
+def playlist_details(uid, playlist):
-        uid = uuid.UUID(uid)
-    except ValueError:
-        flash("Invalid playlist id")
-        return redirect(url_for("frontend.playlist_index"))
-
-    try:
-        playlist = Playlist[uid]
-    except ObjectNotFound:
-        flash("Unknown playlist")
-        return redirect(url_for("frontend.playlist_index"))
-
     return render_template("playlist.html", playlist=playlist)
 
 
 @frontend.route("/playlist/<uid>/export")
-def playlist_export(uid):
+@resolve_and_inject_playlist
-    try:
+def playlist_export(uid, playlist):
-        uid = uuid.UUID(uid)
-    except ValueError:
-        flash("Invalid playlist id")
-        return redirect(url_for("frontend.playlist_index"))
-
-    try:
-        playlist = Playlist[uid]
-    except ObjectNotFound:
-        flash("Unknown playlist")
-        return redirect(url_for("frontend.playlist_index"))
-
     return Response(
         render_template("playlist_export.m3u", playlist=playlist),
         mimetype="audio/mpegurl",
@@ -65,20 +63,9 @@ def playlist_export(uid):
 
 
 @frontend.route("/playlist/<uid>", methods=["POST"])
-def playlist_update(uid):
+@resolve_and_inject_playlist
-    try:
+def playlist_update(uid, playlist):
-        uid = uuid.UUID(uid)
+    if playlist.user_id != request.user.id:
-    except ValueError:
-        flash("Invalid playlist id")
-        return redirect(url_for("frontend.playlist_index"))
-
-    try:
-        playlist = Playlist[uid]
-    except ObjectNotFound:
-        flash("Unknown playlist")
-        return redirect(url_for("frontend.playlist_index"))
-
-    if playlist.user.id != request.user.id:
         flash("You're not allowed to edit this playlist")
     elif not request.form.get("name"):
         flash("Missing playlist name")
@@ -92,29 +79,19 @@ def playlist_update(uid):
             "on",
             "checked",
         )
+        playlist.save()
         flash("Playlist updated.")
 
     return playlist_details(str(uid))
 
 
 @frontend.route("/playlist/del/<uid>")
-def playlist_delete(uid):
+@resolve_and_inject_playlist
-    try:
+def playlist_delete(uid, playlist):
-        uid = uuid.UUID(uid)
+    if playlist.user_id != request.user.id:
-    except ValueError:
-        flash("Invalid playlist id")
-        return redirect(url_for("frontend.playlist_index"))
-
-    try:
-        playlist = Playlist[uid]
-    except ObjectNotFound:
-        flash("Unknown playlist")
-        return redirect(url_for("frontend.playlist_index"))
-
-    if playlist.user.id != request.user.id:
         flash("You're not allowed to delete this playlist")
     else:
-        playlist.delete()
+        playlist.delete_instance()
         flash("Playlist deleted")
 
     return redirect(url_for("frontend.playlist_index"))

supysonic/frontend/user.py

@@ -1,7 +1,7 @@
 # This file is part of Supysonic.
 # Supysonic is a Python implementation of the Subsonic server API.
 #
-# Copyright (C) 2013-2018 Alban 'spl0k' Féron
+# Copyright (C) 2013-2022 Alban 'spl0k' Féron
 #
 # Distributed under terms of the GNU AGPLv3 license.
 
@@ -10,9 +10,8 @@ import logging
 from flask import flash, redirect, render_template, request, session, url_for
 from flask import current_app
 from functools import wraps
-from pony.orm import ObjectNotFound
 
-from ..db import User
+from ..db import ClientPrefs, User
 from ..lastfm import LastFm
 from ..managers.user import UserManager
 
@@ -39,7 +38,7 @@ def me_or_uuid(f, arg="uid"):
             except ValueError as e:
                 flash(str(e), "error")
                 return redirect(url_for("frontend.index"))
-            except ObjectNotFound:
+            except User.DoesNotExist:
                 flash("No such user", "error")
                 return redirect(url_for("frontend.index"))
 
@@ -91,7 +90,7 @@ def update_clients(uid, user):
     logger.debug(clients_opts)
 
     for client, opts in clients_opts.items():
-        prefs = user.clients.select(lambda c: c.client_name == client).first()
+        prefs = user.clients.where(ClientPrefs.client_name == client).first()
         if prefs is None:
             continue
 
@@ -102,13 +101,14 @@ def update_clients(uid, user):
             "selected",
             "1",
         ]:
-            prefs.delete()
+            prefs.delete_instance()
             continue
 
         prefs.format = opts["format"] if "format" in opts and opts["format"] else None
         prefs.bitrate = (
             int(opts["bitrate"]) if "bitrate" in opts and opts["bitrate"] else None
         )
+        prefs.save()
 
     flash("Clients preferences updated.")
     return user_profile(uid, user)
@@ -122,7 +122,7 @@ def change_username_form(uid):
     except ValueError as e:
         flash(str(e), "error")
         return redirect(url_for("frontend.index"))
-    except ObjectNotFound:
+    except User.DoesNotExist:
         flash("No such user", "error")
         return redirect(url_for("frontend.index"))
 
@@ -137,7 +137,7 @@ def change_username_post(uid):
     except ValueError as e:
         flash(str(e), "error")
         return redirect(url_for("frontend.index"))
-    except ObjectNotFound:
+    except User.DoesNotExist:
         flash("No such user", "error")
         return redirect(url_for("frontend.index"))
 
@@ -145,9 +145,13 @@ def change_username_post(uid):
     if username in ("", None):
         flash("The username is required")
         return render_template("change_username.html", user=user)
-    if user.name != username and User.get(name=username) is not None:
+    if user.name != username:
-        flash("This name is already taken")
+        try:
-        return render_template("change_username.html", user=user)
+            User.get(name=username)
+            flash("This name is already taken")
+            return render_template("change_username.html", user=user)
+        except User.DoesNotExist:
+            pass
 
     if request.form.get("admin") is None:
         admin = False
@@ -157,6 +161,7 @@ def change_username_post(uid):
     if user.name != username or user.admin != admin:
         user.name = username
         user.admin = admin
+        user.save()
         flash(f"User '{username}' updated.")
     else:
         flash(f"No changes for '{username}'.")
@@ -262,7 +267,7 @@ def del_user(uid):
         flash("Deleted user")
     except ValueError as e:
         flash(str(e), "error")
-    except ObjectNotFound:
+    except User.DoesNotExist:
         flash("No such user", "error")
 
     return redirect(url_for("frontend.user_index"))

tests/frontend/test_folder.py

@@ -1,14 +1,12 @@
 # This file is part of Supysonic.
 # Supysonic is a Python implementation of the Subsonic server API.
 #
-# Copyright (C) 2017-2019 Alban 'spl0k' Féron
+# Copyright (C) 2017-2022 Alban 'spl0k' Féron
 #
 # Distributed under terms of the GNU AGPLv3 license.
 
 import unittest
 
-from pony.orm import db_session
-
 from supysonic.db import Folder
 
 from .frontendtestbase import FrontendTestBase
@@ -53,18 +51,15 @@ class FolderTestCase(FrontendTestBase):
             follow_redirects=True,
         )
         self.assertIn("created", rv.data)
-        with db_session:
+        self.assertEqual(Folder.select().count(), 1)
-            self.assertEqual(Folder.select().count(), 1)
 
     def test_delete(self):
-        with db_session:
+        folder = Folder.create(name="folder", path="tests/assets", root=True)
-            folder = Folder(name="folder", path="tests/assets", root=True)
 
         self._login("bob", "B0b")
         rv = self.client.get("/folder/del/" + str(folder.id), follow_redirects=True)
         self.assertIn("There's nothing much to see", rv.data)
-        with db_session:
+        self.assertEqual(Folder.select().count(), 1)
-            self.assertEqual(Folder.select().count(), 1)
         self._logout()
 
         self._login("alice", "Alic3")
@@ -74,12 +69,10 @@ class FolderTestCase(FrontendTestBase):
         self.assertIn("No such folder", rv.data)
         rv = self.client.get("/folder/del/" + str(folder.id), follow_redirects=True)
         self.assertIn("Music folders", rv.data)
-        with db_session:
+        self.assertEqual(Folder.select().count(), 0)
-            self.assertEqual(Folder.select().count(), 0)
 
     def test_scan(self):
-        with db_session:
+        folder = Folder.create(name="folder", path="tests/assets/folder", root=True)
-            folder = Folder(name="folder", path="tests/assets/folder", root=True)
 
         self._login("alice", "Alic3")
 

tests/frontend/test_login.py

@@ -1,7 +1,7 @@
 # This file is part of Supysonic.
 # Supysonic is a Python implementation of the Subsonic server API.
 #
-# Copyright (C) 2013-2017 Alban 'spl0k' Féron
+# Copyright (C) 2013-2022 Alban 'spl0k' Féron
 #                    2017 Óscar García Amor
 #
 # Distributed under terms of the GNU AGPLv3 license.
@@ -9,8 +9,6 @@
 import unittest
 import uuid
 
-from pony.orm import db_session
-
 from supysonic.db import User
 
 from .frontendtestbase import FrontendTestBase
@@ -50,9 +48,8 @@ class LoginTestCase(FrontendTestBase):
 
     def test_root_with_valid_session(self):
         # Root with valid session
-        with db_session:
+        with self.client.session_transaction() as sess:
-            with self.client.session_transaction() as sess:
+            sess["userid"] = User.get(name="alice").id
-                sess["userid"] = User.get(name="alice").id
         rv = self.client.get("/", follow_redirects=True)
         self.assertIn("alice", rv.data)
         self.assertIn("Log out", rv.data)

tests/frontend/test_playlist.py

@@ -1,15 +1,13 @@
 # This file is part of Supysonic.
 # Supysonic is a Python implementation of the Subsonic server API.
 #
-# Copyright (C) 2017-2018 Alban 'spl0k' Féron
+# Copyright (C) 2017-2022 Alban 'spl0k' Féron
 #
 # Distributed under terms of the GNU AGPLv3 license.
 
 import unittest
 import uuid
 
-from pony.orm import db_session
-
 from supysonic.db import Folder, Artist, Album, Track, Playlist, User
 
 from .frontendtestbase import FrontendTestBase
@@ -19,28 +17,28 @@ class PlaylistTestCase(FrontendTestBase):
     def setUp(self):
         super().setUp()
 
-        with db_session:
+        folder = Folder.create(name="Root", path="tests/assets", root=True)
-            folder = Folder(name="Root", path="tests/assets", root=True)
+        artist = Artist.create(name="Artist!")
-            artist = Artist(name="Artist!")
+        album = Album.create(name="Album!", artist=artist)
-            album = Album(name="Album!", artist=artist)
 
-            track = Track(
+        track = Track.create(
-                path="tests/assets/23bytes",
+            path="tests/assets/23bytes",
-                title="23bytes",
+            title="23bytes",
-                artist=artist,
+            artist=artist,
-                album=album,
+            album=album,
-                folder=folder,
+            folder=folder,
-                root_folder=folder,
+            root_folder=folder,
-                duration=2,
+            duration=2,
-                disc=1,
+            disc=1,
-                number=1,
+            number=1,
-                bitrate=320,
+            bitrate=320,
-                last_modification=0,
+            last_modification=0,
-            )
+        )
 
-            playlist = Playlist(name="Playlist!", user=User.get(name="alice"))
+        playlist = Playlist.create(name="Playlist!", user=User.get(name="alice"))
-            for _ in range(4):
+        for _ in range(4):
-                playlist.add(track)
+            playlist.add(track)
+        playlist.save()
 
         self.playlistid = playlist.id
 
@@ -80,8 +78,7 @@ class PlaylistTestCase(FrontendTestBase):
         )
         self.assertNotIn("updated", rv.data)
         self.assertIn("Missing", rv.data)
-        with db_session:
+        self.assertEqual(Playlist[self.playlistid].name, "Playlist!")
-            self.assertEqual(Playlist[self.playlistid].name, "Playlist!")
 
         rv = self.client.post(
             "/playlist/" + str(self.playlistid),
@@ -90,10 +87,9 @@ class PlaylistTestCase(FrontendTestBase):
         )
         self.assertIn("updated", rv.data)
         self.assertNotIn("not allowed", rv.data)
-        with db_session:
+        playlist = Playlist[self.playlistid]
-            playlist = Playlist[self.playlistid]
+        self.assertEqual(playlist.name, "abc")
-            self.assertEqual(playlist.name, "abc")
+        self.assertTrue(playlist.public)
-            self.assertTrue(playlist.public)
 
     def test_delete(self):
         self._login("bob", "B0b")
@@ -107,8 +103,7 @@ class PlaylistTestCase(FrontendTestBase):
             "/playlist/del/" + str(self.playlistid), follow_redirects=True
         )
         self.assertIn("not allowed", rv.data)
-        with db_session:
+        self.assertEqual(Playlist.select().count(), 1)
-            self.assertEqual(Playlist.select().count(), 1)
         self._logout()
 
         self._login("alice", "Alic3")
@@ -116,8 +111,7 @@ class PlaylistTestCase(FrontendTestBase):
             "/playlist/del/" + str(self.playlistid), follow_redirects=True
         )
         self.assertIn("deleted", rv.data)
-        with db_session:
+        self.assertEqual(Playlist.select().count(), 0)
-            self.assertEqual(Playlist.select().count(), 0)
 
 
 if __name__ == "__main__":

tests/frontend/test_user.py

@@ -9,7 +9,6 @@ import unittest
 import uuid
 
 from flask import escape
-from pony.orm import db_session
 
 from supysonic.db import User, ClientPrefs
 
@@ -20,8 +19,7 @@ class UserTestCase(FrontendTestBase):
     def setUp(self):
         super().setUp()
 
-        with db_session:
+        self.users = {u.name: u.id for u in User.select()}
-            self.users = {u.name: u.id for u in User.select()}
 
     def test_index(self):
         self._login("bob", "B0b")
@@ -44,8 +42,7 @@ class UserTestCase(FrontendTestBase):
         self.assertIn("bob", rv.data)
         self._logout()
 
-        with db_session:
+        ClientPrefs.create(user=User[self.users["bob"]], client_name="tests")
-            ClientPrefs(user=User[self.users["bob"]], client_name="tests")
 
         self._login("bob", "B0b")
         rv = self.client.get("/user/" + str(self.users["alice"]), follow_redirects=True)
@@ -66,21 +63,18 @@ class UserTestCase(FrontendTestBase):
         self.client.post("/user/me", data={"n_": "o"})
         self.client.post("/user/me", data={"inexisting_client": "setting"})
 
-        with db_session:
+        ClientPrefs.create(user=User[self.users["alice"]], client_name="tests")
-            ClientPrefs(user=User[self.users["alice"]], client_name="tests")
 
         rv = self.client.post(
             "/user/me", data={"tests_format": "mp3", "tests_bitrate": 128}
         )
         self.assertIn("updated", rv.data)
-        with db_session:
+        prefs = ClientPrefs[User[self.users["alice"]], "tests"]
-            prefs = ClientPrefs[User[self.users["alice"]], "tests"]
+        self.assertEqual(prefs.format, "mp3")
-            self.assertEqual(prefs.format, "mp3")
+        self.assertEqual(prefs.bitrate, 128)
-            self.assertEqual(prefs.bitrate, 128)
 
         self.client.post("/user/me", data={"tests_delete": 1})
-        with db_session:
+        self.assertEqual(ClientPrefs.select().count(), 0)
-            self.assertEqual(ClientPrefs.select().count(), 0)
 
     def test_change_username_get(self):
         self._login("bob", "B0b")
@@ -116,13 +110,11 @@ class UserTestCase(FrontendTestBase):
         )
         self.assertIn("updated", rv.data)
         self.assertIn("b0b", rv.data)
-        with db_session:
+        bob = User[self.users["bob"]]
-            bob = User[self.users["bob"]]
+        self.assertEqual(bob.name, "b0b")
-            self.assertEqual(bob.name, "b0b")
+        self.assertTrue(bob.admin)
-            self.assertTrue(bob.admin)
         rv = self.client.post(path, data={"user": "alice"}, follow_redirects=True)
-        with db_session:
+        self.assertEqual(User[self.users["bob"]].name, "b0b")
-            self.assertEqual(User[self.users["bob"]].name, "b0b")
 
     def test_change_mail_get(self):
         self._login("alice", "Alic3")
@@ -208,8 +200,7 @@ class UserTestCase(FrontendTestBase):
             data={"user": "alice", "passwd": "passwd", "passwd_confirm": "passwd"},
         )
         self.assertIn(escape("User 'alice' exists"), rv.data)
-        with db_session:
+        self.assertEqual(User.select().count(), 2)
-            self.assertEqual(User.select().count(), 2)
 
         rv = self.client.post(
             "/user/add",
@@ -222,8 +213,7 @@ class UserTestCase(FrontendTestBase):
             follow_redirects=True,
         )
         self.assertIn("added", rv.data)
-        with db_session:
+        self.assertEqual(User.select().count(), 3)
-            self.assertEqual(User.select().count(), 3)
         self._logout()
         rv = self._login("user", "passwd")
         self.assertIn("Logged in", rv.data)
@@ -234,8 +224,7 @@ class UserTestCase(FrontendTestBase):
         self._login("bob", "B0b")
         rv = self.client.get(path, follow_redirects=True)
         self.assertIn("There's nothing much to see", rv.data)
-        with db_session:
+        self.assertEqual(User.select().count(), 2)
-            self.assertEqual(User.select().count(), 2)
         self._logout()
 
         self._login("alice", "Alic3")
@@ -245,8 +234,7 @@ class UserTestCase(FrontendTestBase):
         self.assertIn("No such user", rv.data)
         rv = self.client.get(path, follow_redirects=True)
         self.assertIn("Deleted", rv.data)
-        with db_session:
+        self.assertEqual(User.select().count(), 1)
-            self.assertEqual(User.select().count(), 1)
         self._logout()
         rv = self._login("bob", "B0b")
         self.assertIn("Wrong username or password", rv.data)