126 lines
5.0 KiB
Python
126 lines
5.0 KiB
Python
# Copyright 2022 (c) Anna Schumaker
|
|
"""Test our custom db Connection object."""
|
|
import pathlib
|
|
import emmental.db
|
|
import tests.util
|
|
import unittest.mock
|
|
|
|
|
|
class TestConnection(tests.util.TestCase):
|
|
"""Test case for our database connection manager."""
|
|
|
|
def test_paths(self):
|
|
"""Check that path constants are pointing to the right places."""
|
|
dir = pathlib.Path(emmental.db.__file__).parent
|
|
self.assertEqual(emmental.db.SQL_V1_SCRIPT, dir / "emmental.sql")
|
|
self.assertEqual(emmental.db.SQL_V2_SCRIPT, dir / "upgrade-v2.sql")
|
|
|
|
def test_connection(self):
|
|
"""Check that the connection manager is initialized properly."""
|
|
self.assertIsInstance(self.sql, emmental.db.connection.Connection)
|
|
|
|
def test_version(self):
|
|
"""Test checking the database schema version."""
|
|
cur = self.sql("PRAGMA user_version")
|
|
self.assertEqual(cur.fetchone()["user_version"], 2)
|
|
|
|
def test_version_too_new(self):
|
|
"""Test failing when the database version is too new."""
|
|
self.sql._Connection__check_version()
|
|
|
|
self.sql("PRAGMA user_version = 3")
|
|
with self.assertRaises(Exception) as e:
|
|
self.sql._Connection__check_version()
|
|
self.assertEqual(str(e.exception), "Unsupported data version: 3")
|
|
|
|
def test_close(self):
|
|
"""Check closing the connection."""
|
|
self.sql.settings.queue.running = True
|
|
for tbl in self.sql.playlist_tables():
|
|
tbl.queue.running = True
|
|
|
|
self.sql.close()
|
|
self.assertFalse(self.sql.connected)
|
|
self.assertFalse(self.sql.settings.queue.running)
|
|
for tbl in self.sql.playlist_tables():
|
|
self.assertFalse(tbl.queue.running)
|
|
|
|
self.sql.close()
|
|
|
|
def test_tables(self):
|
|
"""Check that the connection has pointers to our tables."""
|
|
self.assertIsInstance(self.sql.settings, emmental.db.settings.Table)
|
|
self.assertIsInstance(self.sql.playlists, emmental.db.playlists.Table)
|
|
self.assertIsInstance(self.sql.artists, emmental.db.artists.Table)
|
|
self.assertIsInstance(self.sql.albums, emmental.db.albums.Table)
|
|
self.assertIsInstance(self.sql.media, emmental.db.media.Table)
|
|
self.assertIsInstance(self.sql.genres, emmental.db.genres.Table)
|
|
self.assertIsInstance(self.sql.decades, emmental.db.decades.Table)
|
|
self.assertIsInstance(self.sql.years, emmental.db.years.Table)
|
|
self.assertIsInstance(self.sql.libraries, emmental.db.libraries.Table)
|
|
self.assertIsInstance(self.sql.tracks, emmental.db.tracks.Table)
|
|
|
|
self.assertEqual(self.sql.albums.queue, self.sql.artists.queue)
|
|
self.assertEqual(self.sql.media.queue, self.sql.artists.queue)
|
|
self.assertEqual(self.sql.years.queue, self.sql.decades.queue)
|
|
|
|
self.assertListEqual([tbl for tbl in self.sql.playlist_tables()],
|
|
[self.sql.playlists, self.sql.artists,
|
|
self.sql.albums, self.sql.media,
|
|
self.sql.genres, self.sql.decades,
|
|
self.sql.years, self.sql.libraries])
|
|
|
|
def test_load(self):
|
|
"""Check that calling load() loads the tables."""
|
|
idle_tables = [tbl for tbl in self.sql.playlist_tables()] + \
|
|
[self.sql.tracks]
|
|
|
|
table_loaded = unittest.mock.Mock()
|
|
self.sql.connect("table-loaded", table_loaded)
|
|
|
|
self.sql.load()
|
|
self.assertTrue(self.sql.settings.loaded)
|
|
for tbl in idle_tables:
|
|
self.assertFalse(tbl.loaded)
|
|
for tbl in idle_tables:
|
|
tbl.queue.complete()
|
|
self.assertTrue(tbl.loaded)
|
|
|
|
calls = [unittest.mock.call(self.sql, tbl)
|
|
for tbl in [self.sql.settings] + idle_tables]
|
|
table_loaded.assert_has_calls(calls)
|
|
|
|
def test_filter(self):
|
|
"""Check filtering the playlist tables."""
|
|
for tbl in self.sql.playlist_tables():
|
|
tbl.filter = unittest.mock.Mock()
|
|
|
|
self.sql.filter("*glob*")
|
|
for tbl in self.sql.playlist_tables():
|
|
tbl.filter.assert_called_with("*glob*")
|
|
|
|
def test_set_active_playlist(self):
|
|
"""Check setting the active playlist."""
|
|
table = tests.util.playlist.MockTable(self.sql)
|
|
plist1 = table.create(name="Playlist 1")
|
|
plist2 = table.create(name="Playlist 2")
|
|
self.assertIsNone(self.sql.active_playlist)
|
|
|
|
self.sql.set_active_playlist(plist1)
|
|
self.assertEqual(self.sql.active_playlist, plist1)
|
|
self.assertTrue(plist1.active)
|
|
|
|
self.sql.set_active_playlist(plist2)
|
|
self.assertEqual(self.sql.active_playlist, plist2)
|
|
self.assertFalse(plist1.active)
|
|
self.assertTrue(plist2.active)
|
|
|
|
notify = unittest.mock.Mock()
|
|
self.sql.connect("notify::active-playlist", notify)
|
|
self.sql.set_active_playlist(plist2)
|
|
notify.assert_not_called()
|
|
|
|
self.sql.set_active_playlist(None)
|
|
self.assertIsNone(self.sql.active_playlist)
|
|
self.assertFalse(plist2.active)
|