[catapult] - Copy Telemetry's gsutilz over to third_party.

third_party/gsutil/third_party/protorpc/demos/tunes_db/server/tunes_db_test.py

Index: third_party/gsutil/third_party/protorpc/demos/tunes_db/server/tunes_db_test.py
diff --git a/third_party/gsutil/third_party/protorpc/demos/tunes_db/server/tunes_db_test.py b/third_party/gsutil/third_party/protorpc/demos/tunes_db/server/tunes_db_test.py
new file mode 100755
index 0000000000000000000000000000000000000000..e086de885b9fdcd2f135e9a66f62a897ce6f4998
--- /dev/null
+++ b/third_party/gsutil/third_party/protorpc/demos/tunes_db/server/tunes_db_test.py
@@ -0,0 +1,346 @@
+#!/usr/bin/python2.4
+#
+# Copyright 2010 Google Inc. All Rights Reserved.
+
+"""Tests for google3.apphosting.demos.tunes_db.server.tunes_db."""
+
+__author__ = 'rafek@google.com (Rafe Kaplan)'
+
+import unittest
+
+from protorpc import message_types
+from protorpc import descriptor
+
+import datastore_test_util
+import model
+import tunes_db
+
+
+class MusicLibraryServiceTest(datastore_test_util.DatastoreTest):
+  """Music library service tests."""
+
+  def setUp(self):
+    """Set up music library."""
+    super(MusicLibraryServiceTest, self).setUp()
+
+    self.service = tunes_db.MusicLibraryService()
+
+    # Create a bunch of artists.
+    #
+    # Definitions are not defined in order to test that search order is
+    # correct.
+    self.aem = model.ArtistInfo(name='Amadou et Miriam')
+    self.aem.put()
+
+    self.abba = model.ArtistInfo(name='Abba')
+    self.abba.put()
+
+    self.furnaces = model.ArtistInfo(name='The Fiery Furnaces')
+    self.furnaces.put()
+
+    self.go_team = model.ArtistInfo(name='  The Go! Team  ')
+    self.go_team.put()
+
+    self.wendy_carlos = model.ArtistInfo(name='Wendy Carlos')
+    self.wendy_carlos.put()
+
+    self.beatles = model.ArtistInfo(name='The Beatles')
+    self.beatles.put()
+
+    # Albums.
+
+    # The Go! Team.
+    self.get_it_together = model.AlbumInfo(name='Get It Together',
+                                           released=2000,
+                                           artist=self.go_team,
+                                           parent=self.go_team)
+    self.get_it_together.put()
+
+    self.thunder_lightening_strike = model.AlbumInfo(
+        name='Thunder, Lightning, Strike',
+        released=2004,
+        artist=self.go_team,
+        parent=self.go_team)
+    self.thunder_lightening_strike.put()
+
+    self.proof_of_youth = model.AlbumInfo(name='Proof of Youth',
+                                          released=2007,
+                                          artist=self.go_team,
+                                          parent=self.go_team)
+    self.proof_of_youth.put()
+
+    # The Beatles.
+    self.help = model.AlbumInfo(name='Help',
+                                released=1965,
+                                artist=self.beatles,
+                                parent=self.beatles)
+    self.help.put()
+
+    self.yellow_submarine = model.AlbumInfo(name='Yellow Submarine',
+                                            released=1969,
+                                            artist=self.beatles,
+                                            parent=self.beatles)
+    self.yellow_submarine.put()
+
+  def AssertArtistMatches(self, model_instance, message):
+    """Helper function to assert that artist message matches a model instance.
+
+    Args:
+      model_instance: Datastore model instance of artist to check against.
+      message: Artist message to check match for.
+    """
+    self.assertEquals(str(model_instance.key()), message.artist_id)
+    self.assertEquals(str(model_instance.name), message.name)
+
+  def AssertAlbumMatches(self, model_instance, message):
+    """Helper function to assert that album message matches a model instance.
+
+    Args:
+      model_instance: Datastore model instance of album to check against.
+      message: Album message to check match for.
+    """
+    self.assertEquals(model_instance.name, message.name)
+    self.assertEquals(str(model_instance.key()), message.album_id)
+    self.assertEquals(model_instance.released, message.released)
+    self.assertEquals(str(model_instance.artist.key()), message.artist_id)
+
+  def testAddArtist(self):
+    """Test the add_artist remote method."""
+    request = tunes_db.AddArtistRequest()
+    request.name = 'Elvis Costello'
+
+    response = self.service.add_artist(request)
+    response.check_initialized()
+
+    elvis = model.ArtistInfo.get(response.artist_id)
+    self.assertEquals('Elvis Costello', elvis.name)
+
+  def testUpdateArtist(self):
+    """Test updating artist."""
+    artist = tunes_db.Artist()
+    artist.artist_id = str(self.wendy_carlos.key())
+    artist.name = u'Walter Carlos'
+    request = tunes_db.UpdateArtistRequest()
+    request.artist = artist
+
+    response = self.service.update_artist(request)
+    response.check_initialized()
+
+    self.assertTrue(response.artist_updated)
+    walter_carlos = model.ArtistInfo.get(self.wendy_carlos.key())
+    self.assertEquals(u'Walter Carlos', walter_carlos.name)
+
+  def testUpdateArtist_NotFound(self):
+    """Test updating artist when artist no longer exists."""
+    wendy_carlos_key = self.wendy_carlos.key()
+    self.wendy_carlos.delete()
+
+    artist = tunes_db.Artist()
+    artist.artist_id = str(wendy_carlos_key)
+    artist.name = u'Walter Carlos'
+    request = tunes_db.UpdateArtistRequest()
+    request.artist = artist
+
+    response = self.service.update_artist(request)
+    response.check_initialized()
+
+    self.assertFalse(response.artist_updated)
+    self.assertEquals(None, model.ArtistInfo.get(wendy_carlos_key))
+
+  def testDeleteArtist(self):
+    """Test the delete_artist remote method."""
+    request = tunes_db.DeleteArtistRequest()
+    request.artist_id = str(self.go_team.key())
+
+    response = self.service.delete_artist(request)
+    response.check_initialized()
+    self.assertEquals(True, response.artist_deleted)
+
+    self.assertEquals(None, model.AlbumInfo.get(self.go_team.key()))
+    self.assertEquals(None,
+                      model.AlbumInfo.gql('WHERE artist = :1',
+                                          self.go_team.key()).get())
+
+    response = self.service.delete_artist(request)
+    self.assertEquals(False, response.artist_deleted)
+
+  def testFetchArtist(self):
+    """Test the fetch_artist remote method."""
+    request = tunes_db.FetchArtistRequest()
+    request.artist_id = str(self.wendy_carlos.key())
+
+    response = self.service.fetch_artist(request)
+    response.check_initialized()
+
+    self.AssertArtistMatches(self.wendy_carlos, response.artist)
+
+  def testSearchArtist_NothingFound(self):
+    """Test the search_artists remote method when no artists are found."""
+    request = tunes_db.SearchArtistsRequest()
+    request.name_prefix = u'Duke'
+
+    response = self.service.search_artists(request)
+    response.check_initialized()
+    self.assertFalse(hasattr(response, 'artist_count'))
+
+  def testSearchArtist_All(self):
+    """Test searching all artists in the music library."""
+    request = tunes_db.SearchArtistsRequest()
+    request.fetch_size = 3
+
+    response = self.service.search_artists(request)
+    response.check_initialized()
+
+    self.assertEquals(3, len(response.artists))
+    self.AssertArtistMatches(self.abba, response.artists[0])
+    self.AssertArtistMatches(self.aem, response.artists[1])
+    self.AssertArtistMatches(self.beatles, response.artists[2])
+
+    request = tunes_db.SearchArtistsRequest()
+    request.continuation = response.continuation
+    request.fetch_size = 3
+    response = self.service.search_artists(request)
+    response.check_initialized()
+
+    self.assertEquals(3, len(response.artists))
+    self.AssertArtistMatches(self.furnaces, response.artists[0])
+    self.AssertArtistMatches(self.go_team, response.artists[1])
+    self.AssertArtistMatches(self.wendy_carlos, response.artists[2])
+
+    request = tunes_db.SearchArtistsRequest()
+    request.continuation = response.continuation
+    response = self.service.search_artists(request)
+    response.check_initialized()
+
+    self.assertEquals([], response.artists)
+
+  def testSearchArtist_NamePrefix(self):
+    """Test searching artists and matching by name prefix."""
+    request = tunes_db.SearchArtistsRequest()
+    request.name_prefix = u' { tHe!  '
+    request.fetch_size = 2
+
+    response = self.service.search_artists(request)
+    response.check_initialized()
+
+    self.assertEquals(2, len(response.artists))
+    self.AssertArtistMatches(self.beatles, response.artists[0])
+    self.AssertArtistMatches(self.furnaces, response.artists[1])
+
+    request = tunes_db.SearchArtistsRequest()
+    request.continuation = response.continuation
+
+    response = self.service.search_artists(request)
+    response.check_initialized()
+
+    self.assertEquals(1, len(response.artists))
+    self.assertEquals(None, response.continuation)
+    self.AssertArtistMatches(self.go_team, response.artists[0])
+
+  def testAddAlbum(self):
+    """Test creating an album."""
+    request = tunes_db.AddAlbumRequest()
+    request.artist_id = str(self.furnaces.key())
+    request.name = u'Blueberry Boat'
+    request.released = 2004
+
+    response = self.service.add_album(request)
+    response.check_initialized()
+
+    blueberry = model.AlbumInfo.get(response.album_id)
+    self.assertEquals(self.furnaces.key(), blueberry.artist.key())
+    self.assertEquals('Blueberry Boat', blueberry.name)
+    self.assertEquals(2004, blueberry.released)
+
+  def testUpdateAlbum(self):
+    """Test updating album."""
+    album = tunes_db.Album()
+    album.album_id = str(self.proof_of_youth.key())
+    album.name = u'Proof of Age'
+    album.released = 1908
+    request = tunes_db.UpdateAlbumRequest()
+    request.album = album
+
+    response = self.service.update_album(request)
+    response.check_initialized()
+
+    self.assertTrue(response.album_updated)
+    proof_of_age = model.AlbumInfo.get(self.proof_of_youth.key())
+    self.assertEquals(u'Proof of Age', proof_of_age.name)
+    self.assertEquals(1908, proof_of_age.released)
+
+  def testUpdateAlbum_NotFound(self):
+    """Test updating album when album no longer exists."""
+    proof_of_youth_key = self.proof_of_youth.key()
+    self.proof_of_youth.delete()
+
+    album = tunes_db.Album()
+    album.album_id = str(proof_of_youth_key)
+    album.name = u'Proof of Age'
+    album.released = 1908
+    request = tunes_db.UpdateAlbumRequest()
+    request.album = album
+
+    response = self.service.update_album(request)
+    response.check_initialized()
+
+    self.assertFalse(response.album_updated)
+    self.assertEquals(None, model.AlbumInfo.get(proof_of_youth_key))
+
+  def testSearchAlbum_All(self):
+    """Test searching all albums in library."""
+    request = tunes_db.SearchAlbumsRequest()
+    request.fetch_size = 2
+
+    response = self.service.search_albums(request)
+    response.check_initialized()
+
+    self.assertEquals(2, len(response.albums))
+    self.AssertAlbumMatches(self.get_it_together, response.albums[0])
+    self.AssertAlbumMatches(self.help, response.albums[1])
+
+    request = tunes_db.SearchAlbumsRequest()
+    request.continuation = response.continuation
+
+    response = self.service.search_albums(request)
+    response.check_initialized()
+
+    self.assertEquals(2, len(response.albums))
+    self.AssertAlbumMatches(self.proof_of_youth, response.albums[0])
+    self.AssertAlbumMatches(self.thunder_lightening_strike, response.albums[1])
+
+    request = tunes_db.SearchAlbumsRequest()
+    request.continuation = response.continuation
+
+    response = self.service.search_albums(request)
+    response.check_initialized()
+    self.assertEquals(1, len(response.albums))
+    self.assertEquals(None, response.continuation)
+    self.AssertAlbumMatches(self.yellow_submarine, response.albums[0])
+
+  def testSearchAlbum_FilterArtist(self):
+    """Test searching albums for specific artist in library."""
+    request = tunes_db.SearchAlbumsRequest()
+    request.fetch_size = 2
+    request.artist_id = unicode(self.go_team.key())
+
+    response = self.service.search_albums(request)
+    response.check_initialized()
+
+    self.assertEquals(2, len(response.albums))
+    self.AssertAlbumMatches(self.get_it_together, response.albums[0])
+    self.AssertAlbumMatches(self.proof_of_youth, response.albums[1])
+
+    request = tunes_db.SearchAlbumsRequest()
+    request.continuation = response.continuation
+
+    response = self.service.search_albums(request)
+    response.check_initialized()
+
+    self.assertEquals(1, len(response.albums))
+    self.assertEquals(None, response.continuation)
+    self.AssertAlbumMatches(self.thunder_lightening_strike, response.albums[0])
+
+
+if __name__ == '__main__':
+  unittest.main()