add tests for the main frontend (API), and a few fixes related to that

server/djrandom/frontend/__init__.py

@@ -21,7 +21,9 @@ def shutdown_dbsession(exception=None):
 def require_auth(f):
     @wraps(f)
     def check_auth_wrapper(*args, **kwargs):
-        if 'userid' in session:
+        if app.config.get('TESTING'):
+            g.userid = 'testuser'
+        elif 'userid' in session:
             g.userid = session['userid']
         elif 'X-Key' in request.headers:
             user = User.find_with_api_key(

server/djrandom/frontend/api_views.py

@@ -20,6 +20,7 @@ def all_artists_json():
 @app.route('/json/albums/<artist>')
 @require_auth
 def artist_albums_json(artist):
+    print 'artist: %s' % repr(artist)
     albums = [x[0] for x in 
               Session.query(distinct(MP3.album)).filter_by(artist=artist)]
     return jsonify(albums=albums)
@@ -28,6 +29,7 @@ def artist_albums_json(artist):
 @app.route('/json/album/<artist>/<album>')
 @require_auth
 def album_songs_json(artist, album):
+    print 'artist, album: %s / %s' % (repr(artist), repr(album))
     songs = [x.to_dict() for x in MP3.get_songs_for_album(artist, album)]
     return jsonify(songs=songs)
 
@@ -125,7 +127,6 @@ def playlist_info_json(uuid):
 @app.route('/json/playlist/by_title/<uuid>/<title>')
 @require_auth
 def playlist_info_by_title_json(uuid, title):
-    title = urllib.unquote(title)
     playlist = Playlist.get_by_title(uuid, title)
     if not playlist:
         abort(404)

server/djrandom/model/mp3.py

@@ -39,7 +39,7 @@ class MP3(Base):
     artist = Column(Unicode(256))
     title = Column(Unicode(256))
     album = Column(Unicode(256))
-    track_num = Column(Integer(3))
+    track_num = Column(Integer())
     genre = Column(Unicode(64))
     uploaded_at = Column(DateTime())
     play_count = Column(Integer(), default=0)

server/djrandom/test/__init__.py

@@ -15,11 +15,14 @@ class DbTestCase(mox.MoxTestBase):
         self.mox.StubOutWithMock(indexer.Indexer, 'add_mp3')
         self.mox.StubOutWithMock(indexer.Indexer, 'del_mp3')
         self.mox.StubOutWithMock(indexer.Indexer, 'commit')
-        database.init_db('sqlite://', 'http://solr/')
+        self.engine = database.init_db(
+            'sqlite:///%s' % os.path.join(self.tmpdir, 'data.db'),
+            'http://solr/')
 
     def tearDown(self):
-        database.Session.remove()
         mox.MoxTestBase.tearDown(self)
+        database.Session.remove()
+        database.Base.metadata.drop_all(self.engine)
         shutil.rmtree(self.tmpdir)
 
 
@@ -46,12 +49,17 @@ class SolrTestCase(mox.MoxTestBase):
 class WsgiTestCase(DbTestCase):
 
     FLASK_APP = None
-    DATA = []
 
     def setUp(self):
         DbTestCase.setUp(self)
-        for item in self.DATA:
-            database.Session.add(item)
-        database.Session.commit()
         self.FLASK_APP.config['TESTING'] = True
         self.app = self.FLASK_APP.test_client()
+
+    def _load_data(self, data):
+        sess = database.Session()
+        for item in data:
+            print 'adding item', str(item)
+            sess.add(item)
+        sess.commit()
+
+

server/djrandom/test/test_frontend_api.py

+import urllib
+import json
+from datetime import datetime
+from djrandom.test import WsgiTestCase
+from djrandom import frontend
+from djrandom.database import Session
+from djrandom.frontend import svcs
+from djrandom.model.mp3 import MP3, PlayLog
+from djrandom.model.playlist import Playlist
+
+
+class FrontendApiTest(WsgiTestCase):
+
+    FLASK_APP = frontend.app
+
+
+    def setUp(self):
+        WsgiTestCase.setUp(self)
+        self.searcher = svcs['searcher'] = self.mox.CreateMockAnything()
+        self.markov = svcs['markov'] = self.mox.CreateMockAnything()
+
+        self._load_data(
+            [MP3(sha1='1',
+                 artist=u'artist 1',
+                 album=u'album',
+                 title=u'song 1',
+                 path='/root/1',
+                 play_count=3,
+                 state=MP3.READY),
+             MP3(sha1='2',
+                 artist=u'artist 1',
+                 album=u'album 2',
+                 title=u'song 2',
+                 path='/root/2',
+                 state=MP3.READY),
+             MP3(sha1='3',
+                 artist=u'artist 2',
+                 album=u'album',
+                 title=u'song 3',
+                 path='/root/3',
+                 state=MP3.READY),
+             MP3(sha1='4',
+                 path='/root/4',
+                 state=MP3.INCOMING),
+             Playlist(uuid='1234',
+                      userid='testuser',
+                      play_count=1,
+                      contents='1,2'),
+             Playlist(uuid='2345',
+                      title=u'my playlist',
+                      userid='testuser',
+                      contents='2,3')
+             ])
+
+    def _jsonget(self, url, expected_status=200):
+        rv = self.app.get(url)
+        self.assertEquals(expected_status, rv.status_code)
+        return json.loads(rv.data)
+
+    def _jsonpost(self, url, data, expected_status=200):
+        rv = self.app.post(url, data=data)
+        self.assertEquals(expected_status, rv.status_code)
+        return json.loads(rv.data)
+
+    def test_all_artists_json(self):
+        rv = self._jsonget('/json/artists')
+        self.assertTrue('artists' in rv)
+        self.assertEquals([u'artist 1', u'artist 2'],
+                          sorted(rv['artists']))
+        
+    def test_artist_albums_json(self):
+        rv = self._jsonget('/json/albums/artist%201')
+        self.assertTrue('albums' in rv)
+        self.assertEquals(set([u'album', u'album 2']),
+                          set(rv['albums']))
+
+    def test_album_songs_json(self):
+        rv = self._jsonget('/json/album/artist%201/album%202')
+        self.assertTrue('songs' in rv)
+        mp3 = MP3.query.get('2')
+        self.assertEquals([mp3.to_dict()], rv['songs'])
+
+    def test_song_info_json(self):
+        rv = self._jsonget('/json/song/1')
+        mp3 = MP3.query.get('1')
+        self.assertEquals(mp3.to_dict(), rv)
+
+    def test_json_search(self):
+        self.searcher.search('my query', n=100).AndReturn(
+            [(1.0, '1'), (0.5, '2')])
+        self.mox.ReplayAll()
+
+        rv = self._jsonget('/json/search?q=my+query')
+        self.assertTrue('results' in rv)
+        self.assertEquals(
+            [{'score': 1, 'sha1': '1'},
+             {'score': 0.5, 'sha1': '2'}],
+            rv['results'])
+
+    def test_more_like_these_json(self):
+        self.searcher.more_like_these(['1', '2'], 5).AndReturn(
+            ['3'])
+        self.mox.ReplayAll()
+
+        rv = self._jsonpost('/json/morelikethese', 
+                            {'h': '1,2', 'n': '5'})
+        self.assertTrue('results' in rv)
+        self.assertEquals(['3'], rv['results'])
+
+    def test_most_played_json(self):
+        plog = PlayLog(sha1='1', userid='testuser', stamp=datetime.now())
+        Session.add(plog)
+        Session.commit()
+
+        rv = self._jsonget('/json/most_played')
+        self.assertTrue('results' in rv)
+        self.assertTrue(len(rv['results']) > 0)
+        self.assertEquals(
+            {'sha1': '1', 'count': 1},
+            rv['results'][0])
+
+    #def test_never_played_json(self):
+    #    rv = self._jsonget('/json/never_played')
+    #    self.assertTrue('results' in rv)
+    #    self.assertTrue(len(rv['results']) > 0)
+    #    self.assertEquals(
+    #        set(['2', '3']),
+    #        set(rv['results']))
+
+    def test_markov_json(self):
+        self.markov.generate_sequence(['2', '3'], 2, 2).AndReturn(
+            ['1', '2'])
+        self.mox.ReplayAll()
+
+        rv = self._jsonpost('/json/markov',
+                            {'h': '1,2,3', 'n': '2'})
+        self.assertTrue('results' in rv)
+        self.assertEquals(['1', '2'], rv['results'])
+
+    def test_play_callback(self):
+        rv = self._jsonpost('/json/playing',
+                            {'cur': '3',
+                             'prev': '1,2'})
+        self.assertEquals({'status': True}, rv)
+
+        plog = PlayLog.query.first()
+        self.assertEquals('3', plog.sha1)
+        self.assertEquals('testuser', plog.userid)
+        self.assertEquals('1,2', plog.prev)
+
+        mp3 = MP3.query.get('3')
+        self.assertEquals(1, mp3.play_count)
+
+    def test_playlist_info_json(self):
+        rv = self._jsonget('/json/playlist/get/1234')
+        pl = Playlist.query.get('1234')
+        self.assertEquals(pl.to_dict(), rv)
+
+    def test_playlist_info_by_title_json(self):
+        rv = self._jsonget('/json/playlist/by_title/testuser/my%20playlist')
+        pl = Playlist.query.get('2345')
+        self.assertEquals(pl.to_dict(), rv)
+
+    def test_playlist_list_by_uuid_json(self):
+        rv = self._jsonget('/json/playlist/list/testuser')
+        self.assertTrue('results' in rv)
+        self.assertEquals(2, len(rv['results']))
+        self.assertEquals(
+            set(['1234', '2345']),
+            set([x['uuid'] for x in rv['results']]))

server/djrandom/test/test_receiver.py

@@ -9,11 +9,10 @@ class ReceiverTest(WsgiTestCase):
 
     FLASK_APP = receiver.app
 
-    DATA = [MP3(sha1='1234')]
-
     def setUp(self):
         WsgiTestCase.setUp(self)
         receiver.storage_root = self.tmpdir
+        self._load_data([MP3(sha1='1234')])
 
     def test_check_existing(self):
         rv = self.app.get('/check/1234')