tests/test_simadb.py

   1 # coding: utf-8
   2
   3 import datetime
   4 import unittest
   5 import os
   6
   7 from sima.lib.simadb import SimaDB
   8 from sima.lib.track import Track
   9 from sima.lib.meta import Album, Artist, MetaContainer
  10
  11
  12 DEVOLT = {
  13     'album': 'Grey',
  14     'albumartist': 'Devolt',
  15     'albumartistsort': 'Devolt',
  16     'artist': 'Devolt',
  17     'genre': ['Rock'],
  18     'date': '2011-12-01',
  19     'disc': '1/1',
  20     'file': 'music/Devolt/2011-Grey/03-Devolt - Crazy.mp3',
  21     'last-modified': '2012-04-02T20:48:59Z',
  22     'musicbrainz_albumartistid': 'd8e7e3e2-49ab-4f7c-b148-fc946d521f99',
  23     'musicbrainz_albumid': 'ea2ef2cf-59e1-443a-817e-9066e3e0be4b',
  24     'musicbrainz_artistid': 'd8e7e3e2-49ab-4f7c-b148-fc946d521f99',
  25     'musicbrainz_trackid': 'fabf8fc9-2ae5-49c9-8214-a839c958d872',
  26     'time': '220',
  27     'duration': '220.000',
  28     'title': 'Crazy',
  29     'track': '3/6'}
  30
  31 DB_FILE = 'file::memory:?cache=shared'
  32 KEEP_FILE = True  # File db in file to ease debug
  33 if KEEP_FILE:
  34     DB_FILE = '/dev/shm/unittest.sqlite'
  35 CURRENT = datetime.datetime.utcnow()
  36 IN_THE_PAST = CURRENT - datetime.timedelta(hours=1)
  37
  38
  39 class Main(unittest.TestCase):
  40     """Deal with database creation and purge between tests"""
  41
  42     @classmethod
  43     def setUpClass(self):
  44         self.db = SimaDB(db_path=DB_FILE)
  45
  46     def setUp(self):
  47         # Maintain a connection to keep the database (when stored in memory)
  48         self.conn = self.db.get_database_connection()
  49         self.db.drop_all()
  50         self.db.create_db()
  51
  52     def tearDown(self):
  53         if not KEEP_FILE:
  54             self.db.drop_all()
  55         self.conn.close()
  56
  57     @classmethod
  58     def tearDownClass(self):
  59         if KEEP_FILE:
  60             return
  61         if os.path.isfile(DB_FILE):
  62             os.unlink(DB_FILE)
  63
  64
  65 class Test_00DB(Main):
  66
  67     def test_00_recreation(self):
  68         self.db.create_db()
  69
  70     def test_01_add_track(self):
  71         trk = Track(**DEVOLT)
  72         trk_id = self.db.get_track(trk)
  73         self.assertEqual(trk_id, self.db.get_track(trk),
  74                          'Same track, same record')
  75
  76     def test_02_history(self):
  77         # set records in the past to ease purging then
  78         last = CURRENT - datetime.timedelta(hours=1)
  79         trk = Track(**DEVOLT)
  80         self.db.add_history(trk, date=last)
  81         self.db.add_history(trk, date=last)
  82         hist = self.db.fetch_history()
  83         self.assertEqual(len(hist), 1, 'same track results in a single record')
  84
  85         trk_foo = Track(file="/foo/bar/baz.flac")
  86         self.db.add_history(trk_foo, date=last)
  87         hist = self.db.fetch_history()
  88         self.assertEqual(len(hist), 2)
  89
  90         self.db.add_history(trk, date=last)
  91         hist = self.db.fetch_history()
  92         self.assertEqual(len(hist), 2)
  93         self.db.purge_history(duration=0)
  94         hist = self.db.fetch_history()
  95         self.assertEqual(len(hist), 0)
  96
  97         # Controls we got history in the right order
  98         # recent first, oldest last
  99         hist = list()
 100         for i in range(1, 5):  # starts at 1 to ensure records are in the past
 101             trk = Track(file=f'/foo/bar.{i}', name=f'{i}-baz',
 102                         album='foolbum', genre=f'{i}')
 103             hist.append(trk)
 104             last = CURRENT - datetime.timedelta(minutes=i)
 105             self.db.add_history(trk, date=last)
 106         hist_records = self.db.fetch_history()
 107         self.assertEqual(hist, hist_records)
 108         self.db.purge_history(duration=0)
 109
 110     def test_history_to_tracks(self):
 111         tr = dict(**DEVOLT)
 112         tr.pop('file')
 113         trk01 = Track(file='01', **tr)
 114         self.db.add_history(trk01, CURRENT-datetime.timedelta(minutes=1))
 115         #
 116         tr.pop('musicbrainz_artistid')
 117         trk02 = Track(file='02', **tr)
 118         self.db.add_history(trk02, CURRENT-datetime.timedelta(minutes=2))
 119         #
 120         tr.pop('musicbrainz_albumid')
 121         trk03 = Track(file='03', **tr)
 122         self.db.add_history(trk03, CURRENT-datetime.timedelta(minutes=3))
 123         #
 124         tr.pop('musicbrainz_albumartistid')
 125         trk04 = Track(file='04', **tr)
 126         self.db.add_history(trk04, CURRENT-datetime.timedelta(minutes=4))
 127         #
 128         tr.pop('musicbrainz_trackid')
 129         trk05 = Track(file='05', **tr)
 130         self.db.add_history(trk05, CURRENT-datetime.timedelta(minutes=5))
 131         history = self.db.fetch_history()
 132         self.assertEqual(len(history), 5)
 133         # Controls history ordering, recent first
 134         self.assertEqual(history, [trk01, trk02, trk03, trk04, trk05])
 135
 136     def test_history_to_artists(self):
 137         tr = dict(**DEVOLT)
 138         tr.pop('file')
 139         tr.pop('musicbrainz_artistid')
 140         #
 141         trk01 = Track(file='01', **tr)
 142         self.db.add_history(trk01, CURRENT-datetime.timedelta(hours=1))
 143         #
 144         trk02 = Track(file='02', **tr)
 145         self.db.add_history(trk02, CURRENT-datetime.timedelta(hours=1))
 146         self.db.add_history(trk02, CURRENT-datetime.timedelta(hours=1))
 147         #
 148         trk03 = Track(file='03', **tr)
 149         self.db.add_history(trk03, CURRENT-datetime.timedelta(hours=1))
 150         # got multiple tracks, same artist/album, got artist/album history len = 1
 151         art_history = self.db.fetch_artists_history()
 152         alb_history = self.db.fetch_albums_history()
 153         self.assertEqual(len(art_history), 1)
 154         self.assertEqual(len(alb_history), 1)
 155         self.assertEqual(art_history, [trk01.Artist])
 156
 157         # Now add new artist to history
 158         trk04 = Track(file='04', artist='New Art')
 159         trk05 = Track(file='05', artist='New² Art')
 160         self.db.add_history(trk04, CURRENT-datetime.timedelta(minutes=3))
 161         self.db.add_history(trk03, CURRENT-datetime.timedelta(minutes=2))
 162         self.db.add_history(trk05, CURRENT-datetime.timedelta(minutes=1))
 163         art_history = self.db.fetch_artists_history()
 164         # Now we should have 4 artists in history
 165         self.assertEqual(len(art_history), 4)
 166         # Controling order, recent first
 167         self.assertEqual([trk05.artist, trk03.artist,
 168                          trk04.artist, trk03.artist],
 169                          art_history)
 170
 171     def test_04_filtering_history(self):
 172         # Controls artist history filtering
 173         for i in range(0, 5):
 174             trk = Track(file=f'/foo/bar.{i}', name=f'{i}-baz',
 175                         artist=f'{i}-art', album=f'{i}-lbum')
 176             last = CURRENT - datetime.timedelta(minutes=i)
 177             self.db.add_history(trk, date=last)
 178             if i == 5:  # bounce latest record
 179                 self.db.add_history(trk, date=last)
 180         art_history = self.db.fetch_artists_history()
 181         # Already checked but to be sure, we should have 5 artists in history
 182         self.assertEqual(len(art_history), 5)
 183         for needle in ['4-art', Artist(name='4-art')]:
 184             art_history = self.db.fetch_artists_history(needle=needle)
 185             self.assertEqual(art_history, [needle])
 186         needle = Artist(name='not-art')
 187         art_history = self.db.fetch_artists_history(needle=needle)
 188         self.assertEqual(art_history, [])
 189         # Controls artists history filtering
 190         #   for a list of Artist objects
 191         needle_list = [Artist(name='3-art'), Artist(name='4-art')]
 192         art_history = self.db.fetch_artists_history(needle=needle_list)
 193         self.assertEqual(art_history, needle_list)
 194         #   for a MetaContainer
 195         needle_meta = MetaContainer(needle_list)
 196         art_history = self.db.fetch_artists_history(needle=needle_meta)
 197         self.assertEqual(art_history, needle_list)
 198         #   for a list of string (Meta object handles Artist/str comparison)
 199         art_history = self.db.fetch_artists_history(['3-art', '4-art'])
 200         self.assertEqual(art_history, needle_list)
 201
 202         # Controls album history filtering
 203         needle = Artist(name='4-art')
 204         alb_history = self.db.fetch_albums_history(needle=needle)
 205         self.assertEqual(alb_history, [Album(name='4-lbum')])
 206
 207     def test_05_triggers(self):
 208         self.db.purge_history(duration=0)
 209         tracks_ids = list()
 210         #  Add a first track
 211         track = Track(file='/baz/bar.baz', name='baz', artist='fooart',
 212                       albumartist='not-same', album='not-same',)
 213         self.db.get_track(track)
 214         # Set 6 more records from same artist but not same album
 215         for i in range(1, 6):  # starts at 1 to ensure records are in the past
 216             trk = Track(file=f'/foo/{i}', name=f'{i}', artist='fooart',
 217                         albumartist='fooalbart', album='foolbum',)
 218             # Add track, save its DB id
 219             tracks_ids.append(self.db.get_track(trk))
 220             # set records in the past to ease purging then
 221             last = CURRENT - datetime.timedelta(minutes=i)
 222             self.db.add_history(trk, date=last)  # Add to history
 223         conn = self.db.get_database_connection()
 224         # for tid in tracks_ids:
 225         for tid in tracks_ids[:-1]:
 226             # Delete lastest record
 227             conn.execute('DELETE FROM history WHERE history.track = ?',
 228                          (tid,))
 229             c = conn.execute('SELECT albums.name FROM albums;')
 230             # There are still albums records (still a history using it)
 231             self.assertIn((trk.album,), c.fetchall())
 232         # purging last entry in history for album == trk.album
 233         conn.execute('DELETE FROM history WHERE history.track = ?',
 234                      (tracks_ids[-1],))
 235         # triggers purge other tables if possible
 236         conn.execute('SELECT albums.name FROM albums;')
 237         albums = c.fetchall()
 238         # No more "foolbum" in the table albums
 239         self.assertNotIn(('foolbum',), albums)
 240         # There is still "fooart" though
 241         c = conn.execute('SELECT artists.name FROM artists;')
 242         artists = c.fetchall()
 243         # No more "foolbum" in the table albums
 244         self.assertIn(('fooart',), artists)
 245         conn.close()
 246
 247
 248 class Test_01BlockList(Main):
 249
 250     def test_blocklist_addition(self):
 251         tracks_ids = list()
 252         # Set 6 records, same album
 253         for i in range(1, 6):  # starts at 1 to ensure records are in the past
 254             trk = Track(file=f'/foo/{i}', name=f'{i}', artist='fooart',
 255                         albumartist='fooalbart', album='foolbum',)
 256             # Add track, save its DB id
 257             tracks_ids.append(self.db.get_track(trk))
 258             # set records in the past to ease purging then
 259             last = CURRENT - datetime.timedelta(minutes=i)
 260             self.db.add_history(trk, date=last)  # Add to history
 261             if i == 1:
 262                 self.db.get_bl_track(trk)
 263             if i == 2:
 264                 self.db.get_bl_track(trk)
 265                 self.db.get_bl_album(Album(name=trk.album))
 266             if i == 3:
 267                 self.db.get_bl_artist(trk.Artist)
 268
 269     def test_blocklist_triggers_00(self):
 270         trk01 = Track(file='01', name='01', artist='artist A', album='album A')
 271         blart01_id = self.db.get_bl_artist(trk01.Artist)
 272         blalb01_id = self.db.get_bl_album(Album(name=trk01.album, mbid=trk01.musicbrainz_albumid))
 273         conn = self.db.get_database_connection()
 274         self.db._remove_blocklist_id(blart01_id, with_connection=conn)
 275         self.db._remove_blocklist_id(blalb01_id, with_connection=conn)
 276         albums = conn.execute('SELECT albums.name FROM albums;').fetchall()
 277         artists = conn.execute('SELECT artists.name FROM artists;').fetchall()
 278         conn.close()
 279         self.assertNotIn((trk01.album,), albums)
 280         self.assertNotIn((trk01.artist,), artists)
 281
 282     def test_blocklist_triggers_01(self):
 283         trk01 = Track(file='01', name='01', artist='artist A', album='album A')
 284         trk02 = Track(file='02', name='01', artist='artist A', album='album B')
 285         trk01_id = self.db.get_bl_track(trk01)
 286         trk02_id = self.db.get_bl_track(trk02)
 287         self.db.add_history(trk01, IN_THE_PAST)
 288         self.db._remove_blocklist_id(trk01_id)
 289         # bl trk01 removed:
 290         # albums/artists table not affected since trk01_id still in history
 291         conn = self.db.get_database_connection()
 292         albums = conn.execute('SELECT albums.name FROM albums;').fetchall()
 293         artists = conn.execute('SELECT artists.name FROM artists;').fetchall()
 294         self.assertIn(('album A',), albums)
 295         self.assertIn(('artist A',), artists)
 296         self.db.purge_history(0)
 297         # remove last reference to trk01
 298         albums = conn.execute('SELECT albums.name FROM albums;').fetchall()
 299         self.assertNotIn(('album A',), albums)
 300         self.assertIn(('artist A',), artists)
 301         # remove trk02
 302         self.db._remove_blocklist_id(trk02_id)
 303         albums = conn.execute('SELECT albums.name FROM albums;').fetchall()
 304         artists = conn.execute('SELECT artists.name FROM artists;').fetchall()
 305         self.assertNotIn(('album B',), albums)
 306         self.assertNotIn(('artist A'), artists)
 307         conn.close()
 308
 309
 310 class Test_02Genre(Main):
 311
 312     def test_genre(self):
 313         conn = self.db.get_database_connection()
 314         self.db.get_genre('Post-Rock', with_connection=conn)
 315         genres = list()
 316         for i in range(1, 15):  # starts at 1 to ensure records are in the past
 317             trk = Track(file=f'/foo/bar.{i}', name=f'{i}-baz',
 318                         album='foolbum', artist=f'{i}-art', genre=f'{i}')
 319             genres.append(f'{i}')
 320             last = CURRENT - datetime.timedelta(minutes=i)
 321             self.db.add_history(trk, date=last)
 322         genre_hist = self.db.fetch_genres_history(limit=10)
 323         self.assertEqual([g[0] for g in genre_hist], genres[:10])
 324
 325     def test_null_genres(self):
 326         conn = self.db.get_database_connection()
 327         genres = list()
 328         for i in range(1, 2):  # starts at 1 to ensure records are in the past
 329             trk = Track(file=f'/foo/bar.{i}', name=f'{i}-baz',
 330                         album='foolbum', artist=f'{i}-art')
 331             last = CURRENT - datetime.timedelta(minutes=i)
 332             self.db.add_history(trk, date=last)
 333         genre_hist = self.db.fetch_genres_history(limit=10)
 334         self.assertEqual(genre_hist, [])
 335
 336 # VIM MODLINE
 337 # vim: ai ts=4 sw=4 sts=4 expandtab fileencoding=utf8