db: Give the database a 'loaded' property

This can be checked or connected to so other parts of the application
can easily know if all database tables have been loaded or not.

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
Anna Schumaker 2024-03-13 11:20:15 -04:00
parent 58a1df1d1d
commit a4e0968ef4
2 changed files with 26 additions and 6 deletions

View File

@ -26,6 +26,7 @@ class Connection(connection.Connection):
"""Connect to the database."""
active_playlist = GObject.Property(type=playlist.Playlist)
loaded = GObject.Property(type=bool, default=False)
def __init__(self):
"""Initialize a sqlite connection."""
@ -44,6 +45,12 @@ class Connection(connection.Connection):
self.tracks = tracks.Table(self)
def __check_loaded(self) -> None:
for tbl in list(self.playlist_tables()) + [self.tracks]:
if tbl.loaded is False:
return
self.loaded = True
def __check_version(self) -> None:
user_version = self("PRAGMA user_version").fetchone()["user_version"]
match user_version:
@ -99,3 +106,4 @@ class Connection(connection.Connection):
def table_loaded(self, tbl: table.Table) -> None:
"""Signal that a table has been loaded."""
tbl.loaded = True
self.__check_loaded()

View File

@ -72,22 +72,34 @@ class TestConnection(tests.util.TestCase):
def test_load(self):
"""Check that calling load() loads the tables."""
idle_tables = [tbl for tbl in self.sql.playlist_tables()] + \
[self.sql.tracks]
plist_tables = list(self.sql.playlist_tables())
all_tables = [self.sql.settings] + plist_tables + [self.sql.tracks]
table_loaded = unittest.mock.Mock()
self.sql.connect("table-loaded", table_loaded)
self.assertFalse(self.sql.loaded)
notify_loaded = unittest.mock.Mock()
self.sql.connect("notify::loaded", notify_loaded)
self.sql.load()
self.assertTrue(self.sql.settings.loaded)
for tbl in idle_tables:
notify_loaded.assert_not_called()
for tbl in all_tables[1:]:
self.assertFalse(tbl.loaded)
for tbl in idle_tables:
for tbl in plist_tables:
tbl.queue.complete()
self.assertTrue(tbl.loaded)
self.assertFalse(self.sql.loaded)
notify_loaded.assert_not_called()
calls = [unittest.mock.call(self.sql, tbl)
for tbl in [self.sql.settings] + idle_tables]
self.sql.tracks.queue.complete()
self.assertTrue(self.sql.tracks.loaded)
self.assertTrue(self.sql.loaded)
notify_loaded.assert_called()
calls = [unittest.mock.call(self.sql, tbl) for tbl in all_tables]
table_loaded.assert_has_calls(calls)
def test_filter(self):