1 from twisted.internet import reactor

     2 from twisted.internet.defer import inlineCallbacks

     3 

     4 from Queue import deque

     5 

     6 from iro.model.schema import User, Offer as DBOffer, Userright

     7 from iro.model.message import SMS

     8 from iro.model.status import Status

     9 from iro.model.offer import offers

    10 from iro.model.job import exJobs

    11 

    12 from iro.controller.task import createJob, Task

    13 from iro.controller.pool  import taskPool, dbPool

    14 

    15 from iro.offer import Offer, Provider

    16 

    17 from iro.error import NoRouteForTask

    18 from iro.telnumber import Telnumber

    19 

    20 from .dbtestcase import DBTestCase

    21 

    22 class TaskTestCase(DBTestCase):

    23     def setUp(self):

    24         DBTestCase.setUp(self)

    25         dbPool.start(reactor)

    26 

    27     def tearDown(self):

    28         exJobs.clear()

    29         offers.clear()

    30         dbPool.pool.stop()

    31         taskPool.pool.q.queue = deque()

    32         DBTestCase.tearDown(self)

    33  

    34 class TestTasks(TaskTestCase):

    35    

    36     @inlineCallbacks

    37     def testCreateSMS(self):

    38         with self.session() as session:

    39             u = User(name='test',apikey='abcdef123456789')

    40             session.add(u)

    41         

    42         job = yield createJob(u,[Telnumber('0123325456')],SMS('sms'),[])

    43         

    44         self.assertEqual(taskPool.pool.q.qsize(),1)

    45         

    46         self.assertEqual(job.tasks.keys(),[Telnumber('0123325456')])

    47         self.assertIsInstance(job.tasks[Telnumber('0123325456')], Task)

    48 

    49     @inlineCallbacks

    50     def testRun(self):

    51         with self.session() as session:

    52             u = User(name='test',apikey='abcdef123456789')

    53             session.add(u)

    54             o=DBOffer(name="test", provider="bla", route="basic", typ="sms")

    55             u.rights.append(Userright(o)) 

    56         

    57         p=Provider(name="p", config={}, typs={"sms":["test",]})

    58         def send(typ,route,recipient,message):

    59             return Status(provider=p, route=route)

    60         p.send=send

    61         offers["test"] = Offer("test",provider=p, route="test", typ="sms")

    62 

    63         exjob = yield exJobs.create(u, [Telnumber('123456789')], SMS('test'), ['test'])

    64         

    65         task=Task(Telnumber('123456789'), exjob)

    66         ret = yield task._run()

    67         self.assertIsInstance(ret, Status)

    68         self.assertEqual(ret.provider, p)

    69         self.assertEqual(ret.route, "test")

    70 

    71     @inlineCallbacks

    72     def testNoRoute(self):

    73         with self.session() as session:

    74             u = User(name='test',apikey='abcdef123456789')

    75             session.add(u)

    76 

    77         exjob = yield exJobs.create(u, [Telnumber('123456789')], SMS('test'), [])

    78         

    79         task=Task(Telnumber('123456789'), exjob)

    80         d = task._run()

    81         self.assertFailure(d, NoRouteForTask)

    82 

    83 

    84 class CompletetaskTests(TaskTestCase):

    85     def setUp(self):

    86         TaskTestCase.setUp(self)

    87         taskPool.start(reactor)

    88 

    89     def tearDown(self):

    90         taskPool.pool.stop()

    91         TaskTestCase.tearDown(self)

    92 

    93 #    def testStart(self):

    94 #        pass

    95 

    96 

    97     #def testCreateFax(self):

    98     #    with self.session() as session:

    99     #        u = User(name='test',apikey='abcdef123456789')

   100     #        session.add(u)

   101     #    job = createJob(u,[Telnumber('0123325456')],Fax('header','fax',[]),[])

   102 

   103     #    self.assertEqual(taskPool.pool.q.qsize(),1)

   104 

   105     #def testCreateMail(self):

   106     #    with self.session() as session:

   107     #        u = User(name='test',apikey='abcdef123456789')

   108     #        session.add(u)

   109     #    job = createJob(u,[],Mail('sub','body','t@t.de'),[])