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

Multiple account types: disco get_items implemented and tested

Browse Source 
darcs-hash:20070131175146-86b55-de5924fb265a0f611c053303de114e8667d6f881.gz
This commit is contained in:
David Rousselie
2007-01-31 18:51:46 +01:00
parent 95f645fd16
commit cfa8da9655
5 changed files with 96 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
run_tests.py
View File

@@ -62,7 +62,7 @@ if __name__ == '__main__':
jcl_suite = unittest.TestSuite()
# jcl_suite.addTest(FeederComponent_TestCase('test_handle_tick'))
# jcl_suite.addTest(JCLComponent_TestCase('test_handle_set_register_new_field_mandatory'))
# jcl_suite.addTest(JCLComponent_TestCase('test_disco_get_items_2types_with_node2'))
# jcl_suite = unittest.TestSuite((component_suite))
# jcl_suite = unittest.TestSuite((presence_account_suite))
jcl_suite = unittest.TestSuite((component_suite, \

12
src/jcl/jabber/component.py
View File

@@ -278,7 +278,7 @@ class JCLComponent(Component, object):
if match is not None:
account_type = match.group(1)
DiscoItem(disco_items, \
JID(account_type + "@" + unicode(self.jid)), \
self.jid, \
account_type, \
account_type)
else:
@@ -290,7 +290,8 @@ class JCLComponent(Component, object):
if account_class is not None:
self._list_accounts(disco_items, \
account_class, \
base_from_jid)
base_from_jid,
account_type = nodes[0])
else:
print >> sys.stderr, "Error: " + account_class.__name__ \
+ " class not in account_classes"
@@ -649,15 +650,18 @@ class JCLComponent(Component, object):
return _account_class
return None
def _list_accounts(self, disco_items, _account_class, base_from_jid):
def _list_accounts(self, disco_items, _account_class, base_from_jid, account_type = ""):
"""List accounts in disco_items for given _account_class and user jid"""
if account_type != "":
account_type = account_type + "/"
self.db_connect()
for _account in _account_class.select(_account_class.q.user_jid == \
base_from_jid):
self.__logger.debug(str(_account))
DiscoItem(disco_items, \
JID(_account.jid), \
_account.name, _account.long_name)
account_type + _account.name, \
_account.long_name)
self.db_disconnect()
def _send_presence_available(self, _account, show, lang_class):

101
tests/jcl/jabber/test_component.py
View File

@@ -44,7 +44,7 @@ from jcl.model.account import Account
from jcl.lang import Lang
from jcl.jabber.x import DataForm
from tests.jcl.model.account import ExampleAccount
from tests.jcl.model.account import ExampleAccount, Example2Account
DB_PATH = "/tmp/test.db"
DB_URL = DB_PATH# + "?debug=1&debugThreading=1"
@@ -125,12 +125,14 @@ class JCLComponent_TestCase(unittest.TestCase):
account.hub.threadConnection = connectionForURI('sqlite://' + DB_URL)
Account.createTable(ifNotExists = True)
ExampleAccount.createTable(ifNotExists = True)
Example2Account.createTable(ifNotExists = True)
del account.hub.threadConnection
self.max_tick_count = 1
self.saved_time_handler = None
def tearDown(self):
account.hub.threadConnection = connectionForURI('sqlite://' + DB_URL)
Example2Account.dropTable(ifExists = True)
ExampleAccount.dropTable(ifExists = True)
Account.dropTable(ifExists = True)
del TheURIOpener.cachedURIs['sqlite://' + DB_URL]
@@ -348,56 +350,103 @@ class JCLComponent_TestCase(unittest.TestCase):
disco_items = self.comp.disco_get_items("account1", info_query)
self.assertEquals(disco_items.get_items(), [])
# TODO : test get_items with multiple account_classes
def test_disco_get_items_2types_no_node(self):
self.comp.account_classes = [ExampleAccount, Example2Account]
account.hub.threadConnection = connectionForURI('sqlite://' + DB_URL)
account1 = Account(user_jid = "user1@test.com", \
name = "account1", \
jid = "account1@jcl.test.com")
account11 = ExampleAccount(user_jid = "user1@test.com", \
name = "account11", \
jid = "account11@jcl.test.com")
account21 = Example2Account(user_jid = "user1@test.com", \
name = "account21", \
jid = "account21@jcl.test.com")
del account.hub.threadConnection
info_query = Iq(stanza_type = "get", \
from_jid = "user1@test.com")
disco_items = self.comp.disco_get_items(None, info_query)
self.assertEquals(len(disco_items.get_items()), 2)
disco_item = disco_items.get_items()[0]
self.assertEquals(disco_item.get_jid(), self.comp.jid)
self.assertEquals(disco_item.get_node(), "Example")
self.assertEquals(disco_item.get_name(), "Example")
disco_item = disco_items.get_items()[1]
self.assertEquals(disco_item.get_jid(), self.comp.jid)
self.assertEquals(disco_item.get_node(), "Example2")
self.assertEquals(disco_item.get_name(), "Example2")
# Be careful, account_classes cannot contains parent classes
#
def test_disco_get_items_2types_with_node(self):
self.comp.account_classes = [ExampleAccount, Example2Account]
account.hub.threadConnection = connectionForURI('sqlite://' + DB_URL)
account11 = ExampleAccount(user_jid = "user1@test.com", \
name = "account11", \
jid = "account11@jcl.test.com")
account12 = ExampleAccount(user_jid = "user2@test.com", \
name = "account12", \
jid = "account12@jcl.test.com")
account21 = Example2Account(user_jid = "user1@test.com", \
name = "account21", \
jid = "account21@jcl.test.com")
account22 = Example2Account(user_jid = "user2@test.com", \
name = "account22", \
jid = "account22@jcl.test.com")
del account.hub.threadConnection
info_query = Iq(stanza_type = "get", \
from_jid = "user1@test.com")
disco_items = self.comp.disco_get_items("Example", info_query)
self.assertEquals(len(disco_items.get_items()), 1)
disco_item = disco_items.get_items()[0]
self.assertEquals(disco_item.get_jid(), account1.jid)
self.assertEquals(disco_item.get_node(), account1.name)
self.assertEquals(disco_item.get_name(), account1.long_name)
self.assertEquals(disco_item.get_jid(), account11.jid)
self.assertEquals(disco_item.get_node(), "Example/" + account11.name)
self.assertEquals(disco_item.get_name(), account11.long_name)
# TODO
def test_disco_get_items_2types_with_node(self):
def test_disco_get_items_2types_with_node2(self):
self.comp.account_classes = [ExampleAccount, Example2Account]
account.hub.threadConnection = connectionForURI('sqlite://' + DB_URL)
account1 = Account(user_jid = "user1@test.com", \
name = "account1", \
jid = "account1@jcl.test.com")
account11 = ExampleAccount(user_jid = "user1@test.com", \
name = "account11", \
jid = "account11@jcl.test.com")
account12 = ExampleAccount(user_jid = "user2@test.com", \
name = "account12", \
jid = "account12@jcl.test.com")
account21 = Example2Account(user_jid = "user1@test.com", \
name = "account21", \
jid = "account21@jcl.test.com")
account22 = Example2Account(user_jid = "user2@test.com", \
name = "account22", \
jid = "account22@jcl.test.com")
del account.hub.threadConnection
info_query = Iq(stanza_type = "get", \
from_jid = "user1@test.com")
disco_items = self.comp.disco_get_items("account1", info_query)
self.assertEquals(disco_items.get_items(), [])
from_jid = "user2@test.com")
disco_items = self.comp.disco_get_items("Example2", info_query)
self.assertEquals(len(disco_items.get_items()), 1)
disco_item = disco_items.get_items()[0]
self.assertEquals(disco_item.get_jid(), account22.jid)
self.assertEquals(disco_item.get_node(), "Example2/" + account22.name)
self.assertEquals(disco_item.get_name(), account22.long_name)
# TODO
def test_disco_get_items_2types_with_long_node(self):
self.comp.account_classes = [ExampleAccount, Example2Account]
account.hub.threadConnection = connectionForURI('sqlite://' + DB_URL)
account1 = Account(user_jid = "user1@test.com", \
name = "account1", \
jid = "account1@jcl.test.com")
account1 = ExampleAccount(user_jid = "user1@test.com", \
name = "account1", \
jid = "account1@jcl.test.com")
del account.hub.threadConnection
info_query = Iq(stanza_type = "get", \
from_jid = "user1@test.com")
disco_items = self.comp.disco_get_items("type1/account1", info_query)
disco_items = self.comp.disco_get_items("Example/account1", info_query)
self.assertEquals(disco_items.get_items(), [])
# TODO
def test_disco_get_items_2types_with_long_node2(self):
self.comp.account_classes = [ExampleAccount, Example2Account]
account.hub.threadConnection = connectionForURI('sqlite://' + DB_URL)
account1 = Account(user_jid = "user1@test.com", \
name = "account1", \
jid = "account1@jcl.test.com")
account1 = Example2Account(user_jid = "user1@test.com", \
name = "account1", \
jid = "account1@jcl.test.com")
del account.hub.threadConnection
info_query = Iq(stanza_type = "get", \
from_jid = "user1@test.com")
disco_items = self.comp.disco_get_items("type2/account1", info_query)
disco_items = self.comp.disco_get_items("Example2/account1", info_query)
self.assertEquals(disco_items.get_items(), [])
def test_handle_get_version(self):

4
tests/jcl/jabber/test_feeder.py
View File

@@ -38,7 +38,7 @@ from jcl.jabber.feeder import FeederComponent, Feeder, Sender
from jcl.model.account import Account
from jcl.model import account
from tests.jcl.model.account import ExampleAccount
from tests.jcl.model.account import ExampleAccount, Example2Account
DB_PATH = "/tmp/test.db"
DB_URL = DB_PATH #+ "?debug=1&debugThreading=1"
@@ -55,12 +55,14 @@ class FeederComponent_TestCase(JCLComponent_TestCase):
account.hub.threadConnection = connectionForURI('sqlite://' + DB_URL)
Account.createTable(ifNotExists = True)
ExampleAccount.createTable(ifNotExists = True)
Example2Account.createTable(ifNotExists = True)
del account.hub.threadConnection
def tearDown(self):
account.hub.threadConnection = connectionForURI('sqlite://' + DB_URL)
Account.dropTable(ifExists = True)
ExampleAccount.dropTable(ifExists = True)
Example2Account.dropTable(ifExists = True)
del TheURIOpener.cachedURIs['sqlite://' + DB_URL]
account.hub.threadConnection.close()
del account.hub.threadConnection

9
tests/jcl/model/account.py
View File

@@ -58,6 +58,15 @@ class ExampleAccount(Account):
get_register_fields = classmethod(_get_register_fields)
class Example2Account(Account):
test_new_int = IntCol(default = 42)
def _get_register_fields(cls):
return Account.get_register_fields() + \
[("test_new_int", "text-single", None, account_int_post_func, \
lambda field_name: 43)]
get_register_fields = classmethod(_get_register_fields)
class PresenceAccountExample(PresenceAccount):
DO_SOMETHING_ELSE = 2
possibles_actions = [PresenceAccount.DO_NOTHING, \