diff --git a/app/dbconfig.py b/app/dbconfig.py index 64f94dd..be78662 100644 --- a/app/dbconfig.py +++ b/app/dbconfig.py @@ -43,7 +43,7 @@ def Session() -> Generator[scoped_session, None, None]: file = frame.filename function = frame.function lineno = frame.lineno - Session = scoped_session(sessionmaker(bind=engine, future=True)) + Session = scoped_session(sessionmaker(bind=engine)) log.debug(f"SqlA: session acquired [{hex(id(Session))}]") log.debug( f"Session acquisition: {file}:{function}:{lineno} " f"[{hex(id(Session))}]" diff --git a/app/models.py b/app/models.py index dbfe401..a261b83 100644 --- a/app/models.py +++ b/app/models.py @@ -97,6 +97,40 @@ class NoteColours(Base): f"colour={self.colour}>" ) + def __init__( + self, + session: scoped_session, + substring: str, + colour: str, + enabled: bool = True, + is_regex: bool = False, + is_casesensitive: bool = False, + order: Optional[int] = 0, + ) -> None: + self.substring = substring + self.colour = colour + self.enabled = enabled + self.is_regex = is_regex + self.is_casesensitive = is_casesensitive + self.order = order + + session.add(self) + session.flush() + + @classmethod + def get_all(cls, session: scoped_session) -> Sequence["NoteColours"]: + """ + Return all records + """ + + return ( + session.execute( + select(cls) + ) + .scalars() + .all() + ) + @staticmethod def get_colour(session: scoped_session, text: str) -> Optional[str]: """ @@ -442,7 +476,7 @@ class PlaylistRows(Base): @classmethod def deep_row( cls, session: scoped_session, playlist_id: int, row_number: int - ) ->"PlaylistRows": + ) -> "PlaylistRows": """ Return a playlist row that includes full track and lastplayed data for given playlist_id and row @@ -453,7 +487,7 @@ class PlaylistRows(Base): .options(joinedload(cls.track)) .where( PlaylistRows.playlist_id == playlist_id, - PlaylistRows.plr_rownum == row_number + PlaylistRows.plr_rownum == row_number, ) # .options(joinedload(Tracks.playdates)) ) @@ -731,7 +765,7 @@ class Tracks(Base): fade_at: int, silence_at: int, mtime: int, - bitrate: int + bitrate: int, ): self.path = path self.title = title @@ -764,9 +798,11 @@ class Tracks(Base): """ try: - return session.execute( - select(Tracks).where(Tracks.path == path) - ).unique().scalar_one() + return ( + session.execute(select(Tracks).where(Tracks.path == path)) + .unique() + .scalar_one() + ) except NoResultFound: return None diff --git a/conftest.py b/conftest.py index e6c0b36..92b1374 100644 --- a/conftest.py +++ b/conftest.py @@ -1,41 +1,47 @@ # https://itnext.io/setting-up-transactional-tests-with-pytest-and-sqlalchemy-b2d726347629 import pytest - -# Flake8 doesn't like the sys.append within imports -# import sys -# sys.path.append("app") +import helpers from sqlalchemy import create_engine from sqlalchemy.orm import scoped_session, sessionmaker +from app.models import Base, Tracks + @pytest.fixture(scope="session") -def connection(): +def db_engine(): engine = create_engine( - "mysql+mysqldb://musicmuster_testing:musicmuster_testing@" - "localhost/musicmuster_testing" + "mysql+mysqldb://musicmuster_testing:musicmuster_testing@localhost/dev_musicmuster_testing" ) - return engine.connect() + Base.metadata.create_all(engine) + yield engine + engine.dispose() -@pytest.fixture(scope="session") -def setup_database(connection): - from app.models import Base # noqa E402 - - Base.metadata.bind = connection - Base.metadata.create_all() - # seed_database() - - yield - - Base.metadata.drop_all() - - -@pytest.fixture -def session(setup_database, connection): +@pytest.fixture(scope='function') +def session(db_engine): + connection = db_engine.connect() transaction = connection.begin() - yield scoped_session( - sessionmaker(autocommit=False, autoflush=False, bind=connection) - ) + Session = sessionmaker(bind=connection) + session = scoped_session(Session) + yield session + session.remove() transaction.rollback() + connection.close() + + +@pytest.fixture(scope='function') +def track1(session): + track_path = "testdata/isa.mp3" + metadata = helpers.get_file_metadata(track_path) + track = Tracks(session, **metadata) + return track + + +@pytest.fixture(scope='function') +def track2(session): + track_path = "testdata/mom.mp3" + metadata = helpers.get_file_metadata(track_path) + track = Tracks(session, **metadata) + return track diff --git a/pyproject.toml b/pyproject.toml index e50a727..0927b3b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,7 +53,7 @@ build-backend = "poetry.core.masonry.api" mypy_path = "/home/kae/git/musicmuster/app" [tool.pytest.ini_options] -addopt = "--exitfirst --showlocals --capture=no" +addopts = "--exitfirst --showlocals --capture=no" pythonpath = [".", "app"] [tool.vulture] diff --git a/test_models.py b/test_models.py index 9cd3ce2..d15f519 100644 --- a/test_models.py +++ b/test_models.py @@ -1,8 +1,9 @@ import os.path +import helpers + from app.models import ( NoteColours, - Notes, Playdates, Playlists, Tracks, @@ -12,6 +13,7 @@ from app.models import ( def test_notecolours_get_colour(session): """Create a colour record and retrieve all colours""" + print(">>>text_notcolours_get_colour") note_colour = "#0bcdef" NoteColours(session, substring="substring", colour=note_colour) @@ -24,6 +26,7 @@ def test_notecolours_get_colour(session): def test_notecolours_get_all(session): """Create two colour records and retrieve them all""" + print(">>>text_notcolours_get_all") note1_colour = "#1bcdef" note2_colour = "#20ff00" NoteColours(session, substring="note1", colour=note1_colour) @@ -52,172 +55,84 @@ def test_notecolours_get_colour_match(session): assert result == note_colour -def test_notes_creation(session): - # We need a playlist - playlist = Playlists(session, "my playlist") - - note_text = "note text" - note = Notes(session, playlist.id, 0, note_text) - assert note - - notes = session.query(Notes).all() - assert len(notes) == 1 - assert notes[0].note == note_text - - -def test_notes_delete(session): - # We need a playlist - playlist = Playlists(session, "my playlist") - - note_text = "note text" - note = Notes(session, playlist.id, 0, note_text) - assert note - - notes = session.query(Notes).all() - assert len(notes) == 1 - assert notes[0].note == note_text - - note.delete_note(session) - notes = session.query(Notes).all() - assert len(notes) == 0 - - -def test_notes_update_row_only(session): - # We need a playlist - playlist = Playlists(session, "my playlist") - - note_text = "note text" - note = Notes(session, playlist.id, 0, note_text) - new_row = 10 - - note.update_note(session, new_row) - - notes = session.query(Notes).all() - assert len(notes) == 1 - assert notes[0].row == new_row - - -def test_notes_update_text(session): - # We need a playlist - playlist = Playlists(session, "my playlist") - - note_text = "note text" - note = Notes(session, playlist.id, 0, note_text) - new_text = "This is new" - new_row = 0 - - note.update_note(session, new_row, new_text) - notes = session.query(Notes).all() - - assert len(notes) == 1 - assert notes[0].note == new_text - assert notes[0].row == new_row - - -def test_playdates_add_playdate(session): +def test_playdates_add_playdate(session, track1): """Test playdate and last_played retrieval""" - # We need a track - track_path = "/a/b/c" - track = Tracks(session, track_path) - - playdate = Playdates(session, track.id) + playdate = Playdates(session, track1.id) assert playdate - last_played = Playdates.last_played(session, track.id) + last_played = Playdates.last_played(session, track1.id) assert abs((playdate.lastplayed - last_played).total_seconds()) < 2 -def test_playdates_remove_track(session): - """Test removing a track from a playdate""" - - # We need a track - track_path = "/a/b/c" - track = Tracks(session, track_path) - - Playdates.remove_track(session, track.id) - - last_played = Playdates.last_played(session, track.id) - assert last_played is None - - def test_playlist_create(session): playlist = Playlists(session, "my playlist") assert playlist -def test_playlist_add_note(session): - note_text = "my note" +# def test_playlist_add_track(session, track): +# # We need a playlist +# playlist = Playlists(session, "my playlist") - playlist = Playlists(session, "my playlist") +# row = 17 - assert len(playlist.notes) == 1 - playlist_note = playlist.notes[0] - assert playlist_note.note == note_text +# playlist.add_track(session, track.id, row) + +# assert len(playlist.tracks) == 1 +# playlist_track = playlist.tracks[row] +# assert playlist_track.path == track_path -def test_playlist_add_track(session): - # We need a playlist - playlist = Playlists(session, "my playlist") +# def test_playlist_tracks(session): +# # We need a playlist +# playlist = Playlists(session, "my playlist") - # We need a track - track_path = "/a/b/c" - track = Tracks(session, track_path) +# # We need two tracks +# track1_path = "/a/b/c" +# track1_row = 17 +# track1 = Tracks(session, track1_path) - row = 17 +# track2_path = "/x/y/z" +# track2_row = 29 +# track2 = Tracks(session, track2_path) - playlist.add_track(session, track.id, row) +# playlist.add_track(session, track1.id, track1_row) +# playlist.add_track(session, track2.id, track2_row) - assert len(playlist.tracks) == 1 - playlist_track = playlist.tracks[row] - assert playlist_track.path == track_path +# tracks = playlist.tracks +# assert tracks[track1_row] == track1 +# assert tracks[track2_row] == track2 -def test_playlist_tracks(session): - # We need a playlist - playlist = Playlists(session, "my playlist") +# def test_playlist_notes(session): +# # We need a playlist +# playlist = Playlists(session, "my playlist") - # We need two tracks - track1_path = "/a/b/c" - track1_row = 17 - track1 = Tracks(session, track1_path) +# # We need two notes +# note1_text = "note1 text" +# note1_row = 11 +# _ = Notes(session, playlist.id, note1_row, note1_text) - track2_path = "/x/y/z" - track2_row = 29 - track2 = Tracks(session, track2_path) +# note2_text = "note2 text" +# note2_row = 19 +# _ = Notes(session, playlist.id, note2_row, note2_text) - playlist.add_track(session, track1.id, track1_row) - playlist.add_track(session, track2.id, track2_row) - - tracks = playlist.tracks - assert tracks[track1_row] == track1 - assert tracks[track2_row] == track2 - - -def test_playlist_notes(session): - # We need a playlist - playlist = Playlists(session, "my playlist") - - # We need two notes - note1_text = "note1 text" - note1_row = 11 - _ = Notes(session, playlist.id, note1_row, note1_text) - - note2_text = "note2 text" - note2_row = 19 - _ = Notes(session, playlist.id, note2_row, note2_text) - - notes = playlist.notes - assert note1_text in [n.note for n in notes] - assert note1_row in [n.row for n in notes] - assert note2_text in [n.note for n in notes] - assert note2_row in [n.row for n in notes] +# notes = playlist.notes +# assert note1_text in [n.note for n in notes] +# assert note1_row in [n.row for n in notes] +# assert note2_text in [n.note for n in notes] +# assert note2_row in [n.row for n in notes] def test_playlist_open_and_close(session): # We need a playlist playlist = Playlists(session, "my playlist") + assert len(Playlists.get_open(session)) == 0 + assert len(Playlists.get_closed(session)) == 1 + + playlist.mark_open(session, tab_index=0) + assert len(Playlists.get_open(session)) == 1 assert len(Playlists.get_closed(session)) == 0 @@ -226,11 +141,6 @@ def test_playlist_open_and_close(session): assert len(Playlists.get_open(session)) == 0 assert len(Playlists.get_closed(session)) == 1 - playlist.mark_open(session) - - assert len(Playlists.get_open(session)) == 1 - assert len(Playlists.get_closed(session)) == 0 - def test_playlist_get_all_and_by_id(session): # We need two playlists @@ -243,250 +153,34 @@ def test_playlist_get_all_and_by_id(session): assert len(all_playlists) == 2 assert p1_name in [p.name for p in all_playlists] assert p2_name in [p.name for p in all_playlists] - assert Playlists.get_by_id(session, playlist1.id).name == p1_name + assert session.get(Playlists, playlist1.id).name == p1_name -def test_playlist_remove_tracks(session): - # Need two playlists and three tracks - p1_name = "playlist one" - playlist1 = Playlists(session, p1_name) - p2_name = "playlist two" - playlist2 = Playlists(session, p2_name) - - track1_path = "/a/b/c" - track1 = Tracks(session, track1_path) - track2_path = "/m/n/o" - track2 = Tracks(session, track2_path) - track3_path = "/x/y/z" - track3 = Tracks(session, track3_path) - - # Add all tracks to both playlists - for p in [playlist1, playlist2]: - for t in [track1, track2, track3]: - p.add_track(session, t.id) - - assert len(playlist1.tracks) == 3 - assert len(playlist2.tracks) == 3 - - playlist1.remove_track(session, 1) - assert len(playlist1.tracks) == 2 - # Check the track itself still exists - original_track = Tracks.get_by_id(session, track1.id) - assert original_track - - playlist1.remove_all_tracks(session) - assert len(playlist1.tracks) == 0 - assert len(playlist2.tracks) == 3 - - -def test_playlist_get_track_playlists(session): - # Need two playlists and two tracks - p1_name = "playlist one" - playlist1 = Playlists(session, p1_name) - p2_name = "playlist two" - playlist2 = Playlists(session, p2_name) - - track1_path = "/a/b/c" - track1 = Tracks(session, track1_path) - track2_path = "/m/n/o" - track2 = Tracks(session, track2_path) - - # Put track1 in both playlists, track2 only in playlist1 - playlist1.add_track(session, track1.id) - playlist2.add_track(session, track1.id) - playlist1.add_track(session, track2.id) - - playlists_track1 = track1.playlists - playlists_track2 = track2.playlists - assert p1_name in [a.playlist.name for a in playlists_track1] - assert p2_name in [a.playlist.name for a in playlists_track1] - assert p1_name in [a.playlist.name for a in playlists_track2] - assert p2_name not in [a.playlist.name for a in playlists_track2] - - -def test_playlist_move_track(session): - # We need two playlists - p1_name = "playlist one" - p2_name = "playlist two" - playlist1 = Playlists(session, p1_name) - playlist2 = Playlists(session, p2_name) - +def test_tracks_get_all_tracks(session, track1, track2): # Need two tracks - track1_row = 17 - track1_path = "/a/b/c" - track1 = Tracks(session, track1_path) - track2_row = 29 - track2_path = "/m/n/o" - track2 = Tracks(session, track2_path) - # Add both to playlist1 and check - playlist1.add_track(session, track1.id, track1_row) - playlist1.add_track(session, track2.id, track2_row) - - tracks = playlist1.tracks - assert tracks[track1_row] == track1 - assert tracks[track2_row] == track2 - - # Move track2 to playlist2 and check - playlist1.move_track(session, [track2_row], playlist2) - - tracks1 = playlist1.tracks - tracks2 = playlist2.tracks - assert len(tracks1) == 1 - assert len(tracks2) == 1 - assert tracks1[track1_row] == track1 - assert tracks2[0] == track2 + result = [a.path for a in Tracks.get_all(session)] + assert track1.path in result + assert track2.path in result -def test_tracks_get_all_paths(session): - # Need two tracks - track1_path = "/a/b/c" - _ = Tracks(session, track1_path) - track2_path = "/m/n/o" - _ = Tracks(session, track2_path) +def test_tracks_by_path(session, track1): - result = Tracks.get_all_paths(session) - assert track1_path in result - assert track2_path in result + assert Tracks.get_by_path(session, track1.path) is track1 -def test_tracks_get_all_tracks(session): - # Need two tracks - track1_path = "/a/b/c" - track2_path = "/m/n/o" +def test_tracks_by_id(session, track1): - result = Tracks.get_all_tracks(session) - assert track1_path in [a.path for a in result] - assert track2_path in [a.path for a in result] + assert session.get(Tracks, track1.id) is track1 -def test_tracks_by_filename(session): - track1_path = "/a/b/c" +def test_tracks_search_artists(session, track1): + track1_artist = "Fleetwood Mac" - track1 = Tracks(session, track1_path) - assert Tracks.get_by_filename(session, os.path.basename(track1_path)) is track1 - - -def test_tracks_by_path(session): - track1_path = "/a/b/c" - - track1 = Tracks(session, track1_path) - assert Tracks.get_by_path(session, track1_path) is track1 - - -def test_tracks_by_id(session): - track1_path = "/a/b/c" - - track1 = Tracks(session, track1_path) - assert Tracks.get_by_id(session, track1.id) is track1 - - -def test_tracks_rescan(session): - # Get test track - test_track_path = "./testdata/isa.mp3" - test_track_data = "./testdata/isa.py" - - track = Tracks(session, test_track_path) - track.rescan(session) - - # Get test data - with open(test_track_data) as f: - testdata = eval(f.read()) - - # Re-read the track - track_read = Tracks.get_by_path(session, test_track_path) - - assert track_read.duration == testdata["duration"] - assert track_read.start_gap == testdata["leading_silence"] - # Silence detection can vary, so ± 1 second is OK - assert track_read.fade_at < testdata["fade_at"] + 1000 - assert track_read.fade_at > testdata["fade_at"] - 1000 - assert track_read.silence_at < testdata["trailing_silence"] + 1000 - assert track_read.silence_at > testdata["trailing_silence"] - 1000 - - -def test_tracks_remove_by_path(session): - track1_path = "/a/b/c" - - assert len(Tracks.get_all_tracks(session)) == 1 - Tracks.remove_by_path(session, track1_path) - assert len(Tracks.get_all_tracks(session)) == 0 - - -def test_tracks_search_artists(session): - track1_path = "/a/b/c" - track1_artist = "Artist One" - track1 = Tracks(session, track1_path) - track1.artist = track1_artist - - track2_path = "/m/n/o" - track2_artist = "Artist Two" - track2 = Tracks(session, track2_path) - track2.artist = track2_artist - - session.commit() - - artist_first_word = track1_artist.split()[0].lower() - assert len(Tracks.search_artists(session, artist_first_word)) == 2 assert len(Tracks.search_artists(session, track1_artist)) == 1 -def test_tracks_search_titles(session): - track1_path = "/a/b/c" - track1_title = "Title One" - track1 = Tracks(session, track1_path) - track1.title = track1_title +def test_tracks_search_titles(session, track1): + track1_title = "I'm So Afraid" - track2_path = "/m/n/o" - track2_title = "Title Two" - track2 = Tracks(session, track2_path) - track2.title = track2_title - - session.commit() - - title_first_word = track1_title.split()[0].lower() - assert len(Tracks.search_titles(session, title_first_word)) == 2 assert len(Tracks.search_titles(session, track1_title)) == 1 - - -def test_tracks_update_lastplayed(session): - track1_path = "/a/b/c" - track1 = Tracks(session, track1_path) - - assert track1.lastplayed is None - track1.update_lastplayed(session, track1.id) - assert track1.lastplayed is not None - - -def test_tracks_update_info(session): - path = "/a/b/c" - artist = "The Beatles" - title = "Help!" - newinfo = "abcdef" - - track1 = Tracks(session, path) - track1.artist = artist - track1.title = title - - test1 = Tracks.get_by_id(session, track1.id) - assert test1.artist == artist - assert test1.title == title - assert test1.path == path - - track1.path = newinfo - test2 = Tracks.get_by_id(session, track1.id) - assert test2.artist == artist - assert test2.title == title - assert test2.path == newinfo - - track1.artist = newinfo - test2 = Tracks.get_by_id(session, track1.id) - assert test2.artist == newinfo - assert test2.title == title - assert test2.path == newinfo - - track1.title = newinfo - test3 = Tracks.get_by_id(session, track1.id) - assert test3.artist == newinfo - assert test3.title == newinfo - assert test3.path == newinfo