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

Scanner, CLI and watcher are on a pony

Browse Source
This commit is contained in:
spl0k 2017-12-17 23:25:34 +01:00
parent a4b9a97271
commit 2428ffeb57
8 changed files with 262 additions and 241 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
bin/supysonic-cli
View File

@ -12,9 +12,11 @@ import sys
from supysonic.cli import SupysonicCLI from supysonic.cli import SupysonicCLI
from supysonic.config import IniConfig from supysonic.config import IniConfig
from supysonic.db import get_database, release_database
if __name__ == "__main__": if __name__ == "__main__":
config = IniConfig.from_common_locations() config = IniConfig.from_common_locations()
db = get_database(config.BASE['database_uri'])
cli = SupysonicCLI(config) cli = SupysonicCLI(config)
if len(sys.argv) > 1: if len(sys.argv) > 1:
@ -22,3 +24,5 @@ if __name__ == "__main__":
else: else:
cli.cmdloop() cli.cmdloop()
release_database(db)

46
supysonic/cli.py
View File

@ -25,7 +25,9 @@ import getpass
import sys import sys
import time import time
from .db import get_store, Folder, User from pony.orm import db_session
from .db import Folder, User
from .managers.folder import FolderManager from .managers.folder import FolderManager
from .managers.user import UserManager from .managers.user import UserManager
from .scanner import Scanner from .scanner import Scanner
@ -105,8 +107,6 @@ class SupysonicCLI(cmd.Cmd):
for action, subparser in getattr(self.__class__, command + '_subparsers').choices.iteritems(): for action, subparser in getattr(self.__class__, command + '_subparsers').choices.iteritems():
setattr(self, 'help_{} {}'.format(command, action), subparser.print_help) setattr(self, 'help_{} {}'.format(command, action), subparser.print_help)
self.__store = get_store(config.BASE['database_uri'])
def write_line(self, line = ''): def write_line(self, line = ''):
self.stdout.write(line + '\n') self.stdout.write(line + '\n')
@ -148,44 +148,49 @@ class SupysonicCLI(cmd.Cmd):
folder_scan_parser.add_argument('folders', metavar = 'folder', nargs = '*', help = 'Folder(s) to be scanned. If ommitted, all folders are scanned') folder_scan_parser.add_argument('folders', metavar = 'folder', nargs = '*', help = 'Folder(s) to be scanned. If ommitted, all folders are scanned')
folder_scan_parser.add_argument('-f', '--force', action = 'store_true', help = "Force scan of already know files even if they haven't changed") folder_scan_parser.add_argument('-f', '--force', action = 'store_true', help = "Force scan of already know files even if they haven't changed")
@db_session
def folder_list(self): def folder_list(self):
self.write_line('Name\t\tPath\n----\t\t----') self.write_line('Name\t\tPath\n----\t\t----')
self.write_line('\n'.join('{0: <16}{1}'.format(f.name, f.path) for f in self.__store.find(Folder, Folder.root == True))) self.write_line('\n'.join('{0: <16}{1}'.format(f.name, f.path) for f in Folder.select(lambda f: f.root)))
def folder_add(self, name, path): def folder_add(self, name, path):
ret = FolderManager.add(self.__store, name, path) ret = FolderManager.add(name, path)
if ret != FolderManager.SUCCESS: if ret != FolderManager.SUCCESS:
self.write_error_line(FolderManager.error_str(ret)) self.write_error_line(FolderManager.error_str(ret))
else: else:
self.write_line("Folder '{}' added".format(name)) self.write_line("Folder '{}' added".format(name))
def folder_delete(self, name): def folder_delete(self, name):
ret = FolderManager.delete_by_name(self.__store, name) ret = FolderManager.delete_by_name(name)
if ret != FolderManager.SUCCESS: if ret != FolderManager.SUCCESS:
self.write_error_line(FolderManager.error_str(ret)) self.write_error_line(FolderManager.error_str(ret))
else: else:
self.write_line("Deleted folder '{}'".format(name)) self.write_line("Deleted folder '{}'".format(name))
@db_session
def folder_scan(self, folders, force): def folder_scan(self, folders, force):
extensions = self.__config.BASE['scanner_extensions'] extensions = self.__config.BASE['scanner_extensions']
if extensions: if extensions:
extensions = extensions.split(' ') extensions = extensions.split(' ')
scanner = Scanner(self.__store, force = force, extensions = extensions)
scanner = Scanner(force = force, extensions = extensions)
if folders: if folders:
folders = map(lambda n: self.__store.find(Folder, Folder.name == n, Folder.root == True).one() or n, folders) fstrs = folders
if any(map(lambda f: isinstance(f, basestring), folders)): folders = Folder.select(lambda f: f.root and f.name in fstrs)[:]
self.write_line("No such folder(s): " + ' '.join(f for f in folders if isinstance(f, basestring))) notfound = set(fstrs) - set(map(lambda f: f.name, folders))
for folder in filter(lambda f: isinstance(f, Folder), folders): if notfound:
self.write_line("No such folder(s): " + ' '.join(notfound))
for folder in folders:
scanner.scan(folder, TimedProgressDisplay(folder.name, self.stdout)) scanner.scan(folder, TimedProgressDisplay(folder.name, self.stdout))
self.write_line() self.write_line()
else: else:
for folder in self.__store.find(Folder, Folder.root == True): for folder in Folder.select(lambda f: f.root):
scanner.scan(folder, TimedProgressDisplay(folder.name, self.stdout)) scanner.scan(folder, TimedProgressDisplay(folder.name, self.stdout))
self.write_line() self.write_line()
scanner.finish() scanner.finish()
added, deleted = scanner.stats() added, deleted = scanner.stats()
self.__store.commit()
self.write_line("Scanning done") self.write_line("Scanning done")
self.write_line('Added: %i artists, %i albums, %i tracks' % (added[0], added[1], added[2])) self.write_line('Added: %i artists, %i albums, %i tracks' % (added[0], added[1], added[2]))
@ -208,9 +213,10 @@ class SupysonicCLI(cmd.Cmd):
user_pass_parser.add_argument('name', help = 'Name/login of the user to which change the password') user_pass_parser.add_argument('name', help = 'Name/login of the user to which change the password')
user_pass_parser.add_argument('password', nargs = '?', help = 'New password') user_pass_parser.add_argument('password', nargs = '?', help = 'New password')
@db_session
def user_list(self): def user_list(self):
self.write_line('Name\t\tAdmin\tEmail\n----\t\t-----\t-----') self.write_line('Name\t\tAdmin\tEmail\n----\t\t-----\t-----')
self.write_line('\n'.join('{0: <16}{1}\t{2}'.format(u.name, '*' if u.admin else '', u.mail) for u in self.__store.find(User))) self.write_line('\n'.join('{0: <16}{1}\t{2}'.format(u.name, '*' if u.admin else '', u.mail) for u in User.select()))
def user_add(self, name, admin, password, email): def user_add(self, name, admin, password, email):
if not password: if not password:
@ -219,24 +225,24 @@ class SupysonicCLI(cmd.Cmd):
if password != confirm: if password != confirm:
self.write_error_line("Passwords don't match") self.write_error_line("Passwords don't match")
return return
status = UserManager.add(self.__store, name, password, email, admin) status = UserManager.add(name, password, email, admin)
if status != UserManager.SUCCESS: if status != UserManager.SUCCESS:
self.write_error_line(UserManager.error_str(status)) self.write_error_line(UserManager.error_str(status))
def user_delete(self, name): def user_delete(self, name):
ret = UserManager.delete_by_name(self.__store, name) ret = UserManager.delete_by_name(name)
if ret != UserManager.SUCCESS: if ret != UserManager.SUCCESS:
self.write_error_line(UserManager.error_str(ret)) self.write_error_line(UserManager.error_str(ret))
else: else:
self.write_line("Deleted user '{}'".format(name)) self.write_line("Deleted user '{}'".format(name))
@db_session
def user_setadmin(self, name, off): def user_setadmin(self, name, off):
user = self.__store.find(User, User.name == name).one() user = User.get(name = name)
if not user: if user is None:
self.write_error_line('No such user') self.write_error_line('No such user')
else: else:
user.admin = not off user.admin = not off
self.__store.commit()
self.write_line("{0} '{1}' admin rights".format('Revoked' if off else 'Granted', name)) self.write_line("{0} '{1}' admin rights".format('Revoked' if off else 'Granted', name))
def user_changepass(self, name, password): def user_changepass(self, name, password):
@ -246,7 +252,7 @@ class SupysonicCLI(cmd.Cmd):
if password != confirm: if password != confirm:
self.write_error_line("Passwords don't match") self.write_error_line("Passwords don't match")
return return
status = UserManager.change_password2(self.__store, name, password) status = UserManager.change_password2(name, password)
if status != UserManager.SUCCESS: if status != UserManager.SUCCESS:
self.write_error_line(UserManager.error_str(status)) self.write_error_line(UserManager.error_str(status))
else: else:

2
supysonic/db.py
View File

@ -153,7 +153,7 @@ class Track(db.Entity):
number = Required(int) number = Required(int)
title = Required(str) title = Required(str)
year = Optional(int) year = Optional(int)
genre = Optional(str) genre = Optional(str, nullable = True)
duration = Required(int) duration = Required(int)
album = Required(Album, column = 'album_id') album = Required(Album, column = 'album_id')

175
supysonic/scanner.py
View File

@ -23,6 +23,8 @@ import mimetypes
import mutagen import mutagen
import time import time
from pony.orm import db_session
from .db import Folder, Artist, Album, Track, User from .db import Folder, Artist, Album, Track, User
from .db import StarredFolder, StarredArtist, StarredAlbum, StarredTrack from .db import StarredFolder, StarredArtist, StarredAlbum, StarredTrack
from .db import RatingFolder, RatingTrack from .db import RatingFolder, RatingTrack
@ -67,8 +69,9 @@ class Scanner:
progress_callback(current, total) progress_callback(current, total)
# Remove files that have been deleted # Remove files that have been deleted
for track in [ t for t in self.__store.find(Track, Track.root_folder_id == folder.id) if not self.__is_valid_path(t.path) ]: for track in Track.select(lambda t: t.root_folder == folder):
self.remove_file(track.path) if not self.__is_valid_path(track.path):
self.remove_file(track.path)
# Update cover art info # Update cover art info
folders = [ folder ] folders = [ folder ]
@ -79,25 +82,32 @@ class Scanner:
folder.last_scan = int(time.time()) folder.last_scan = int(time.time())
@db_session
def finish(self): def finish(self):
for album in [ a for a in self.__albums_to_check if not a.tracks.count() ]: for album in Album.select(lambda a: a.id in self.__albums_to_check):
self.__artists_to_check.add(album.artist) if not album.tracks.is_empty():
continue
self.__artists_to_check.add(album.artist.id)
self.__deleted_albums += 1 self.__deleted_albums += 1
album.delete() album.delete()
self.__albums_to_check.clear() self.__albums_to_check.clear()
for artist in [ a for a in self.__artists_to_check if not a.albums.count() and not a.tracks.count() ]: for artist in Artist.select(lambda a: a.id in self.__artists_to_check):
if not artist.albums.is_empty() or not artist.tracks.is_empty():
continue
self.__deleted_artists += 1 self.__deleted_artists += 1
artist.delete() artist.delete()
self.__artists_to_check.clear() self.__artists_to_check.clear()
while self.__folders_to_check: while self.__folders_to_check:
folder = self.__folders_to_check.pop() folder = Folder[self.__folders_to_check.pop()]
if folder.root: if folder.root:
continue continue
if not folder.tracks.count() and not folder.children.count(): if folder.tracks.is_empty() and folder.children.is_empty():
self.__folders_to_check.add(folder.parent) self.__folders_to_check.add(folder.parent.id)
folder.delete() folder.delete()
def __is_valid_path(self, path): def __is_valid_path(self, path):
@ -107,13 +117,13 @@ class Scanner:
return True return True
return os.path.splitext(path)[1][1:].lower() in self.__extensions return os.path.splitext(path)[1][1:].lower() in self.__extensions
@db_session
def scan_file(self, path): def scan_file(self, path):
if not isinstance(path, basestring): if not isinstance(path, basestring):
raise TypeError('Expecting string, got ' + str(type(path))) raise TypeError('Expecting string, got ' + str(type(path)))
tr = self.__store.find(Track, Track.path == path).one() tr = Track.get(path = path)
add = False if tr is not None:
if tr:
if not self.__force and not int(os.path.getmtime(path)) > tr.last_modification: if not self.__force and not int(os.path.getmtime(path)) > tr.last_modification:
return return
@ -121,55 +131,55 @@ class Scanner:
if not tag: if not tag:
self.remove_file(path) self.remove_file(path)
return return
trdict = {}
else: else:
tag = self.__try_load_tag(path) tag = self.__try_load_tag(path)
if not tag: if not tag:
return return
tr = Track() trdict = { 'path': path }
tr.path = path
add = True
artist = self.__try_read_tag(tag, 'artist', '') artist = self.__try_read_tag(tag, 'artist')
album = self.__try_read_tag(tag, 'album', '') if not artist:
return
album = self.__try_read_tag(tag, 'album', '[non-album tracks]')
albumartist = self.__try_read_tag(tag, 'albumartist', artist) albumartist = self.__try_read_tag(tag, 'albumartist', artist)
tr.disc = self.__try_read_tag(tag, 'discnumber', 1, lambda x: int(x[0].split('/')[0])) trdict['disc'] = self.__try_read_tag(tag, 'discnumber', 1, lambda x: int(x[0].split('/')[0]))
tr.number = self.__try_read_tag(tag, 'tracknumber', 1, lambda x: int(x[0].split('/')[0])) trdict['number'] = self.__try_read_tag(tag, 'tracknumber', 1, lambda x: int(x[0].split('/')[0]))
tr.title = self.__try_read_tag(tag, 'title', '') trdict['title'] = self.__try_read_tag(tag, 'title', '')
tr.year = self.__try_read_tag(tag, 'date', None, lambda x: int(x[0].split('-')[0])) trdict['year'] = self.__try_read_tag(tag, 'date', None, lambda x: int(x[0].split('-')[0]))
tr.genre = self.__try_read_tag(tag, 'genre') trdict['genre'] = self.__try_read_tag(tag, 'genre')
tr.duration = int(tag.info.length) trdict['duration'] = int(tag.info.length)
tr.bitrate = (tag.info.bitrate if hasattr(tag.info, 'bitrate') else int(os.path.getsize(path) * 8 / tag.info.length)) / 1000 trdict['bitrate'] = (tag.info.bitrate if hasattr(tag.info, 'bitrate') else int(os.path.getsize(path) * 8 / tag.info.length)) / 1000
tr.content_type = mimetypes.guess_type(path, False)[0] or 'application/octet-stream' trdict['content_type'] = mimetypes.guess_type(path, False)[0] or 'application/octet-stream'
tr.last_modification = os.path.getmtime(path) trdict['last_modification'] = int(os.path.getmtime(path))
tralbum = self.__find_album(albumartist, album) tralbum = self.__find_album(albumartist, album)
trartist = self.__find_artist(artist) trartist = self.__find_artist(artist)
if add: if tr is None:
trroot = self.__find_root_folder(path) trdict['root_folder'] = self.__find_root_folder(path)
trfolder = self.__find_folder(path) trdict['folder'] = self.__find_folder(path)
trdict['album'] = tralbum
trdict['artist'] = trartist
# Set the references at the very last as searching for them will cause the added track to be flushed, even if Track(**trdict)
# it is incomplete, causing not null constraints errors.
tr.album = tralbum
tr.artist = trartist
tr.folder = trfolder
tr.root_folder = trroot
self.__store.add(tr)
self.__added_tracks += 1 self.__added_tracks += 1
else: else:
if tr.album.id != tralbum.id: if tr.album.id != tralbum.id:
self.__albums_to_check.add(tr.album) self.__albums_to_check.add(tr.album.id)
tr.album = tralbum trdict['album'] = tralbum
if tr.artist.id != trartist.id: if tr.artist.id != trartist.id:
self.__artists_to_check.add(tr.artist) self.__artists_to_check.add(tr.artist.id)
tr.artist = trartist trdict['artist'] = trartist
tr.set(**trdict)
@db_session
def remove_file(self, path): def remove_file(self, path):
if not isinstance(path, basestring): if not isinstance(path, basestring):
raise TypeError('Expecting string, got ' + str(type(path))) raise TypeError('Expecting string, got ' + str(type(path)))
@ -178,12 +188,13 @@ class Scanner:
if not tr: if not tr:
return return
self.__folders_to_check.add(tr.folder) self.__folders_to_check.add(tr.folder.id)
self.__albums_to_check.add(tr.album) self.__albums_to_check.add(tr.album.id)
self.__artists_to_check.add(tr.artist) self.__artists_to_check.add(tr.artist.id)
self.__deleted_tracks += 1 self.__deleted_tracks += 1
tr.delete() tr.delete()
@db_session
def move_file(self, src_path, dst_path): def move_file(self, src_path, dst_path):
if not isinstance(src_path, basestring): if not isinstance(src_path, basestring):
raise TypeError('Expecting string, got ' + str(type(src_path))) raise TypeError('Expecting string, got ' + str(type(src_path)))
@ -193,16 +204,18 @@ class Scanner:
if src_path == dst_path: if src_path == dst_path:
return return
tr = self.__store.find(Track, Track.path == src_path).one() tr = Track.get(path = src_path)
if not tr: if tr is None:
return return
self.__folders_to_check.add(tr.folder) self.__folders_to_check.add(tr.folder.id)
tr_dst = self.__store.find(Track, Track.path == dst_path).one() tr_dst = Track.get(path = dst_path)
if tr_dst: if tr_dst is not None:
tr.root_folder = tr_dst.root_folder root = tr_dst.root_folder
tr.folder = tr_dst.folder folder = tr_dst.folder
self.remove_file(dst_path) self.remove_file(dst_path)
tr.root_folder = root
tr.folder = folder
else: else:
root = self.__find_root_folder(dst_path) root = self.__find_root_folder(dst_path)
folder = self.__find_folder(dst_path) folder = self.__find_folder(dst_path)
@ -212,70 +225,48 @@ class Scanner:
def __find_album(self, artist, album): def __find_album(self, artist, album):
ar = self.__find_artist(artist) ar = self.__find_artist(artist)
al = ar.albums.find(name = album).one() al = ar.albums.select(lambda a: a.name == album).first()
if al: if al:
return al return al
al = Album() al = Album(name = album, artist = ar)
al.name = album
al.artist = ar
self.__store.add(al)
self.__added_albums += 1 self.__added_albums += 1
return al return al
def __find_artist(self, artist): def __find_artist(self, artist):
ar = self.__store.find(Artist, Artist.name == artist).one() ar = Artist.get(name = artist)
if ar: if ar:
return ar return ar
ar = Artist() ar = Artist(name = artist)
ar.name = artist
self.__store.add(ar)
self.__added_artists += 1 self.__added_artists += 1
return ar return ar
def __find_root_folder(self, path): def __find_root_folder(self, path):
path = os.path.dirname(path) path = os.path.dirname(path)
db = self.__store.get_database().__module__[len('storm.databases.'):] for folder in Folder.select(lambda f: f.root):
folders = self.__store.find(Folder, Like(path, Concat(Folder.path, u'%', db)), Folder.root == True) if path.startswith(folder.path):
count = folders.count() return folder
if count > 1:
raise Exception("Found multiple root folders for '{}'.".format(path)) raise Exception("Couldn't find the root folder for '{}'.\nDon't scan files that aren't located in a defined music folder".format(path))
elif count == 0:
raise Exception("Couldn't find the root folder for '{}'.\nDon't scan files that aren't located in a defined music folder".format(path))
return folders.one()
def __find_folder(self, path): def __find_folder(self, path):
children = []
drive, _ = os.path.splitdrive(path)
path = os.path.dirname(path) path = os.path.dirname(path)
folders = self.__store.find(Folder, Folder.path == path) while path != drive and path != '/':
count = folders.count() folder = Folder.get(path = path)
if count > 1: if folder is not None:
raise Exception("Found multiple folders for '{}'.".format(path)) break
elif count == 1:
return folders.one()
db = self.__store.get_database().__module__[len('storm.databases.'):] children.append(dict(root = False, name = os.path.basename(path), path = path))
folder = self.__store.find(Folder, Like(path, Concat(Folder.path, os.sep + u'%', db))).order_by(Folder.path).last() path = os.path.dirname(path)
full_path = folder.path assert folder is not None
path = path[len(folder.path) + 1:] while children:
folder = Folder(parent = folder, **children.pop())
for name in path.split(os.sep):
full_path = os.path.join(full_path, name)
fold = Folder()
fold.root = False
fold.name = name
fold.path = full_path
fold.parent = folder
self.__store.add(fold)
folder = fold
return folder return folder

33
supysonic/watcher.py
View File

@ -22,12 +22,13 @@ import logging
import time import time
from logging.handlers import TimedRotatingFileHandler from logging.handlers import TimedRotatingFileHandler
from pony.orm import db_session
from signal import signal, SIGTERM, SIGINT from signal import signal, SIGTERM, SIGINT
from threading import Thread, Condition, Timer from threading import Thread, Condition, Timer
from watchdog.observers import Observer from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler from watchdog.events import PatternMatchingEventHandler
from . import db from .db import get_database, release_database, Folder
from .scanner import Scanner from .scanner import Scanner
OP_SCAN = 1 OP_SCAN = 1
@ -109,12 +110,11 @@ class Event(object):
return self.__src return self.__src
class ScannerProcessingQueue(Thread): class ScannerProcessingQueue(Thread):
def __init__(self, database_uri, delay, logger): def __init__(self, delay, logger):
super(ScannerProcessingQueue, self).__init__() super(ScannerProcessingQueue, self).__init__()
self.__logger = logger self.__logger = logger
self.__timeout = delay self.__timeout = delay
self.__database_uri = database_uri
self.__cond = Condition() self.__cond = Condition()
self.__timer = None self.__timer = None
self.__queue = {} self.__queue = {}
@ -138,8 +138,7 @@ class ScannerProcessingQueue(Thread):
continue continue
self.__logger.debug("Instantiating scanner") self.__logger.debug("Instantiating scanner")
store = db.get_store(self.__database_uri) scanner = Scanner()
scanner = Scanner(store)
item = self.__next_item() item = self.__next_item()
while item: while item:
@ -155,8 +154,6 @@ class ScannerProcessingQueue(Thread):
item = self.__next_item() item = self.__next_item()
scanner.finish() scanner.finish()
store.commit()
store.close()
self.__logger.debug("Freeing scanner") self.__logger.debug("Freeing scanner")
del scanner del scanner
@ -208,6 +205,7 @@ class SupysonicWatcher(object):
def __init__(self, config): def __init__(self, config):
self.__config = config self.__config = config
self.__running = True self.__running = True
self.__db = get_database(config.BASE['database_uri'])
def run(self): def run(self):
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -227,22 +225,22 @@ class SupysonicWatcher(object):
} }
logger.setLevel(mapping.get(self.__config.DAEMON['log_level'].upper(), logging.NOTSET)) logger.setLevel(mapping.get(self.__config.DAEMON['log_level'].upper(), logging.NOTSET))
store = db.get_store(self.__config.BASE['database_uri']) with db_session:
folders = store.find(db.Folder, db.Folder.root == True) folders = Folder.select(lambda f: f.root)
shouldrun = folders.exists()
if not folders.count(): if not shouldrun:
logger.info("No folder set. Exiting.") logger.info("No folder set. Exiting.")
store.close() release_database(self.__db)
return return
queue = ScannerProcessingQueue(self.__config.BASE['database_uri'], self.__config.DAEMON['wait_delay'], logger) queue = ScannerProcessingQueue(self.__config.DAEMON['wait_delay'], logger)
handler = SupysonicWatcherEventHandler(self.__config.BASE['scanner_extensions'], queue, logger) handler = SupysonicWatcherEventHandler(self.__config.BASE['scanner_extensions'], queue, logger)
observer = Observer() observer = Observer()
for folder in folders: with db_session:
logger.info("Starting watcher for %s", folder.path) for folder in folders:
observer.schedule(handler, folder.path, recursive = True) logger.info("Starting watcher for %s", folder.path)
store.close() observer.schedule(handler, folder.path, recursive = True)
try: try:
signal(SIGTERM, self.__terminate) signal(SIGTERM, self.__terminate)
@ -260,6 +258,7 @@ class SupysonicWatcher(object):
observer.join() observer.join()
queue.stop() queue.stop()
queue.join() queue.join()
release_database(self.__db)
def stop(self): def stop(self):
self.__running = False self.__running = False

40
tests/base/test_cli.py
View File

@ -15,27 +15,22 @@ import tempfile
import unittest import unittest
from contextlib import contextmanager from contextlib import contextmanager
from pony.orm import db_session
from StringIO import StringIO from StringIO import StringIO
from supysonic.db import Folder, User, get_store from supysonic.db import Folder, User, get_database, release_database
from supysonic.cli import SupysonicCLI from supysonic.cli import SupysonicCLI
from ..testbase import TestConfig from ..testbase import TestConfig
class CLITestCase(unittest.TestCase): class CLITestCase(unittest.TestCase):
""" Really basic tests. Some even don't check anything but are juste there for coverage """ """ Really basic tests. Some even don't check anything but are just there for coverage """
def setUp(self): def setUp(self):
conf = TestConfig(False, False) conf = TestConfig(False, False)
self.__dbfile = tempfile.mkstemp()[1] self.__dbfile = tempfile.mkstemp()[1]
conf.BASE['database_uri'] = 'sqlite:///' + self.__dbfile conf.BASE['database_uri'] = 'sqlite:///' + self.__dbfile
self.__store = get_store(conf.BASE['database_uri']) self.__store = get_database(conf.BASE['database_uri'], True)
with io.open('schema/sqlite.sql', 'r') as sql:
schema = sql.read()
for statement in schema.split(';'):
self.__store.execute(statement)
self.__store.commit()
self.__stdout = StringIO() self.__stdout = StringIO()
self.__stderr = StringIO() self.__stderr = StringIO()
@ -44,7 +39,7 @@ class CLITestCase(unittest.TestCase):
def tearDown(self): def tearDown(self):
self.__stdout.close() self.__stdout.close()
self.__stderr.close() self.__stderr.close()
self.__store.close() release_database(self.__store)
os.unlink(self.__dbfile) os.unlink(self.__dbfile)
@contextmanager @contextmanager
@ -59,9 +54,10 @@ class CLITestCase(unittest.TestCase):
with self._tempdir() as d: with self._tempdir() as d:
self.__cli.onecmd('folder add tmpfolder ' + d) self.__cli.onecmd('folder add tmpfolder ' + d)
f = self.__store.find(Folder).one() with db_session:
self.assertIsNotNone(f) f = Folder.select().first()
self.assertEqual(f.path, d) self.assertIsNotNone(f)
self.assertEqual(f.path, d)
def test_folder_add_errors(self): def test_folder_add_errors(self):
with self._tempdir() as d: with self._tempdir() as d:
@ -71,14 +67,17 @@ class CLITestCase(unittest.TestCase):
self.__cli.onecmd('folder add f1 ' + d) self.__cli.onecmd('folder add f1 ' + d)
self.__cli.onecmd('folder add f3 /invalid/path') self.__cli.onecmd('folder add f3 /invalid/path')
self.assertEqual(self.__store.find(Folder).count(), 1) with db_session:
self.assertEqual(Folder.select().count(), 1)
def test_folder_delete(self): def test_folder_delete(self):
with self._tempdir() as d: with self._tempdir() as d:
self.__cli.onecmd('folder add tmpfolder ' + d) self.__cli.onecmd('folder add tmpfolder ' + d)
self.__cli.onecmd('folder delete randomfolder') self.__cli.onecmd('folder delete randomfolder')
self.__cli.onecmd('folder delete tmpfolder') self.__cli.onecmd('folder delete tmpfolder')
self.assertEqual(self.__store.find(Folder).count(), 0)
with db_session:
self.assertEqual(Folder.select().count(), 0)
def test_folder_list(self): def test_folder_list(self):
with self._tempdir() as d: with self._tempdir() as d:
@ -97,13 +96,17 @@ class CLITestCase(unittest.TestCase):
def test_user_add(self): def test_user_add(self):
self.__cli.onecmd('user add -p Alic3 alice') self.__cli.onecmd('user add -p Alic3 alice')
self.__cli.onecmd('user add -p alice alice') self.__cli.onecmd('user add -p alice alice')
self.assertEqual(self.__store.find(User).count(), 1)
with db_session:
self.assertEqual(User.select().count(), 1)
def test_user_delete(self): def test_user_delete(self):
self.__cli.onecmd('user add -p Alic3 alice') self.__cli.onecmd('user add -p Alic3 alice')
self.__cli.onecmd('user delete alice') self.__cli.onecmd('user delete alice')
self.__cli.onecmd('user delete bob') self.__cli.onecmd('user delete bob')
self.assertEqual(self.__store.find(User).count(), 0)
with db_session:
self.assertEqual(User.select().count(), 0)
def test_user_list(self): def test_user_list(self):
self.__cli.onecmd('user add -p Alic3 alice') self.__cli.onecmd('user add -p Alic3 alice')
@ -114,7 +117,8 @@ class CLITestCase(unittest.TestCase):
self.__cli.onecmd('user add -p Alic3 alice') self.__cli.onecmd('user add -p Alic3 alice')
self.__cli.onecmd('user setadmin alice') self.__cli.onecmd('user setadmin alice')
self.__cli.onecmd('user setadmin bob') self.__cli.onecmd('user setadmin bob')
self.assertTrue(self.__store.find(User, User.name == 'alice').one().admin) with db_session:
self.assertTrue(User.get(name = 'alice').admin)
def test_user_changepass(self): def test_user_changepass(self):
self.__cli.onecmd('user add -p Alic3 alice') self.__cli.onecmd('user add -p Alic3 alice')

126
tests/base/test_scanner.py
View File

@ -16,6 +16,7 @@ import tempfile
import unittest import unittest
from contextlib import contextmanager from contextlib import contextmanager
from pony.orm import db_session, commit
from supysonic import db from supysonic import db
from supysonic.managers.folder import FolderManager from supysonic.managers.folder import FolderManager
@ -23,133 +24,158 @@ from supysonic.scanner import Scanner
class ScannerTestCase(unittest.TestCase): class ScannerTestCase(unittest.TestCase):
def setUp(self): def setUp(self):
self.store = db.get_store('sqlite:') self.store = db.get_database('sqlite:', True)
with io.open('schema/sqlite.sql', 'r') as f:
for statement in f.read().split(';'):
self.store.execute(statement)
FolderManager.add(self.store, 'folder', os.path.abspath('tests/assets')) FolderManager.add('folder', os.path.abspath('tests/assets'))
self.folder = self.store.find(db.Folder).one() with db_session:
self.assertIsNotNone(self.folder) folder = db.Folder.select().first()
self.assertIsNotNone(folder)
self.folderid = folder.id
self.scanner = Scanner(self.store) self.scanner = Scanner()
self.scanner.scan(self.folder) self.scanner.scan(folder)
def tearDown(self): def tearDown(self):
self.scanner.finish() self.scanner.finish()
self.store.close() db.release_database(self.store)
@contextmanager @contextmanager
def __temporary_track_copy(self): def __temporary_track_copy(self):
track = self.store.find(db.Track).one() track = db.Track.select().first()
with tempfile.NamedTemporaryFile(dir = os.path.dirname(track.path)) as tf: with tempfile.NamedTemporaryFile(dir = os.path.dirname(track.path)) as tf:
with io.open(track.path, 'rb') as f: with io.open(track.path, 'rb') as f:
tf.write(f.read()) tf.write(f.read())
yield tf yield tf
@db_session
def test_scan(self): def test_scan(self):
self.assertEqual(self.store.find(db.Track).count(), 1) self.assertEqual(db.Track.select().count(), 1)
self.assertRaises(TypeError, self.scanner.scan, None) self.assertRaises(TypeError, self.scanner.scan, None)
self.assertRaises(TypeError, self.scanner.scan, 'string') self.assertRaises(TypeError, self.scanner.scan, 'string')
@db_session
def test_progress(self): def test_progress(self):
def progress(processed, total): def progress(processed, total):
self.assertIsInstance(processed, int) self.assertIsInstance(processed, int)
self.assertIsInstance(total, int) self.assertIsInstance(total, int)
self.assertLessEqual(processed, total) self.assertLessEqual(processed, total)
self.scanner.scan(self.folder, progress) self.scanner.scan(db.Folder[self.folderid], progress)
@db_session
def test_rescan(self): def test_rescan(self):
self.scanner.scan(self.folder) self.scanner.scan(db.Folder[self.folderid])
self.assertEqual(self.store.find(db.Track).count(), 1) commit()
self.assertEqual(db.Track.select().count(), 1)
@db_session
def test_force_rescan(self): def test_force_rescan(self):
self.scanner = Scanner(self.store, True) self.scanner = Scanner(True)
self.scanner.scan(self.folder) self.scanner.scan(db.Folder[self.folderid])
self.assertEqual(self.store.find(db.Track).count(), 1) commit()
self.assertEqual(db.Track.select().count(), 1)
@db_session
def test_scan_file(self): def test_scan_file(self):
track = self.store.find(db.Track).one() track = db.Track.select().first()
self.assertRaises(TypeError, self.scanner.scan_file, None) self.assertRaises(TypeError, self.scanner.scan_file, None)
self.assertRaises(TypeError, self.scanner.scan_file, track) self.assertRaises(TypeError, self.scanner.scan_file, track)
self.scanner.scan_file('/some/inexistent/path') self.scanner.scan_file('/some/inexistent/path')
self.assertEqual(self.store.find(db.Track).count(), 1) commit()
self.assertEqual(db.Track.select().count(), 1)
@db_session
def test_remove_file(self): def test_remove_file(self):
track = self.store.find(db.Track).one() track = db.Track.select().first()
self.assertRaises(TypeError, self.scanner.remove_file, None) self.assertRaises(TypeError, self.scanner.remove_file, None)
self.assertRaises(TypeError, self.scanner.remove_file, track) self.assertRaises(TypeError, self.scanner.remove_file, track)
self.scanner.remove_file('/some/inexistent/path') self.scanner.remove_file('/some/inexistent/path')
self.assertEqual(self.store.find(db.Track).count(), 1) commit()
self.assertEqual(db.Track.select().count(), 1)
self.scanner.remove_file(track.path) self.scanner.remove_file(track.path)
self.scanner.finish() self.scanner.finish()
self.assertEqual(self.store.find(db.Track).count(), 0) commit()
self.assertEqual(self.store.find(db.Album).count(), 0) self.assertEqual(db.Track.select().count(), 0)
self.assertEqual(self.store.find(db.Artist).count(), 0) self.assertEqual(db.Album.select().count(), 0)
self.assertEqual(db.Artist.select().count(), 0)
@db_session
def test_move_file(self): def test_move_file(self):
track = self.store.find(db.Track).one() track = db.Track.select().first()
self.assertRaises(TypeError, self.scanner.move_file, None, 'string') self.assertRaises(TypeError, self.scanner.move_file, None, 'string')
self.assertRaises(TypeError, self.scanner.move_file, track, 'string') self.assertRaises(TypeError, self.scanner.move_file, track, 'string')
self.assertRaises(TypeError, self.scanner.move_file, 'string', None) self.assertRaises(TypeError, self.scanner.move_file, 'string', None)
self.assertRaises(TypeError, self.scanner.move_file, 'string', track) self.assertRaises(TypeError, self.scanner.move_file, 'string', track)
self.scanner.move_file('/some/inexistent/path', track.path) self.scanner.move_file('/some/inexistent/path', track.path)
self.assertEqual(self.store.find(db.Track).count(), 1) commit()
self.assertEqual(db.Track.select().count(), 1)
self.scanner.move_file(track.path, track.path) self.scanner.move_file(track.path, track.path)
self.assertEqual(self.store.find(db.Track).count(), 1) commit()
self.assertEqual(db.Track.select().count(), 1)
self.assertRaises(Exception, self.scanner.move_file, track.path, '/some/inexistent/path') self.assertRaises(Exception, self.scanner.move_file, track.path, '/some/inexistent/path')
with self.__temporary_track_copy() as tf: with self.__temporary_track_copy() as tf:
self.scanner.scan(self.folder) self.scanner.scan(db.Folder[self.folderid])
self.assertEqual(self.store.find(db.Track).count(), 2) commit()
self.assertEqual(db.Track.select().count(), 2)
self.scanner.move_file(tf.name, track.path) self.scanner.move_file(tf.name, track.path)
self.assertEqual(self.store.find(db.Track).count(), 1) commit()
self.assertEqual(db.Track.select().count(), 1)
track = self.store.find(db.Track).one() track = db.Track.select().first()
new_path = os.path.abspath(os.path.join(os.path.dirname(track.path), '..', 'silence.mp3')) new_path = os.path.abspath(os.path.join(os.path.dirname(track.path), '..', 'silence.mp3'))
self.scanner.move_file(track.path, new_path) self.scanner.move_file(track.path, new_path)
self.assertEqual(self.store.find(db.Track).count(), 1) commit()
self.assertEqual(db.Track.select().count(), 1)
self.assertEqual(track.path, new_path) self.assertEqual(track.path, new_path)
@db_session
def test_rescan_corrupt_file(self): def test_rescan_corrupt_file(self):
track = self.store.find(db.Track).one() track = db.Track.select().first()
self.scanner = Scanner(self.store, True) self.scanner = Scanner(True)
with self.__temporary_track_copy() as tf: with self.__temporary_track_copy() as tf:
self.scanner.scan(self.folder) self.scanner.scan(db.Folder[self.folderid])
self.assertEqual(self.store.find(db.Track).count(), 2) commit()
self.assertEqual(db.Track.select().count(), 2)
tf.seek(0, 0) tf.seek(0, 0)
tf.write('\x00' * 4096) tf.write('\x00' * 4096)
tf.truncate() tf.truncate()
self.scanner.scan(self.folder) self.scanner.scan(db.Folder[self.folderid])
self.assertEqual(self.store.find(db.Track).count(), 1) commit()
self.assertEqual(db.Track.select().count(), 1)
@db_session
def test_rescan_removed_file(self): def test_rescan_removed_file(self):
track = self.store.find(db.Track).one() track = db.Track.select().first()
with self.__temporary_track_copy() as tf: with self.__temporary_track_copy() as tf:
self.scanner.scan(self.folder) self.scanner.scan(db.Folder[self.folderid])
self.assertEqual(self.store.find(db.Track).count(), 2) commit()
self.assertEqual(db.Track.select().count(), 2)
self.scanner.scan(self.folder) self.scanner.scan(db.Folder[self.folderid])
self.assertEqual(self.store.find(db.Track).count(), 1) commit()
self.assertEqual(db.Track.select().count(), 1)
@db_session
def test_scan_tag_change(self): def test_scan_tag_change(self):
self.scanner = Scanner(self.store, True) self.scanner = Scanner(True)
folder = db.Folder[self.folderid]
with self.__temporary_track_copy() as tf: with self.__temporary_track_copy() as tf:
self.scanner.scan(self.folder) self.scanner.scan(folder)
copy = self.store.find(db.Track, db.Track.path == tf.name).one() commit()
copy = db.Track.get(path = tf.name)
self.assertEqual(copy.artist.name, 'Some artist') self.assertEqual(copy.artist.name, 'Some artist')
self.assertEqual(copy.album.name, 'Awesome album') self.assertEqual(copy.album.name, 'Awesome album')
@ -158,12 +184,12 @@ class ScannerTestCase(unittest.TestCase):
tags['album'] = 'Crappy album' tags['album'] = 'Crappy album'
tags.save() tags.save()
self.scanner.scan(self.folder) self.scanner.scan(folder)
self.scanner.finish() self.scanner.finish()
self.assertEqual(copy.artist.name, 'Renamed artist') self.assertEqual(copy.artist.name, 'Renamed artist')
self.assertEqual(copy.album.name, 'Crappy album') self.assertEqual(copy.album.name, 'Crappy album')
self.assertIsNotNone(self.store.find(db.Artist, db.Artist.name == 'Some artist').one()) self.assertIsNotNone(db.Artist.get(name = 'Some artist'))
self.assertIsNotNone(self.store.find(db.Album, db.Album.name == 'Awesome album').one()) self.assertIsNotNone(db.Album.get(name = 'Awesome album'))
def test_stats(self): def test_stats(self):
self.assertEqual(self.scanner.stats(), ((1,1,1),(0,0,0))) self.assertEqual(self.scanner.stats(), ((1,1,1),(0,0,0)))

77
tests/base/test_watcher.py
View File

@ -18,9 +18,10 @@ import time
import unittest import unittest
from contextlib import contextmanager from contextlib import contextmanager
from pony.orm import db_session
from threading import Thread from threading import Thread
from supysonic.db import get_store, Track, Artist from supysonic.db import get_database, release_database, Track, Artist
from supysonic.managers.folder import FolderManager from supysonic.managers.folder import FolderManager
from supysonic.watcher import SupysonicWatcher from supysonic.watcher import SupysonicWatcher
@ -38,29 +39,13 @@ class WatcherTestConfig(TestConfig):
self.BASE['database_uri'] = db_uri self.BASE['database_uri'] = db_uri
class WatcherTestBase(unittest.TestCase): class WatcherTestBase(unittest.TestCase):
@contextmanager
def _get_store(self):
store = None
try:
store = get_store('sqlite:///' + self.__dbfile)
yield store
store.commit()
store.close()
except:
store.rollback()
store.close()
raise
def setUp(self): def setUp(self):
self.__dbfile = tempfile.mkstemp()[1] self.__dbfile = tempfile.mkstemp()[1]
conf = WatcherTestConfig('sqlite:///' + self.__dbfile) dburi = 'sqlite:///' + self.__dbfile
self.__sleep_time = conf.DAEMON['wait_delay'] + 1 release_database(get_database(dburi, True))
with self._get_store() as store: conf = WatcherTestConfig(dburi)
with io.open('schema/sqlite.sql', 'r') as sql: self.__sleep_time = conf.DAEMON['wait_delay'] + 1
schema = sql.read()
for statement in schema.split(';'):
store.execute(statement)
self.__watcher = SupysonicWatcher(conf) self.__watcher = SupysonicWatcher(conf)
self.__thread = Thread(target = self.__watcher.run) self.__thread = Thread(target = self.__watcher.run)
@ -82,6 +67,12 @@ class WatcherTestBase(unittest.TestCase):
def _sleep(self): def _sleep(self):
time.sleep(self.__sleep_time) time.sleep(self.__sleep_time)
@contextmanager
def _tempdbrebind(self):
db = get_database('sqlite:///' + self.__dbfile)
try: yield
finally: release_database(db)
class NothingToWatchTestCase(WatcherTestBase): class NothingToWatchTestCase(WatcherTestBase):
def test_spawn_useless_watcher(self): def test_spawn_useless_watcher(self):
self._start() self._start()
@ -93,8 +84,7 @@ class WatcherTestCase(WatcherTestBase):
def setUp(self): def setUp(self):
super(WatcherTestCase, self).setUp() super(WatcherTestCase, self).setUp()
self.__dir = tempfile.mkdtemp() self.__dir = tempfile.mkdtemp()
with self._get_store() as store: FolderManager.add('Folder', self.__dir)
FolderManager.add(store, 'Folder', self.__dir)
self._start() self._start()
def tearDown(self): def tearDown(self):
@ -115,9 +105,9 @@ class WatcherTestCase(WatcherTestBase):
shutil.copyfile('tests/assets/folder/silence.mp3', path) shutil.copyfile('tests/assets/folder/silence.mp3', path)
return path return path
@db_session
def assertTrackCountEqual(self, expected): def assertTrackCountEqual(self, expected):
with self._get_store() as store: self.assertEqual(Track.select().count(), expected)
self.assertEqual(store.find(Track).count(), expected)
def test_add(self): def test_add(self):
self._addfile() self._addfile()
@ -128,7 +118,8 @@ class WatcherTestCase(WatcherTestBase):
def test_add_nowait_stop(self): def test_add_nowait_stop(self):
self._addfile() self._addfile()
self._stop() self._stop()
self.assertTrackCountEqual(1) with self._tempdbrebind():
self.assertTrackCountEqual(1)
def test_add_multiple(self): def test_add_multiple(self):
self._addfile() self._addfile()
@ -136,46 +127,46 @@ class WatcherTestCase(WatcherTestBase):
self._addfile() self._addfile()
self.assertTrackCountEqual(0) self.assertTrackCountEqual(0)
self._sleep() self._sleep()
with self._get_store() as store: with db_session:
self.assertEqual(store.find(Track).count(), 3) self.assertEqual(Track.select().count(), 3)
self.assertEqual(store.find(Artist).count(), 1) self.assertEqual(Artist.select().count(), 1)
def test_change(self): def test_change(self):
path = self._addfile() path = self._addfile()
self._sleep() self._sleep()
trackid = None trackid = None
with self._get_store() as store: with db_session:
self.assertEqual(store.find(Track).count(), 1) self.assertEqual(Track.select().count(), 1)
self.assertEqual(store.find(Artist, Artist.name == 'Some artist').count(), 1) self.assertEqual(Artist.select(lambda a: a.name == 'Some artist').count(), 1)
trackid = store.find(Track).one().id trackid = Track.select().first().id
tags = mutagen.File(path, easy = True) tags = mutagen.File(path, easy = True)
tags['artist'] = 'Renamed' tags['artist'] = 'Renamed'
tags.save() tags.save()
self._sleep() self._sleep()
with self._get_store() as store: with db_session:
self.assertEqual(store.find(Track).count(), 1) self.assertEqual(Track.select().count(), 1)
self.assertEqual(store.find(Artist, Artist.name == 'Some artist').count(), 0) self.assertEqual(Artist.select(lambda a: a.name == 'Some artist').count(), 0)
self.assertEqual(store.find(Artist, Artist.name == 'Renamed').count(), 1) self.assertEqual(Artist.select(lambda a: a.name == 'Renamed').count(), 1)
self.assertEqual(store.find(Track).one().id, trackid) self.assertEqual(Track.select().first().id, trackid)
def test_rename(self): def test_rename(self):
path = self._addfile() path = self._addfile()
self._sleep() self._sleep()
trackid = None trackid = None
with self._get_store() as store: with db_session:
self.assertEqual(store.find(Track).count(), 1) self.assertEqual(Track.select().count(), 1)
trackid = store.find(Track).one().id trackid = Track.select().first().id
newpath = self._temppath() newpath = self._temppath()
shutil.move(path, newpath) shutil.move(path, newpath)
self._sleep() self._sleep()
with self._get_store() as store: with db_session:
track = store.find(Track).one() track = Track.select().first()
self.assertIsNotNone(track) self.assertIsNotNone(track)
self.assertNotEqual(track.path, path) self.assertNotEqual(track.path, path)
self.assertEqual(track.path, newpath) self.assertEqual(track.path, newpath)