<filename>mindstormsy-robot/waveapi/robot_test.py
#!/usr/bin/python2.4
#
# Copyright (C) 2009 Google Inc.
#
# 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":["<EMAIL>"],'
               '"waveletId":"test.com!conv+root",'
               '"waveId":"test.com!wdykLROk*11",'
               '"parentBlipId":null,'
               '"version":3,'
               '"creator":"<EMAIL>",'
               '"content":"\\nContent!",'
               '"blipId":"wdykLROk*13","'
               'annotations":[{"range":{"start":0,"end":1},'
                   '"name":"user/e/<EMAIL>","value":"David"}],'
               '"elements":{},'
               '"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":["<EMAIL>","<EMAIL>"],'
                 '"creator":"<EMAIL>",'
                 '"version":5}')

EVENTS_JSON = ('[{"timestamp":1242079611003,'
                '"modifiedBy":"<EMAIL>",'
                '"properties":{"participantsRemoved":[],'
                    '"participantsAdded":["<EMAIL>"]},'
                '"type":"WAVELET_PARTICIPANTS_CHANGED"}]')

TEST_JSON = '{"blips":%s,"wavelet":%s,"events":%s}' % (
    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"}]

NEW_WAVE_JSON_OLD = [{'data':
                      [{'data':
                        {'waveletId': 'googlewave.com!conv+root',
                         'blipId': 'b+VqQXQbZkCP1',
                         'waveId': 'googlewave.com!w+VqQXQbZkCP0'},
                         'id': 'wavelet.create1265055048410'}],
                      'id': 'op10'}];

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.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)
    self.robot.submit = lambda x: NEW_WAVE_JSON_OLD
    new_wave = self.robot.new_wave('googlewave.com', submit=True)
    self.assertEqual('googlewave.com!w+VqQXQbZkCP0', 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': ['<EMAIL>']},
                       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('<b>Hello</b>')

    self.robot.register_handler(events.WaveletParticipantsChanged,
                                check)
    json = self.robot.process_events(TEST_JSON)
    operations = simplejson.loads(json)
    expected = set([ops.ROBOT_NOTIFY_CAPABILITIES_HASH,
                    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_CAPABILITIES_HASH,
                      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 = (
        '<?xml version="1.0"?>\n'
        '<w:robot xmlns:w="http://wave.google.com/extensions/robots/1.0">\n'
        '<w:version>1</w:version>\n'
        '<w:protocolversion>%s</w:protocolversion>\n'
        '<w:capabilities>\n</w:capabilities>\n'
        '</w:robot>\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 = (
        '<?xml version="1.0"?>\n'
        '<w:robot xmlns:w="http://wave.google.com/extensions/robots/1.0">\n'
        '<w:version>1</w:version>\n'
        '<w:protocolversion>%s</w:protocolversion>\n'
        '<w:capabilities>\n</w:capabilities>\n'
        '</w:robot>\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 = (
        '<?xml version="1.0"?>\n'
        '<w:robot xmlns:w="http://wave.google.com/extensions/robots/1.0">\n'
        '<w:version>1</w:version>\n'
        '<w:consumer_key>consumer</w:consumer_key>\n'
        '<w:protocolversion>%s</w:protocolversion>\n'
        '<w:capabilities>\n</w:capabilities>\n'
        '</w:robot>\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 = (
        '<?xml version="1.0"?>\n'
        '<w:robot xmlns:w="http://wave.google.com/extensions/robots/1.0">\n'
        '<w:version>1</w:version>\n'
        '<w:protocolversion>%s</w:protocolversion>\n'
        '<w:capabilities>\n'
        '  <w:capability name="%s" context="SELF,ROOT"/>\n'
        '</w:capabilities>\n'
        '</w:robot>\n') % (ops.PROTOCOL_VERSION, events.BlipSubmitted.type)
    xml = self.robot.capabilities_xml()
    self.assertStringsEqual(expected, xml)


if __name__ == '__main__':
  unittest.main()

<filename>mindstormsy-robot/waveapi/robot_test.py
#!/usr/bin/python2.4
#
# Copyright (C) 2009 Google Inc.
#
# 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":["<EMAIL>"],'
               '"waveletId":"test.com!conv+root",'
               '"waveId":"test.com!wdykLROk*11",'
               '"parentBlipId":null,'
               '"version":3,'
               '"creator":"<EMAIL>",'
               '"content":"\\nContent!",'
               '"blipId":"wdykLROk*13","'
               'annotations":[{"range":{"start":0,"end":1},'
                   '"name":"user/e/<EMAIL>","value":"David"}],'
               '"elements":{},'
               '"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":["<EMAIL>","<EMAIL>"],'
                 '"creator":"<EMAIL>",'
                 '"version":5}')

EVENTS_JSON = ('[{"timestamp":1242079611003,'
                '"modifiedBy":"<EMAIL>",'
                '"properties":{"participantsRemoved":[],'
                    '"participantsAdded":["<EMAIL>"]},'
                '"type":"WAVELET_PARTICIPANTS_CHANGED"}]')

TEST_JSON = '{"blips":%s,"wavelet":%s,"events":%s}' % (
    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"}]

NEW_WAVE_JSON_OLD = [{'data':
                      [{'data':
                        {'waveletId': 'googlewave.com!conv+root',
                         'blipId': 'b+VqQXQbZkCP1',
                         'waveId': 'googlewave.com!w+VqQXQbZkCP0'},
                         'id': 'wavelet.create1265055048410'}],
                      'id': 'op10'}];

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.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)
    self.robot.submit = lambda x: NEW_WAVE_JSON_OLD
    new_wave = self.robot.new_wave('googlewave.com', submit=True)
    self.assertEqual('googlewave.com!w+VqQXQbZkCP0', 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': ['<EMAIL>']},
                       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('<b>Hello</b>')

    self.robot.register_handler(events.WaveletParticipantsChanged,
                                check)
    json = self.robot.process_events(TEST_JSON)
    operations = simplejson.loads(json)
    expected = set([ops.ROBOT_NOTIFY_CAPABILITIES_HASH,
                    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_CAPABILITIES_HASH,
                      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 = (
        '<?xml version="1.0"?>\n'
        '<w:robot xmlns:w="http://wave.google.com/extensions/robots/1.0">\n'
        '<w:version>1</w:version>\n'
        '<w:protocolversion>%s</w:protocolversion>\n'
        '<w:capabilities>\n</w:capabilities>\n'
        '</w:robot>\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 = (
        '<?xml version="1.0"?>\n'
        '<w:robot xmlns:w="http://wave.google.com/extensions/robots/1.0">\n'
        '<w:version>1</w:version>\n'
        '<w:protocolversion>%s</w:protocolversion>\n'
        '<w:capabilities>\n</w:capabilities>\n'
        '</w:robot>\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 = (
        '<?xml version="1.0"?>\n'
        '<w:robot xmlns:w="http://wave.google.com/extensions/robots/1.0">\n'
        '<w:version>1</w:version>\n'
        '<w:consumer_key>consumer</w:consumer_key>\n'
        '<w:protocolversion>%s</w:protocolversion>\n'
        '<w:capabilities>\n</w:capabilities>\n'
        '</w:robot>\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 = (
        '<?xml version="1.0"?>\n'
        '<w:robot xmlns:w="http://wave.google.com/extensions/robots/1.0">\n'
        '<w:version>1</w:version>\n'
        '<w:protocolversion>%s</w:protocolversion>\n'
        '<w:capabilities>\n'
        '  <w:capability name="%s" context="SELF,ROOT"/>\n'
        '</w:capabilities>\n'
        '</w:robot>\n') % (ops.PROTOCOL_VERSION, events.BlipSubmitted.type)
    xml = self.robot.capabilities_xml()
    self.assertStringsEqual(expected, xml)


if __name__ == '__main__':
  unittest.main()

#!/usr/bin/python2.4 # # Copyright (C) 2009 Google Inc. # # 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. Tests for testing the basic parsing of json in robots. # Test some basic properties; the rest should be covered by # ops.CreateContext. # there should be one operation indicating the current version: # setup_oauth doesn't work during testing, so heavy handed setting of # properties it is: