9 from iro.model.job import exJobs, ExJob, Status |
9 from iro.model.job import exJobs, ExJob, Status |
10 from iro.model.offer import offers |
10 from iro.model.offer import offers |
11 from iro.model.pool import data |
11 from iro.model.pool import data |
12 |
12 |
13 from iro.model.message import SMS, Fax, Mail |
13 from iro.model.message import SMS, Fax, Mail |
14 from iro.model.schema import Job, User |
14 from iro.model.schema import Job, User, Offer as DBOffer, Userright |
15 from iro.telnumber import Telnumber |
15 from iro.telnumber import Telnumber |
16 |
16 |
17 |
17 |
18 from iro.offer.offer import Offer |
18 from iro.offer.offer import Offer |
19 from iro.validate import vInteger |
19 from iro.validate import vInteger |
20 from iro.error import NoRoute |
20 from iro.error import NoRoute |
21 |
21 |
22 from .dbtestcase import DBTestCase |
22 from .dbtestcase import DBTestCase |
23 |
23 |
24 class DumpPool(): |
24 class DummyPool(): |
25 def run(self, f,*a,**k): |
25 def run(self, f,*a,**k): |
26 return f(*a,**k) |
26 return f(*a,**k) |
27 |
27 |
28 |
28 |
29 class exJobsTest(DBTestCase): |
29 class exJobsTest(DBTestCase): |
30 '''tests for exJobs''' |
30 '''tests for exJobs''' |
31 |
31 |
32 def setUp(self): |
32 def setUp(self): |
33 DBTestCase.setUp(self) |
33 DBTestCase.setUp(self) |
34 self.pool = data.pool |
34 self.pool = data.pool |
35 data.pool = DumpPool() |
35 data.pool = DummyPool() |
36 |
36 |
37 def tearDown(self): |
37 def tearDown(self): |
38 exJobs.clear() |
38 exJobs.clear() |
39 data.pool = self.pool |
39 data.pool = self.pool |
40 self.pool = None |
40 self.pool = None |
48 job = exJobs.create(u, [Telnumber('123456789')], SMS('test'), ['test']) |
48 job = exJobs.create(u, [Telnumber('123456789')], SMS('test'), ['test']) |
49 self.assertIsInstance(job, ExJob) |
49 self.assertIsInstance(job, ExJob) |
50 self.assertTrue(vInteger(job.dbjob, None, minv=0 )) |
50 self.assertTrue(vInteger(job.dbjob, None, minv=0 )) |
51 self.assertEqual(job.message, SMS('test')) |
51 self.assertEqual(job.message, SMS('test')) |
52 self.assertEqual(job.recipients, [Telnumber('123456789')]) |
52 self.assertEqual(job.recipients, [Telnumber('123456789')]) |
53 self.assertEqual(job.offers,['test']) |
53 self.assertEqual(job.offers,[]) |
54 self.assertEqual(job.tasks,{}) |
54 self.assertEqual(job.tasks,{}) |
55 |
55 |
56 with self.session() as session: |
56 with self.session() as session: |
57 j = session.query(Job.id).all() |
57 j = session.query(Job.id).all() |
58 self.assertEqual(j,[(job.dbjob,)]) |
58 self.assertEqual(j,[(job.dbjob,)]) |
59 |
59 |
60 self.assertEqual(exJobs[job.dbjob],job) |
60 self.assertEqual(exJobs[job.dbjob],job) |
|
61 |
|
62 def testCreate2(self): |
|
63 with self.session() as session: |
|
64 u = User(name='test',apikey='abcdef123456789') |
|
65 session.add(u) |
|
66 o=DBOffer(name="test", provider="bla", route="basic", typ="sms") |
|
67 u.rights.append(Userright(o)) |
|
68 |
|
69 job = exJobs.create(u, [Telnumber('123456789')], SMS('test'), ['test']) |
|
70 self.assertEqual(job.offers,['test']) |
61 |
71 |
62 def testGet(self): |
72 def testGet(self): |
63 with self.session() as session: |
73 with self.session() as session: |
64 u = User(name='test',apikey='abcdef123456789') |
74 u = User(name='test',apikey='abcdef123456789') |
65 session.add(u) |
75 session.add(u) |
117 @inlineCallbacks |
127 @inlineCallbacks |
118 def testRun(self): |
128 def testRun(self): |
119 with self.session() as session: |
129 with self.session() as session: |
120 u = User(name='test',apikey='abcdef123456789') |
130 u = User(name='test',apikey='abcdef123456789') |
121 session.add(u) |
131 session.add(u) |
|
132 o=DBOffer(name="test", provider="bla", route="basic", typ="sms") |
|
133 u.rights.append(Userright(o)) |
122 |
134 |
123 offers["test"] = Offer("test") |
135 offers["test"] = Offer("test") |
124 |
136 |
125 exjob = yield exJobs.create(u, [Telnumber('123456789')], SMS('test'), ['test']) |
137 exjob = yield exJobs.create(u, [Telnumber('123456789')], SMS('test'), ['test']) |
126 |
138 |