test coverage

This commit is contained in:
evilchili 2022-12-07 23:10:41 -08:00
parent 70f55105de
commit af33e91232
5 changed files with 204 additions and 74 deletions

View File

@ -10,7 +10,7 @@ EDITOR_TEMPLATE = """
{name}:
description: {description}
entries:
{entries}
{entries}
# ------------------------------------------------------------------------------
#

View File

@ -27,3 +27,9 @@ class PlaylistValidationError(Exception):
"""
An error was discovered in the playlist definition.
"""
class TrackNotFoundError(Exception):
"""
The specified track doesn't exist.
"""

View File

@ -5,7 +5,7 @@ from typing import Union, List
from groove import db
from groove.editor import PlaylistEditor, EDITOR_TEMPLATE
from groove.exceptions import PlaylistValidationError
from groove.exceptions import PlaylistValidationError, TrackNotFoundError
from slugify import slugify
from sqlalchemy import func, delete
@ -19,14 +19,16 @@ class Playlist:
CRUD operations and convenience methods for playlists.
"""
def __init__(self,
slug: str,
name: str,
session: Session,
slug: str = '',
description: str = '',
create_ok=True):
self._session = session
self._slug = slug
self._name = name
if not self._name:
raise PlaylistValidationError("Name cannot be empty.")
self._slug = slug or slugify(name)
self._description = description
self._entries = []
self._record = None
@ -79,11 +81,16 @@ class Playlist:
return self._session
@property
def record(self) -> Union[Row, None]:
def record(self) -> Union[Row, None, bool]:
"""
Cache the playlist row from the database and return it. Optionally create it if it doesn't exist.
Returns:
None: If we've never tried to get or create the record
False: False if we've tried and failed to get the record
Row: The record as it appears in the database.
"""
if not self._record:
if self._record is None:
self.get_or_create()
return self._record
@ -98,7 +105,7 @@ class Playlist:
db.entry,
db.track
).filter(
(db.playlist.c.id == self.record.id)
(db.playlist.c.id == self._record.id)
).filter(
db.entry.c.playlist_id == db.playlist.c.id
).filter(
@ -144,15 +151,50 @@ class Playlist:
Retrieve tracks from the database that match the specified path fragments. The exceptions NoResultFound and
MultipleResultsFound are expected in the case of no matches and multiple matches, respectively.
"""
return [self.session.query(db.track).filter(db.track.c.relpath.ilike(f"%{path}%")).one() for path in paths]
for path in paths:
try:
yield self.session.query(db.track).filter(
db.track.c.relpath.ilike(f"%{path}%")
).one()
except NoResultFound:
raise TrackNotFoundError(f'Could not find track for path "{path}"')
def _get_tracks_by_artist_and_title(self, entries: List[tuple]) -> List:
return [
self.session.query(db.track).filter(
for (artist, title) in entries:
try:
yield self.session.query(db.track).filter(
db.track.c.artist == artist, db.track.c.title == title
).one()
for (artist, title) in entries
]
except NoResultFound: # pragma: no cover
raise TrackNotFoundError(f'Could not find track "{artist}": "{title}"')
def _get(self):
try:
return self.session.query(db.playlist).filter(
db.playlist.c.slug == self.slug
).one()
except NoResultFound:
logging.debug(f"Could not find a playlist with slug {self.slug}.")
return False
def _insert(self, values):
stmt = db.playlist.insert(values)
results = self.session.execute(stmt)
self.session.commit()
logging.debug(f"Inserted playlist with slug {self.slug}")
return self.session.query(db.playlist).filter(
db.playlist.c.id == results.inserted_primary_key[0]
).one()
def _update(self, values):
stmt = db.playlist.update().where(
db.playlist.c.id == self._record.id
).values(values)
self.session.execute(stmt)
self.session.commit()
return self.session.query(db.playlist).filter(
db.playlist.c.id == self._record.id
).one()
def edit(self):
edits = self.editor.edit(self)
@ -182,11 +224,11 @@ class Playlist:
"""
logging.debug(f"Attempting to add tracks matching: {paths}")
try:
return self.create_entries(self._get_tracks_by_path(paths))
except NoResultFound:
return self.create_entries(list(self._get_tracks_by_path(paths)))
except NoResultFound: # pragma: no cover
logging.error("One or more of the specified paths do not match any tracks in the database.")
return 0
except MultipleResultsFound:
except MultipleResultsFound: # pragma: no cover
logging.error("One or more of the specified paths matches multiple tracks in the database.")
return 0
@ -210,40 +252,14 @@ class Playlist:
return plid
def get_or_create(self, create_ok: bool = False) -> Row:
try:
if self._record is None:
self._record = self._get()
return
except NoResultFound:
logging.debug(f"Could not find a playlist with slug {self.slug}.")
if not self._record:
if self.deleted:
raise RuntimeError("Object has been deleted.")
raise PlaylistValidationError("Object has been deleted.")
if self._create_ok or create_ok:
self.save()
def _get(self):
return self.session.query(db.playlist).filter(
db.playlist.c.slug == self.slug
).one()
def _insert(self, values):
stmt = db.playlist.insert(values)
results = self.session.execute(stmt)
self.session.commit()
logging.debug(f"Inserted playlist with slug {self.slug}")
return self.session.query(db.playlist).filter(
db.playlist.c.id == results.inserted_primary_key[0]
).one()
def _update(self, values):
stmt = db.playlist.update().where(
db.playlist.c.id == self._record.id
).values(values)
self.session.execute(stmt)
self.session.commit()
return self.session.query(db.playlist).filter(
db.playlist.c.id == self._record.id
).one()
def save(self) -> Row:
values = {
'slug': self.slug,
@ -265,10 +281,6 @@ class Playlist:
self.session.execute(stmt)
return self.create_entries(self.entries)
def load(self):
self.get_or_create(create_ok=False)
return self
def create_entries(self, tracks: List[Row]) -> int:
"""
Append a list of tracks to a playlist by populating the entries table with records referencing the playlist and
@ -322,7 +334,7 @@ class Playlist:
return pl
@classmethod
def from_yaml(cls, source, session):
def from_yaml(cls, source, session, create_ok=False):
try:
name = list(source.keys())[0].strip()
description = (source[name]['description'] or '').strip()
@ -331,13 +343,13 @@ class Playlist:
name=name,
description=description,
session=session,
create_ok=create_ok
)
except (IndexError, KeyError):
PlaylistValidationError("The specified source was not a valid playlist.")
pl._entries = pl._get_tracks_by_artist_and_title(entries=[
pl._entries = list(pl._get_tracks_by_artist_and_title(entries=[
list(entry.items())[0] for entry in source[name]['entries']
])
]))
except (IndexError, KeyError):
raise PlaylistValidationError("The specified source was not a valid playlist.")
return pl
def __eq__(self, obj):

View File

@ -51,9 +51,9 @@ def db(in_memory_db):
# create tracks
query = insert(groove.db.track)
in_memory_db.execute(query, [
{'id': 1, 'relpath': 'UNKLE/Psyence Fiction/01 Guns Blazing (Drums of Death, Part 1).flac'},
{'id': 2, 'relpath': 'UNKLE/Psyence Fiction/02 UNKLE (Main Title Theme).flac'},
{'id': 3, 'relpath': 'UNKLE/Psyence Fiction/03 Bloodstain.flac'}
{'id': 1, 'artist': 'UNKLE', 'title': 'Guns Blazing', 'relpath': 'UNKLE/Psyence Fiction/01 Guns Blazing (Drums of Death, Part 1).flac'},
{'id': 2, 'artist': 'UNKLE', 'title': 'UNKLE', 'relpath': 'UNKLE/Psyence Fiction/02 UNKLE (Main Title Theme).flac'},
{'id': 3, 'artist': 'UNKLE', 'title': 'Bloodstain', 'relpath': 'UNKLE/Psyence Fiction/03 Bloodstain.flac'}
])
# create playlists
@ -82,9 +82,4 @@ def db(in_memory_db):
@pytest.fixture(scope='function')
def empty_playlist(db):
return Playlist(
name='an empty playlist fixture',
slug='an-empty-playlist',
description='a fixture',
session=db
)
return Playlist.by_slug('empty-playlist', session=db)

View File

@ -1,15 +1,38 @@
import pytest
from groove import playlist
from groove.exceptions import PlaylistValidationError
import yaml
from unittest.mock import MagicMock
from groove import playlist, editor
from groove.exceptions import PlaylistValidationError, TrackNotFoundError
# 166-167, 200, 203-204, 227-228, 253->255, 255->exit, 270, 346-347
def test_create(empty_playlist):
assert empty_playlist.record.id
def test_get_no_create(in_memory_db):
pl = playlist.Playlist(name='something', session=in_memory_db, create_ok=False)
# assert twice to ensure we cache the first db query result.
assert pl.get_or_create() is None
assert pl.get_or_create() is None
def test_exists(db):
pl = playlist.Playlist(name='something', session=db, create_ok=True)
assert pl.exists
def test_exists_deleted(empty_playlist):
assert empty_playlist.exists
empty_playlist.delete()
assert not empty_playlist.exists
@pytest.mark.parametrize('vals, expected', [
(dict(name='missing-the-slug', ), TypeError),
(dict(name='missing-the-slug', slug=''), PlaylistValidationError),
(dict(slug='missing-the-name', name=''), PlaylistValidationError),
])
def test_create_missing_fields(vals, expected, db):
@ -26,7 +49,8 @@ def test_add(tracks, empty_playlist):
def test_add_no_matches(empty_playlist):
assert empty_playlist.add(('no match', )) == 0
with pytest.raises(TrackNotFoundError):
empty_playlist.add(('no match', ))
def test_add_multiple_matches(empty_playlist):
@ -49,7 +73,7 @@ def test_delete_playlist_not_exist(db):
def test_cannot_create_after_delete(db, empty_playlist):
empty_playlist.delete()
with pytest.raises(RuntimeError):
with pytest.raises(PlaylistValidationError):
assert empty_playlist.record
assert not empty_playlist.exists
@ -71,3 +95,96 @@ def test_playlist_formatted(db, empty_playlist):
assert repr(empty_playlist)
assert empty_playlist.as_string
assert empty_playlist.as_dict
def test_playlist_equality(db):
pl = playlist.Playlist(name='foo', slug='foo', session=db, create_ok=False)
pl2 = playlist.Playlist(name='foo', slug='foo', session=db, create_ok=False)
assert pl == pl2
pl.save()
assert pl == pl2
def test_playlist_inequality(db):
pl = playlist.Playlist(name='foo', slug='foo', session=db, create_ok=False)
pl2 = playlist.Playlist(name='bar', slug='foo', session=db, create_ok=False)
assert pl != pl2
pl.save()
assert pl != pl2
def test_playlist_inequality_tracks_differ(db):
pl = playlist.Playlist.from_yaml({
'foo': {
'description': '',
'entries': []
}
}, db)
pl2 = playlist.Playlist.from_yaml({
'foo': {
'description': '',
'entries': [
{'UNKLE': 'Guns Blazing'},
]
}
}, db)
assert pl != pl2
def test_as_yaml(db):
expected = {
'playlist one': {
'description': 'the first one',
'entries': [
{'UNKLE': 'Guns Blazing'},
{'UNKLE': 'UNKLE'},
{'UNKLE': 'Bloodstain'},
]
}
}
pl = playlist.Playlist.by_slug('playlist-one', db)
assert yaml.safe_load(pl.as_yaml) == expected
def test_from_yaml(db):
pl = playlist.Playlist.by_slug('playlist-one', db)
pl2 = playlist.Playlist.from_yaml(yaml.safe_load(pl.as_yaml), db)
assert pl2 == pl
@pytest.mark.parametrize('src', [
{'missing description': {'entries': []}},
{'missing entries': {'description': 'foo'}},
{'empty': {}},
{'': {'description': 'no name', 'entries': []}},
])
def test_from_yaml_missing_values(src, db):
with pytest.raises(PlaylistValidationError):
playlist.Playlist.from_yaml(src, db)
@pytest.mark.parametrize('edits, expected', [
({'edited': {'description': 'the edited one', 'entries': []}}, 'edited'),
({'empty playlist': {'description': 'no tracks', 'entries': []}}, 'empty playlist'),
(None, 'empty playlist'),
])
def test_edit(monkeypatch, edits, expected, empty_playlist):
monkeypatch.setattr(
empty_playlist._editor, 'edit', MagicMock(spec=editor, return_value=edits)
)
empty_playlist.edit()
assert empty_playlist.name == expected
@pytest.mark.parametrize('slug', [None, ''])
def test_save_no_slug(slug, empty_playlist):
empty_playlist._slug = slug
with pytest.raises(PlaylistValidationError):
empty_playlist.save()@pytest.mark.parametrize('slug', [None, ''])
@pytest.mark.parametrize('name', [None, ''])
def test_save_no_name(name, empty_playlist):
empty_playlist._name = name
with pytest.raises(PlaylistValidationError):
empty_playlist.save()