|
1 from twisted.internet import reactor |
|
2 from twisted.internet.defer import inlineCallbacks |
|
3 |
|
4 import io |
|
5 |
|
6 from iro.model import offer |
|
7 from iro.config import configParser |
|
8 from iro.offer import Offer, providers as OfferProviders, Provider |
|
9 from iro.model.schema import User, Offer as DBOffer, Userright |
|
10 from iro.controller.pool import dbPool |
|
11 from iro.error import NoProvider |
|
12 from ..test_helpers.dbtestcase import DBTestCase |
|
13 |
|
14 class TestOffers(DBTestCase): |
|
15 '''test config class''' |
|
16 def setUp(self): |
|
17 DBTestCase.setUp(self) |
|
18 dbPool.start(reactor) |
|
19 |
|
20 def tearDown(self): |
|
21 for s in configParser.sections(): |
|
22 configParser.remove_section(s) |
|
23 |
|
24 dbPool.pool.stop() |
|
25 offer.offers.clear() |
|
26 |
|
27 try: |
|
28 del(OfferProviders["test"]) |
|
29 except KeyError: |
|
30 pass |
|
31 |
|
32 offer.providers.clear() |
|
33 DBTestCase.tearDown(self) |
|
34 |
|
35 def testReloadList(self): |
|
36 '''test if loadOffers will be fired by reloading Config''' |
|
37 self.assertTrue(offer.loadOffers in configParser.reloadList) |
|
38 |
|
39 @inlineCallbacks |
|
40 def testExtendProviderUnknown(self): |
|
41 '''test the extendProvider Function (route unknown)''' |
|
42 with self.session() as session: |
|
43 u = User(name='test',apikey='abcdef123456789') |
|
44 session.add(u) |
|
45 |
|
46 ret = yield offer.extendProvider(u, "sms", ["blub"]) |
|
47 self.assertEqual(ret, []) |
|
48 |
|
49 @inlineCallbacks |
|
50 def testExtendProviderKnown(self): |
|
51 '''test the extendProvider Function (route known)''' |
|
52 with self.session() as session: |
|
53 u = User(name='test',apikey='abcdef123456789') |
|
54 session.add(u) |
|
55 o=DBOffer(name="blub", provider="bla", route="basic", typ="sms") |
|
56 u.rights.append(Userright(o)) |
|
57 |
|
58 ret = yield offer.extendProvider(u, "sms", ["blub"]) |
|
59 self.assertEqual(ret, ["blub"]) |
|
60 |
|
61 ret = yield offer.extendProvider(u, "sms", ["blub", "blub"]) |
|
62 self.assertEqual(ret, ["blub"]) |
|
63 |
|
64 @inlineCallbacks |
|
65 def testExtendProviderProvider(self): |
|
66 '''test the extendProvider Function (provider known)''' |
|
67 with self.session() as session: |
|
68 u = User(name='test',apikey='abcdef123456789') |
|
69 session.add(u) |
|
70 o=DBOffer(name="oh", provider="bla", route="a", typ="sms") |
|
71 u.rights.append(Userright(o)) |
|
72 |
|
73 offer.providers={"bla":Provider("bla", {"sms":["a","b","c"]})} |
|
74 |
|
75 for l in [['bla'],['oh'],['oh','bla'],['bla','oh']]: |
|
76 ret = yield offer.extendProvider(u, "sms", l) |
|
77 self.assertEqual(ret, ["oh"]) |
|
78 |
|
79 @inlineCallbacks |
|
80 def testExtendProviderDouble(self): |
|
81 '''test the extendProvider Function (provider and name doubles)''' |
|
82 with self.session() as session: |
|
83 u = User(name='test',apikey='abcdef123456789') |
|
84 session.add(u) |
|
85 o=DBOffer(name="oh", provider="bla", route="a", typ="sms") |
|
86 u.rights.append(Userright(o)) |
|
87 o=DBOffer(name="a", provider="bla", route="b", typ="sms") |
|
88 u.rights.append(Userright(o)) |
|
89 |
|
90 offer.providers={"bla":Provider("bla", {"sms":["a","b","c"]})} |
|
91 |
|
92 ret = yield offer.extendProvider(u, "sms", ["bla"]) |
|
93 self.assertEqual(ret, ["oh","a"]) |
|
94 |
|
95 ret = yield offer.extendProvider(u, "sms", ["a","bla"]) |
|
96 self.assertEqual(ret, ["a","oh"]) |
|
97 |
|
98 ret = yield offer.extendProvider(u, "sms", ["bla", "a"]) |
|
99 self.assertEqual(ret, ["oh","a"]) |
|
100 |
|
101 @inlineCallbacks |
|
102 def testExtendProviderDefault(self): |
|
103 '''test the extendProvider Function (default)''' |
|
104 with self.session() as session: |
|
105 u = User(name='test',apikey='abcdef123456789') |
|
106 session.add(u) |
|
107 o=DBOffer(name="blub3", provider="bla", route="basic", typ="sms") |
|
108 u.rights.append(Userright(o,default=2)) |
|
109 o=DBOffer(name="blub", provider="bla", route="basic", typ="sms") |
|
110 u.rights.append(Userright(o,default=1)) |
|
111 o=DBOffer(name="blub2", provider="bla2", route="basic", typ="sms") |
|
112 u.rights.append(Userright(o)) |
|
113 o=DBOffer(name="blub4", provider="bla", route="basic", typ="sms") |
|
114 u.rights.append(Userright(o,default=3)) |
|
115 |
|
116 ret = yield offer.extendProvider(u, "sms", ["default"]) |
|
117 self.assertEqual(ret, ["blub","blub3",'blub4']) |
|
118 |
|
119 ret = yield offer.extendProvider(u, "sms", "default") |
|
120 self.assertEqual(ret, ["blub","blub3", 'blub4']) |
|
121 |
|
122 |
|
123 @inlineCallbacks |
|
124 def testLoadOffers(self): |
|
125 |
|
126 class TestProvider(Provider): |
|
127 def __init__(self,name): |
|
128 Provider.__init__(self,name,{"sms":["a",]}) |
|
129 |
|
130 with self.session() as session: |
|
131 session.add(DBOffer(name="oh", provider="p", route="a", typ="sms")) |
|
132 |
|
133 sample_config = """[p] |
|
134 typ = test |
|
135 """ |
|
136 configParser.readfp(io.BytesIO(sample_config)) |
|
137 OfferProviders["test"]=TestProvider |
|
138 yield offer.loadOffers() |
|
139 self.assertEqual(offer.offers.keys(),["oh"]) |
|
140 self.assertEqual(offer.providers.keys(),["p"]) |
|
141 self.assertEqual(offer.providers["p"].name,"p") |
|
142 self.assertEqual(offer.providers["p"].typ,"test") |
|
143 self.assertEqual(offer.offers["oh"],Offer(provider=offer.providers["p"], name="oh", route="a", typ="sms")) |
|
144 |
|
145 def testLoadOffersUnknown(self): |
|
146 with self.session() as session: |
|
147 session.add(DBOffer(name="oh", provider="p", route="a", typ="sms")) |
|
148 |
|
149 sample_config = """[p] |
|
150 typ = unknown |
|
151 """ |
|
152 configParser.readfp(io.BytesIO(sample_config)) |
|
153 d = offer.loadOffers() |
|
154 self.assertFailure(d, NoProvider) |
|
155 return d |
|
156 |