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

Test error in handle_tick

Browse Source 
Add NotImplementedError in virtual methods
Test NotImplementError raising for JCLComponent handle_tick method

darcs-hash:20061010062453-86b55-91546440555f22e80d5a5e3dd4eb73d0d40b3dc1.gz
This commit is contained in:
David Rousselie
2006-10-10 08:24:53 +02:00
parent 26395d1f63
commit 7c8726b5c9
3 changed files with 54 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -125,9 +125,6 @@ class JCLComponent(Component):
self.spool_dir += "/" + unicode(self.jid)
self.running = True
self.connect()
## TODO : workaround to make test_run pass on FeederComponent
# time.sleep(1)
##
timer_thread = threading.Thread(target = self.time_handler, \
name = "TimerThread")
timer_thread.start()
@@ -183,7 +180,6 @@ class JCLComponent(Component):
connectionForURI(self.db_connection_str)
def db_disconnect(self):
# account.hub.threadConnection.close()
del account.hub.threadConnection
@@ -195,7 +191,9 @@ class JCLComponent(Component):
"""
self.__logger.info("Timer thread started...")
try:
while self.running:
while (self.running and self.stream \
and not self.stream.eof \
and self.stream.socket is not None):
self.handle_tick()
self.__logger.debug("Resetting alarm signal")
time.sleep(self.time_unit)
@@ -474,6 +472,18 @@ class JCLComponent(Component):
"""
return account.name + u"@" + unicode(self.jid)
def get_reg_form(self, lang_class, account_class):
"""Return register form based on language and account class
"""
# TODO
pass
def get_reg_form_init(self, lang_class, account):
"""Return register form for an existing account (update)
"""
# TODO
pass
###########################################################################
# Virtual methods
###########################################################################
@@ -481,16 +491,4 @@ class JCLComponent(Component):
"""Virtual method
Called regularly
"""
pass
def get_reg_form(self, lang_class, account_class):
"""Virtual method
Return register form based on language and account class
"""
pass
def get_reg_form_init(self, lang_class, account):
"""Virtual method
Return register form for an existing account (update)
"""
pass
raise NotImplementedError

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

@@ -99,6 +99,11 @@ class MockStream(object):
def close(self):
pass
class MockStreamNoConnect(MockStream):
def connect(self):
self.connection_started = True
self.eof = True
class JCLComponent_TestCase(unittest.TestCase):
def setUp(self):
if os.path.exists(DB_PATH):
@@ -122,21 +127,14 @@ class JCLComponent_TestCase(unittest.TestCase):
def test_constructor(self):
account.hub.threadConnection = connectionForURI('sqlite://' + DB_URL)
self.assertTrue(Account._connection.tableExists("account"))
if os.path.exists(DB_PATH):
print DB_PATH + " exists cons"
del account.hub.threadConnection
def test_run(self):
self.comp.time_unit = 1
self.comp.stream = MockStream()
self.comp.stream_class = MockStream
run_thread = threading.Thread(target = self.comp.run, \
name = "run_thread")
run_thread.start()
time.sleep(1)
self.comp.stream = MockStreamNoConnect()
self.comp.stream_class = MockStreamNoConnect
self.comp.run()
self.assertTrue(self.comp.stream.connection_started)
self.comp.running = False
time.sleep(JCLComponent.timeout + 1)
threads = threading.enumerate()
self.assertEquals(len(threads), 1)
self.assertTrue(self.comp.stream.connection_stopped)
@@ -156,6 +154,7 @@ class JCLComponent_TestCase(unittest.TestCase):
self.comp.time_unit = 1
self.max_tick_count = 2
self.comp.handle_tick = self.__handle_tick_test_time_handler
self.comp.stream = MockStream()
self.comp.running = True
self.comp.time_handler()
self.assertEquals(self.max_tick_count, 0)
@@ -260,6 +259,6 @@ class JCLComponent_TestCase(unittest.TestCase):
pass
def test_handle_tick(self):
self.comp.handle_tick()
self.assertTrue(True)
self.assertRaises(NotImplementedError, self.comp.handle_tick)

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

@@ -23,12 +23,15 @@
import unittest
import os
import threading
import time
from sqlobject import *
from sqlobject.dbconnection import TheURIOpener
from tests.jcl.jabber.test_component import JCLComponent_TestCase
from tests.jcl.jabber.test_component import JCLComponent_TestCase, MockStream
from jcl.jabber.component import JCLComponent
from jcl.jabber.feeder import FeederComponent, Feeder, Sender
from jcl.model.account import Account
from jcl.model import account
@@ -55,10 +58,27 @@ class FeederComponent_TestCase(JCLComponent_TestCase):
if os.path.exists(DB_PATH):
os.unlink(DB_PATH)
def test_constructor(self):
account.hub.threadConnection = connectionForURI('sqlite://' + DB_URL)
self.assertTrue(Account._connection.tableExists("account"))
del account.hub.threadConnection
def test_run(self):
self.comp.time_unit = 1
self.comp.stream = MockStream()
self.comp.stream_class = MockStream
run_thread = threading.Thread(target = self.comp.run, \
name = "run_thread")
run_thread.start()
time.sleep(1)
self.comp.running = False
self.assertTrue(self.comp.stream.connection_started)
time.sleep(JCLComponent.timeout + 1)
threads = threading.enumerate()
self.assertEquals(len(threads), 1)
self.assertTrue(self.comp.stream.connection_stopped)
if self.comp.queue.qsize():
raise self.comp.queue.get(0)
def test_handle_tick(self):
# TODO
self.comp.handle_tick()
self.assertTrue(True)
class Feeder_TestCase(unittest.TestCase):
def setUp(self):