2 from twisted.internet.defer import inlineCallbacks |
2 from twisted.internet.defer import inlineCallbacks |
3 |
3 |
4 import io |
4 import io |
5 |
5 |
6 from iro.model import offer |
6 from iro.model import offer |
7 from iro.config import config |
7 from iro.config import configParser |
8 from iro.offer import Offer, providers as OfferProviders, Provider |
8 from iro.offer import Offer, providers as OfferProviders, Provider |
9 from iro.model.schema import User, Offer as DBOffer, Userright |
9 from iro.model.schema import User, Offer as DBOffer, Userright |
10 from iro.controller.pool import dbPool |
10 from iro.controller.pool import dbPool |
11 from iro.error import NoProvider |
11 from iro.error import NoProvider |
12 from .dbtestcase import DBTestCase |
12 from .dbtestcase import DBTestCase |
32 offer.providers.clear() |
32 offer.providers.clear() |
33 DBTestCase.tearDown(self) |
33 DBTestCase.tearDown(self) |
34 |
34 |
35 def testReloadList(self): |
35 def testReloadList(self): |
36 '''test if loadOffers will be fired by reloading Config''' |
36 '''test if loadOffers will be fired by reloading Config''' |
37 self.assertTrue(offer.loadOffers in config.reloadList) |
37 self.assertTrue(offer.loadOffers in configParser.reloadList) |
38 |
38 |
39 @inlineCallbacks |
39 @inlineCallbacks |
40 def testExtendProviderUnknown(self): |
40 def testExtendProviderUnknown(self): |
41 '''test the extendProvider Function (route unknown)''' |
41 '''test the extendProvider Function (route unknown)''' |
42 with self.session() as session: |
42 with self.session() as session: |
70 u = User(name='test',apikey='abcdef123456789') |
70 u = User(name='test',apikey='abcdef123456789') |
71 session.add(u) |
71 session.add(u) |
72 o=DBOffer(name="oh", provider="bla", route="a", typ="sms") |
72 o=DBOffer(name="oh", provider="bla", route="a", typ="sms") |
73 u.rights.append(Userright(o)) |
73 u.rights.append(Userright(o)) |
74 |
74 |
75 offer.providers={"bla":Provider("bla", [], {"sms":["a","b","c"]})} |
75 offer.providers={"bla":Provider("bla", {"sms":["a","b","c"]})} |
76 |
76 |
77 for l in [['bla'],['oh'],['oh','bla'],['bla','oh']]: |
77 for l in [['bla'],['oh'],['oh','bla'],['bla','oh']]: |
78 ret = yield offer.extendProvider(u, "sms", l) |
78 ret = yield offer.extendProvider(u, "sms", l) |
79 self.assertEqual(ret, ["oh"]) |
79 self.assertEqual(ret, ["oh"]) |
80 |
80 |
87 o=DBOffer(name="oh", provider="bla", route="a", typ="sms") |
87 o=DBOffer(name="oh", provider="bla", route="a", typ="sms") |
88 u.rights.append(Userright(o)) |
88 u.rights.append(Userright(o)) |
89 o=DBOffer(name="a", provider="bla", route="b", typ="sms") |
89 o=DBOffer(name="a", provider="bla", route="b", typ="sms") |
90 u.rights.append(Userright(o)) |
90 u.rights.append(Userright(o)) |
91 |
91 |
92 offer.providers={"bla":Provider("bla", [], {"sms":["a","b","c"]})} |
92 offer.providers={"bla":Provider("bla", {"sms":["a","b","c"]})} |
93 |
93 |
94 ret = yield offer.extendProvider(u, "sms", ["bla"]) |
94 ret = yield offer.extendProvider(u, "sms", ["bla"]) |
95 self.assertEqual(ret, ["oh","a"]) |
95 self.assertEqual(ret, ["oh","a"]) |
96 |
96 |
97 ret = yield offer.extendProvider(u, "sms", ["a","bla"]) |
97 ret = yield offer.extendProvider(u, "sms", ["a","bla"]) |
102 |
102 |
103 @inlineCallbacks |
103 @inlineCallbacks |
104 def testLoadOffers(self): |
104 def testLoadOffers(self): |
105 |
105 |
106 class TestProvider(Provider): |
106 class TestProvider(Provider): |
107 def __init__(self,name,config): |
107 def __init__(self,name): |
108 Provider.__init__(self,name,config,{"sms":["a",]}) |
108 Provider.__init__(self,name,{"sms":["a",]}) |
109 self.loadConfig() |
|
110 |
109 |
111 with self.session() as session: |
110 with self.session() as session: |
112 session.add(DBOffer(name="oh", provider="p", route="a", typ="sms")) |
111 session.add(DBOffer(name="oh", provider="p", route="a", typ="sms")) |
113 |
112 |
114 sample_config = """[p] |
113 sample_config = """[p] |
115 typ = test |
114 typ = test |
116 """ |
115 """ |
117 config.readfp(io.BytesIO(sample_config)) |
116 configParser.readfp(io.BytesIO(sample_config)) |
118 OfferProviders["test"]=TestProvider |
117 OfferProviders["test"]=TestProvider |
119 yield offer.loadOffers() |
118 yield offer.loadOffers() |
120 self.assertEqual(offer.offers.keys(),["oh"]) |
119 self.assertEqual(offer.offers.keys(),["oh"]) |
121 self.assertEqual(offer.providers.keys(),["p"]) |
120 self.assertEqual(offer.providers.keys(),["p"]) |
122 self.assertEqual(offer.providers["p"].name,"p") |
121 self.assertEqual(offer.providers["p"].name,"p") |
128 session.add(DBOffer(name="oh", provider="p", route="a", typ="sms")) |
127 session.add(DBOffer(name="oh", provider="p", route="a", typ="sms")) |
129 |
128 |
130 sample_config = """[p] |
129 sample_config = """[p] |
131 typ = unknown |
130 typ = unknown |
132 """ |
131 """ |
133 config.readfp(io.BytesIO(sample_config)) |
132 configParser.readfp(io.BytesIO(sample_config)) |
134 d = offer.loadOffers() |
133 d = offer.loadOffers() |
135 self.assertFailure(d, NoProvider) |
134 self.assertFailure(d, NoProvider) |
136 return d |
135 return d |
137 |
136 |