from sima.lib.track import Track
+class SimaDBError(Exception):
+ """
+ Exceptions.
+ """
+
+
class SimaDB:
"SQLite management"
'CREATE TABLE IF NOT EXISTS history (id INTEGER PRIMARY KEY, '
'last_play TIMESTAMP, track integer, '
'FOREIGN KEY(track) REFERENCES tracks(id))')
+ connection.execute( # BLOCKLIST
+ 'CREATE TABLE IF NOT EXISTS blocklist (id INTEGER PRIMARY KEY, '
+ 'artist INTEGER, album INTEGER, track INTEGER, '
+ 'FOREIGN KEY(artist) REFERENCES artists(id), '
+ 'FOREIGN KEY(album) REFERENCES albums(id), '
+ 'FOREIGN KEY(track) REFERENCES tracks(id))')
# Create cleanup triggers:
- # Tracks table
+ # DELETE history → Tracks table
connection.execute('''
- CREATE TRIGGER IF NOT EXISTS cleanup_tracks
- AFTER DELETE ON history
- WHEN ((SELECT count(*) FROM history WHERE track=old.id) = 0)
- BEGIN
- DELETE FROM tracks WHERE id = old.id;
- END;
- ''')
- # Artists table
+ CREATE TRIGGER IF NOT EXISTS del_history_cleanup_tracks
+ AFTER DELETE ON history
+ WHEN ((SELECT count(*) FROM history WHERE track=old.track) = 0 AND
+ (SELECT count(*) FROM blocklist WHERE track=old.track) = 0)
+ BEGIN
+ DELETE FROM tracks WHERE id = old.track;
+ END;
+ ''')
+ # DELETE Tracks → Artists table
connection.execute('''
- CREATE TRIGGER IF NOT EXISTS cleanup_artists
- AFTER DELETE ON tracks
- WHEN ((SELECT count(*) FROM tracks WHERE artist=old.artist) = 0)
- BEGIN
- DELETE FROM artists WHERE id = old.artist;
- END;
- ''')
- # Albums table
+ CREATE TRIGGER IF NOT EXISTS del_tracks_cleanup_artists
+ AFTER DELETE ON tracks
+ WHEN ((SELECT count(*) FROM tracks WHERE artist=old.artist) = 0 AND
+ (SELECT count(*) FROM blocklist WHERE artist=old.artist) = 0)
+ BEGIN
+ DELETE FROM artists WHERE id = old.artist;
+ END;
+ ''')
+ # DELETE Tracks → Albums table
connection.execute('''
- CREATE TRIGGER IF NOT EXISTS cleanup_albums
- AFTER DELETE ON tracks
- WHEN ((SELECT count(*) FROM tracks WHERE album=old.album) = 0)
- BEGIN
- DELETE FROM albums WHERE id = old.album;
- END;
- ''')
- # AlbumArtists table
+ CREATE TRIGGER IF NOT EXISTS del_tracks_cleanup_albums
+ AFTER DELETE ON tracks
+ WHEN ((SELECT count(*) FROM tracks WHERE album=old.album) = 0 AND
+ (SELECT count(*) FROM blocklist WHERE album=old.album) = 0)
+ BEGIN
+ DELETE FROM albums WHERE id = old.album;
+ END;
+ ''')
+ # DELETE Tracks → cleanup AlbumArtists table
connection.execute('''
- CREATE TRIGGER IF NOT EXISTS cleanup_albumartists
- AFTER DELETE ON tracks
- WHEN ((SELECT count(*) FROM tracks WHERE albumartist=old.albumartist) = 0)
- BEGIN
- DELETE FROM albumartists WHERE id = old.albumartist;
- END;
- ''')
+ CREATE TRIGGER IF NOT EXISTS del_tracks_cleanup_albumartists
+ AFTER DELETE ON tracks
+ WHEN ((SELECT count(*) FROM tracks WHERE albumartist=old.albumartist) = 0)
+ BEGIN
+ DELETE FROM albumartists WHERE id = old.albumartist;
+ END;
+ ''')
+ # DELETE blocklist → Tracks table
+ connection.execute('''
+ CREATE TRIGGER IF NOT EXISTS del_blocklist_cleanup_tracks
+ AFTER DELETE ON blocklist
+ WHEN ((SELECT count(*) FROM history WHERE track=old.track) = 0 AND
+ (SELECT count(*) FROM blocklist WHERE track=old.track) = 0)
+ BEGIN
+ DELETE FROM tracks WHERE id = old.track;
+ END;
+ ''')
+ self.close_database_connection(connection)
+
+ def drop_all(self):
+ connection = self.get_database_connection()
+ rows = connection.execute(
+ "SELECT name FROM sqlite_master WHERE type='table'")
+ for r in rows.fetchall():
+ connection.execute(f'DROP TABLE IF EXISTS {r[0]}')
+ connection.close()
+
+ def _remove_blocklist_id(self, blid):
+ """Remove id"""
+ connection = self.get_database_connection()
+ connection.execute('DELETE FROM blocklist'
+ ' WHERE blocklist.id = ?', (blid,))
+ connection.commit()
self.close_database_connection(connection)
def _get_album(self, album, connection):
"""Get a track from Tracks table, add if not existing,
Attention: use Track() object!!
if not in database insert new entry."""
+ if not track.file:
+ raise SimaDBError('Got a track with no file attribute: %r' % track)
if with_connection:
connection = with_connection
else:
connection.commit()
self.close_database_connection(connection)
- def get_history(self, duration=__HIST_DURATION__):
+ def fetch_artists_history(self, duration=__HIST_DURATION__):
+ date = datetime.utcnow() - timedelta(hours=duration)
+ connection = self.get_database_connection()
+ connection.row_factory = sqlite3.Row
+ rows = connection.execute("""
+ SELECT artists.name AS name,
+ artists.mbid as mbid
+ FROM history
+ JOIN tracks ON history.track = tracks.id
+ LEFT OUTER JOIN artists ON tracks.artist = artists.id
+ WHERE history.last_play > ?
+ ORDER BY history.last_play DESC""", (date.isoformat(' '),))
+ hist = list()
+ for row in rows:
+ if hist and hist[-1] == Album(**row): # remove consecutive dupes
+ continue
+ hist.append(Album(**row))
+ connection.close()
+ return hist
+
+ def fetch_history(self, duration=__HIST_DURATION__):
+ """Fetches tracks history, more recent first
+ :param int duration: How long ago to fetch history from
+ """
date = datetime.utcnow() - timedelta(hours=duration)
connection = self.get_database_connection()
connection.row_factory = sqlite3.Row
connection.close()
return hist
+ def get_bl_track(self, track, with_connection=None, add=True):
+ """Add a track to blocklist
+ :param sima.lib.track.Track track: Track object to add to blocklist
+ :param sqlite3.Connection with_connection: sqlite3.Connection to reuse, else create a new one
+ :param bool add: Default is to add a new record, set to False to fetch associated record"""
+ if with_connection:
+ connection = with_connection
+ else:
+ connection = self.get_database_connection()
+ track_id = self.get_track(track, with_connection=connection, add=True)
+ rows = connection.execute(
+ "SELECT * FROM blocklist WHERE track = ?", (track_id,))
+ if not rows.fetchone():
+ if not add:
+ return None
+ connection.execute('INSERT INTO blocklist (track) VALUES (?)',
+ (track_id,))
+ connection.commit()
+ rows = connection.execute(
+ "SELECT * FROM blocklist WHERE track = ?", (track_id,))
+ return rows.fetchone()[0]
+
+ def get_bl_album(self, album, with_connection=None, add=True):
+ """Add an album to blocklist
+ :param sima.lib.meta.Album: Album object to add to blocklist
+ :param sqlite3.Connection with_connection: sqlite3.Connection to reuse, ele create a new one
+ :param bool add: Default is to add a new record, set to False to fetch associated record"""
+ if with_connection:
+ connection = with_connection
+ else:
+ connection = self.get_database_connection()
+ album_id = self.get_album(album, with_connection=connection, add=True)
+ rows = connection.execute(
+ "SELECT * FROM blocklist WHERE album = ?", (album_id,))
+ if not rows.fetchone():
+ if not add:
+ return None
+ connection.execute('INSERT INTO blocklist (album) VALUES (?)',
+ (album_id,))
+ connection.commit()
+ rows = connection.execute(
+ "SELECT * FROM blocklist WHERE album = ?", (album_id,))
+ return rows.fetchone()
+
+ def get_bl_artist(self, artist, with_connection=None, add=True):
+ """Add an artist to blocklist
+ :param sima.lib.meta.Artist: Artist object to add to blocklist
+ :param sqlite3.Connection with_connection: sqlite3.Connection to reuse, ele create a new one
+ :param bool add: Default is to add a new record, set to False to fetch associated record"""
+ if with_connection:
+ connection = with_connection
+ else:
+ connection = self.get_database_connection()
+ artist_id = self.get_artist(artist, with_connection=connection, add=True)
+ rows = connection.execute(
+ "SELECT * FROM blocklist WHERE artist = ?", (artist_id,))
+ if not rows.fetchone():
+ if not add:
+ return None
+ connection.execute('INSERT INTO blocklist (artist) VALUES (?)',
+ (artist_id,))
+ connection.commit()
+ rows = connection.execute(
+ "SELECT * FROM blocklist WHERE artist = ?", (artist_id,))
+ return rows.fetchone()
+
def main():
+ DEVOLT = {
+ 'album': 'Grey',
+ 'albumartist': 'Devolt',
+ 'artist': 'Devolt',
+ 'date': '2011-12-01',
+ 'file': 'music/Devolt/2011-Grey/03-Devolt - Crazy.mp3',
+ 'musicbrainz_albumartistid': 'd8e7e3e2-49ab-4f7c-b148-fc946d521f99',
+ 'musicbrainz_albumid': 'ea2ef2cf-59e1-443a-817e-9066e3e0be4b',
+ 'musicbrainz_artistid': 'd8e7e3e2-49ab-4f7c-b148-fc946d521f99',
+ 'musicbrainz_trackid': 'fabf8fc9-2ae5-49c9-8214-a839c958d872',
+ 'duration': '220.000',
+ 'title': 'Crazy'}
db = SimaDB('/dev/shm/test.sqlite')
db.create_db()
+ db.add_history(Track(**DEVOLT))
+ DEVOLT['file'] = 'foo'
+ print(db.get_bl_track(Track(**DEVOLT)))
+ db.add_history(Track(**DEVOLT))
# VIM MODLINE
# vim: ai ts=4 sw=4 sts=4 expandtab fileencoding=utf8
+++ /dev/null
-# coding: utf-8
-
-import unittest
-import os
-import datetime
-
-from sima.lib.db import SimaDB
-from sima.lib.track import Track
-
-
-DEVOLT = {
- 'album': 'Grey',
- 'albumartist': 'Devolt',
- 'albumartistsort': 'Devolt',
- 'artist': 'Devolt',
- 'date': '2011-12-01',
- 'disc': '1/1',
- 'file': 'music/Devolt/2011-Grey/03-Devolt - Crazy.mp3',
- 'last-modified': '2012-04-02T20:48:59Z',
- 'musicbrainz_albumartistid': 'd8e7e3e2-49ab-4f7c-b148-fc946d521f99',
- 'musicbrainz_albumid': 'ea2ef2cf-59e1-443a-817e-9066e3e0be4b',
- 'musicbrainz_artistid': 'd8e7e3e2-49ab-4f7c-b148-fc946d521f99',
- 'musicbrainz_trackid': 'fabf8fc9-2ae5-49c9-8214-a839c958d872',
- 'time': '220',
- 'duration': '220.000',
- 'title': 'Crazy',
- 'track': '3/6'}
-
-
-class Main_TestDB(unittest.TestCase):
- db_file = 'file::memory:?cache=shared'
- #db_file = '/dev/shm/unittest.sqlite'
-
- @classmethod
- def setUpClass(self):
- self.db = SimaDB(db_path=self.db_file)
- # Maintain a connection to keep the database between test cases
- self.conn = self.db.get_database_connection()
-
- @classmethod
- def tearDownClass(self):
- self.conn.close()
-
-
-class TestDB(Main_TestDB):
-
- def test_00_recreation(self):
- self.db.create_db()
-
- def test_01_add_track(self):
- trk = Track(**DEVOLT)
- trk_id = self.db.get_track(trk)
- self.assertEqual(trk_id, self.db.get_track(trk),
- 'Same track, same record')
-
- def test_02_history(self):
- curr = datetime.datetime.utcnow()
- # set records in the past to ease purging then
- last = curr - datetime.timedelta(hours=1)
- trk = Track(**DEVOLT)
- self.db.add_history(trk, date=last)
- self.db.add_history(trk, date=last)
- hist = self.db.get_history()
- self.assertEqual(len(hist), 1, 'same track results in a single record')
-
- trk_foo = Track(file="/foo/bar/baz.flac")
- self.db.add_history(trk_foo, date=last)
- hist = self.db.get_history()
- self.assertEqual(len(hist), 2)
-
- self.db.add_history(trk, date=last)
- hist = self.db.get_history()
- self.assertEqual(len(hist), 2)
- self.db.purge_history(duration=0)
- hist = self.db.get_history()
- self.assertEqual(len(hist), 0)
-
- # Controls we got history in the right order
- # recent first, oldest last
- hist = list()
- for i in range(1, 5): # starts at 1 to ensure records are in the past
- trk = Track(file=f'/foo/bar.{i}', name='{i}-baz', album='foolbum')
- hist.append(trk)
- last = curr - datetime.timedelta(minutes=i)
- self.db.add_history(trk, date=last)
- hist_records = self.db.get_history()
- self.assertEqual(hist, hist_records)
- self.db.purge_history(duration=0)
-
- def test_04_triggers(self):
- self.db.purge_history(duration=0)
- curr = datetime.datetime.utcnow()
- tracks_ids = list()
- # Set 4 records, same album
- for i in range(1, 6): # starts at 1 to ensure records are in the past
- trk = Track(file=f'/foo/{i}', name=f'{i}', artist='fooart',
- albumartist='fooalbart', album='foolbum',)
- tracks_ids.append(self.db.get_track(trk)) # Add track, save its DB id
- # set records in the past to ease purging then
- last = curr - datetime.timedelta(minutes=i)
- self.db.add_history(trk, date=last) # Add to history
- conn = self.db.get_database_connection()
- # Add another track not related (not same album)
- track = Track(file='/baz/bar.baz', name='baz', artist='fooart',
- albumartist='not-same', album='not-same',)
- self.db.get_track(track)
- # for tid in tracks_ids:
- for tid in tracks_ids[:-1]:
- # Delete lastest record
- conn.execute('DELETE FROM history WHERE history.track = ?', (tid,))
- c = conn.execute('SELECT albums.name FROM albums;')
- # There are still albums records (still a history using it)
- self.assertIn((trk.album,), c.fetchall())
- # purging last entry in history or album == trk.album
- c.execute('DELETE FROM history WHERE history.track = ?',
- (tracks_ids[-1],))
- # triggers purge other tables if possible
- c.execute('SELECT albums.name FROM albums;')
- albums = c.fetchall()
- self.assertNotIn(('foolbum',), albums)
- conn.close()
-
-
-# VIM MODLINE
-# vim: ai ts=4 sw=4 sts=4 expandtab fileencoding=utf8
--- /dev/null
+# coding: utf-8
+
+import datetime
+import unittest
+import os
+
+from sima.lib.db import SimaDB
+from sima.lib.track import Track
+from sima.lib.meta import Album
+
+
+DEVOLT = {
+ 'album': 'Grey',
+ 'albumartist': 'Devolt',
+ 'albumartistsort': 'Devolt',
+ 'artist': 'Devolt',
+ 'date': '2011-12-01',
+ 'disc': '1/1',
+ 'file': 'music/Devolt/2011-Grey/03-Devolt - Crazy.mp3',
+ 'last-modified': '2012-04-02T20:48:59Z',
+ 'musicbrainz_albumartistid': 'd8e7e3e2-49ab-4f7c-b148-fc946d521f99',
+ 'musicbrainz_albumid': 'ea2ef2cf-59e1-443a-817e-9066e3e0be4b',
+ 'musicbrainz_artistid': 'd8e7e3e2-49ab-4f7c-b148-fc946d521f99',
+ 'musicbrainz_trackid': 'fabf8fc9-2ae5-49c9-8214-a839c958d872',
+ 'time': '220',
+ 'duration': '220.000',
+ 'title': 'Crazy',
+ 'track': '3/6'}
+
+DB_FILE = 'file::memory:?cache=shared'
+KEEP_FILE = False # File db in file to ease debug
+if KEEP_FILE:
+ DB_FILE = '/dev/shm/unittest.sqlite'
+CURRENT = datetime.datetime.utcnow()
+IN_THE_PAST = CURRENT - datetime.timedelta(hours=1)
+
+
+class Main(unittest.TestCase):
+ """Deal with database creation and purge between tests"""
+
+ @classmethod
+ def setUpClass(self):
+ self.db = SimaDB(db_path=DB_FILE)
+
+ def setUp(self):
+ # Maintain a connection to keep the database (when stored in memory)
+ self.conn = self.db.get_database_connection()
+ self.db.drop_all()
+ self.db.create_db()
+
+ def tearDown(self):
+ if not KEEP_FILE:
+ self.db.drop_all()
+ self.conn.close()
+
+ @classmethod
+ def tearDownClass(self):
+ if KEEP_FILE:
+ return
+ if os.path.isfile(DB_FILE):
+ os.unlink(DB_FILE)
+
+
+class Test_00DB(Main):
+
+ def test_00_recreation(self):
+ self.db.create_db()
+
+ def test_01_add_track(self):
+ trk = Track(**DEVOLT)
+ trk_id = self.db.get_track(trk)
+ self.assertEqual(trk_id, self.db.get_track(trk),
+ 'Same track, same record')
+
+ def test_02_history(self):
+ # set records in the past to ease purging then
+ last = CURRENT - datetime.timedelta(hours=1)
+ trk = Track(**DEVOLT)
+ self.db.add_history(trk, date=last)
+ self.db.add_history(trk, date=last)
+ hist = self.db.fetch_history()
+ self.assertEqual(len(hist), 1, 'same track results in a single record')
+
+ trk_foo = Track(file="/foo/bar/baz.flac")
+ self.db.add_history(trk_foo, date=last)
+ hist = self.db.fetch_history()
+ self.assertEqual(len(hist), 2)
+
+ self.db.add_history(trk, date=last)
+ hist = self.db.fetch_history()
+ self.assertEqual(len(hist), 2)
+ self.db.purge_history(duration=0)
+ hist = self.db.fetch_history()
+ self.assertEqual(len(hist), 0)
+
+ # Controls we got history in the right order
+ # recent first, oldest last
+ hist = list()
+ for i in range(1, 5): # starts at 1 to ensure records are in the past
+ trk = Track(file=f'/foo/bar.{i}', name='{i}-baz', album='foolbum')
+ hist.append(trk)
+ last = CURRENT - datetime.timedelta(minutes=i)
+ self.db.add_history(trk, date=last)
+ hist_records = self.db.fetch_history()
+ self.assertEqual(hist, hist_records)
+ self.db.purge_history(duration=0)
+
+ def test_history_to_tracks(self):
+ tr = dict(**DEVOLT)
+ tr.pop('file')
+ trk01 = Track(file='01', **tr)
+ self.db.add_history(trk01, CURRENT-datetime.timedelta(minutes=1))
+ #
+ tr.pop('musicbrainz_artistid')
+ trk02 = Track(file='02', **tr)
+ self.db.add_history(trk02, CURRENT-datetime.timedelta(minutes=2))
+ #
+ tr.pop('musicbrainz_albumid')
+ trk03 = Track(file='03', **tr)
+ self.db.add_history(trk03, CURRENT-datetime.timedelta(minutes=3))
+ #
+ tr.pop('musicbrainz_albumartistid')
+ trk04 = Track(file='04', **tr)
+ self.db.add_history(trk04, CURRENT-datetime.timedelta(minutes=4))
+ #
+ tr.pop('musicbrainz_trackid')
+ trk05 = Track(file='05', **tr)
+ self.db.add_history(trk05, CURRENT-datetime.timedelta(minutes=5))
+ history = self.db.fetch_history()
+ self.assertEqual(len(history), 5)
+ # Controls history ordering, recent first
+ self.assertEqual(history, [trk01, trk02, trk03, trk04, trk05])
+
+ def test_history_to_artists(self):
+ tr = dict(**DEVOLT)
+ tr.pop('file')
+ tr.pop('musicbrainz_artistid')
+ #
+ trk01 = Track(file='01', **tr)
+ self.db.add_history(trk01, CURRENT-datetime.timedelta(hours=1))
+ #
+ trk02 = Track(file='02', **tr)
+ self.db.add_history(trk02, CURRENT-datetime.timedelta(hours=1))
+ self.db.add_history(trk02, CURRENT-datetime.timedelta(hours=1))
+ #
+ trk03 = Track(file='03', **tr)
+ self.db.add_history(trk03, CURRENT-datetime.timedelta(hours=1))
+ # got multiple tracks, same artist, got artist history len == 1
+ art_history = self.db.fetch_artists_history()
+ self.assertEqual(len(art_history), 1)
+ self.assertEqual(art_history, [trk01.Artist])
+
+ # Now add new artist to history
+ trk04 = Track(file='04', artist='New Art')
+ trk05 = Track(file='05', artist='New² Art')
+ self.db.add_history(trk04, CURRENT-datetime.timedelta(minutes=3))
+ self.db.add_history(trk03, CURRENT-datetime.timedelta(minutes=2))
+ self.db.add_history(trk05, CURRENT-datetime.timedelta(minutes=1))
+ art_history = self.db.fetch_artists_history()
+ # Now we should have 4 artists in history
+ self.assertEqual(len(art_history), 4)
+ # Controling order, recent first
+ self.assertEqual([trk05.artist, trk03.artist,
+ trk04.artist, trk03.artist],
+ art_history)
+
+ def test_04_triggers(self):
+ self.db.purge_history(duration=0)
+ tracks_ids = list()
+ # Add a first track
+ track = Track(file='/baz/bar.baz', name='baz', artist='fooart',
+ albumartist='not-same', album='not-same',)
+ self.db.get_track(track)
+ # Set 6 more records from same artist but not same album
+ for i in range(1, 6): # starts at 1 to ensure records are in the past
+ trk = Track(file=f'/foo/{i}', name=f'{i}', artist='fooart',
+ albumartist='fooalbart', album='foolbum',)
+ # Add track, save its DB id
+ tracks_ids.append(self.db.get_track(trk))
+ # set records in the past to ease purging then
+ last = CURRENT - datetime.timedelta(minutes=i)
+ self.db.add_history(trk, date=last) # Add to history
+ conn = self.db.get_database_connection()
+ # for tid in tracks_ids:
+ for tid in tracks_ids[:-1]:
+ # Delete lastest record
+ conn.execute('DELETE FROM history WHERE history.track = ?',
+ (tid,))
+ c = conn.execute('SELECT albums.name FROM albums;')
+ # There are still albums records (still a history using it)
+ self.assertIn((trk.album,), c.fetchall())
+ # purging last entry in history for album == trk.album
+ conn.execute('DELETE FROM history WHERE history.track = ?',
+ (tracks_ids[-1],))
+ # triggers purge other tables if possible
+ conn.execute('SELECT albums.name FROM albums;')
+ albums = c.fetchall()
+ # No more "foolbum" in the table albums
+ self.assertNotIn(('foolbum',), albums)
+ # There is still "fooart" though
+ c = conn.execute('SELECT artists.name FROM artists;')
+ artists = c.fetchall()
+ # No more "foolbum" in the table albums
+ self.assertIn(('fooart',), artists)
+ conn.close()
+
+
+class Test_01BlockList(Main):
+
+ def test_blocklist_addition(self):
+ tracks_ids = list()
+ # Set 6 records, same album
+ for i in range(1, 6): # starts at 1 to ensure records are in the past
+ trk = Track(file=f'/foo/{i}', name=f'{i}', artist='fooart',
+ albumartist='fooalbart', album='foolbum',)
+ # Add track, save its DB id
+ tracks_ids.append(self.db.get_track(trk))
+ # set records in the past to ease purging then
+ last = CURRENT - datetime.timedelta(minutes=i)
+ self.db.add_history(trk, date=last) # Add to history
+ if i == 1:
+ self.db.get_bl_track(trk)
+ if i == 2:
+ self.db.get_bl_track(trk)
+ self.db.get_bl_album(Album(name=trk.album))
+ if i == 3:
+ self.db.get_bl_artist(trk.Artist)
+
+ def test_blocklist_triggers(self):
+ trk01 = Track(file='01', name='01', artist='artist A', album='album A')
+ trk02 = Track(file='02', name='01', artist='artist A', album='album B')
+ trk01_id = self.db.get_bl_track(trk01)
+ trk02_id = self.db.get_bl_track(trk02)
+ self.db.add_history(trk01, IN_THE_PAST)
+ self.db._remove_blocklist_id(trk01_id)
+ # bl trk01 removed:
+ # albums/artists table not affected since trk01_id still in history
+ conn = self.db.get_database_connection()
+ albums = conn.execute('SELECT albums.name FROM albums;').fetchall()
+ artists = conn.execute('SELECT artists.name FROM artists;').fetchall()
+ self.assertIn(('album A',), albums)
+ self.assertIn(('artist A',), artists)
+ self.db.purge_history(0)
+ # remove last reference to trk01
+ albums = conn.execute('SELECT albums.name FROM albums;').fetchall()
+ self.assertNotIn(('album A',), albums)
+ self.assertIn(('artist A',), artists)
+ # remove trk02
+ self.db._remove_blocklist_id(trk02_id)
+ albums = conn.execute('SELECT albums.name FROM albums;').fetchall()
+ artists = conn.execute('SELECT artists.name FROM artists;').fetchall()
+ self.assertNotIn(('album B',), albums)
+ self.assertNotIn(('artist A'), artists)
+ conn.close()
+
+
+# VIM MODLINE
+# vim: ai ts=4 sw=4 sts=4 expandtab fileencoding=utf8
home / dev / repositories / mpd-sima.git / commitdiff