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
  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 = False  # 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, got artist history len == 1
 149         art_history = self.db.fetch_artists_history()
 150         self.assertEqual(len(art_history), 1)
 151         self.assertEqual(art_history, [trk01.Artist])
 152
 153         # Now add new artist to history
 154         trk04 = Track(file='04', artist='New Art')
 155         trk05 = Track(file='05', artist='New² Art')
 156         self.db.add_history(trk04, CURRENT-datetime.timedelta(minutes=3))
 157         self.db.add_history(trk03, CURRENT-datetime.timedelta(minutes=2))
 158         self.db.add_history(trk05, CURRENT-datetime.timedelta(minutes=1))
 159         art_history = self.db.fetch_artists_history()
 160         # Now we should have 4 artists in history
 161         self.assertEqual(len(art_history), 4)
 162         # Controling order, recent first
 163         self.assertEqual([trk05.artist, trk03.artist,
 164                          trk04.artist, trk03.artist],
 165                          art_history)
 166
 167     def test_04_triggers(self):
 168         self.db.purge_history(duration=0)
 169         tracks_ids = list()
 170         #  Add a first track
 171         track = Track(file='/baz/bar.baz', name='baz', artist='fooart',
 172                       albumartist='not-same', album='not-same',)
 173         self.db.get_track(track)
 174         # Set 6 more records from same artist but not same album
 175         for i in range(1, 6):  # starts at 1 to ensure records are in the past
 176             trk = Track(file=f'/foo/{i}', name=f'{i}', artist='fooart',
 177                         albumartist='fooalbart', album='foolbum',)
 178             # Add track, save its DB id
 179             tracks_ids.append(self.db.get_track(trk))
 180             # set records in the past to ease purging then
 181             last = CURRENT - datetime.timedelta(minutes=i)
 182             self.db.add_history(trk, date=last)  # Add to history
 183         conn = self.db.get_database_connection()
 184         # for tid in tracks_ids:
 185         for tid in tracks_ids[:-1]:
 186             # Delete lastest record
 187             conn.execute('DELETE FROM history WHERE history.track = ?',
 188                          (tid,))
 189             c = conn.execute('SELECT albums.name FROM albums;')
 190             # There are still albums records (still a history using it)
 191             self.assertIn((trk.album,), c.fetchall())
 192         # purging last entry in history for album == trk.album
 193         conn.execute('DELETE FROM history WHERE history.track = ?',
 194                      (tracks_ids[-1],))
 195         # triggers purge other tables if possible
 196         conn.execute('SELECT albums.name FROM albums;')
 197         albums = c.fetchall()
 198         # No more "foolbum" in the table albums
 199         self.assertNotIn(('foolbum',), albums)
 200         # There is still "fooart" though
 201         c = conn.execute('SELECT artists.name FROM artists;')
 202         artists = c.fetchall()
 203         # No more "foolbum" in the table albums
 204         self.assertIn(('fooart',), artists)
 205         conn.close()
 206
 207
 208 class Test_01BlockList(Main):
 209
 210     def test_blocklist_addition(self):
 211         tracks_ids = list()
 212         # Set 6 records, 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             if i == 1:
 222                 self.db.get_bl_track(trk)
 223             if i == 2:
 224                 self.db.get_bl_track(trk)
 225                 self.db.get_bl_album(Album(name=trk.album))
 226             if i == 3:
 227                 self.db.get_bl_artist(trk.Artist)
 228
 229     def test_blocklist_triggers(self):
 230         trk01 = Track(file='01', name='01', artist='artist A', album='album A')
 231         trk02 = Track(file='02', name='01', artist='artist A', album='album B')
 232         trk01_id = self.db.get_bl_track(trk01)
 233         trk02_id = self.db.get_bl_track(trk02)
 234         self.db.add_history(trk01, IN_THE_PAST)
 235         self.db._remove_blocklist_id(trk01_id)
 236         # bl trk01 removed:
 237         # albums/artists table not affected since trk01_id still in history
 238         conn = self.db.get_database_connection()
 239         albums = conn.execute('SELECT albums.name FROM albums;').fetchall()
 240         artists = conn.execute('SELECT artists.name FROM artists;').fetchall()
 241         self.assertIn(('album A',), albums)
 242         self.assertIn(('artist A',), artists)
 243         self.db.purge_history(0)
 244         # remove last reference to trk01
 245         albums = conn.execute('SELECT albums.name FROM albums;').fetchall()
 246         self.assertNotIn(('album A',), albums)
 247         self.assertIn(('artist A',), artists)
 248         # remove trk02
 249         self.db._remove_blocklist_id(trk02_id)
 250         albums = conn.execute('SELECT albums.name FROM albums;').fetchall()
 251         artists = conn.execute('SELECT artists.name FROM artists;').fetchall()
 252         self.assertNotIn(('album B',), albums)
 253         self.assertNotIn(('artist A'), artists)
 254         conn.close()
 255
 256
 257 # VIM MODLINE
 258 # vim: ai ts=4 sw=4 sts=4 expandtab fileencoding=utf8