slixmpp/tests/test_stream_xep_0050.py

import time
import logging

import unittest
from slixmpp.test import SlixTest
from slixmpp.xmlstream import ElementBase, register_stanza_plugin


class TestAdHocCommands(SlixTest):

    def setUp(self):
        self.stream_start(mode='client',
                          plugins=['xep_0030', 'xep_0004', 'xep_0050'])

        # Real session IDs don't make for nice tests, so use
        # a dummy value.
        self.xmpp['xep_0050'].new_session = lambda: '_sessionid_'

    def tearDown(self):
        self.stream_close()

    def testInitialPayloadCommand(self):
        """Test a command with an initial payload."""

        class TestPayload(ElementBase):
            name = 'foo'
            namespace = 'test'
            interfaces = {'bar'}
            plugin_attrib = name

        Command = self.xmpp['xep_0050'].stanza.Command
        register_stanza_plugin(Command, TestPayload, iterable=True)

        def handle_command(iq, session):
            initial = session['payload']
            logging.debug(initial)
            new_payload = TestPayload()
            if initial:
                new_payload['bar'] = 'Received: %s' % initial['bar']
            else:
                new_payload['bar'] = 'Failed'

            logging.debug(initial)

            session['payload'] = new_payload
            session['next'] = None
            session['has_next'] = False

            return session

        self.xmpp['xep_0050'].add_command('tester@localhost', 'foo',
                                          'Do Foo', handle_command)

        self.recv("""
          <iq id="11" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="execute">
              <foo xmlns="test" bar="baz" />
            </command>
          </iq>
        """)

        self.send("""
          <iq id="11" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="completed"
                     sessionid="_sessionid_">
              <foo xmlns="test" bar="Received: baz" />
            </command>
          </iq>
        """)

    def testZeroStepCommand(self):
        """Test running a command with no steps."""

        def handle_command(iq, session):
            form = self.xmpp['xep_0004'].make_form(ftype='result')
            form.addField(var='foo', ftype='text-single',
                          label='Foo', value='bar')

            session['payload'] = form
            session['next'] = None
            session['has_next'] = False

            return session

        self.xmpp['xep_0050'].add_command('tester@localhost', 'foo',
                                          'Do Foo', handle_command)

        self.recv("""
          <iq id="11" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="execute" />
          </iq>
        """)

        self.send("""
          <iq id="11" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="completed"
                     sessionid="_sessionid_">
              <x xmlns="jabber:x:data" type="result">
                <field var="foo" label="Foo" type="text-single">
                  <value>bar</value>
                </field>
              </x>
            </command>
          </iq>
        """)

    def testOneStepCommand(self):
        """Test running a single step command."""
        results = []

        def handle_command(iq, session):

            def handle_form(form, session):
                results.append(form.get_values()['foo'])
                session['payload'] = None

            form = self.xmpp['xep_0004'].make_form('form')
            form.addField(var='foo', ftype='text-single', label='Foo')

            session['payload'] = form
            session['next'] = handle_form
            session['has_next'] = False

            return session

        self.xmpp['xep_0050'].add_command('tester@localhost', 'foo',
                                          'Do Foo', handle_command)

        self.recv("""
          <iq id="11" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="execute" />
          </iq>
        """)

        self.send("""
          <iq id="11" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="executing"
                     sessionid="_sessionid_">
              <actions>
                <complete />
              </actions>
              <x xmlns="jabber:x:data" type="form">
                <field var="foo" label="Foo" type="text-single" />
              </x>
            </command>
          </iq>
        """)

        self.recv("""
          <iq id="12" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="complete"
                     sessionid="_sessionid_">
              <x xmlns="jabber:x:data" type="submit">
                <field var="foo" label="Foo" type="text-single">
                  <value>blah</value>
                </field>
              </x>
            </command>
          </iq>
        """)

        self.send("""
          <iq id="12" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="completed"
                     sessionid="_sessionid_" />
          </iq>
        """)

        self.assertEqual(results, ['blah'],
                "Command handler was not executed: %s" % results)

    def testTwoStepCommand(self):
        """Test using a two-stage command."""
        results = []

        def handle_command(iq, session):

            def handle_step2(form, session):
                results.append(form.get_values()['bar'])
                session['payload'] = None

            def handle_step1(form, session):
                results.append(form.get_values()['foo'])

                form = self.xmpp['xep_0004'].make_form('form')
                form.addField(var='bar', ftype='text-single', label='Bar')

                session['payload'] = form
                session['next'] = handle_step2
                session['has_next'] = False

                return session

            form = self.xmpp['xep_0004'].make_form('form')
            form.addField(var='foo', ftype='text-single', label='Foo')

            session['payload'] = form
            session['next'] = handle_step1
            session['has_next'] = True

            return session

        self.xmpp['xep_0050'].add_command('tester@localhost', 'foo',
                                          'Do Foo', handle_command)

        self.recv("""
          <iq id="11" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="execute" />
          </iq>
        """)

        self.send("""
          <iq id="11" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="executing"
                     sessionid="_sessionid_">
              <actions>
                <next />
              </actions>
              <x xmlns="jabber:x:data" type="form">
                <field var="foo" label="Foo" type="text-single" />
              </x>
            </command>
          </iq>
        """)

        self.recv("""
          <iq id="12" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="next"
                     sessionid="_sessionid_">
              <x xmlns="jabber:x:data" type="submit">
                <field var="foo" label="Foo" type="text-single">
                  <value>blah</value>
                </field>
              </x>
            </command>
          </iq>
        """)

        self.send("""
          <iq id="12" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="executing"
                     sessionid="_sessionid_">
              <actions>
                <complete />
              </actions>
              <x xmlns="jabber:x:data" type="form">
                <field var="bar" label="Bar" type="text-single" />
              </x>
            </command>
          </iq>
        """)

        self.recv("""
          <iq id="13" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="complete"
                     sessionid="_sessionid_">
              <x xmlns="jabber:x:data" type="submit">
                <field var="bar" label="Bar" type="text-single">
                  <value>meh</value>
                </field>
              </x>
            </command>
          </iq>
        """)
        self.send("""
          <iq id="13" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="completed"
                     sessionid="_sessionid_" />
          </iq>
        """)

        self.assertEqual(results, ['blah', 'meh'],
                "Command handler was not executed: %s" % results)

    def testCancelCommand(self):
        """Test canceling command."""
        results = []

        def handle_command(iq, session):

            def handle_form(form, session):
                results.append(form['values']['foo'])

            def handle_cancel(iq, session):
                results.append('canceled')

            form = self.xmpp['xep_0004'].make_form('form')
            form.addField(var='foo', ftype='text-single', label='Foo')

            session['payload'] = form
            session['next'] = handle_form
            session['cancel'] = handle_cancel
            session['has_next'] = False

            return session

        self.xmpp['xep_0050'].add_command('tester@localhost', 'foo',
                                          'Do Foo', handle_command)

        self.recv("""
          <iq id="11" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="execute" />
          </iq>
        """)

        self.send("""
          <iq id="11" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="executing"
                     sessionid="_sessionid_">
              <actions>
                <complete />
              </actions>
              <x xmlns="jabber:x:data" type="form">
                <field var="foo" label="Foo" type="text-single" />
              </x>
            </command>
          </iq>
        """)

        self.recv("""
          <iq id="12" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="cancel"
                     sessionid="_sessionid_">
              <x xmlns="jabber:x:data" type="submit">
                <field var="foo" label="Foo" type="text-single">
                  <value>blah</value>
                </field>
              </x>
            </command>
          </iq>
        """)

        self.send("""
          <iq id="12" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="canceled"
                     sessionid="_sessionid_" />
          </iq>
        """)

        self.assertEqual(results, ['canceled'],
                "Cancelation handler not executed: %s" % results)

    def testCommandNote(self):
        """Test adding notes to commands."""

        def handle_command(iq, session):
            form = self.xmpp['xep_0004'].make_form(ftype='result')
            form.addField(var='foo', ftype='text-single',
                          label='Foo', value='bar')

            session['payload'] = form
            session['next'] = None
            session['has_next'] = False
            session['notes'] = [('info', 'testing notes')]

            return session

        self.xmpp['xep_0050'].add_command('tester@localhost', 'foo',
                                          'Do Foo', handle_command)

        self.recv("""
          <iq id="11" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="execute" />
          </iq>
        """)

        self.send("""
          <iq id="11" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="completed"
                     sessionid="_sessionid_">
              <note type="info">testing notes</note>
              <x xmlns="jabber:x:data" type="result">
                <field var="foo" label="Foo" type="text-single">
                  <value>bar</value>
                </field>
              </x>
            </command>
          </iq>
        """)



    def testMultiPayloads(self):
        """Test using commands with multiple payloads."""
        results = []

        def handle_command(iq, session):

            def handle_form(forms, session):
                for form in forms:
                    results.append(form.get_values()['FORM_TYPE'])
                session['payload'] = None

            form1 = self.xmpp['xep_0004'].make_form('form')
            form1.addField(var='FORM_TYPE', ftype='hidden', value='form_1')
            form1.addField(var='foo', ftype='text-single', label='Foo')

            form2 = self.xmpp['xep_0004'].make_form('form')
            form2.addField(var='FORM_TYPE', ftype='hidden', value='form_2')
            form2.addField(var='foo', ftype='text-single', label='Foo')

            session['payload'] = [form1, form2]
            session['next'] = handle_form
            session['has_next'] = False

            return session

        self.xmpp['xep_0050'].add_command('tester@localhost', 'foo',
                                          'Do Foo', handle_command)

        self.recv("""
          <iq id="11" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="execute" />
          </iq>
        """)

        self.send("""
          <iq id="11" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="executing"
                     sessionid="_sessionid_">
              <actions>
                <complete />
              </actions>
              <x xmlns="jabber:x:data" type="form">
                <field var="FORM_TYPE" type="hidden">
                  <value>form_1</value>
                </field>
                <field var="foo" label="Foo" type="text-single" />
              </x>
              <x xmlns="jabber:x:data" type="form">
                <field var="FORM_TYPE" type="hidden">
                  <value>form_2</value>
                </field>
                <field var="foo" label="Foo" type="text-single" />
              </x>
            </command>
          </iq>
        """)

        self.recv("""
          <iq id="12" type="set" to="tester@localhost" from="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     action="complete"
                     sessionid="_sessionid_">
             <x xmlns="jabber:x:data" type="submit">
                <field var="FORM_TYPE" type="hidden">
                  <value>form_1</value>
                </field>
                <field var="foo" type="text-single">
                  <value>bar</value>
                </field>
              </x>
              <x xmlns="jabber:x:data" type="submit">
                <field var="FORM_TYPE" type="hidden">
                  <value>form_2</value>
                </field>
                <field var="foo" type="text-single">
                  <value>bar</value>
                </field>
              </x>
            </command>
          </iq>
        """)

        self.send("""
          <iq id="12" type="result" to="foo@bar">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="foo"
                     status="completed"
                     sessionid="_sessionid_" />
          </iq>
        """)

        self.assertEqual(results, [['form_1'], ['form_2']],
                "Command handler was not executed: %s" % results)

    def testClientAPI(self):
        """Test using client-side API for commands."""
        results = []

        def handle_complete(iq, session):
            for item in session['custom_data']:
                results.append(item)

        def handle_step2(iq, session):
            form = self.xmpp['xep_0004'].make_form(ftype='submit')
            form.addField(var='bar', value='123')

            session['custom_data'].append('baz')
            session['payload'] = form
            session['next'] = handle_complete
            self.xmpp['xep_0050'].complete_command(session)

        def handle_step1(iq, session):
            form = self.xmpp['xep_0004'].make_form(ftype='submit')
            form.addField(var='foo', value='42')

            session['custom_data'].append('bar')
            session['payload'] = form
            session['next'] = handle_step2
            self.xmpp['xep_0050'].continue_command(session)

        session = {'custom_data': ['foo'],
                   'next': handle_step1}

        self.xmpp['xep_0050'].start_command(
                'foo@example.com',
                'test_client',
                session)

        self.send("""
          <iq id="1" to="foo@example.com" type="set">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     action="execute" />
          </iq>
        """)

        self.recv("""
          <iq id="1" from="foo@example.com" type="result">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     sessionid="_sessionid_"
                     status="executing">
              <x xmlns="jabber:x:data" type="form">
                <field var="foo" type="text-single" />
              </x>
            </command>
          </iq>
        """)

        self.send("""
          <iq id="2" to="foo@example.com" type="set">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     sessionid="_sessionid_"
                     action="next">
              <x xmlns="jabber:x:data" type="submit">
                <field var="foo">
                  <value>42</value>
                </field>
              </x>
            </command>
          </iq>
        """)

        self.recv("""
          <iq id="2" from="foo@example.com" type="result">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     sessionid="_sessionid_"
                     status="executing">
              <x xmlns="jabber:x:data" type="form">
                <field var="bar" type="text-single" />
              </x>
            </command>
          </iq>
        """)

        self.send("""
          <iq id="3" to="foo@example.com" type="set">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     sessionid="_sessionid_"
                     action="complete">
              <x xmlns="jabber:x:data" type="submit">
                <field var="bar">
                  <value>123</value>
                </field>
              </x>
            </command>
          </iq>
        """)

        self.recv("""
          <iq id="3" from="foo@example.com" type="result">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     sessionid="_sessionid_"
                     status="completed" />
          </iq>
        """)

        self.assertTrue(results == ['foo', 'bar', 'baz'],
                'Incomplete command workflow: %s' % results)

    def testClientAPICancel(self):
        """Test using client-side cancel API for commands."""
        results = []

        def handle_canceled(iq, session):
            for item in session['custom_data']:
                results.append(item)

        def handle_step1(iq, session):
            session['custom_data'].append('bar')
            session['next'] = handle_canceled
            self.xmpp['xep_0050'].cancel_command(session)

        session = {'custom_data': ['foo'],
                   'next': handle_step1}

        self.xmpp['xep_0050'].start_command(
                'foo@example.com',
                'test_client',
                session)

        self.send("""
          <iq id="1" to="foo@example.com" type="set">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     action="execute" />
          </iq>
        """)

        self.recv("""
          <iq id="1" to="foo@example.com" type="result">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     sessionid="_sessionid_"
                     status="executing">
              <x xmlns="jabber:x:data" type="form">
                <field var="foo" type="text-single" />
              </x>
            </command>
          </iq>
        """)

        self.send("""
          <iq id="2" to="foo@example.com" type="set">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     sessionid="_sessionid_"
                     action="cancel" />
          </iq>
        """)

        self.recv("""
          <iq id="2" to="foo@example.com" type="result">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     sessionid="_sessionid_"
                     status="canceled" />
          </iq>
        """)

        self.assertTrue(results == ['foo', 'bar'],
                'Incomplete command workflow: %s' % results)

    def testClientAPIError(self):
        """Test using client-side error API for commands."""
        results = []

        def handle_error(iq, session):
            for item in session['custom_data']:
                results.append(item)

        session = {'custom_data': ['foo'],
                   'error': handle_error}

        self.xmpp['xep_0050'].start_command(
                'foo@example.com',
                'test_client',
                session)

        self.send("""
          <iq id="1" to="foo@example.com" type="set">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     action="execute" />
          </iq>
        """)

        self.recv("""
          <iq id="1" to="foo@example.com" type="error">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     action="execute" />
            <error type='cancel'>
              <item-not-found xmlns='urn:ietf:params:xml:ns:xmpp-stanzas'/>
            </error>
          </iq>
        """)

        self.assertTrue(results == ['foo'],
                'Incomplete command workflow: %s' % results)

    def testClientAPIErrorStrippedResponse(self):
        """Test errors that don't include the command substanza."""
        results = []

        def handle_error(iq, session):
            for item in session['custom_data']:
                results.append(item)

        session = {'custom_data': ['foo'],
                   'error': handle_error}

        self.xmpp['xep_0050'].start_command(
                'foo@example.com',
                'test_client',
                session)

        self.send("""
          <iq id="1" to="foo@example.com" type="set">
            <command xmlns="http://jabber.org/protocol/commands"
                     node="test_client"
                     action="execute" />
          </iq>
        """)

        self.recv("""
          <iq id="1" to="foo@example.com" type="error">
            <error type='cancel'>
              <item-not-found xmlns='urn:ietf:params:xml:ns:xmpp-stanzas' />
            </error>
          </iq>
        """)

        self.assertTrue(results == ['foo'],
                'Incomplete command workflow: %s' % results)


suite = unittest.TestLoader().loadTestsFromTestCase(TestAdHocCommands)