#!/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 robot module.""" import unittest import events import ops import robot import simplejson BLIP_JSON = ('{"wdykLROk*13":' '{"lastModifiedTime":1242079608457,' '"contributors":["someguy@test.com"],' '"waveletId":"test.com!conv+root",' '"waveId":"test.com!wdykLROk*11",' '"parentBlipId":null,' '"version":3,' '"creator":"someguy@test.com",' '"content":"\\nContent!",' '"blipId":"wdykLROk*13","' 'annotations":[{"range":{"start":0,"end":1},' '"name":"user/e/davidbyttow@google.com","value":"David"}],' '"elements":{},' '"threadId": "",' '"childBlipIds":[]}' '}') WAVELET_JSON = ('{"lastModifiedTime":1242079611003,' '"title":"A title",' '"waveletId":"test.com!conv+root",' '"rootBlipId":"wdykLROk*13",' '"dataDocuments":null,' '"creationTime":1242079608457,' '"waveId":"test.com!wdykLROk*11",' '"participants":["someguy@test.com","monty@appspot.com"],' '"creator":"someguy@test.com",' '"rootThread": ' '{"id":"", "location": "-1", "blipIds": ["wdykLROk*13"]},' '"version":5}') EVENTS_JSON = ('[{"timestamp":1242079611003,' '"modifiedBy":"someguy@test.com",' '"properties":{"participantsRemoved":[],' '"participantsAdded":["monty@appspot.com"]},' '"type":"WAVELET_PARTICIPANTS_CHANGED"}]') TEST_JSON = '{"blips":%s,"wavelet":%s,"events":%s, "threads": {}}' % ( BLIP_JSON, WAVELET_JSON, EVENTS_JSON) NEW_WAVE_JSON = [{"data": {"waveletId": "wavesandbox.com!conv+root", "blipId": "b+LrODcLZkDlu", "waveId": "wavesandbox.com!w+LrODcLZkDlt"}, "id": "op2"}] class TestRobot(unittest.TestCase): """Tests for testing the basic parsing of json in robots.""" def setUp(self): self.robot = robot.Robot('Testy') def testCreateWave(self): self.robot.get_waveservice().submit = lambda x: NEW_WAVE_JSON new_wave = self.robot.new_wave('wavesandbox.com', submit=True) self.assertEqual('wavesandbox.com!w+LrODcLZkDlt', new_wave.wave_id) def testEventParsing(self): def check(event, wavelet): # Test some basic properties; the rest should be covered by # ops.CreateContext. root = wavelet.root_blip self.assertEqual(1, len(wavelet.blips)) self.assertEqual('wdykLROk*13', root.blip_id) self.assertEqual('test.com!wdykLROk*11', root.wave_id) self.assertEqual('test.com!conv+root', root.wavelet_id) self.assertEqual('WAVELET_PARTICIPANTS_CHANGED', event.type) self.assertEqual({'participantsRemoved': [], 'participantsAdded': ['monty@appspot.com']}, event.properties) self.robot.test_called = True self.robot.test_called = False self.robot.register_handler(events.WaveletParticipantsChanged, check) json = self.robot.process_events(TEST_JSON) self.assertTrue(self.robot.test_called) operations = simplejson.loads(json) # there should be one operation indicating the current version: self.assertEqual(1, len(operations)) def testWrongEventsIgnored(self): self.robot.test_called = True def check(event, wavelet): called = True self.robot.test_called = False self.robot.register_handler(events.BlipSubmitted, check) self.robot.process_events(TEST_JSON) self.assertFalse(self.robot.test_called) def testOperationParsing(self): def check(event, wavelet): wavelet.reply() wavelet.title = 'new title' wavelet.root_blip.append_markup('Hello') self.robot.register_handler(events.WaveletParticipantsChanged, check) json = self.robot.process_events(TEST_JSON) operations = simplejson.loads(json) expected = set([ops.ROBOT_NOTIFY, ops.WAVELET_APPEND_BLIP, ops.WAVELET_SET_TITLE, ops.DOCUMENT_APPEND_MARKUP]) methods = [operation['method'] for operation in operations] for method in methods: self.assertTrue(method in expected) expected.remove(method) self.assertEquals(0, len(expected)) def testSerializeWavelets(self): wavelet = self.robot.blind_wavelet(TEST_JSON) serialized = wavelet.serialize() unserialized = self.robot.blind_wavelet(serialized) self.assertEquals(wavelet.creator, unserialized.creator) self.assertEquals(wavelet.creation_time, unserialized.creation_time) self.assertEquals(wavelet.last_modified_time, unserialized.last_modified_time) self.assertEquals(wavelet.root_blip.blip_id, unserialized.root_blip.blip_id) self.assertEquals(wavelet.title, unserialized.title) self.assertEquals(wavelet.wave_id, unserialized.wave_id) self.assertEquals(wavelet.wavelet_id, unserialized.wavelet_id) self.assertEquals(wavelet.domain, unserialized.domain) def testProxiedBlindWavelet(self): def handler(event, wavelet): blind_wavelet = self.robot.blind_wavelet(TEST_JSON, 'proxyid') blind_wavelet.reply() blind_wavelet.submit_with(wavelet) self.robot.register_handler(events.WaveletParticipantsChanged, handler) json = self.robot.process_events(TEST_JSON) operations = simplejson.loads(json) self.assertEqual(2, len(operations)) self.assertEquals(ops.ROBOT_NOTIFY, operations[0]['method']) self.assertEquals(ops.WAVELET_APPEND_BLIP, operations[1]['method']) self.assertEquals('proxyid', operations[1]['params']['proxyingFor']) def testCapabilitiesHashIncludesContextAndFilter(self): robot1 = robot.Robot('Robot1') robot1.register_handler(events.WaveletSelfAdded, lambda: '') robot2 = robot.Robot('Robot2') robot2.register_handler(events.WaveletSelfAdded, lambda: '', context=events.Context.ALL) self.assertNotEqual(robot1.capabilities_hash(), robot2.capabilities_hash()) robot3 = robot.Robot('Robot3') robot2.register_handler(events.WaveletSelfAdded, lambda: '', context=events.Context.ALL, filter="foo") self.assertNotEqual(robot1.capabilities_hash(), robot2.capabilities_hash()) self.assertNotEqual(robot1.capabilities_hash(), robot3.capabilities_hash()) self.assertNotEqual(robot2.capabilities_hash(), robot3.capabilities_hash()) class TestGetCapabilitiesXml(unittest.TestCase): def setUp(self): self.robot = robot.Robot('Testy') self.robot.capabilities_hash = lambda: '1' def assertStringsEqual(self, s1, s2): self.assertEqual(s1, s2, 'Strings differ:\n%s--\n%s' % (s1, s2)) def testDefault(self): expected = ( '\n' '\n' '1\n' '%s\n' '\n\n' '\n') % ops.PROTOCOL_VERSION xml = self.robot.capabilities_xml() self.assertStringsEqual(expected, xml) def testUrls(self): profile_robot = robot.Robot( 'Testy', image_url='http://example.com/image.png', profile_url='http://example.com/profile.xml') profile_robot.capabilities_hash = lambda: '1' expected = ( '\n' '\n' '1\n' '%s\n' '\n\n' '\n') % ops.PROTOCOL_VERSION xml = profile_robot.capabilities_xml() self.assertStringsEqual(expected, xml) def testConsumerKey(self): # setup_oauth doesn't work during testing, so heavy handed setting of # properties it is: self.robot._consumer_key = 'consumer' expected = ( '\n' '\n' '1\n' 'consumer\n' '%s\n' '\n\n' '\n') % ops.PROTOCOL_VERSION xml = self.robot.capabilities_xml() self.assertStringsEqual(expected, xml) def testCapsAndEvents(self): self.robot.register_handler(events.BlipSubmitted, None, context=[events.Context.SELF, events.Context.ROOT]) expected = ( '\n' '\n' '1\n' '%s\n' '\n' ' \n' '\n' '\n') % (ops.PROTOCOL_VERSION, events.BlipSubmitted.type) xml = self.robot.capabilities_xml() self.assertStringsEqual(expected, xml) if __name__ == '__main__': unittest.main()