From 9c754908e645834a53ce3906aac76a0aa92e98f4 Mon Sep 17 00:00:00 2001 From: kaliko Date: Thu, 29 Apr 2021 16:03:27 +0200 Subject: [PATCH] Add drop_all, fetch_artists*, get_bl_* methods --- sima/lib/db.py | 214 +++++++++++++++++++++++++++++------ tests/test_db.py | 125 --------------------- tests/test_simadb.py | 258 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 439 insertions(+), 158 deletions(-) delete mode 100644 tests/test_db.py create mode 100644 tests/test_simadb.py diff --git a/sima/lib/db.py b/sima/lib/db.py index 6d7836a..9ce9082 100644 --- a/sima/lib/db.py +++ b/sima/lib/db.py @@ -32,6 +32,12 @@ from sima.lib.meta import Artist, Album from sima.lib.track import Track +class SimaDBError(Exception): + """ + Exceptions. + """ + + class SimaDB: "SQLite management" @@ -80,43 +86,78 @@ class SimaDB: '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): @@ -250,6 +291,8 @@ class SimaDB: """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: @@ -325,7 +368,30 @@ class SimaDB: 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 @@ -349,10 +415,92 @@ class SimaDB: 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 diff --git a/tests/test_db.py b/tests/test_db.py deleted file mode 100644 index 2419d44..0000000 --- a/tests/test_db.py +++ /dev/null @@ -1,125 +0,0 @@ -# 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 diff --git a/tests/test_simadb.py b/tests/test_simadb.py new file mode 100644 index 0000000..b71fa1b --- /dev/null +++ b/tests/test_simadb.py @@ -0,0 +1,258 @@ +# 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 -- 2.39.2