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

get-user-password ad-hoc command implementation

Browse Source 
darcs-hash:20070725205420-86b55-4713327f3c21b679acea21ca5be35b01cb186ef2.gz
This commit is contained in:
David Rousselie
2007-07-25 22:54:20 +02:00
parent d8fa96f54c
commit 7c5824dc0b
2 changed files with 341 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -274,29 +274,84 @@ class JCLCommandManager(CommandManager):
result_form.as_xml(command_node) result_form.as_xml(command_node)
return [] return []
def add_form_select_user_jid(self, command_node, lang_class):
result_form = Form(xmlnode_or_type="result",
title="TODO")
result_form.add_field(name="user_jid",
field_type="jid-single",
label=lang_class.field_user_jid)
result_form.as_xml(command_node)
return []
def __add_accounts_to_field(self, user_jids, field, lang_class):
for (account_type, type_label) in \
self.account_manager.list_account_types(lang_class):
account_class = self.account_manager.get_account_class(\
account_type=account_type)
for user_jid in user_jids:
for _account in account.get_accounts(user_jid, account_class):
field.add_option(label=_account.name + " (" + account_type
+ ") (" + user_jid + ")",
values=[_account.name + "/" + user_jid])
def add_form_select_accounts(self, session_context, def add_form_select_accounts(self, session_context,
command_node, lang_class): command_node, lang_class):
""" """
Add a form to select account for user JIDs contained in Add a form to select accounts for user JIDs contained in
session_context[\"user_jids\"] session_context[\"user_jids\"]
""" """
result_form = Form(xmlnode_or_type="result", result_form = Form(xmlnode_or_type="result",
title="TODO") title="TODO")
field = result_form.add_field(name="account_names", field = result_form.add_field(name="account_names",
field_type="list-multi", field_type="list-multi",
label="Account") label="Account") # TODO
for (account_type, type_label) in \ self.__add_accounts_to_field(session_context["user_jids"],
self.account_manager.list_account_types(lang_class): field, lang_class)
account_class = self.account_manager.get_account_class(\
account_type=account_type)
for user_jid in session_context["user_jids"]:
for _account in account.get_accounts(user_jid, account_class):
field.add_option(label=_account.name + " (" + account_type
+ ") (" + user_jid + ")",
values=[_account.name + "/" + user_jid])
result_form.as_xml(command_node) result_form.as_xml(command_node)
return [] return []
def add_form_select_account(self, session_context,
command_node, lang_class):
"""
Add a form to select account for user JIDs contained in
session_context[\"user_jids\"]
"""
result_form = Form(xmlnode_or_type="result",
title="TODO")
field = result_form.add_field(name="account_name",
field_type="list-single",
label="Account") # TODO
self.__add_accounts_to_field([session_context["user_jid"]],
field, lang_class)
result_form.as_xml(command_node)
return []
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 select_user_jid_step_1(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing select_user_jid step 1")
self.add_actions(command_node, [ACTION_NEXT])
return self.add_form_select_user_jid(command_node, lang_class)
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)
def select_account_step_2(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing select_account step 2")
self.add_actions(command_node, [ACTION_PREVIOUS, ACTION_COMPLETE], 1)
return self.add_form_select_account(session_context, command_node,
lang_class)
def execute_list_1(self, info_query, session_context, def execute_list_1(self, info_query, session_context,
command_node, lang_class): command_node, lang_class):
"""Execute command 'list'. List accounts""" """Execute command 'list'. List accounts"""
@@ -362,19 +417,6 @@ class JCLCommandManager(CommandManager):
command_node.setProp("status", STATUS_COMPLETED) command_node.setProp("status", STATUS_COMPLETED)
return to_send return to_send
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 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)
execute_delete_user_1 = select_user_jids_step_1 execute_delete_user_1 = select_user_jids_step_1
execute_delete_user_2 = select_accounts_step_2 execute_delete_user_2 = select_accounts_step_2
@@ -432,7 +474,24 @@ class JCLCommandManager(CommandManager):
command_node.setProp("status", STATUS_COMPLETED) command_node.setProp("status", STATUS_COMPLETED)
return result return result
def execute_get_user_password(self, info_query): execute_get_user_password_1 = select_user_jid_step_1
execute_get_user_password_2 = select_account_step_2
def execute_get_user_password_3(self, info_query, session_context,
command_node, lang_class):
self.__logger.debug("Executing command 'get-user-password' step 3")
result_form = Form(xmlnode_or_type="result")
result_form.add_field(field_type="hidden",
name="FORM_TYPE",
value="http://jabber.org/protocol/admin")
name, user_jid = session_context["account_name"].split("/", 1)[:2]
_account = account.get_account(user_jid, name)
result_form.fields.append(FieldNoType(name="accountjids",
value=user_jid))
result_form.fields.append(FieldNoType(name="password",
value=_account.password))
result_form.as_xml(command_node)
command_node.setProp("status", STATUS_COMPLETED)
return [] return []
def execute_change_user_password(self, info_query): def execute_change_user_password(self, info_query):

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

@@ -105,6 +105,154 @@ class JCLCommandManager_TestCase(unittest.TestCase):
self.assertEquals(children.name, action) self.assertEquals(children.name, action)
children = children.next children = children.next
def test_add_form_select_user_jids(self):
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")
self.command_manager.add_form_select_user_jids(command_node, Lang.en)
user_jid_field = info_query.xpath_eval("c:command/data:x/data:field[1]",
{"c": "http://jabber.org/protocol/commands",
"data": "jabber:x:data"})
self.assertNotEquals(user_jid_field, None)
self.assertEquals(len(user_jid_field), 1)
self.assertEquals(user_jid_field[0].prop("var"), "user_jids")
self.assertEquals(user_jid_field[0].prop("type"), "jid-multi")
self.assertEquals(user_jid_field[0].prop("label"), Lang.en.field_user_jid)
def test_add_form_select_user_jid(self):
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")
self.command_manager.add_form_select_user_jid(command_node, Lang.en)
user_jid_field = info_query.xpath_eval("c:command/data:x/data:field[1]",
{"c": "http://jabber.org/protocol/commands",
"data": "jabber:x:data"})
self.assertNotEquals(user_jid_field, None)
self.assertEquals(len(user_jid_field), 1)
self.assertEquals(user_jid_field[0].prop("var"), "user_jid")
self.assertEquals(user_jid_field[0].prop("type"), "jid-single")
self.assertEquals(user_jid_field[0].prop("label"), Lang.en.field_user_jid)
def test_add_form_select_accounts(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")
session_context = {}
session_context["user_jids"] = ["test1@test.com", "test2@test.com"]
self.command_manager.add_form_select_accounts(session_context,
command_node,
Lang.en)
fields = info_query.xpath_eval("c:command/data:x/data:field",
{"c": "http://jabber.org/protocol/commands",
"data": "jabber:x:data"})
self.assertEquals(len(fields), 1)
self.assertEquals(fields[0].prop("var"), "account_names")
self.assertEquals(fields[0].prop("type"), "list-multi")
self.assertEquals(fields[0].prop("label"), "Account")
options = info_query.xpath_eval("c:command/data:x/data:field[1]/data:option",
{"c": "http://jabber.org/protocol/commands",
"data": "jabber:x:data"})
self.assertEquals(len(options), 4)
self.assertEquals(options[0].prop("label"),
"account11 (Example) (test1@test.com)")
self.assertEquals(options[0].children.name, "value")
self.assertEquals(options[0].children.content,
"account11/test1@test.com")
self.assertEquals(options[1].prop("label"),
"account21 (Example) (test2@test.com)")
self.assertEquals(options[1].children.name, "value")
self.assertEquals(options[1].children.content,
"account21/test2@test.com")
self.assertEquals(options[2].prop("label"),
"account11 (Example) (test2@test.com)")
self.assertEquals(options[2].children.name, "value")
self.assertEquals(options[2].children.content,
"account11/test2@test.com")
self.assertEquals(options[3].prop("label"),
"account12 (Example2) (test1@test.com)")
self.assertEquals(options[3].children.name, "value")
self.assertEquals(options[3].children.content,
"account12/test1@test.com")
def test_add_form_select_account(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")
session_context = {}
session_context["user_jid"] = "test1@test.com"
self.command_manager.add_form_select_account(session_context,
command_node,
Lang.en)
fields = info_query.xpath_eval("c:command/data:x/data:field",
{"c": "http://jabber.org/protocol/commands",
"data": "jabber:x:data"})
self.assertEquals(len(fields), 1)
self.assertEquals(fields[0].prop("var"), "account_name")
self.assertEquals(fields[0].prop("type"), "list-single")
self.assertEquals(fields[0].prop("label"), "Account")
options = info_query.xpath_eval("c:command/data:x/data:field[1]/data:option",
{"c": "http://jabber.org/protocol/commands",
"data": "jabber:x:data"})
self.assertEquals(len(options), 2)
self.assertEquals(options[0].prop("label"),
"account11 (Example) (test1@test.com)")
self.assertEquals(options[0].children.name, "value")
self.assertEquals(options[0].children.content,
"account11/test1@test.com")
self.assertEquals(options[1].prop("label"),
"account12 (Example2) (test1@test.com)")
self.assertEquals(options[1].children.name, "value")
self.assertEquals(options[1].children.content,
"account12/test1@test.com")
def test_execute_add_user(self): def test_execute_add_user(self):
self.comp.account_manager.account_classes = (ExampleAccount, self.comp.account_manager.account_classes = (ExampleAccount,
Example2Account) Example2Account)
@@ -449,84 +597,6 @@ class JCLCommandManager_TestCase(unittest.TestCase):
self.assertEquals(xml_command.prop("sessionid"), session_id) self.assertEquals(xml_command.prop("sessionid"), session_id)
self.assertEquals(xml_command.children, None) self.assertEquals(xml_command.children, None)
def test_add_form_select_user_jids(self):
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")
self.command_manager.add_form_select_user_jids(command_node, Lang.en)
user_jid_field = info_query.xpath_eval("c:command/data:x/data:field[1]",
{"c": "http://jabber.org/protocol/commands",
"data": "jabber:x:data"})
self.assertNotEquals(user_jid_field, None)
self.assertEquals(len(user_jid_field), 1)
self.assertEquals(user_jid_field[0].prop("var"), "user_jids")
self.assertEquals(user_jid_field[0].prop("type"), "jid-multi")
self.assertEquals(user_jid_field[0].prop("label"), Lang.en.field_user_jid)
def test_add_form_select_accounts(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")
session_context = {}
session_context["user_jids"] = ["test1@test.com", "test2@test.com"]
self.command_manager.add_form_select_accounts(session_context,
command_node,
Lang.en)
fields = info_query.xpath_eval("c:command/data:x/data:field",
{"c": "http://jabber.org/protocol/commands",
"data": "jabber:x:data"})
self.assertEquals(len(fields), 1)
self.assertEquals(fields[0].prop("var"), "account_names")
self.assertEquals(fields[0].prop("type"), "list-multi")
self.assertEquals(fields[0].prop("label"), "Account")
options = info_query.xpath_eval("c:command/data:x/data:field[1]/data:option",
{"c": "http://jabber.org/protocol/commands",
"data": "jabber:x:data"})
self.assertEquals(len(options), 4)
self.assertEquals(options[0].prop("label"),
"account11 (Example) (test1@test.com)")
self.assertEquals(options[0].children.name, "value")
self.assertEquals(options[0].children.content,
"account11/test1@test.com")
self.assertEquals(options[1].prop("label"),
"account21 (Example) (test2@test.com)")
self.assertEquals(options[1].children.name, "value")
self.assertEquals(options[1].children.content,
"account21/test2@test.com")
self.assertEquals(options[2].prop("label"),
"account11 (Example) (test2@test.com)")
self.assertEquals(options[2].children.name, "value")
self.assertEquals(options[2].children.content,
"account11/test2@test.com")
self.assertEquals(options[3].prop("label"),
"account12 (Example2) (test1@test.com)")
self.assertEquals(options[3].children.name, "value")
self.assertEquals(options[3].children.content,
"account12/test1@test.com")
def test_execute_delete_user(self): def test_execute_delete_user(self):
self.comp.account_manager.account_classes = (ExampleAccount, self.comp.account_manager.account_classes = (ExampleAccount,
@@ -983,14 +1053,115 @@ class JCLCommandManager_TestCase(unittest.TestCase):
self.assertEquals(presence_component.get_node().prop("type"), self.assertEquals(presence_component.get_node().prop("type"),
"unavailable") "unavailable")
# def test_execute_get_user_password(self): def test_execute_get_user_password(self):
# #TODO : implement command self.comp.account_manager.account_classes = (ExampleAccount,
# info_query = Iq(stanza_type="set", Example2Account)
# from_jid="user1@test.com", model.db_connect()
# to_jid="jcl.test.com") account11 = ExampleAccount(user_jid="test1@test.com",
# result = self.command_manager.execute_add_user(info_query) name="account11",
# self.assertNotEquals(result, None) jid="account11@jcl.test.com")
# self.assertEquals(len(result), 1) account11.password = "pass1"
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")
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#get-user-password")
result = self.command_manager.apply_command_action(info_query,
"http://jabber.org/protocol/admin#get-user-password",
"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#get-user-password")
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-single",
name="user_jid",
value="test1@test.com")
submit_form.as_xml(command_node)
result = self.command_manager.apply_command_action(info_query,
"http://jabber.org/protocol/admin#get-user-password",
"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_jid"],
"test1@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#get-user-password")
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-single",
name="account_name",
value="account11/test1@test.com")
submit_form.as_xml(command_node)
result = self.command_manager.apply_command_action(info_query,
"http://jabber.org/protocol/admin#get-user-password",
"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_name"],
"account11/test1@test.com")
stanza_sent = result
self.assertEquals(len(stanza_sent), 1)
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")
fields = iq_result.xpath_eval("c:command/data:x/data:field",
{"c": "http://jabber.org/protocol/commands",
"data": "jabber:x:data"})
self.assertEquals(len(fields), 3)
self.assertEquals(fields[0].prop("var"), "FORM_TYPE")
self.assertEquals(fields[0].prop("type"), "hidden")
self.assertEquals(fields[0].children.name, "value")
self.assertEquals(fields[0].children.content,
"http://jabber.org/protocol/admin")
self.assertEquals(fields[1].prop("var"), "accountjids")
self.assertEquals(fields[1].children.name, "value")
self.assertEquals(fields[1].children.content,
"test1@test.com")
self.assertEquals(fields[2].prop("var"), "password")
self.assertEquals(fields[2].children.name, "value")
self.assertEquals(fields[2].children.content,
"pass1")
# def test_execute_change_user_password(self): # def test_execute_change_user_password(self):
# #TODO : implement command # #TODO : implement command