tests/test_simadb.py

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