david/jcl
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

end-user-session ad-hoc command implementation

Browse Source 
darcs-hash:20070725200331-86b55-e68cd99383cd82950691fe0724b8b7e481569c10.gz
This commit is contained in:
David Rousselie
2007-07-25 22:03:31 +02:00
parent d82e1f2a0c
commit d8fa96f54c
2 changed files with 142 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
src/jcl/jabber/command.py
View File

@@ -333,7 +333,8 @@ class JCLCommandManager(CommandManager):
result_form.as_xml(command_node)
return []
def execute_add_user_2(self, info_query, session_context, command_node, lang_class):
def execute_add_user_2(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing command 'add-user' step 2")
self.add_actions(command_node, [ACTION_PREVIOUS, ACTION_COMPLETE], 1)
user_jid = session_context["user_jid"]
@@ -361,17 +362,21 @@ class JCLCommandManager(CommandManager):
command_node.setProp("status", STATUS_COMPLETED)
return to_send
def execute_delete_user_1(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing command 'delete-user' step 1")
def select_user_jids_step_1(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing select_user_jids step 1")
self.add_actions(command_node, [ACTION_NEXT])
return self.add_form_select_user_jids(command_node, lang_class)
def execute_delete_user_2(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing command 'delete-user' step 2")
def select_accounts_step_2(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing select_accounts step 2")
self.add_actions(command_node, [ACTION_PREVIOUS, ACTION_COMPLETE], 1)
return self.add_form_select_accounts(session_context, command_node, lang_class)
return self.add_form_select_accounts(session_context, command_node,
lang_class)
execute_delete_user_1 = select_user_jids_step_1
execute_delete_user_2 = select_accounts_step_2
def execute_delete_user_3(self, info_query, session_context,
command_node, lang_class):
@@ -384,17 +389,8 @@ class JCLCommandManager(CommandManager):
command_node.setProp("status", STATUS_COMPLETED)
return result
def execute_disable_user_1(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing command 'disable-user' step 1")
self.add_actions(command_node, [ACTION_NEXT])
return self.add_form_select_user_jids(command_node, lang_class)
def execute_disable_user_2(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing command 'disable-user' step 2")
self.add_actions(command_node, [ACTION_PREVIOUS, ACTION_COMPLETE], 1)
return self.add_form_select_accounts(session_context, command_node, lang_class)
execute_disable_user_1 = select_user_jids_step_1
execute_disable_user_2 = select_accounts_step_2
def execute_disable_user_3(self, info_query, session_context,
command_node, lang_class):
@@ -407,17 +403,8 @@ class JCLCommandManager(CommandManager):
command_node.setProp("status", STATUS_COMPLETED)
return result
def execute_reenable_user_1(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing command 'reenable-user' step 1")
self.add_actions(command_node, [ACTION_NEXT])
return self.add_form_select_user_jids(command_node, lang_class)
def execute_reenable_user_2(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing command 'reenable-user' step 2")
self.add_actions(command_node, [ACTION_PREVIOUS, ACTION_COMPLETE], 1)
return self.add_form_select_accounts(session_context, command_node, lang_class)
execute_reenable_user_1 = select_user_jids_step_1
execute_reenable_user_2 = select_accounts_step_2
def execute_reenable_user_3(self, info_query, session_context,
command_node, lang_class):
@@ -430,8 +417,20 @@ class JCLCommandManager(CommandManager):
command_node.setProp("status", STATUS_COMPLETED)
return result
def execute_end_user_session(self, info_query):
return []
execute_end_user_session_1 = select_user_jids_step_1
execute_end_user_session_2 = select_accounts_step_2
def execute_end_user_session_3(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing command 'end-user-session' step 3")
result = []
for account_name in session_context["account_names"]:
name, user_jid = account_name.split("/", 1)[:2]
_account = account.get_account(user_jid, name)
result += self.component.account_manager.send_presence_unavailable(
_account)
command_node.setProp("status", STATUS_COMPLETED)
return result
def execute_get_user_password(self, info_query):
return []

119
src/jcl/jabber/tests/command.py
View File

@@ -871,14 +871,117 @@ class JCLCommandManager_TestCase(unittest.TestCase):
else:
self.assertFalse(_account.enabled)
# def test_execute_end_user_session(self):
# #TODO : implement command
# info_query = Iq(stanza_type="set",
# from_jid="user1@test.com",
# to_jid="jcl.test.com")
# result = self.command_manager.execute_add_user(info_query)
# self.assertNotEquals(result, None)
# self.assertEquals(len(result), 1)
def test_execute_end_user_session(self):
self.comp.account_manager.account_classes = (ExampleAccount,
Example2Account)
model.db_connect()
account11 = ExampleAccount(user_jid="test1@test.com",
name="account11",
jid="account11@jcl.test.com")
account12 = Example2Account(user_jid="test1@test.com",
name="account12",
jid="account12@jcl.test.com")
account21 = ExampleAccount(user_jid="test2@test.com",
name="account21",
jid="account21@jcl.test.com")
account22 = ExampleAccount(user_jid="test2@test.com",
name="account11",
jid="account11@jcl.test.com")
account31 = ExampleAccount(user_jid="test3@test.com",
name="account31",
jid="account31@jcl.test.com")
account32 = Example2Account(user_jid="test3@test.com",
name="account32",
jid="account32@jcl.test.com")
model.db_disconnect()
info_query = Iq(stanza_type="set",
from_jid="user1@test.com",
to_jid="jcl.test.com")
command_node = info_query.set_new_content(command.COMMAND_NS, "command")
command_node.setProp("node", "http://jabber.org/protocol/admin#end-user-session")
result = self.command_manager.apply_command_action(info_query,
"http://jabber.org/protocol/admin#end-user-session",
"execute")
self.assertNotEquals(result, None)
self.assertEquals(len(result), 1)
xml_command = result[0].xpath_eval("c:command",
{"c": "http://jabber.org/protocol/commands"})[0]
self.assertEquals(xml_command.prop("status"), "executing")
self.assertNotEquals(xml_command.prop("sessionid"), None)
self.__check_actions(result[0], ["next"])
# Second step
info_query = Iq(stanza_type="set",
from_jid="user1@test.com",
to_jid="jcl.test.com")
command_node = info_query.set_new_content(command.COMMAND_NS, "command")
command_node.setProp("node", "http://jabber.org/protocol/admin#end-user-session")
session_id = xml_command.prop("sessionid")
command_node.setProp("sessionid", session_id)
command_node.setProp("action", "next")
submit_form = Form(xmlnode_or_type="submit")
submit_form.add_field(field_type="jid-multi",
name="user_jids",
values=["test1@test.com", "test2@test.com"])
submit_form.as_xml(command_node)
result = self.command_manager.apply_command_action(info_query,
"http://jabber.org/protocol/admin#end-user-session",
"execute")
self.assertNotEquals(result, None)
self.assertEquals(len(result), 1)
xml_command = result[0].xpath_eval("c:command",
{"c": "http://jabber.org/protocol/commands"})[0]
self.assertEquals(xml_command.prop("status"), "executing")
self.assertEquals(xml_command.prop("sessionid"), session_id)
self.__check_actions(result[0], ["prev", "complete"], 1)
context_session = self.command_manager.sessions[session_id][1]
self.assertEquals(context_session["user_jids"],
["test1@test.com", "test2@test.com"])
# Third step
info_query = Iq(stanza_type="set",
from_jid="user1@test.com",
to_jid="jcl.test.com")
command_node = info_query.set_new_content(command.COMMAND_NS, "command")
command_node.setProp("node", "http://jabber.org/protocol/admin#end-user-session")
command_node.setProp("sessionid", session_id)
command_node.setProp("action", "complete")
submit_form = Form(xmlnode_or_type="submit")
submit_form.add_field(field_type="list-multi",
name="account_names",
values=["account11/test1@test.com",
"account11/test2@test.com"])
submit_form.as_xml(command_node)
result = self.command_manager.apply_command_action(info_query,
"http://jabber.org/protocol/admin#end-user-session",
"execute")
xml_command = result[0].xpath_eval("c:command",
{"c": "http://jabber.org/protocol/commands"})[0]
self.assertEquals(xml_command.prop("status"), "completed")
self.assertEquals(xml_command.prop("sessionid"), session_id)
self.__check_actions(result[0])
self.assertEquals(context_session["account_names"],
["account11/test1@test.com",
"account11/test2@test.com"])
stanza_sent = result
self.assertEquals(len(stanza_sent), 3)
iq_result = stanza_sent[0]
self.assertTrue(isinstance(iq_result, Iq))
self.assertEquals(iq_result.get_node().prop("type"), "result")
self.assertEquals(iq_result.get_from(), "jcl.test.com")
self.assertEquals(iq_result.get_to(), "user1@test.com")
presence_component = stanza_sent[1]
self.assertTrue(isinstance(presence_component, Presence))
self.assertEquals(presence_component.get_from(), "account11@jcl.test.com")
self.assertEquals(presence_component.get_to(), "test1@test.com")
self.assertEquals(presence_component.get_node().prop("type"),
"unavailable")
presence_component = stanza_sent[2]
self.assertTrue(isinstance(presence_component, Presence))
self.assertEquals(presence_component.get_from(), "account11@jcl.test.com")
self.assertEquals(presence_component.get_to(), "test2@test.com")
self.assertEquals(presence_component.get_node().prop("type"),
"unavailable")
# def test_execute_get_user_password(self):
# #TODO : implement command