--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/iro/tests/job.py Sun Mar 18 14:06:27 2012 +0100
@@ -0,0 +1,194 @@
+from datetime import datetime
+from decimal import Decimal
+from mock import patch
+
+from iro.model.job import exJobs, ExJob
+from iro.model.pool import data
+from iro.model.message import SMS
+from iro.model.status import Status
+from iro.model.schema import Job, User, Offer as DBOffer, Userright
+
+from iro.controller.task import Task
+
+from iro.offer.provider import Provider
+
+from iro.telnumber import Telnumber
+from iro.validate import vInteger
+
+from ..test_helpers.dbtestcase import DBTestCase
+
+class DummyPool():
+ def run(self, f,*a,**k):
+ return f(*a,**k)
+
+from twisted.python import log
+
+class DummyObserver(object):
+ def __init__(self):
+ self.e=[]
+
+ def start(self):
+ log.addObserver(self.emit)
+
+ def stop(self):
+ log.removeObserver(self.emit)
+
+ def emit(self, eventDict):
+ self.e.append(eventDict)
+
+class JobTestCase(DBTestCase):
+ def setUp(self):
+ DBTestCase.setUp(self)
+ self.pool = data.pool
+ data.pool = DummyPool()
+
+ def tearDown(self):
+ exJobs.clear()
+ data.pool = self.pool
+ self.pool = None
+ DBTestCase.tearDown(self)
+
+
+class exJobsTest(JobTestCase):
+ '''tests for exJobs'''
+
+ def testCreate(self):
+ with self.session() as session:
+ u = User(name='test',apikey='abcdef123456789')
+ session.add(u)
+
+ job = exJobs.create(u, [Telnumber('123456789')], SMS('test'), ['test'])
+ self.assertIsInstance(job, ExJob)
+ self.assertTrue(vInteger(job.dbjob, None, minv=0 ))
+ self.assertEqual(job.message, SMS('test'))
+ self.assertEqual(job.recipients, [Telnumber('123456789')])
+ self.assertEqual(job.offers,[])
+ self.assertEqual(job.tasks,{})
+
+ with self.session() as session:
+ j = session.query(Job.id).all()
+ self.assertEqual(j,[(job.dbjob,)])
+
+ self.assertEqual(exJobs[job.dbjob],job)
+
+ def testCreate2(self):
+ with self.session() as session:
+ u = User(name='test',apikey='abcdef123456789')
+ session.add(u)
+ o=DBOffer(name="test", provider="bla", route="basic", typ="sms")
+ u.rights.append(Userright(o))
+
+ job = exJobs.create(u, [Telnumber('123456789')], SMS('test'), ['test'])
+ self.assertEqual(job.offers,['test'])
+
+ def testGet(self):
+ with self.session() as session:
+ u = User(name='test',apikey='abcdef123456789')
+ session.add(u)
+
+ job = ExJob(None, [Telnumber('123456789')], SMS('test'), ['test'])
+ exJobs[1] = job
+
+ self.assertEqual(len(exJobs), 1)
+ self.assertEqual(job, exJobs[1])
+
+ def testGetFromDB(self):
+ with self.session() as session:
+ u = User(name='test',apikey='abcdef123456789')
+ job = Job( info="info", status="started")
+ u.jobs.append(job)
+ session.add(u)
+
+ with self.session() as session:
+ job = session.merge(job)
+ u = session.merge(u)
+ ejob= ExJob(job.id, [Telnumber('123456789')], SMS('test'), ['test'])
+ exJobs[job.id]=ejob
+ self.assertEqual(job.extend, ejob)
+ self.assertEqual(u.jobs[0].extend,ejob)
+
+ def testUnknownExJob(self):
+ self.assertRaises(KeyError,exJobs.__getitem__,'a1234567890')
+
+
+class StatiTest(JobTestCase):
+ def setUp(self):
+ JobTestCase.setUp(self)
+ self.log = DummyObserver()
+ self.log.start()
+
+ with self.session() as session:
+ u = User(name='test',apikey='abcdef123456789')
+ session.add(u)
+ o=DBOffer(name="test", provider="bla", route="a", typ="sms")
+ u.rights.append(Userright(o))
+
+ self.user = u
+ self.offer = o
+
+ self.provider=Provider("bla", {"sms":["a","b","c"]})
+ self.job = exJobs.create(u, [Telnumber('123456789')], SMS('test'), [])
+
+ def tearDown(self):
+ self.log.stop()
+ JobTestCase.tearDown(self)
+
+
+ def testSetError(self):
+ self.job.setError(Task(Telnumber('123456789'),self),Exception("muhaha"))
+ errors = self.flushLoggedErrors(Exception)
+ self.assertEqual(len(errors), 1)
+ self.assertEqual(self.log.e[0]['why'], "Error: Job(%s) to '0049123456789' failed."%self.job.dbjob)
+
+ with self.session() as session:
+ u = session.merge(self.user)
+ job = u.job(self.job.dbjob)
+ self.assertEqual(job.status,"error")
+
+ def testSetStatus(self):
+ task = Task(Telnumber('123456789'),self.job)
+ status = Status(self.provider,"a")
+ self.job.setStatus(task, status)
+
+ self.assertEqual(self.log.e[0]['message'], ("Job(%s) to '0049123456789' ended sucecessfully via bla:a."%self.job.dbjob,))
+
+ with self.session() as session:
+ u = session.merge(self.user)
+ job = u.job(self.job.dbjob)
+ self.assertEqual(job.status,"sended")
+ self.assertEqual(len(job.messages),0)
+
+ def testMultipleRecipients(self):
+ self.job.recipients.append(Telnumber("01234567890"))
+ task = Task(Telnumber('123456789'),self.job)
+ status = Status(self.provider,"a")
+ self.job.setStatus(task, status)
+
+ with self.session() as session:
+ u = session.merge(self.user)
+ job = u.job(self.job.dbjob)
+ self.assertEqual(job.status,"sending")
+
+ @patch("iro.model.job.datetime")
+ def testCosts(self,p_dt):
+ p_dt.today.return_value = datetime(2000, 1, 2, 3, 4, 5)
+ task = Task(Telnumber('123456789'),self.job)
+ status = Status(self.provider,"a",costs=0.055,exID="12345678",count=1)
+
+ self.job.setStatus(task, status)
+
+ with self.session() as session:
+ u = session.merge(self.user)
+ o = session.merge(self.offer)
+ job = u.job(self.job.dbjob)
+ self.assertEqual(job.status,"sended")
+ self.assertEqual(len(job.messages),1)
+
+ msg = job.messages[0]
+ self.assertEqual(msg.price,Decimal('0.0550'))
+ self.assertEqual(msg.isBilled,False)
+ self.assertEqual(msg.recipient,str(Telnumber('123456789')))
+ self.assertEqual(msg.date,datetime(2000, 1, 2, 3, 4, 5))
+ self.assertEqual(msg.offer,o)
+ self.assertEqual(msg.exID,"12345678")
+ self.assertEqual(msg.count,1)