#!/usr/bin/python2.4 # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Unit tests for the wavelet module.""" import unittest import blip import element import ops import wavelet import simplejson ROBOT_NAME = 'robot@appspot.com' TEST_WAVELET_DATA = { 'creator': ROBOT_NAME, 'creationTime': 100, 'lastModifiedTime': 101, 'participants': [ROBOT_NAME], 'participantsRoles': {ROBOT_NAME: wavelet.Participants.ROLE_FULL}, 'rootBlipId': 'blip-1', 'title': 'Title', 'waveId': 'test.com!w+g3h3im', 'waveletId': 'test.com!root+conv', 'tags': ['tag1', 'tag2'], 'rootThread': { 'id': '', 'location': -1, 'blipIds': ['blip-1'] } } TEST_BLIP_DATA = { 'blipId': TEST_WAVELET_DATA['rootBlipId'], 'childBlipIds': [], 'content': '\ntesting', 'contributors': [TEST_WAVELET_DATA['creator'], 'robot@google.com'], 'creator': TEST_WAVELET_DATA['creator'], 'lastModifiedTime': TEST_WAVELET_DATA['lastModifiedTime'], 'parentBlipId': None, 'waveId': TEST_WAVELET_DATA['waveId'], 'elements': {}, 'waveletId': TEST_WAVELET_DATA['waveletId'], 'replyThreadIds': [], 'threadId': '' } class TestWavelet(unittest.TestCase): """Tests the wavelet class.""" def setUp(self): self.operation_queue = ops.OperationQueue() self.all_blips = {} self.blip = blip.Blip(TEST_BLIP_DATA, self.all_blips, self.operation_queue) self.all_blips[self.blip.blip_id] = self.blip root_thread_data = TEST_WAVELET_DATA.get('rootThread') root_thread = wavelet.BlipThread('', root_thread_data.get('location'), root_thread_data.get('blipIds', []), self.all_blips, self.operation_queue) self.wavelet = wavelet.Wavelet(TEST_WAVELET_DATA, self.all_blips, root_thread, self.operation_queue) self.wavelet.robot_address = ROBOT_NAME def testWaveletProperties(self): w = self.wavelet self.assertEquals(TEST_WAVELET_DATA['creator'], w.creator) self.assertEquals(TEST_WAVELET_DATA['creationTime'], w.creation_time) self.assertEquals(TEST_WAVELET_DATA['lastModifiedTime'], w.last_modified_time) self.assertEquals(len(TEST_WAVELET_DATA['participants']), len(w.participants)) self.assertTrue(TEST_WAVELET_DATA['participants'][0] in w.participants) self.assertEquals(TEST_WAVELET_DATA['rootBlipId'], w.root_blip.blip_id) self.assertEquals(TEST_WAVELET_DATA['title'], w.title) self.assertEquals(TEST_WAVELET_DATA['waveId'], w.wave_id) self.assertEquals(TEST_WAVELET_DATA['waveletId'], w.wavelet_id) self.assertEquals(TEST_WAVELET_DATA['rootThread']['id'], w.root_thread.id) self.assertEquals(TEST_WAVELET_DATA['rootThread']['location'], w.root_thread.location) self.assertEquals(len(TEST_WAVELET_DATA['rootThread']['blipIds']), len(w.root_thread.blips)) self.assertEquals('test.com', w.domain) def testWaveletMethods(self): w = self.wavelet reply = w.reply() self.assertEquals(2, len(w.blips)) w.delete(reply) self.assertEquals(1, len(w.blips)) self.assertEquals(0, len(w.data_documents)) self.wavelet.data_documents['key'] = 'value' self.assert_('key' in w.data_documents) self.assertEquals(1, len(w.data_documents)) for key in w.data_documents: self.assertEquals(key, 'key') self.assertEquals(1, len(w.data_documents.keys())) self.wavelet.data_documents['key'] = None self.assertEquals(0, len(w.data_documents)) num_participants = len(w.participants) w.proxy_for('proxy').reply() self.assertEquals(2, len(w.blips)) # check that the new proxy for participant was added self.assertEquals(num_participants + 1, len(w.participants)) w._robot_address = ROBOT_NAME.replace('@', '+proxy@') w.proxy_for('proxy').reply() self.assertEquals(num_participants + 1, len(w.participants)) self.assertEquals(3, len(w.blips)) def testSetTitle(self): self.blip._content = '\nOld title\n\nContent' self.wavelet.title = 'New title \xd0\xb0\xd0\xb1\xd0\xb2' self.assertEquals(1, len(self.operation_queue)) self.assertEquals('wavelet.setTitle', self.operation_queue.serialize()[1]['method']) self.assertEquals(u'\nNew title \u0430\u0431\u0432\n\nContent', self.blip._content) def testSetTitleAdjustRootBlipWithOneLineProperly(self): self.blip._content = '\nOld title' self.wavelet.title = 'New title' self.assertEquals(1, len(self.operation_queue)) self.assertEquals('wavelet.setTitle', self.operation_queue.serialize()[1]['method']) self.assertEquals('\nNew title\n', self.blip._content) def testSetTitleAdjustEmptyRootBlipProperly(self): self.blip._content = '\n' self.wavelet.title = 'New title' self.assertEquals(1, len(self.operation_queue)) self.assertEquals('wavelet.setTitle', self.operation_queue.serialize()[1]['method']) self.assertEquals('\nNew title\n', self.blip._content) def testTags(self): w = self.wavelet self.assertEquals(2, len(w.tags)) w.tags.append('tag3') self.assertEquals(3, len(w.tags)) w.tags.append('tag3') self.assertEquals(3, len(w.tags)) w.tags.remove('tag1') self.assertEquals(2, len(w.tags)) self.assertEquals('tag2', w.tags[0]) def testParticipantRoles(self): w = self.wavelet self.assertEquals(wavelet.Participants.ROLE_FULL, w.participants.get_role(ROBOT_NAME)) w.participants.set_role(ROBOT_NAME, wavelet.Participants.ROLE_READ_ONLY) self.assertEquals(wavelet.Participants.ROLE_READ_ONLY, w.participants.get_role(ROBOT_NAME)) def testSerialize(self): self.blip.append(element.Gadget('http://test.com', {'a': 3})) self.wavelet.title = 'A wavelet title' self.blip.append(element.Image(url='http://www.google.com/logos/clickortreat1.gif', width=320, height=118)) self.blip.append(element.Attachment(caption='fake', data='fake data')) self.blip.append(element.Line(line_type='li', indent='2')) self.blip.append('bulleted!') self.blip.append(element.Installer( 'http://wave-skynet.appspot.com/public/extensions/areyouin/manifest.xml')) self.wavelet.proxy_for('proxy').reply().append('hi from douwe') inlineBlip = self.blip.insert_inline_blip(5) inlineBlip.append('hello again!') serialized = self.wavelet.serialize() serialized = simplejson.dumps(serialized) self.assertTrue(serialized.find('test.com') > 0) if __name__ == '__main__': unittest.main()