github/supysonic
1
0
Fork 0
mirror of https://github.com/spl0k/supysonic.git synced 2024-12-23 01:16:18 +00:00
Code Issues Releases Wiki Activity

Getting out of the storm on a pony

Browse Source
This commit is contained in:
spl0k 2017-12-14 22:28:49 +01:00
parent 8046457661
commit 6bd61e0388
5 changed files with 401 additions and 381 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
README.md
View File

@ -2,8 +2,9 @@
Supysonic is a Python implementation of the [Subsonic][] server API. Supysonic is a Python implementation of the [Subsonic][] server API.
[![Build Status](https://travis-ci.org/spl0k/supysonic.svg?branch=master)](https://travis-ci.org/spl0k/supysonic) [![Build Status](https://travis-ci.org/spl0k/supysonic.svg?branch=pony)](https://travis-ci.org/spl0k/supysonic)
[![codecov](https://codecov.io/gh/spl0k/supysonic/branch/master/graph/badge.svg)](https://codecov.io/gh/spl0k/supysonic) [![codecov](https://codecov.io/gh/spl0k/supysonic/branch/pony/graph/badge.svg)](https://codecov.io/gh/spl0k/supysonic)
![Python](https://img.shields.io/badge/python-2.7-blue.svg)
Current supported features are: Current supported features are:
* browsing (by folders or tags) * browsing (by folders or tags)
@ -50,27 +51,21 @@ You'll need these to run Supysonic:
* Python 2.7 * Python 2.7
* [Flask](http://flask.pocoo.org/) >= 0.9 * [Flask](http://flask.pocoo.org/) >= 0.9
* [Storm](https://storm.canonical.com/) * [PonyORM](https://ponyorm.com/)
* [Python Imaging Library](https://github.com/python-pillow/Pillow) * [Python Imaging Library](https://github.com/python-pillow/Pillow)
* [simplejson](https://simplejson.readthedocs.io/en/latest/) * [simplejson](https://simplejson.readthedocs.io/en/latest/)
* [requests](http://docs.python-requests.org/) * [requests](http://docs.python-requests.org/)
* [mutagen](https://mutagen.readthedocs.io/en/latest/) * [mutagen](https://mutagen.readthedocs.io/en/latest/)
* [watchdog](https://github.com/gorakhargosh/watchdog) * [watchdog](https://github.com/gorakhargosh/watchdog)
On a Debian-like OS (Debian, Ubuntu, Linux Mint, etc.), you can install them You can install all of them using `pip`:
this way:
$ apt-get install python-flask python-storm python-imaging python-simplesjon python-requests python-mutagen python-watchdog $ pip install -r requirements.txt
You may also need a database specific package: You may also need a database specific package:
* MySQL: `apt install python-mysqldb` * MySQL: `pip install pymysql` or `pip install mysqlclient`
* PostgreSQL: `apt-install python-psycopg2` * PostgreSQL: `pip install psycopg2`
Due to a bug in `storm`, `psycopg2` version 2.5 and later does not work
properly. You can either use version 2.4 or [patch storm][storm] yourself.
[storm]: https://bugs.launchpad.net/storm/+bug/1170063
### Configuration ### Configuration
@ -84,7 +79,7 @@ The sample configuration (`config.sample`) looks like this:
```ini ```ini
[base] [base]
; A Storm database URI. See the 'schema' folder for schema creation scripts ; A database URI. See the 'schema' folder for schema creation scripts
; Default: sqlite:///tmp/supysonic/supysonic.db ; Default: sqlite:///tmp/supysonic/supysonic.db
;database_uri = sqlite:////var/supysonic/supysonic.db ;database_uri = sqlite:////var/supysonic/supysonic.db
;database_uri = mysql://supysonic:supysonic@localhost/supysonic ;database_uri = mysql://supysonic:supysonic@localhost/supysonic

2
config.sample
View File

@ -1,5 +1,5 @@
[base] [base]
; A Storm database URI. See the 'schema' folder for schema creation scripts ; A database URI. See the 'schema' folder for schema creation scripts
; Default: sqlite:///tmp/supysonic/supysonic.db ; Default: sqlite:///tmp/supysonic/supysonic.db
;database_uri = sqlite:////var/supysonic/supysonic.db ;database_uri = sqlite:////var/supysonic/supysonic.db
;database_uri = mysql://supysonic:supysonic@localhost/supysonic ;database_uri = mysql://supysonic:supysonic@localhost/supysonic

2
requirements.txt
View File

@ -1,5 +1,5 @@
flask>=0.9 flask>=0.9
storm pony
Pillow Pillow
simplejson simplejson
requests>=1.0.0 requests>=1.0.0

398
supysonic/db.py
View File

@ -18,45 +18,41 @@
# You should have received a copy of the GNU Affero General Public License # You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>. # along with this program. If not, see <http://www.gnu.org/licenses/>.
from storm.properties import * import time
from storm.references import Reference, ReferenceSet
from storm.database import create_database
from storm.store import Store
from storm.variables import Variable
import uuid, datetime, time
import mimetypes import mimetypes
import os.path import os.path
from datetime import datetime
from pony.orm import Database, Required, Optional, Set, PrimaryKey
from pony.orm import ObjectNotFound
from pony.orm import min, max, avg, sum
from urlparse import urlparse
from uuid import UUID, uuid4
def now(): def now():
return datetime.datetime.now().replace(microsecond = 0) return datetime.now().replace(microsecond = 0)
class UnicodeOrStrVariable(Variable): db = Database()
__slots__ = ()
def parse_set(self, value, from_db): class Folder(db.Entity):
if isinstance(value, unicode): _table_ = 'folder'
return value
elif isinstance(value, str):
return unicode(value)
raise TypeError("Expected unicode, found %r: %r" % (type(value), value))
Unicode.variable_class = UnicodeOrStrVariable id = PrimaryKey(UUID, default = uuid4)
root = Required(bool, default = False)
name = Required(str)
path = Required(str, unique = True)
created = Required(datetime, precision = 0, default = now)
has_cover_art = Required(bool, default = False)
last_scan = Required(int, default = 0)
class Folder(object): parent = Optional(lambda: Folder, reverse = 'children', column = 'parent_id')
__storm_table__ = 'folder' children = Set(lambda: Folder, reverse = 'parent')
id = UUID(primary = True, default_factory = uuid.uuid4) __alltracks = Set(lambda: Track, lazy = True, reverse = 'root_folder') # Never used, hide it. Could be huge, lazy load
root = Bool(default = False) tracks = Set(lambda: Track, reverse = 'folder')
name = Unicode()
path = Unicode() # unique
created = DateTime(default_factory = now)
has_cover_art = Bool(default = False)
last_scan = Int(default = 0)
parent_id = UUID() # nullable stars = Set(lambda: StarredFolder)
parent = Reference(parent_id, id) ratings = Set(lambda: RatingFolder)
children = ReferenceSet(id, parent_id)
def as_subsonic_child(self, user): def as_subsonic_child(self, user):
info = { info = {
@ -67,29 +63,36 @@ class Folder(object):
'created': self.created.isoformat() 'created': self.created.isoformat()
} }
if not self.root: if not self.root:
info['parent'] = str(self.parent_id) info['parent'] = str(self.parent.id)
info['artist'] = self.parent.name info['artist'] = self.parent.name
if self.has_cover_art: if self.has_cover_art:
info['coverArt'] = str(self.id) info['coverArt'] = str(self.id)
starred = Store.of(self).get(StarredFolder, (user.id, self.id)) try:
if starred: starred = StarredFolder[user.id, self.id]
info['starred'] = starred.date.isoformat() info['starred'] = starred.date.isoformat()
except ObjectNotFound: pass
rating = Store.of(self).get(RatingFolder, (user.id, self.id)) try:
if rating: rating = RatingFolder[user.id, self.id]
info['userRating'] = rating.rating info['userRating'] = rating.rating
avgRating = Store.of(self).find(RatingFolder, RatingFolder.rated_id == self.id).avg(RatingFolder.rating) except ObjectNotFound: pass
avgRating = avg(self.ratings.rating)
if avgRating: if avgRating:
info['averageRating'] = avgRating info['averageRating'] = avgRating
return info return info
class Artist(object): class Artist(db.Entity):
__storm_table__ = 'artist' _table_ = 'artist'
id = UUID(primary = True, default_factory = uuid.uuid4) id = PrimaryKey(UUID, default = uuid4)
name = Unicode() # unique name = Required(str, unique = True)
albums = Set(lambda: Album)
tracks = Set(lambda: Track)
stars = Set(lambda: StarredArtist)
def as_subsonic_artist(self, user): def as_subsonic_artist(self, user):
info = { info = {
@ -99,38 +102,42 @@ class Artist(object):
'albumCount': self.albums.count() 'albumCount': self.albums.count()
} }
starred = Store.of(self).get(StarredArtist, (user.id, self.id)) try:
if starred: starred = StarredArtist[user.id, self.id]
info['starred'] = starred.date.isoformat() info['starred'] = starred.date.isoformat()
except ObjectNotFound: pass
return info return info
class Album(object): class Album(db.Entity):
__storm_table__ = 'album' _table_ = 'album'
id = UUID(primary = True, default_factory = uuid.uuid4) id = PrimaryKey(UUID, default = uuid4)
name = Unicode() name = Required(str)
artist_id = UUID() artist = Required(Artist, column = 'artist_id')
artist = Reference(artist_id, Artist.id) tracks = Set(lambda: Track)
stars = Set(lambda: StarredAlbum)
def as_subsonic_album(self, user): def as_subsonic_album(self, user):
info = { info = {
'id': str(self.id), 'id': str(self.id),
'name': self.name, 'name': self.name,
'artist': self.artist.name, 'artist': self.artist.name,
'artistId': str(self.artist_id), 'artistId': str(self.artist.id),
'songCount': self.tracks.count(), 'songCount': self.tracks.count(),
'duration': sum(self.tracks.values(Track.duration)), 'duration': sum(self.tracks.duration),
'created': min(self.tracks.values(Track.created)).isoformat() 'created': min(self.tracks.created).isoformat()
} }
track_with_cover = self.tracks.find(Track.folder_id == Folder.id, Folder.has_cover_art).any() track_with_cover = self.tracks.select(lambda t: t.folder.has_cover_art)[:1][0]
if track_with_cover: if track_with_cover:
info['coverArt'] = str(track_with_cover.folder_id) info['coverArt'] = str(track_with_cover.folder.id)
starred = Store.of(self).get(StarredAlbum, (user.id, self.id)) try:
if starred: starred = StarredAlbum[user.id, self.id]
info['starred'] = starred.date.isoformat() info['starred'] = starred.date.isoformat()
except ObjectNotFound: pass
return info return info
@ -138,41 +145,42 @@ class Album(object):
year = min(map(lambda t: t.year if t.year else 9999, self.tracks)) year = min(map(lambda t: t.year if t.year else 9999, self.tracks))
return '%i%s' % (year, self.name.lower()) return '%i%s' % (year, self.name.lower())
Artist.albums = ReferenceSet(Artist.id, Album.artist_id) class Track(db.Entity):
_table_ = 'track'
class Track(object): id = PrimaryKey(UUID, default = uuid4)
__storm_table__ = 'track' disc = Required(int)
number = Required(int)
title = Required(str)
year = Optional(int)
genre = Optional(str)
duration = Required(int)
id = UUID(primary = True, default_factory = uuid.uuid4) album = Required(Album, column = 'album_id')
disc = Int() artist = Required(Artist, column = 'artist_id')
number = Int()
title = Unicode()
year = Int() # nullable
genre = Unicode() # nullable
duration = Int()
album_id = UUID()
album = Reference(album_id, Album.id)
artist_id = UUID()
artist = Reference(artist_id, Artist.id)
bitrate = Int()
path = Unicode() # unique bitrate = Required(int)
content_type = Unicode()
created = DateTime(default_factory = now)
last_modification = Int()
play_count = Int(default = 0) path = Required(str, unique = True)
last_play = DateTime() # nullable content_type = Required(str)
created = Required(datetime, precision = 0, default = now)
last_modification = Required(int)
root_folder_id = UUID() play_count = Required(int, default = 0)
root_folder = Reference(root_folder_id, Folder.id) last_play = Optional(datetime, precision = 0)
folder_id = UUID()
folder = Reference(folder_id, Folder.id) root_folder = Required(Folder, column = 'root_folder_id')
folder = Required(Folder, column = 'folder_id')
__lastly_played_by = Set(lambda: User) # Never used, hide it
stars = Set(lambda: StarredTrack)
ratings = Set(lambda: RatingTrack)
def as_subsonic_child(self, user, prefs): def as_subsonic_child(self, user, prefs):
info = { info = {
'id': str(self.id), 'id': str(self.id),
'parent': str(self.folder_id), 'parent': str(self.folder.id),
'isDir': False, 'isDir': False,
'title': self.title, 'title': self.title,
'album': self.album.name, 'album': self.album.name,
@ -187,8 +195,8 @@ class Track(object):
'isVideo': False, 'isVideo': False,
'discNumber': self.disc, 'discNumber': self.disc,
'created': self.created.isoformat(), 'created': self.created.isoformat(),
'albumId': str(self.album_id), 'albumId': str(self.album.id),
'artistId': str(self.artist_id), 'artistId': str(self.artist.id),
'type': 'music' 'type': 'music'
} }
@ -197,16 +205,19 @@ class Track(object):
if self.genre: if self.genre:
info['genre'] = self.genre info['genre'] = self.genre
if self.folder.has_cover_art: if self.folder.has_cover_art:
info['coverArt'] = str(self.folder_id) info['coverArt'] = str(self.folder.id)
starred = Store.of(self).get(StarredTrack, (user.id, self.id)) try:
if starred: starred = StarredTrack[user.id, self.id]
info['starred'] = starred.date.isoformat() info['starred'] = starred.date.isoformat()
except ObjectNotFound: pass
rating = Store.of(self).get(RatingTrack, (user.id, self.id)) try:
if rating: rating = RatingTrack[user.id, self.id]
info['userRating'] = rating.rating info['userRating'] = rating.rating
avgRating = Store.of(self).find(RatingTrack, RatingTrack.rated_id == self.id).avg(RatingTrack.rating) except ObjectNotFound: pass
avgRating = avg(self.ratings.rating)
if avgRating: if avgRating:
info['averageRating'] = avgRating info['averageRating'] = avgRating
@ -228,25 +239,31 @@ class Track(object):
def sort_key(self): def sort_key(self):
return (self.album.artist.name + self.album.name + ("%02i" % self.disc) + ("%02i" % self.number) + self.title).lower() return (self.album.artist.name + self.album.name + ("%02i" % self.disc) + ("%02i" % self.number) + self.title).lower()
Folder.tracks = ReferenceSet(Folder.id, Track.folder_id) class User(db.Entity):
Album.tracks = ReferenceSet(Album.id, Track.album_id) _table_ = 'user'
Artist.tracks = ReferenceSet(Artist.id, Track.artist_id)
class User(object): id = PrimaryKey(UUID, default = uuid4)
__storm_table__ = 'user' name = Required(str, unique = True)
mail = Optional(str)
password = Required(str)
salt = Required(str)
admin = Required(bool, default = False)
lastfm_session = Optional(str)
lastfm_status = Required(bool, default = True) # True: ok/unlinked, False: invalid session
id = UUID(primary = True, default_factory = uuid.uuid4) last_play = Optional(Track, column = 'last_play_id')
name = Unicode() # unique last_play_date = Optional(datetime, precision = 0)
mail = Unicode()
password = Unicode()
salt = Unicode()
admin = Bool(default = False)
lastfm_session = Unicode() # nullable
lastfm_status = Bool(default = True) # True: ok/unlinked, False: invalid session
last_play_id = UUID() # nullable clients = Set(lambda: ClientPrefs)
last_play = Reference(last_play_id, Track.id) playlists = Set(lambda: Playlist)
last_play_date = DateTime() # nullable __messages = Set(lambda: ChatMessage, lazy = True) # Never used, hide it
starred_folders = Set(lambda: StarredFolder, lazy = True)
starred_artists = Set(lambda: StarredArtist, lazy = True)
starred_albums = Set(lambda: StarredAlbum, lazy = True)
starred_tracks = Set(lambda: StarredTrack, lazy = True)
folder_ratings = Set(lambda: RatingFolder, lazy = True)
track_ratings = Set(lambda: RatingTrack, lazy = True)
def as_subsonic_user(self): def as_subsonic_user(self):
return { return {
@ -266,72 +283,74 @@ class User(object):
'shareRole': False 'shareRole': False
} }
class ClientPrefs(object): class ClientPrefs(db.Entity):
__storm_table__ = 'client_prefs' _table_ = 'client_prefs'
__storm_primary__ = 'user_id', 'client_name'
user_id = UUID() user = Required(User, column = 'user_id')
client_name = Unicode() client_name = Required(str)
format = Unicode() # nullable PrimaryKey(user, client_name)
bitrate = Int() # nullable format = Optional(str)
bitrate = Optional(int)
class BaseStarred(object): class StarredFolder(db.Entity):
__storm_primary__ = 'user_id', 'starred_id' _table_ = 'starred_folder'
user_id = UUID() user = Required(User, column = 'user_id')
starred_id = UUID() starred = Required(Folder, column = 'starred_id')
date = DateTime(default_factory = now) date = Required(datetime, precision = 0, default = now)
user = Reference(user_id, User.id) PrimaryKey(user, starred)
class StarredFolder(BaseStarred): class StarredArtist(db.Entity):
__storm_table__ = 'starred_folder' _table_ = 'starred_artist'
starred = Reference(BaseStarred.starred_id, Folder.id) user = Required(User, column = 'user_id')
starred = Required(Artist, column = 'starred_id')
date = Required(datetime, precision = 0, default = now)
class StarredArtist(BaseStarred): PrimaryKey(user, starred)
__storm_table__ = 'starred_artist'
starred = Reference(BaseStarred.starred_id, Artist.id) class StarredAlbum(db.Entity):
_table_ = 'starred_album'
class StarredAlbum(BaseStarred): user = Required(User, column = 'user_id')
__storm_table__ = 'starred_album' starred = Required(Album, column = 'starred_id')
date = Required(datetime, precision = 0, default = now)
starred = Reference(BaseStarred.starred_id, Album.id) PrimaryKey(user, starred)
class StarredTrack(BaseStarred): class StarredTrack(db.Entity):
__storm_table__ = 'starred_track' _table_ = 'starred_track'
starred = Reference(BaseStarred.starred_id, Track.id) user = Required(User, column = 'user_id')
starred = Required(Track, column = 'starred_id')
date = Required(datetime, precision = 0, default = now)
class BaseRating(object): PrimaryKey(user, starred)
__storm_primary__ = 'user_id', 'rated_id'
user_id = UUID() class RatingFolder(db.Entity):
rated_id = UUID() _table_ = 'rating_folder'
rating = Int() user = Required(User, column = 'user_id')
rated = Required(Folder, column = 'rated_id')
rating = Required(int)
user = Reference(user_id, User.id) PrimaryKey(user, rated)
class RatingFolder(BaseRating): class RatingTrack(db.Entity):
__storm_table__ = 'rating_folder' _table_ = 'rating_track'
user = Required(User, column = 'user_id')
rated = Required(Track, column = 'rated_id')
rating = Required(int)
rated = Reference(BaseRating.rated_id, Folder.id) PrimaryKey(user, rated)
class RatingTrack(BaseRating): class ChatMessage(db.Entity):
__storm_table__ = 'rating_track' _table_ = 'chat_message'
rated = Reference(BaseRating.rated_id, Track.id) id = PrimaryKey(UUID, default = uuid4)
user = Required(User, column = 'user_id')
class ChatMessage(object): time = Required(int, default = lambda: int(time.time()))
__storm_table__ = 'chat_message' message = Required(str)
id = UUID(primary = True, default_factory = uuid.uuid4)
user_id = UUID()
time = Int(default_factory = lambda: int(time.time()))
message = Unicode()
user = Reference(user_id, User.id)
def responsize(self): def responsize(self):
return { return {
@ -340,24 +359,22 @@ class ChatMessage(object):
'message': self.message 'message': self.message
} }
class Playlist(object): class Playlist(db.Entity):
__storm_table__ = 'playlist' _table_ = 'playlist'
id = UUID(primary = True, default_factory = uuid.uuid4) id = PrimaryKey(UUID, default = uuid4)
user_id = UUID() user = Required(User, column = 'user_id')
name = Unicode() name = Required(str)
comment = Unicode() # nullable comment = Optional(str)
public = Bool(default = False) public = Required(bool, default = False)
created = DateTime(default_factory = now) created = Required(datetime, precision = 0, default = now)
tracks = Unicode() tracks = Optional(str)
user = Reference(user_id, User.id)
def as_subsonic_playlist(self, user): def as_subsonic_playlist(self, user):
tracks = self.get_tracks() tracks = self.get_tracks()
info = { info = {
'id': str(self.id), 'id': str(self.id),
'name': self.name if self.user_id == user.id else '[%s] %s' % (self.user.name, self.name), 'name': self.name if self.user.id == user.id else '[%s] %s' % (self.user.name, self.name),
'owner': self.user.name, 'owner': self.user.name,
'public': self.public, 'public': self.public,
'songCount': len(tracks), 'songCount': len(tracks),
@ -374,38 +391,34 @@ class Playlist(object):
tracks = [] tracks = []
should_fix = False should_fix = False
store = Store.of(self)
for t in self.tracks.split(','): for t in self.tracks.split(','):
try: try:
tid = uuid.UUID(t) tid = UUID(t)
track = store.get(Track, tid) track = Track[tid]
if track: tracks.append(track)
tracks.append(track)
else:
should_fix = True
except: except:
should_fix = True should_fix = True
if should_fix: if should_fix:
self.tracks = ','.join(map(lambda t: str(t.id), tracks)) self.tracks = ','.join(map(lambda t: str(t.id), tracks))
store.commit() db.commit()
return tracks return tracks
def clear(self): def clear(self):
self.tracks = "" self.tracks = ''
def add(self, track): def add(self, track):
if isinstance(track, uuid.UUID): if isinstance(track, UUID):
tid = track tid = track
elif isinstance(track, Track): elif isinstance(track, Track):
tid = track.id tid = track.id
elif isinstance(track, basestring): elif isinstance(track, basestring):
tid = uuid.UUID(track) tid = UUID(track)
if self.tracks and len(self.tracks) > 0: if self.tracks and len(self.tracks) > 0:
self.tracks = "{},{}".format(self.tracks, tid) self.tracks = '{},{}'.format(self.tracks, tid)
else: else:
self.tracks = str(tid) self.tracks = str(tid)
@ -418,8 +431,35 @@ class Playlist(object):
self.tracks = ','.join(t for t in tracks if t) self.tracks = ','.join(t for t in tracks if t)
def get_store(database_uri): def parse_uri(database_uri):
database = create_database(database_uri) if not isinstance(database_uri, basestring):
store = Store(database) raise TypeError('Expecting a string')
return store
uri = urlparse(database_uri)
if uri.scheme == 'sqlite':
path = uri.path
if not path:
path = ':memory:'
elif path[0] == '/':
path = path[1:]
return dict(provider = 'sqlite', filename = path)
elif uri.scheme in ('postgres', 'postgresql'):
return dict(provider = 'postgres', user = uri.username, password = uri.password, host = uri.hostname, database = uri.path[1:])
elif uri.scheme == 'mysql':
return dict(provider = 'mysql', user = uri.username, passwd = uri.password, host = uri.hostname, db = uri.path[1:])
return dict()
def get_database(database_uri, create_tables = False):
db.bind(**parse_uri(database_uri))
db.generate_mapping(create_tables = create_tables)
return db
def release_database(db):
if not isinstance(db, Database):
raise TypeError('Expecting a pony.orm.Database instance')
db.disconnect()
db.provider = None
db.schema = None

357
tests/base/test_db.py
View File

@ -9,41 +9,38 @@
# #
# Distributed under terms of the GNU AGPLv3 license. # Distributed under terms of the GNU AGPLv3 license.
import re
import unittest import unittest
import io, re
from collections import namedtuple
import uuid import uuid
from collections import namedtuple
from pony.orm import db_session
from supysonic import db from supysonic import db
date_regex = re.compile(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}$') date_regex = re.compile(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}$')
class DbTestCase(unittest.TestCase): class DbTestCase(unittest.TestCase):
def setUp(self): def setUp(self):
self.store = db.get_store(u'sqlite:') self.store = db.get_database('sqlite:', True)
with io.open(u'schema/sqlite.sql', u'r') as f:
for statement in f.read().split(u';'):
self.store.execute(statement)
def tearDown(self): def tearDown(self):
self.store.close() db.release_database(self.store)
def create_some_folders(self): def create_some_folders(self):
root_folder = db.Folder() root_folder = db.Folder(
root_folder.root = True root = True,
root_folder.name = u'Root folder' name = 'Root folder',
root_folder.path = u'tests' path = 'tests'
)
child_folder = db.Folder() child_folder = db.Folder(
child_folder.root = False root = False,
child_folder.name = u'Child folder' name = 'Child folder',
child_folder.path = u'tests/assets' path = 'tests/assets',
child_folder.has_cover_art = True has_cover_art = True,
child_folder.parent = root_folder parent = root_folder
)
self.store.add(root_folder)
self.store.add(child_folder)
self.store.commit()
return root_folder, child_folder return root_folder, child_folder
@ -51,244 +48,230 @@ class DbTestCase(unittest.TestCase):
root, child = self.create_some_folders() root, child = self.create_some_folders()
if not artist: if not artist:
artist = db.Artist() artist = db.Artist(name = 'Test artist')
artist.name = u'Test Artist'
if not album: if not album:
album = db.Album() album = db.Album(artist = artist, name = 'Test Album')
album.artist = artist
album.name = u'Test Album'
track1 = db.Track() track1 = db.Track(
track1.title = u'Track Title' title = 'Track Title',
track1.album = album album = album,
track1.artist = artist artist = artist,
track1.disc = 1 disc = 1,
track1.number = 1 number = 1,
track1.duration = 3 duration = 3,
track1.bitrate = 320 bitrate = 320,
track1.path = u'tests/assets/empty' path = 'tests/assets/empty',
track1.content_type = u'audio/mpeg' content_type = 'audio/mpeg',
track1.last_modification = 1234 last_modification = 1234,
track1.root_folder = root root_folder = root,
track1.folder = child folder = child
self.store.add(track1) )
track2 = db.Track() track2 = db.Track(
track2.title = u'One Awesome Song' title = 'One Awesome Song',
track2.album = album album = album,
track2.artist = artist artist = artist,
track2.disc = 1 disc = 1,
track2.number = 2 number = 2,
track2.duration = 5 duration = 5,
track2.bitrate = 96 bitrate = 96,
track2.path = u'tests/assets/empty' path = 'tests/assets/23bytes',
track2.content_type = u'audio/mpeg' content_type = 'audio/mpeg',
track2.last_modification = 1234 last_modification = 1234,
track2.root_folder = root root_folder = root,
track2.folder = child folder = child
self.store.add(track2) )
return track1, track2 return track1, track2
def create_playlist(self): def create_user(self, name = 'Test User'):
user = db.User() return db.User(
user.name = u'Test User' name = name,
user.password = u'secret' password = 'secret',
user.salt = u'ABC+' salt = 'ABC+',
)
playlist = db.Playlist() def create_playlist(self):
playlist.user = user
playlist.name = u'Playlist!' playlist = db.Playlist(
self.store.add(playlist) user = self.create_user(),
name = 'Playlist!'
)
return playlist return playlist
@db_session
def test_folder_base(self): def test_folder_base(self):
root_folder, child_folder = self.create_some_folders() root_folder, child_folder = self.create_some_folders()
self.store.commit()
MockUser = namedtuple(u'User', [ u'id' ]) MockUser = namedtuple('User', [ 'id' ])
user = MockUser(uuid.uuid4()) user = MockUser(uuid.uuid4())
root = root_folder.as_subsonic_child(user) root = root_folder.as_subsonic_child(user)
self.assertIsInstance(root, dict) self.assertIsInstance(root, dict)
self.assertIn(u'id', root) self.assertIn('id', root)
self.assertIn(u'isDir', root) self.assertIn('isDir', root)
self.assertIn(u'title', root) self.assertIn('title', root)
self.assertIn(u'album', root) self.assertIn('album', root)
self.assertIn(u'created', root) self.assertIn('created', root)
self.assertTrue(root[u'isDir']) self.assertTrue(root['isDir'])
self.assertEqual(root[u'title'], u'Root folder') self.assertEqual(root['title'], 'Root folder')
self.assertEqual(root[u'album'], u'Root folder') self.assertEqual(root['album'], 'Root folder')
self.assertRegexpMatches(root['created'], date_regex) self.assertRegexpMatches(root['created'], date_regex)
child = child_folder.as_subsonic_child(user) child = child_folder.as_subsonic_child(user)
self.assertIn(u'parent', child) self.assertIn('parent', child)
self.assertIn(u'artist', child) self.assertIn('artist', child)
self.assertIn(u'coverArt', child) self.assertIn('coverArt', child)
self.assertEqual(child[u'parent'], str(root_folder.id)) self.assertEqual(child['parent'], str(root_folder.id))
self.assertEqual(child[u'artist'], root_folder.name) self.assertEqual(child['artist'], root_folder.name)
self.assertEqual(child[u'coverArt'], child[u'id']) self.assertEqual(child['coverArt'], child['id'])
@db_session
def test_folder_annotation(self): def test_folder_annotation(self):
root_folder, child_folder = self.create_some_folders() root_folder, child_folder = self.create_some_folders()
# Assuming SQLite doesn't enforce foreign key constraints user = self.create_user()
MockUser = namedtuple(u'User', [ u'id' ]) star = db.StarredFolder(
user = MockUser(uuid.uuid4()) user = user,
starred = root_folder
star = db.StarredFolder() )
star.user_id = user.id rating_user = db.RatingFolder(
star.starred_id = root_folder.id user = user,
rated = root_folder,
rating_user = db.RatingFolder() rating = 2
rating_user.user_id = user.id )
rating_user.rated_id = root_folder.id other = self.create_user('Other')
rating_user.rating = 2 rating_other = db.RatingFolder(
user = other,
rating_other = db.RatingFolder() rated = root_folder,
rating_other.user_id = uuid.uuid4() rating = 5
rating_other.rated_id = root_folder.id )
rating_other.rating = 5 self.store.commit()
self.store.add(star)
self.store.add(rating_user)
self.store.add(rating_other)
root = root_folder.as_subsonic_child(user) root = root_folder.as_subsonic_child(user)
self.assertIn(u'starred', root) self.assertIn('starred', root)
self.assertIn(u'userRating', root) self.assertIn('userRating', root)
self.assertIn(u'averageRating', root) self.assertIn('averageRating', root)
self.assertRegexpMatches(root[u'starred'], date_regex) self.assertRegexpMatches(root['starred'], date_regex)
self.assertEqual(root[u'userRating'], 2) self.assertEqual(root['userRating'], 2)
self.assertEqual(root[u'averageRating'], 3.5) self.assertEqual(root['averageRating'], 3.5)
child = child_folder.as_subsonic_child(user) child = child_folder.as_subsonic_child(user)
self.assertNotIn(u'starred', child) self.assertNotIn('starred', child)
self.assertNotIn(u'userRating', child) self.assertNotIn('userRating', child)
@db_session
def test_artist(self): def test_artist(self):
artist = db.Artist() artist = db.Artist(name = 'Test Artist')
artist.name = u'Test Artist'
self.store.add(artist)
# Assuming SQLite doesn't enforce foreign key constraints user = self.create_user()
MockUser = namedtuple(u'User', [ u'id' ]) star = db.StarredArtist(user = user, starred = artist)
user = MockUser(uuid.uuid4()) self.store.commit()
star = db.StarredArtist()
star.user_id = user.id
star.starred_id = artist.id
self.store.add(star)
artist_dict = artist.as_subsonic_artist(user) artist_dict = artist.as_subsonic_artist(user)
self.assertIsInstance(artist_dict, dict) self.assertIsInstance(artist_dict, dict)
self.assertIn(u'id', artist_dict) self.assertIn('id', artist_dict)
self.assertIn(u'name', artist_dict) self.assertIn('name', artist_dict)
self.assertIn(u'albumCount', artist_dict) self.assertIn('albumCount', artist_dict)
self.assertIn(u'starred', artist_dict) self.assertIn('starred', artist_dict)
self.assertEqual(artist_dict[u'name'], u'Test Artist') self.assertEqual(artist_dict['name'], 'Test Artist')
self.assertEqual(artist_dict[u'albumCount'], 0) self.assertEqual(artist_dict['albumCount'], 0)
self.assertRegexpMatches(artist_dict[u'starred'], date_regex) self.assertRegexpMatches(artist_dict['starred'], date_regex)
album = db.Album() db.Album(name = 'Test Artist', artist = artist) # self-titled
album.name = u'Test Artist' # self-titled db.Album(name = 'The Album After The First One', artist = artist)
artist.albums.add(album) self.store.commit()
album = db.Album()
album.name = u'The Album After The Frist One'
artist.albums.add(album)
artist_dict = artist.as_subsonic_artist(user) artist_dict = artist.as_subsonic_artist(user)
self.assertEqual(artist_dict[u'albumCount'], 2) self.assertEqual(artist_dict['albumCount'], 2)
@db_session
def test_album(self): def test_album(self):
artist = db.Artist() artist = db.Artist(name = 'Test Artist')
artist.name = u'Test Artist' album = db.Album(artist = artist, name = 'Test Album')
album = db.Album() user = self.create_user()
album.artist = artist star = db.StarredAlbum(
album.name = u'Test Album' user = user,
starred = album
# Assuming SQLite doesn't enforce foreign key constraints )
MockUser = namedtuple(u'User', [ u'id' ]) self.store.commit()
user = MockUser(uuid.uuid4())
star = db.StarredAlbum()
star.user_id = user.id
star.starred = album
self.store.add(album)
self.store.add(star)
# No tracks, shouldn't be stored under normal circumstances # No tracks, shouldn't be stored under normal circumstances
self.assertRaises(ValueError, album.as_subsonic_album, user) self.assertRaises(ValueError, album.as_subsonic_album, user)
self.create_some_tracks(artist, album) self.create_some_tracks(artist, album)
self.store.commit()
album_dict = album.as_subsonic_album(user) album_dict = album.as_subsonic_album(user)
self.assertIsInstance(album_dict, dict) self.assertIsInstance(album_dict, dict)
self.assertIn(u'id', album_dict) self.assertIn('id', album_dict)
self.assertIn(u'name', album_dict) self.assertIn('name', album_dict)
self.assertIn(u'artist', album_dict) self.assertIn('artist', album_dict)
self.assertIn(u'artistId', album_dict) self.assertIn('artistId', album_dict)
self.assertIn(u'songCount', album_dict) self.assertIn('songCount', album_dict)
self.assertIn(u'duration', album_dict) self.assertIn('duration', album_dict)
self.assertIn(u'created', album_dict) self.assertIn('created', album_dict)
self.assertIn(u'starred', album_dict) self.assertIn('starred', album_dict)
self.assertEqual(album_dict[u'name'], album.name) self.assertEqual(album_dict['name'], album.name)
self.assertEqual(album_dict[u'artist'], artist.name) self.assertEqual(album_dict['artist'], artist.name)
self.assertEqual(album_dict[u'artistId'], str(artist.id)) self.assertEqual(album_dict['artistId'], str(artist.id))
self.assertEqual(album_dict[u'songCount'], 2) self.assertEqual(album_dict['songCount'], 2)
self.assertEqual(album_dict[u'duration'], 8) self.assertEqual(album_dict['duration'], 8)
self.assertRegexpMatches(album_dict[u'created'], date_regex) self.assertRegexpMatches(album_dict['created'], date_regex)
self.assertRegexpMatches(album_dict[u'starred'], date_regex) self.assertRegexpMatches(album_dict['starred'], date_regex)
@db_session
def test_track(self): def test_track(self):
track1, track2 = self.create_some_tracks() track1, track2 = self.create_some_tracks()
self.store.commit()
# Assuming SQLite doesn't enforce foreign key constraints # Assuming SQLite doesn't enforce foreign key constraints
MockUser = namedtuple(u'User', [ u'id' ]) MockUser = namedtuple('User', [ 'id' ])
user = MockUser(uuid.uuid4()) user = MockUser(uuid.uuid4())
track1_dict = track1.as_subsonic_child(user, None) track1_dict = track1.as_subsonic_child(user, None)
self.assertIsInstance(track1_dict, dict) self.assertIsInstance(track1_dict, dict)
self.assertIn(u'id', track1_dict) self.assertIn('id', track1_dict)
self.assertIn(u'parent', track1_dict) self.assertIn('parent', track1_dict)
self.assertIn(u'isDir', track1_dict) self.assertIn('isDir', track1_dict)
self.assertIn(u'title', track1_dict) self.assertIn('title', track1_dict)
self.assertFalse(track1_dict[u'isDir']) self.assertFalse(track1_dict['isDir'])
# ... we'll test the rest against the API XSD. # ... we'll test the rest against the API XSD.
@db_session
def test_user(self): def test_user(self):
user = db.User() user = self.create_user()
user.name = u'Test User' self.store.commit()
user.password = u'secret'
user.salt = u'ABC+'
user_dict = user.as_subsonic_user() user_dict = user.as_subsonic_user()
self.assertIsInstance(user_dict, dict) self.assertIsInstance(user_dict, dict)
@db_session
def test_chat(self): def test_chat(self):
user = db.User() user = self.create_user()
user.name = u'Test User'
user.password = u'secret'
user.salt = u'ABC+'
line = db.ChatMessage() line = db.ChatMessage(
line.user = user user = user,
line.message = u'Hello world!' message = 'Hello world!'
)
self.store.commit()
line_dict = line.responsize() line_dict = line.responsize()
self.assertIsInstance(line_dict, dict) self.assertIsInstance(line_dict, dict)
self.assertIn(u'username', line_dict) self.assertIn('username', line_dict)
self.assertEqual(line_dict[u'username'], user.name) self.assertEqual(line_dict['username'], user.name)
@db_session
def test_playlist(self): def test_playlist(self):
playlist = self.create_playlist() playlist = self.create_playlist()
playlist_dict = playlist.as_subsonic_playlist(playlist.user) playlist_dict = playlist.as_subsonic_playlist(playlist.user)
self.assertIsInstance(playlist_dict, dict) self.assertIsInstance(playlist_dict, dict)
@db_session
def test_playlist_tracks(self): def test_playlist_tracks(self):
playlist = self.create_playlist() playlist = self.create_playlist()
track1, track2 = self.create_some_tracks() track1, track2 = self.create_some_tracks()
@ -307,9 +290,10 @@ class DbTestCase(unittest.TestCase):
playlist.add(str(track1.id)) playlist.add(str(track1.id))
self.assertSequenceEqual(playlist.get_tracks(), [ track1 ]) self.assertSequenceEqual(playlist.get_tracks(), [ track1 ])
self.assertRaises(ValueError, playlist.add, u'some string') self.assertRaises(ValueError, playlist.add, 'some string')
self.assertRaises(NameError, playlist.add, 2345) self.assertRaises(NameError, playlist.add, 2345)
@db_session
def test_playlist_remove_tracks(self): def test_playlist_remove_tracks(self):
playlist = self.create_playlist() playlist = self.create_playlist()
track1, track2 = self.create_some_tracks() track1, track2 = self.create_some_tracks()
@ -329,6 +313,7 @@ class DbTestCase(unittest.TestCase):
playlist.remove_at_indexes([ 1, 1 ]) playlist.remove_at_indexes([ 1, 1 ])
self.assertSequenceEqual(playlist.get_tracks(), [ track2, track1 ]) self.assertSequenceEqual(playlist.get_tracks(), [ track2, track1 ])
@db_session
def test_playlist_fixing(self): def test_playlist_fixing(self):
playlist = self.create_playlist() playlist = self.create_playlist()
track1, track2 = self.create_some_tracks() track1, track2 = self.create_some_tracks()
@ -338,10 +323,10 @@ class DbTestCase(unittest.TestCase):
playlist.add(track2) playlist.add(track2)
self.assertSequenceEqual(playlist.get_tracks(), [ track1, track2 ]) self.assertSequenceEqual(playlist.get_tracks(), [ track1, track2 ])
self.store.remove(track2) track2.delete()
self.assertSequenceEqual(playlist.get_tracks(), [ track1 ]) self.assertSequenceEqual(playlist.get_tracks(), [ track1 ])
playlist.tracks = u'{0},{0},some random garbage,{0}'.format(track1.id) playlist.tracks = '{0},{0},some random garbage,{0}'.format(track1.id)
self.assertSequenceEqual(playlist.get_tracks(), [ track1, track1, track1 ]) self.assertSequenceEqual(playlist.get_tracks(), [ track1, track1, track1 ])
if __name__ == '__main__': if __name__ == '__main__':