X-Git-Url: https://git.kaliko.me/?a=blobdiff_plain;f=tests%2Ftest_simadb.py;fp=tests%2Ftest_simadb.py;h=b71fa1baca973c13e8ac2e2e6581d1a62c0e68f1;hb=9c754908e645834a53ce3906aac76a0aa92e98f4;hp=0000000000000000000000000000000000000000;hpb=9be53ff0c7f6dfeaa62bf0964f7d3a477c10499e;p=mpd-sima.git

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