iro/tests/offer.py
branchdevel
changeset 231 3929338fd17f
parent 204 4eb663df29fb
child 294 0e75bd39767d
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/iro/tests/offer.py	Sun Mar 18 14:06:27 2012 +0100
@@ -0,0 +1,156 @@
+from twisted.internet import reactor
+from twisted.internet.defer import inlineCallbacks
+
+import io
+
+from iro.model import offer
+from iro.config import configParser
+from iro.offer import Offer, providers as OfferProviders, Provider
+from iro.model.schema import User, Offer as DBOffer, Userright
+from iro.controller.pool  import dbPool
+from iro.error import NoProvider
+from ..test_helpers.dbtestcase import DBTestCase
+
+class TestOffers(DBTestCase):
+    '''test config class'''
+    def setUp(self):
+        DBTestCase.setUp(self)
+        dbPool.start(reactor)
+
+    def tearDown(self):
+        for s in configParser.sections():
+            configParser.remove_section(s)
+        
+        dbPool.pool.stop()
+        offer.offers.clear()
+        
+        try:
+            del(OfferProviders["test"])
+        except KeyError:
+            pass
+        
+        offer.providers.clear()
+        DBTestCase.tearDown(self)
+
+    def testReloadList(self):
+        '''test if loadOffers will be fired by reloading Config'''
+        self.assertTrue(offer.loadOffers in configParser.reloadList)
+
+    @inlineCallbacks
+    def testExtendProviderUnknown(self):
+        '''test the extendProvider Function (route unknown)'''
+        with self.session() as session:
+            u = User(name='test',apikey='abcdef123456789')
+            session.add(u)
+
+        ret = yield offer.extendProvider(u, "sms", ["blub"])
+        self.assertEqual(ret, [])
+
+    @inlineCallbacks
+    def testExtendProviderKnown(self):
+        '''test the extendProvider Function (route known)'''
+        with self.session() as session:
+            u = User(name='test',apikey='abcdef123456789')
+            session.add(u)
+            o=DBOffer(name="blub", provider="bla", route="basic", typ="sms")
+            u.rights.append(Userright(o)) 
+
+        ret = yield offer.extendProvider(u, "sms", ["blub"])
+        self.assertEqual(ret, ["blub"])
+
+        ret = yield offer.extendProvider(u, "sms", ["blub", "blub"])
+        self.assertEqual(ret, ["blub"])
+
+    @inlineCallbacks
+    def testExtendProviderProvider(self):
+        '''test the extendProvider Function (provider known)'''
+        with self.session() as session:
+            u = User(name='test',apikey='abcdef123456789')
+            session.add(u)
+            o=DBOffer(name="oh", provider="bla", route="a", typ="sms")
+            u.rights.append(Userright(o)) 
+        
+        offer.providers={"bla":Provider("bla", {"sms":["a","b","c"]})}
+
+        for l in [['bla'],['oh'],['oh','bla'],['bla','oh']]:
+            ret = yield offer.extendProvider(u, "sms", l)
+            self.assertEqual(ret, ["oh"])
+
+    @inlineCallbacks
+    def testExtendProviderDouble(self):
+        '''test the extendProvider Function (provider and name doubles)'''
+        with self.session() as session:
+            u = User(name='test',apikey='abcdef123456789')
+            session.add(u)
+            o=DBOffer(name="oh", provider="bla", route="a", typ="sms")
+            u.rights.append(Userright(o)) 
+            o=DBOffer(name="a", provider="bla", route="b", typ="sms")
+            u.rights.append(Userright(o)) 
+        
+        offer.providers={"bla":Provider("bla",  {"sms":["a","b","c"]})}
+
+        ret = yield offer.extendProvider(u, "sms",  ["bla"])
+        self.assertEqual(ret, ["oh","a"])
+    
+        ret = yield offer.extendProvider(u, "sms",  ["a","bla"])
+        self.assertEqual(ret, ["a","oh"])
+
+        ret = yield offer.extendProvider(u, "sms",  ["bla", "a"])
+        self.assertEqual(ret, ["oh","a"])
+
+    @inlineCallbacks
+    def testExtendProviderDefault(self):
+        '''test the extendProvider Function (default)'''
+        with self.session() as session:
+            u = User(name='test',apikey='abcdef123456789')
+            session.add(u)
+            o=DBOffer(name="blub3", provider="bla", route="basic", typ="sms")
+            u.rights.append(Userright(o,default=2)) 
+            o=DBOffer(name="blub", provider="bla", route="basic", typ="sms")
+            u.rights.append(Userright(o,default=1)) 
+            o=DBOffer(name="blub2", provider="bla2", route="basic", typ="sms")
+            u.rights.append(Userright(o)) 
+            o=DBOffer(name="blub4", provider="bla", route="basic", typ="sms")
+            u.rights.append(Userright(o,default=3)) 
+        
+        ret = yield offer.extendProvider(u, "sms", ["default"])
+        self.assertEqual(ret, ["blub","blub3",'blub4'])
+
+        ret = yield offer.extendProvider(u, "sms", "default")
+        self.assertEqual(ret, ["blub","blub3", 'blub4'])
+
+    
+    @inlineCallbacks
+    def testLoadOffers(self):
+        
+        class TestProvider(Provider):
+            def __init__(self,name):
+                Provider.__init__(self,name,{"sms":["a",]})
+        
+        with self.session() as session:
+            session.add(DBOffer(name="oh", provider="p", route="a", typ="sms"))
+
+        sample_config = """[p]
+typ = test
+"""
+        configParser.readfp(io.BytesIO(sample_config))
+        OfferProviders["test"]=TestProvider
+        yield offer.loadOffers()        
+        self.assertEqual(offer.offers.keys(),["oh"])
+        self.assertEqual(offer.providers.keys(),["p"])
+        self.assertEqual(offer.providers["p"].name,"p")
+        self.assertEqual(offer.providers["p"].typ,"test")
+        self.assertEqual(offer.offers["oh"],Offer(provider=offer.providers["p"], name="oh", route="a", typ="sms"))
+
+    def testLoadOffersUnknown(self):
+        with self.session() as session:
+            session.add(DBOffer(name="oh", provider="p", route="a", typ="sms"))
+
+        sample_config = """[p]
+typ = unknown
+"""
+        configParser.readfp(io.BytesIO(sample_config))
+        d = offer.loadOffers()        
+        self.assertFailure(d, NoProvider)
+        return d
+