1 import unittest |
1 from twisted.internet import reactor |
|
2 |
2 from Queue import deque |
3 from Queue import deque |
3 |
4 |
4 from iro.controller.task import createJob, Task |
5 from iro.controller.task import createJob, Task |
5 from iro.controller.pool import taskPool |
6 from iro.controller.pool import taskPool, dbPool |
6 |
7 |
7 from iro.model.job import exJobs, ExJob |
8 from iro.model.job import exJobs, ExJob |
8 |
9 |
9 from iro.model.pool import data |
10 from iro.model.pool import data |
10 |
11 |
12 from iro.model.schema import Job, User |
13 from iro.model.schema import Job, User |
13 from iro.telnumber import Telnumber |
14 from iro.telnumber import Telnumber |
14 |
15 |
15 from iro.validate import vInteger |
16 from iro.validate import vInteger |
16 |
17 |
17 |
|
18 from .dbtestcase import DBTestCase |
18 from .dbtestcase import DBTestCase |
19 |
|
20 #activates all logging we can get. |
|
21 |
19 |
22 class DumpPool(): |
20 class DumpPool(): |
23 def run(self, f,*a,**k): |
21 def run(self, f,*a,**k): |
24 return f(*a,**k) |
22 return f(*a,**k) |
25 |
23 |
88 |
86 |
89 class JobTest(DBTestCase): |
87 class JobTest(DBTestCase): |
90 |
88 |
91 def setUp(self): |
89 def setUp(self): |
92 DBTestCase.setUp(self) |
90 DBTestCase.setUp(self) |
93 self.pool = data.pool |
91 dbPool.start(reactor) |
94 data.pool = DumpPool() |
|
95 |
92 |
96 def tearDown(self): |
93 def tearDown(self): |
97 exJobs.clear() |
94 exJobs.clear() |
98 data.pool = self.pool |
95 dbPool.pool.stop() |
99 self.pool = None |
|
100 taskPool.pool.q.queue = deque() |
96 taskPool.pool.q.queue = deque() |
101 DBTestCase.tearDown(self) |
97 DBTestCase.tearDown(self) |
102 |
98 |
103 def testCreateSMS(self): |
99 def testCreateSMS(self): |
104 with self.session() as session: |
100 with self.session() as session: |
105 u = User(name='test',apikey='abcdef123456789') |
101 u = User(name='test',apikey='abcdef123456789') |
106 session.add(u) |
102 session.add(u) |
107 exjob = createJob(u,[Telnumber('0123325456')],SMS('sms'),[]) |
103 |
108 self.assertEqual(taskPool.pool.q.qsize(),1) |
104 d = createJob(u,[Telnumber('0123325456')],SMS('sms'),[]) |
109 self.assertEqual(exjob.tasks.keys(),[Telnumber('0123325456')]) |
105 |
110 self.assertIsInstance(exjob.tasks[Telnumber('0123325456')], Task) |
106 def poolsize(result): |
|
107 self.assertEqual(taskPool.pool.q.qsize(),1) |
|
108 return result |
|
109 d.addCallback(poolsize) |
|
110 def keys(result): |
|
111 self.assertEqual(result.tasks.keys(),[Telnumber('0123325456')]) |
|
112 return result |
|
113 d.addCallback(keys) |
|
114 def tasktest(result): |
|
115 self.assertIsInstance(result.tasks[Telnumber('0123325456')], Task) |
|
116 return result |
|
117 d.addCallback(tasktest) |
111 |
118 |
112 |
119 |
113 |
120 |
114 def testCreateFax(self): |
121 return d |
115 with self.session() as session: |
|
116 u = User(name='test',apikey='abcdef123456789') |
|
117 session.add(u) |
|
118 job = createJob(u,[Telnumber('0123325456')],Fax('header','fax',[]),[]) |
|
119 |
122 |
120 self.assertEqual(taskPool.pool.q.qsize(),1) |
123 #def testCreateFax(self): |
|
124 # with self.session() as session: |
|
125 # u = User(name='test',apikey='abcdef123456789') |
|
126 # session.add(u) |
|
127 # job = createJob(u,[Telnumber('0123325456')],Fax('header','fax',[]),[]) |
121 |
128 |
122 def testCreateMail(self): |
129 # self.assertEqual(taskPool.pool.q.qsize(),1) |
123 with self.session() as session: |
|
124 u = User(name='test',apikey='abcdef123456789') |
|
125 session.add(u) |
|
126 job = createJob(u,[],Mail('sub','body','t@t.de'),[]) |
|
127 |
|
128 |
130 |
129 if __name__ == '__main__': |
131 #def testCreateMail(self): |
130 unittest.main() |
132 # with self.session() as session: |
|
133 # u = User(name='test',apikey='abcdef123456789') |
|
134 # session.add(u) |
|
135 # job = createJob(u,[],Mail('sub','body','t@t.de'),[]) |