--- /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
+