diff --git a/emmental/db/__init__.py b/emmental/db/__init__.py index 95cb68d..3aa2cac 100644 --- a/emmental/db/__init__.py +++ b/emmental/db/__init__.py @@ -8,6 +8,7 @@ from . import artists from . import connection from . import decades from . import genres +from . import libraries from . import playlist from . import media from . import playlists @@ -41,6 +42,7 @@ class Connection(connection.Connection): self.genres = genres.Table(self) self.decades = decades.Table(self) self.years = years.Table(self, queue=self.decades.queue) + self.libraries = libraries.Table(self) def close(self) -> None: """Close the database connection.""" @@ -64,7 +66,7 @@ class Connection(connection.Connection): def playlist_tables(self) -> Generator[playlist.Table, None, None]: """Iterate over each playlist table.""" for tbl in [self.playlists, self.artists, self.albums, self.media, - self.genres, self.decades, self.years]: + self.genres, self.decades, self.years, self.libraries]: yield tbl def set_active_playlist(self, plist: playlist.Playlist) -> None: diff --git a/emmental/db/emmental.sql b/emmental/db/emmental.sql index 9fafea8..f48c95b 100644 --- a/emmental/db/emmental.sql +++ b/emmental/db/emmental.sql @@ -305,6 +305,40 @@ CREATE TRIGGER years_delete_trigger AFTER DELETE ON years END; +/******************************* + * * + * Library Paths * + * * + *******************************/ + +CREATE TABLE libraries ( + libraryid INTEGER PRIMARY KEY, + propertyid INTEGER REFERENCES playlist_properties (propertyid) + ON DELETE CASCADE + ON UPDATE CASCADE, + path PATH UNIQUE, + enabled BOOLEAN DEFAULT TRUE, + deleting BOOLEAN DEFAULT FALSE +); + +CREATE VIEW libraries_view AS + SELECT libraryid, propertyid, path, path as name, enabled, active + FROM libraries + JOIN playlist_properties USING (propertyid); + +CREATE TRIGGER libraries_insert_trigger AFTER INSERT ON libraries + BEGIN + INSERT INTO playlist_properties (active) VALUES (False); + UPDATE libraries SET propertyid = last_insert_rowid() + WHERE libraryid = NEW.libraryid; + END; + +CREATE TRIGGER libraries_delete_trigger AFTER DELETE ON libraries + BEGIN + DELETE FROM playlist_properties WHERE propertyid = OLD.propertyid; + END; + + /****************************************** * * * Create Default Playlists * diff --git a/emmental/db/libraries.py b/emmental/db/libraries.py new file mode 100644 index 0000000..f9bfb7a --- /dev/null +++ b/emmental/db/libraries.py @@ -0,0 +1,93 @@ +# Copyright 2022 (c) Anna Schumaker +"""A custom Gio.ListModel for working with libraries.""" +import pathlib +import sqlite3 +from gi.repository import GObject +from . import idle +from . import playlist + + +class Library(playlist.Playlist): + """Our custom Library with path and enabled properties.""" + + libraryid = GObject.Property(type=int) + path = GObject.Property(type=GObject.TYPE_PYOBJECT) + enabled = GObject.Property(type=bool, default=True) + deleting = GObject.Property(type=bool, default=False) + + queue = GObject.Property(type=idle.Queue) + online = GObject.Property(type=bool, default=False) + + def __init__(self, **kwargs): + """Initialize our Library object.""" + super().__init__(queue=idle.Queue(), **kwargs) + + def __queue_delete(self) -> bool: + self.table.delete(self) + return True + + def do_update(self, column: str) -> bool: + """Update a Library playlist.""" + match column: + case "online": self.table.notify_online(self) + case _: return super().do_update(column) + return True + + def delete(self) -> bool: + """Delete this Library.""" + if self.deleting is False: + self.deleting = True + self.queue.push(self.__queue_delete) + return True + return False + + @property + def primary_key(self) -> int: + """Get this library's primary key.""" + return self.libraryid + + +class Table(playlist.Table): + """Our Library ListModel.""" + + def do_construct(self, **kwargs) -> Library: + """Construct a new library.""" + return Library(**kwargs) + + def do_sql_delete(self, library: Library) -> sqlite3.Cursor: + """Delete a library.""" + return self.sql("DELETE FROM libraries WHERE libraryid=?", + library.libraryid) + + def do_sql_insert(self, path: pathlib.Path) -> sqlite3.Cursor: + """Create a new library.""" + if cur := self.sql("INSERT INTO libraries (path) VALUES (?)", path): + return self.sql("SELECT * FROM libraries_view WHERE libraryid=?", + cur.lastrowid) + + def do_sql_glob(self, glob: str) -> sqlite3.Cursor: + """Search for libraries matching the search text.""" + return self.sql("""SELECT libraryid FROM libraries_view + WHERE name GLOB ?""", glob) + + def do_sql_select_all(self) -> sqlite3.Cursor: + """Load libraries from the database.""" + return self.sql("SELECT * FROM libraries_view") + + def do_sql_select_one(self, path: pathlib.Path) -> sqlite3.Cursor: + """Look up a library by path.""" + return self.sql("SELECT libraryid FROM libraries WHERE path=?", path) + + def do_sql_update(self, library: Library, column: str, newval) -> bool: + """Update a Library playlist.""" + return self.sql(f"UPDATE libraries SET {column}=? WHERE rowid=?", + newval, library.libraryid) + + def notify_online(self, library: Library) -> None: + """Notify that a library's online status has changed.""" + if not library.online or self.loaded: + self.emit("library-online", library) + + @GObject.Signal(arg_types=(Library,)) + def library_online(self, library: Library) -> None: + """Signal that a library online status has changed.""" diff --git a/tests/db/test_db.py b/tests/db/test_db.py index 5b702c3..13410f5 100644 --- a/tests/db/test_db.py +++ b/tests/db/test_db.py @@ -47,6 +47,7 @@ class TestConnection(tests.util.TestCase): 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.assertEqual(self.sql.albums.queue, self.sql.artists.queue) self.assertEqual(self.sql.media.queue, self.sql.artists.queue) @@ -56,7 +57,7 @@ class TestConnection(tests.util.TestCase): [self.sql.playlists, self.sql.artists, self.sql.albums, self.sql.media, self.sql.genres, self.sql.decades, - self.sql.years]) + self.sql.years, self.sql.libraries]) def test_load(self): """Check that calling load() loads the tables.""" diff --git a/tests/db/test_libraries.py b/tests/db/test_libraries.py new file mode 100644 index 0000000..0e36b29 --- /dev/null +++ b/tests/db/test_libraries.py @@ -0,0 +1,186 @@ +# Copyright 2022 (c) Anna Schumaker +"""Tests our library Gio.ListModel.""" +import pathlib +import emmental.db +import tests.util +import unittest.mock + + +class TestLibraryObject(tests.util.TestCase): + """Tests our library object.""" + + def setUp(self): + """Set up common variables.""" + super().setUp() + self.table = self.sql.libraries + self.path = pathlib.Path("/a/b/c") + self.library = emmental.db.libraries.Library(table=self.table, + libraryid=123, + propertyid=456, + path=self.path, + name=str(self.path)) + + def test_init(self): + """Test that the Library is set up properly.""" + self.assertIsInstance(self.library, emmental.db.playlist.Playlist) + self.assertIsInstance(self.library.queue, emmental.db.idle.Queue) + + self.assertEqual(self.library.table, self.table) + self.assertEqual(self.library.propertyid, 456) + self.assertEqual(self.library.libraryid, 123) + self.assertEqual(self.library.primary_key, 123) + self.assertEqual(self.library.path, self.path) + self.assertTrue(self.library.enabled) + self.assertFalse(self.library.deleting) + self.assertFalse(self.library.online) + self.assertIsNone(self.library.parent) + + def test_delete(self): + """Test deleting a Library path.""" + with unittest.mock.patch.object(self.table, "delete") as mock_delete: + with unittest.mock.patch.object(self.table, "update"): + self.assertTrue(self.library.delete()) + self.assertTrue(self.library.deleting) + mock_delete.assert_not_called() + self.assertEqual(self.library.queue[0], + (self.library._Library__queue_delete,)) + + self.assertFalse(self.library.delete()) + self.assertEqual(self.library.queue.total, 1) + + self.library.queue.complete() + mock_delete.assert_called_with(self.library) + + def test_online(self): + """Test that changing the online property notifies the table.""" + with unittest.mock.patch.object(self.table, + "notify_online") as mock_notify: + self.library.online = True + mock_notify.assert_called_with(self.library) + + +class TestLibraryTable(tests.util.TestCase): + """Tests our library table.""" + + def setUp(self): + """Set up common variables.""" + tests.util.TestCase.setUp(self) + self.table = self.sql.libraries + + def test_init(self): + """Test that the library model is configured correctly.""" + self.assertIsInstance(self.table, emmental.db.playlist.Table) + self.assertEqual(len(self.table), 0) + + def test_construct(self): + """Test constructing a new library.""" + library = self.table.construct(propertyid=1, libraryid=1, + path=pathlib.Path("/a/b/c"), + name="/a/b/c") + self.assertIsInstance(library, emmental.db.libraries.Library) + self.assertEqual(library.table, self.table) + self.assertEqual(library.propertyid, 1) + self.assertEqual(library.libraryid, 1) + self.assertEqual(library.path, pathlib.Path("/a/b/c")) + self.assertEqual(library.name, "/a/b/c") + self.assertFalse(library.active) + + def test_create(self): + """Test creating a library.""" + library = self.table.create(pathlib.Path("/a/b/c")) + self.assertIsInstance(library, emmental.db.libraries.Library) + self.assertEqual(library.path, pathlib.Path("/a/b/c")) + + cur = self.sql("SELECT COUNT(path) FROM libraries") + self.assertEqual(cur.fetchone()["COUNT(path)"], 1) + self.assertEqual(len(self.table.store), 1) + self.assertEqual(self.table.store.get_item(0), library) + + cur = self.sql("""SELECT COUNT(*) FROM playlist_properties + WHERE propertyid=?""", library.propertyid) + self.assertEqual(cur.fetchone()["COUNT(*)"], 1) + + self.assertIsNone(self.table.create("/a/b/c")) + + def test_delete(self): + """Test deleting a library.""" + library = self.table.create(pathlib.Path("/a/b/c")) + self.assertTrue(self.table.delete(library)) + self.assertIsNone(self.table.index(library)) + + cur = self.sql("SELECT COUNT(path) FROM libraries") + self.assertEqual(cur.fetchone()["COUNT(path)"], 0) + self.assertEqual(len(self.table), 0) + self.assertIsNone(self.table.get_item(0)) + + cur = self.sql("""SELECT COUNT(*) FROM playlist_properties + WHERE propertyid=?""", library.propertyid) + self.assertEqual(cur.fetchone()["COUNT(*)"], 0) + + self.assertFalse(self.table.delete(library)) + + def test_filter(self): + """Test filtering the library model.""" + self.table.create(pathlib.Path("/a/b/c")) + self.table.create(pathlib.Path("/a/b/d")) + + self.table.filter("*c", now=True) + self.assertSetEqual(self.table.get_filter().keys, {1}) + self.table.filter("*a/b*", now=True) + self.assertSetEqual(self.table.get_filter().keys, {1, 2}) + + def test_load(self): + """Test loading libraries from the database.""" + self.table.create("/a/b/c") + self.table.create("/a/b/d").enabled = False + + libraries2 = emmental.db.libraries.Table(self.sql) + self.assertEqual(len(libraries2), 0) + + libraries2.load(now=True) + self.assertEqual(len(libraries2), 2) + + self.assertEqual(libraries2.get_item(0).libraryid, 1) + self.assertEqual(libraries2.get_item(0).path, pathlib.Path("/a/b/c")) + self.assertTrue(libraries2.get_item(0).enabled) + + self.assertEqual(libraries2.get_item(1).libraryid, 2) + self.assertEqual(libraries2.get_item(1).path, pathlib.Path("/a/b/d")) + self.assertFalse(libraries2.get_item(1).enabled) + + def test_lookup(self): + """Test looking up a library.""" + library = self.table.create(pathlib.Path("/a/b/c")) + self.assertEqual(self.table.lookup(pathlib.Path("/a/b/c/")), library) + self.assertIsNone(self.table.lookup(pathlib.Path("/no/library/path"))) + + def test_update(self): + """Test updating genre attributes.""" + library = self.table.create("/a/b/c") + library.active = True + library.enabled = False + + row = self.sql("""SELECT active, enabled FROM libraries_view + WHERE libraryid=?""", library.libraryid).fetchone() + self.assertTrue(row["active"]) + self.assertFalse(row["enabled"]) + + def test_library_online(self): + """Test the library-online signal.""" + library = self.table.create(pathlib.Path("/a/b/c")) + callback = unittest.mock.Mock() + self.table.connect("library-online", callback) + + library.online = True + callback.assert_not_called() + library.online = False + callback.assert_called_with(self.table, library) + + callback.reset_mock() + self.table.loaded = True + library.online = True + callback.assert_called_with(self.table, library) + + callback.reset_mock() + library.online = False + callback.assert_called_with(self.table, library)