summaryrefslogtreecommitdiff
path: root/sleekxmpp/plugins/xep_0050
diff options
context:
space:
mode:
authorFlorent Le Coz <louiz@louiz.org>2014-07-17 14:19:04 +0200
committerFlorent Le Coz <louiz@louiz.org>2014-07-17 14:19:04 +0200
commit5ab77c745270d7d5c016c1dc7ef2a82533a4b16e (patch)
tree259377cc666f8b9c7954fc4e7b8f7a912bcfe101 /sleekxmpp/plugins/xep_0050
parente5582694c07236e6830c20361840360a1dde37f3 (diff)
downloadslixmpp-5ab77c745270d7d5c016c1dc7ef2a82533a4b16e.tar.gz
slixmpp-5ab77c745270d7d5c016c1dc7ef2a82533a4b16e.tar.bz2
slixmpp-5ab77c745270d7d5c016c1dc7ef2a82533a4b16e.tar.xz
slixmpp-5ab77c745270d7d5c016c1dc7ef2a82533a4b16e.zip
Rename to slixmpp
Diffstat (limited to 'sleekxmpp/plugins/xep_0050')
-rw-r--r--sleekxmpp/plugins/xep_0050/__init__.py19
-rw-r--r--sleekxmpp/plugins/xep_0050/adhoc.py688
-rw-r--r--sleekxmpp/plugins/xep_0050/stanza.py185
3 files changed, 0 insertions, 892 deletions
diff --git a/sleekxmpp/plugins/xep_0050/__init__.py b/sleekxmpp/plugins/xep_0050/__init__.py
deleted file mode 100644
index 640b182d..00000000
--- a/sleekxmpp/plugins/xep_0050/__init__.py
+++ /dev/null
@@ -1,19 +0,0 @@
-"""
- SleekXMPP: The Sleek XMPP Library
- Copyright (C) 2011 Nathanael C. Fritz, Lance J.T. Stout
- This file is part of SleekXMPP.
-
- See the file LICENSE for copying permission.
-"""
-
-from sleekxmpp.plugins.base import register_plugin
-
-from sleekxmpp.plugins.xep_0050.stanza import Command
-from sleekxmpp.plugins.xep_0050.adhoc import XEP_0050
-
-
-register_plugin(XEP_0050)
-
-
-# Retain some backwards compatibility
-xep_0050 = XEP_0050
diff --git a/sleekxmpp/plugins/xep_0050/adhoc.py b/sleekxmpp/plugins/xep_0050/adhoc.py
deleted file mode 100644
index e5594c3f..00000000
--- a/sleekxmpp/plugins/xep_0050/adhoc.py
+++ /dev/null
@@ -1,688 +0,0 @@
-"""
- SleekXMPP: The Sleek XMPP Library
- Copyright (C) 2011 Nathanael C. Fritz, Lance J.T. Stout
- This file is part of SleekXMPP.
-
- See the file LICENSE for copying permission.
-"""
-
-import logging
-import time
-
-from sleekxmpp import Iq
-from sleekxmpp.exceptions import IqError
-from sleekxmpp.xmlstream.handler import Callback
-from sleekxmpp.xmlstream.matcher import StanzaPath
-from sleekxmpp.xmlstream import register_stanza_plugin, JID
-from sleekxmpp.plugins import BasePlugin
-from sleekxmpp.plugins.xep_0050 import stanza
-from sleekxmpp.plugins.xep_0050 import Command
-from sleekxmpp.plugins.xep_0004 import Form
-
-
-log = logging.getLogger(__name__)
-
-
-class XEP_0050(BasePlugin):
-
- """
- XEP-0050: Ad-Hoc Commands
-
- XMPP's Adhoc Commands provides a generic workflow mechanism for
- interacting with applications. The result is similar to menu selections
- and multi-step dialogs in normal desktop applications. Clients do not
- need to know in advance what commands are provided by any particular
- application or agent. While adhoc commands provide similar functionality
- to Jabber-RPC, adhoc commands are used primarily for human interaction.
-
- Also see <http://xmpp.org/extensions/xep-0050.html>
-
- Configuration Values:
- threaded -- Indicates if command events should be threaded.
- Defaults to True.
-
- Events:
- command_execute -- Received a command with action="execute"
- command_next -- Received a command with action="next"
- command_complete -- Received a command with action="complete"
- command_cancel -- Received a command with action="cancel"
-
- Attributes:
- threaded -- Indicates if command events should be threaded.
- Defaults to True.
- commands -- A dictionary mapping JID/node pairs to command
- names and handlers.
- sessions -- A dictionary or equivalent backend mapping
- session IDs to dictionaries containing data
- relevant to a command's session.
-
- Methods:
- plugin_init -- Overrides base_plugin.plugin_init
- post_init -- Overrides base_plugin.post_init
- new_session -- Return a new session ID.
- prep_handlers -- Placeholder. May call with a list of handlers
- to prepare them for use with the session storage
- backend, if needed.
- set_backend -- Replace the default session storage with some
- external storage mechanism, such as a database.
- The provided backend wrapper must be able to
- act using the same syntax as a dictionary.
- add_command -- Add a command for use by external entitites.
- get_commands -- Retrieve a list of commands provided by a
- remote agent.
- send_command -- Send a command request to a remote agent.
- start_command -- Command user API: initiate a command session
- continue_command -- Command user API: proceed to the next step
- cancel_command -- Command user API: cancel a command
- complete_command -- Command user API: finish a command
- terminate_command -- Command user API: delete a command's session
- """
-
- name = 'xep_0050'
- description = 'XEP-0050: Ad-Hoc Commands'
- dependencies = set(['xep_0030', 'xep_0004'])
- stanza = stanza
- default_config = {
- 'threaded': True,
- 'session_db': None
- }
-
- def plugin_init(self):
- """Start the XEP-0050 plugin."""
- self.sessions = self.session_db
- if self.sessions is None:
- self.sessions = {}
-
- self.commands = {}
-
- self.xmpp.register_handler(
- Callback("Ad-Hoc Execute",
- StanzaPath('iq@type=set/command'),
- self._handle_command))
-
- register_stanza_plugin(Iq, Command)
- register_stanza_plugin(Command, Form)
-
- self.xmpp.add_event_handler('command_execute',
- self._handle_command_start,
- threaded=self.threaded)
- self.xmpp.add_event_handler('command_next',
- self._handle_command_next,
- threaded=self.threaded)
- self.xmpp.add_event_handler('command_cancel',
- self._handle_command_cancel,
- threaded=self.threaded)
- self.xmpp.add_event_handler('command_complete',
- self._handle_command_complete,
- threaded=self.threaded)
-
- def plugin_end(self):
- self.xmpp.del_event_handler('command_execute',
- self._handle_command_start)
- self.xmpp.del_event_handler('command_next',
- self._handle_command_next)
- self.xmpp.del_event_handler('command_cancel',
- self._handle_command_cancel)
- self.xmpp.del_event_handler('command_complete',
- self._handle_command_complete)
- self.xmpp.remove_handler('Ad-Hoc Execute')
- self.xmpp['xep_0030'].del_feature(feature=Command.namespace)
- self.xmpp['xep_0030'].set_items(node=Command.namespace, items=tuple())
-
- def session_bind(self, jid):
- self.xmpp['xep_0030'].add_feature(Command.namespace)
- self.xmpp['xep_0030'].set_items(node=Command.namespace, items=tuple())
-
- def set_backend(self, db):
- """
- Replace the default session storage dictionary with
- a generic, external data storage mechanism.
-
- The replacement backend must be able to interact through
- the same syntax and interfaces as a normal dictionary.
-
- Arguments:
- db -- The new session storage mechanism.
- """
- self.sessions = db
-
- def prep_handlers(self, handlers, **kwargs):
- """
- Prepare a list of functions for use by the backend service.
-
- Intended to be replaced by the backend service as needed.
-
- Arguments:
- handlers -- A list of function pointers
- **kwargs -- Any additional parameters required by the backend.
- """
- pass
-
- # =================================================================
- # Server side (command provider) API
-
- def add_command(self, jid=None, node=None, name='', handler=None):
- """
- Make a new command available to external entities.
-
- Access control may be implemented in the provided handler.
-
- Command workflow is done across a sequence of command handlers. The
- first handler is given the initial Iq stanza of the request in order
- to support access control. Subsequent handlers are given only the
- payload items of the command. All handlers will receive the command's
- session data.
-
- Arguments:
- jid -- The JID that will expose the command.
- node -- The node associated with the command.
- name -- A human readable name for the command.
- handler -- A function that will generate the response to the
- initial command request, as well as enforcing any
- access control policies.
- """
- if jid is None:
- jid = self.xmpp.boundjid
- elif not isinstance(jid, JID):
- jid = JID(jid)
- item_jid = jid.full
-
- self.xmpp['xep_0030'].add_identity(category='automation',
- itype='command-list',
- name='Ad-Hoc commands',
- node=Command.namespace,
- jid=jid)
- self.xmpp['xep_0030'].add_item(jid=item_jid,
- name=name,
- node=Command.namespace,
- subnode=node,
- ijid=jid)
- self.xmpp['xep_0030'].add_identity(category='automation',
- itype='command-node',
- name=name,
- node=node,
- jid=jid)
- self.xmpp['xep_0030'].add_feature(Command.namespace, None, jid)
-
- self.commands[(item_jid, node)] = (name, handler)
-
- def new_session(self):
- """Return a new session ID."""
- return str(time.time()) + '-' + self.xmpp.new_id()
-
- def _handle_command(self, iq):
- """Raise command events based on the command action."""
- self.xmpp.event('command_%s' % iq['command']['action'], iq)
-
- def _handle_command_start(self, iq):
- """
- Process an initial request to execute a command.
-
- Arguments:
- iq -- The command execution request.
- """
- sessionid = self.new_session()
- node = iq['command']['node']
- key = (iq['to'].full, node)
- name, handler = self.commands.get(key, ('Not found', None))
- if not handler:
- log.debug('Command not found: %s, %s', key, self.commands)
-
- payload = []
- for stanza in iq['command']['substanzas']:
- payload.append(stanza)
-
- if len(payload) == 1:
- payload = payload[0]
-
- interfaces = set([item.plugin_attrib for item in payload])
- payload_classes = set([item.__class__ for item in payload])
-
- initial_session = {'id': sessionid,
- 'from': iq['from'],
- 'to': iq['to'],
- 'node': node,
- 'payload': payload,
- 'interfaces': interfaces,
- 'payload_classes': payload_classes,
- 'notes': None,
- 'has_next': False,
- 'allow_complete': False,
- 'allow_prev': False,
- 'past': [],
- 'next': None,
- 'prev': None,
- 'cancel': None}
-
- session = handler(iq, initial_session)
-
- self._process_command_response(iq, session)
-
- def _handle_command_next(self, iq):
- """
- Process a request for the next step in the workflow
- for a command with multiple steps.
-
- Arguments:
- iq -- The command continuation request.
- """
- sessionid = iq['command']['sessionid']
- session = self.sessions.get(sessionid)
-
- if session:
- handler = session['next']
- interfaces = session['interfaces']
- results = []
- for stanza in iq['command']['substanzas']:
- if stanza.plugin_attrib in interfaces:
- results.append(stanza)
- if len(results) == 1:
- results = results[0]
-
- session = handler(results, session)
-
- self._process_command_response(iq, session)
- else:
- raise XMPPError('item-not-found')
-
- def _handle_command_prev(self, iq):
- """
- Process a request for the prev step in the workflow
- for a command with multiple steps.
-
- Arguments:
- iq -- The command continuation request.
- """
- sessionid = iq['command']['sessionid']
- session = self.sessions.get(sessionid)
-
- if session:
- handler = session['prev']
- interfaces = session['interfaces']
- results = []
- for stanza in iq['command']['substanzas']:
- if stanza.plugin_attrib in interfaces:
- results.append(stanza)
- if len(results) == 1:
- results = results[0]
-
- session = handler(results, session)
-
- self._process_command_response(iq, session)
- else:
- raise XMPPError('item-not-found')
-
- def _process_command_response(self, iq, session):
- """
- Generate a command reply stanza based on the
- provided session data.
-
- Arguments:
- iq -- The command request stanza.
- session -- A dictionary of relevant session data.
- """
- sessionid = session['id']
-
- payload = session['payload']
- if payload is None:
- payload = []
- if not isinstance(payload, list):
- payload = [payload]
-
- interfaces = session.get('interfaces', set())
- payload_classes = session.get('payload_classes', set())
-
- interfaces.update(set([item.plugin_attrib for item in payload]))
- payload_classes.update(set([item.__class__ for item in payload]))
-
- session['interfaces'] = interfaces
- session['payload_classes'] = payload_classes
-
- self.sessions[sessionid] = session
-
- for item in payload:
- register_stanza_plugin(Command, item.__class__, iterable=True)
-
- iq.reply()
- iq['command']['node'] = session['node']
- iq['command']['sessionid'] = session['id']
-
- if session['next'] is None:
- iq['command']['actions'] = []
- iq['command']['status'] = 'completed'
- elif session['has_next']:
- actions = ['next']
- if session['allow_complete']:
- actions.append('complete')
- if session['allow_prev']:
- actions.append('prev')
- iq['command']['actions'] = actions
- iq['command']['status'] = 'executing'
- else:
- iq['command']['actions'] = ['complete']
- iq['command']['status'] = 'executing'
-
- iq['command']['notes'] = session['notes']
-
- for item in payload:
- iq['command'].append(item)
-
- iq.send()
-
- def _handle_command_cancel(self, iq):
- """
- Process a request to cancel a command's execution.
-
- Arguments:
- iq -- The command cancellation request.
- """
- node = iq['command']['node']
- sessionid = iq['command']['sessionid']
-
- session = self.sessions.get(sessionid)
-
- if session:
- handler = session['cancel']
- if handler:
- handler(iq, session)
- del self.sessions[sessionid]
- iq.reply()
- iq['command']['node'] = node
- iq['command']['sessionid'] = sessionid
- iq['command']['status'] = 'canceled'
- iq['command']['notes'] = session['notes']
- iq.send()
- else:
- raise XMPPError('item-not-found')
-
-
- def _handle_command_complete(self, iq):
- """
- Process a request to finish the execution of command
- and terminate the workflow.
-
- All data related to the command session will be removed.
-
- Arguments:
- iq -- The command completion request.
- """
- node = iq['command']['node']
- sessionid = iq['command']['sessionid']
- session = self.sessions.get(sessionid)
-
- if session:
- handler = session['next']
- interfaces = session['interfaces']
- results = []
- for stanza in iq['command']['substanzas']:
- if stanza.plugin_attrib in interfaces:
- results.append(stanza)
- if len(results) == 1:
- results = results[0]
-
- if handler:
- handler(results, session)
-
- del self.sessions[sessionid]
-
- iq.reply()
- iq['command']['node'] = node
- iq['command']['sessionid'] = sessionid
- iq['command']['actions'] = []
- iq['command']['status'] = 'completed'
- iq['command']['notes'] = session['notes']
- iq.send()
- else:
- raise XMPPError('item-not-found')
-
- # =================================================================
- # Client side (command user) API
-
- def get_commands(self, jid, **kwargs):
- """
- Return a list of commands provided by a given JID.
-
- Arguments:
- jid -- The JID to query for commands.
- local -- If true, then the query is for a JID/node
- combination handled by this Sleek instance and
- no stanzas need to be sent.
- Otherwise, a disco stanza must be sent to the
- remove JID to retrieve the items.
- ifrom -- Specifiy the sender's JID.
- block -- If true, block and wait for the stanzas' reply.
- timeout -- The time in seconds to block while waiting for
- a reply. If None, then wait indefinitely.
- callback -- Optional callback to execute when a reply is
- received instead of blocking and waiting for
- the reply.
- iterator -- If True, return a result set iterator using
- the XEP-0059 plugin, if the plugin is loaded.
- Otherwise the parameter is ignored.
- """
- return self.xmpp['xep_0030'].get_items(jid=jid,
- node=Command.namespace,
- **kwargs)
-
- def send_command(self, jid, node, ifrom=None, action='execute',
- payload=None, sessionid=None, flow=False, **kwargs):
- """
- Create and send a command stanza, without using the provided
- workflow management APIs.
-
- Arguments:
- jid -- The JID to send the command request or result.
- node -- The node for the command.
- ifrom -- Specify the sender's JID.
- action -- May be one of: execute, cancel, complete,
- or cancel.
- payload -- Either a list of payload items, or a single
- payload item such as a data form.
- sessionid -- The current session's ID value.
- flow -- If True, process the Iq result using the
- command workflow methods contained in the
- session instead of returning the response
- stanza itself. Defaults to False.
- block -- Specify if the send call will block until a
- response is received, or a timeout occurs.
- Defaults to True.
- timeout -- The length of time (in seconds) to wait for a
- response before exiting the send call
- if blocking is used. Defaults to
- sleekxmpp.xmlstream.RESPONSE_TIMEOUT
- callback -- Optional reference to a stream handler
- function. Will be executed when a reply
- stanza is received if flow=False.
- """
- iq = self.xmpp.Iq()
- iq['type'] = 'set'
- iq['to'] = jid
- iq['from'] = ifrom
- iq['command']['node'] = node
- iq['command']['action'] = action
- if sessionid is not None:
- iq['command']['sessionid'] = sessionid
- if payload is not None:
- if not isinstance(payload, list):
- payload = [payload]
- for item in payload:
- iq['command'].append(item)
- if not flow:
- return iq.send(**kwargs)
- else:
- if kwargs.get('block', True):
- try:
- result = iq.send(**kwargs)
- except IqError as err:
- result = err.iq
- self._handle_command_result(result)
- else:
- iq.send(block=False, callback=self._handle_command_result)
-
- def start_command(self, jid, node, session, ifrom=None, block=False):
- """
- Initiate executing a command provided by a remote agent.
-
- The default workflow provided is non-blocking, but a blocking
- version may be used with block=True.
-
- The provided session dictionary should contain:
- next -- A handler for processing the command result.
- error -- A handler for processing any error stanzas
- generated by the request.
-
- Arguments:
- jid -- The JID to send the command request.
- node -- The node for the desired command.
- session -- A dictionary of relevant session data.
- ifrom -- Optionally specify the sender's JID.
- block -- If True, block execution until a result
- is received. Defaults to False.
- """
- session['jid'] = jid
- session['node'] = node
- session['timestamp'] = time.time()
- session['block'] = block
- if 'payload' not in session:
- session['payload'] = None
-
- iq = self.xmpp.Iq()
- iq['type'] = 'set'
- iq['to'] = jid
- iq['from'] = ifrom
- session['from'] = ifrom
- iq['command']['node'] = node
- iq['command']['action'] = 'execute'
- if session['payload'] is not None:
- payload = session['payload']
- if not isinstance(payload, list):
- payload = list(payload)
- for stanza in payload:
- iq['command'].append(stanza)
- sessionid = 'client:pending_' + iq['id']
- session['id'] = sessionid
- self.sessions[sessionid] = session
- if session['block']:
- try:
- result = iq.send(block=True)
- except IqError as err:
- result = err.iq
- self._handle_command_result(result)
- else:
- iq.send(block=False, callback=self._handle_command_result)
-
- def continue_command(self, session, direction='next'):
- """
- Execute the next action of the command.
-
- Arguments:
- session -- All stored data relevant to the current
- command session.
- """
- sessionid = 'client:' + session['id']
- self.sessions[sessionid] = session
-
- self.send_command(session['jid'],
- session['node'],
- ifrom=session.get('from', None),
- action=direction,
- payload=session.get('payload', None),
- sessionid=session['id'],
- flow=True,
- block=session['block'])
-
- def cancel_command(self, session):
- """
- Cancel the execution of a command.
-
- Arguments:
- session -- All stored data relevant to the current
- command session.
- """
- sessionid = 'client:' + session['id']
- self.sessions[sessionid] = session
-
- self.send_command(session['jid'],
- session['node'],
- ifrom=session.get('from', None),
- action='cancel',
- payload=session.get('payload', None),
- sessionid=session['id'],
- flow=True,
- block=session['block'])
-
- def complete_command(self, session):
- """
- Finish the execution of a command workflow.
-
- Arguments:
- session -- All stored data relevant to the current
- command session.
- """
- sessionid = 'client:' + session['id']
- self.sessions[sessionid] = session
-
- self.send_command(session['jid'],
- session['node'],
- ifrom=session.get('from', None),
- action='complete',
- payload=session.get('payload', None),
- sessionid=session['id'],
- flow=True,
- block=session['block'])
-
- def terminate_command(self, session):
- """
- Delete a command's session after a command has completed
- or an error has occured.
-
- Arguments:
- session -- All stored data relevant to the current
- command session.
- """
- sessionid = 'client:' + session['id']
- try:
- del self.sessions[sessionid]
- except Exception as e:
- log.error("Error deleting adhoc command session: %s" % e.message)
-
- def _handle_command_result(self, iq):
- """
- Process the results of a command request.
-
- Will execute the 'next' handler stored in the session
- data, or the 'error' handler depending on the Iq's type.
-
- Arguments:
- iq -- The command response.
- """
- sessionid = 'client:' + iq['command']['sessionid']
- pending = False
-
- if sessionid not in self.sessions:
- pending = True
- pendingid = 'client:pending_' + iq['id']
- if pendingid not in self.sessions:
- return
- sessionid = pendingid
-
- session = self.sessions[sessionid]
- sessionid = 'client:' + iq['command']['sessionid']
- session['id'] = iq['command']['sessionid']
-
- self.sessions[sessionid] = session
-
- if pending:
- del self.sessions[pendingid]
-
- handler_type = 'next'
- if iq['type'] == 'error':
- handler_type = 'error'
- handler = session.get(handler_type, None)
- if handler:
- handler(iq, session)
- elif iq['type'] == 'error':
- self.terminate_command(session)
-
- if iq['command']['status'] == 'completed':
- self.terminate_command(session)
diff --git a/sleekxmpp/plugins/xep_0050/stanza.py b/sleekxmpp/plugins/xep_0050/stanza.py
deleted file mode 100644
index 2367c77b..00000000
--- a/sleekxmpp/plugins/xep_0050/stanza.py
+++ /dev/null
@@ -1,185 +0,0 @@
-"""
- SleekXMPP: The Sleek XMPP Library
- Copyright (C) 2011 Nathanael C. Fritz, Lance J.T. Stout
- This file is part of SleekXMPP.
-
- See the file LICENSE for copying permission.
-"""
-
-from sleekxmpp.xmlstream import ElementBase, ET
-
-
-class Command(ElementBase):
-
- """
- XMPP's Adhoc Commands provides a generic workflow mechanism for
- interacting with applications. The result is similar to menu selections
- and multi-step dialogs in normal desktop applications. Clients do not
- need to know in advance what commands are provided by any particular
- application or agent. While adhoc commands provide similar functionality
- to Jabber-RPC, adhoc commands are used primarily for human interaction.
-
- Also see <http://xmpp.org/extensions/xep-0050.html>
-
- Example command stanzas:
- <iq type="set">
- <command xmlns="http://jabber.org/protocol/commands"
- node="run_foo"
- action="execute" />
- </iq>
-
- <iq type="result">
- <command xmlns="http://jabber.org/protocol/commands"
- node="run_foo"
- sessionid="12345"
- status="executing">
- <actions>
- <complete />
- </actions>
- <note type="info">Information!</note>
- <x xmlns="jabber:x:data">
- <field var="greeting"
- type="text-single"
- label="Greeting" />
- </x>
- </command>
- </iq>
-
- Stanza Interface:
- action -- The action to perform.
- actions -- The set of allowable next actions.
- node -- The node associated with the command.
- notes -- A list of tuples for informative notes.
- sessionid -- A unique identifier for a command session.
- status -- May be one of: canceled, completed, or executing.
-
- Attributes:
- actions -- A set of allowed action values.
- statuses -- A set of allowed status values.
- next_actions -- A set of allowed next action names.
-
- Methods:
- get_action -- Return the requested action.
- get_actions -- Return the allowable next actions.
- set_actions -- Set the allowable next actions.
- del_actions -- Remove the current set of next actions.
- get_notes -- Return a list of informative note data.
- set_notes -- Set informative notes.
- del_notes -- Remove any note data.
- add_note -- Add a single note.
- """
-
- name = 'command'
- namespace = 'http://jabber.org/protocol/commands'
- plugin_attrib = 'command'
- interfaces = set(('action', 'sessionid', 'node',
- 'status', 'actions', 'notes'))
- actions = set(('cancel', 'complete', 'execute', 'next', 'prev'))
- statuses = set(('canceled', 'completed', 'executing'))
- next_actions = set(('prev', 'next', 'complete'))
-
- def get_action(self):
- """
- Return the value of the action attribute.
-
- If the Iq stanza's type is "set" then use a default
- value of "execute".
- """
- if self.parent()['type'] == 'set':
- return self._get_attr('action', default='execute')
- return self._get_attr('action')
-
- def set_actions(self, values):
- """
- Assign the set of allowable next actions.
-
- Arguments:
- values -- A list containing any combination of:
- 'prev', 'next', and 'complete'
- """
- self.del_actions()
- if values:
- self._set_sub_text('{%s}actions' % self.namespace, '', True)
- actions = self.find('{%s}actions' % self.namespace)
- for val in values:
- if val in self.next_actions:
- action = ET.Element('{%s}%s' % (self.namespace, val))
- actions.append(action)
-
- def get_actions(self):
- """
- Return the set of allowable next actions.
- """
- actions = set()
- actions_xml = self.find('{%s}actions' % self.namespace)
- if actions_xml is not None:
- for action in self.next_actions:
- action_xml = actions_xml.find('{%s}%s' % (self.namespace,
- action))
- if action_xml is not None:
- actions.add(action)
- return actions
-
- def del_actions(self):
- """
- Remove all allowable next actions.
- """
- self._del_sub('{%s}actions' % self.namespace)
-
- def get_notes(self):
- """
- Return a list of note information.
-
- Example:
- [('info', 'Some informative data'),
- ('warning', 'Use caution'),
- ('error', 'The command ran, but had errors')]
- """
- notes = []
- notes_xml = self.findall('{%s}note' % self.namespace)
- for note in notes_xml:
- notes.append((note.attrib.get('type', 'info'),
- note.text))
- return notes
-
- def set_notes(self, notes):
- """
- Add multiple notes to the command result.
-
- Each note is a tuple, with the first item being one of:
- 'info', 'warning', or 'error', and the second item being
- any human readable message.
-
- Example:
- [('info', 'Some informative data'),
- ('warning', 'Use caution'),
- ('error', 'The command ran, but had errors')]
-
-
- Arguments:
- notes -- A list of tuples of note information.
- """
- self.del_notes()
- for note in notes:
- self.add_note(note[1], note[0])
-
- def del_notes(self):
- """
- Remove all notes associated with the command result.
- """
- notes_xml = self.findall('{%s}note' % self.namespace)
- for note in notes_xml:
- self.xml.remove(note)
-
- def add_note(self, msg='', ntype='info'):
- """
- Add a single note annotation to the command.
-
- Arguments:
- msg -- A human readable message.
- ntype -- One of: 'info', 'warning', 'error'
- """
- xml = ET.Element('{%s}note' % self.namespace)
- xml.attrib['type'] = ntype
- xml.text = msg
- self.xml.append(xml)