--- a/iro/model/decorators.py Thu Mar 01 21:00:31 2012 +0100
+++ b/iro/model/decorators.py Fri Mar 02 02:50:17 2012 +0100
@@ -1,15 +1,36 @@
+import types
from decorator import decorator
from .user import vUser
+from .schema import Offer
from .dbdefer import dbdefer
from .pool import runInDBPool
from ..error import ValidateException
def vRoute(typ):
- @decorator
- def wrapper(f, *args, **kwargs):
- return f(*args, **kwargs)
+ @dbdefer
+ def wrapper(session, value, field, allowString=True, allowList=True):
+ str_ = False
+ ret = []
+
+ if type(value) is types.StringType:
+ if not allowString:
+ raise ValidateException(field=field,msg='%s must be a list of routes.'%field)
+ str_=True
+ value=[value]
+ elif not allowList:
+ raise ValidateException(field=field,msg='%s must be a route - No list of routes.'%field)
+
+ routes =[o.name for o in Offer.routes(session,typ)]
+ for v in value:
+ if v not in routes:
+ raise ValidateException(field=field,msg='Route %s is not valid.'%v)
+ if v not in ret:
+ ret.append(v)
+ if str_:
+ return ret[0]
+ return ret
return wrapper
@dbdefer
--- a/iro/model/schema.py Thu Mar 01 21:00:31 2012 +0100
+++ b/iro/model/schema.py Fri Mar 02 02:50:17 2012 +0100
@@ -58,6 +58,11 @@
def get(cls, session, provider, route, typ):
return session.query(cls).filter(and_(cls.provider==provider, cls.route==route, cls.typ==typ)).first()
+ @classmethod
+ def routes(cls, session, typ):
+ return session.query(cls).filter_by(typ=typ)
+
+
class Message(Base):
"""Wenn ein Vorgang von Iro Kosten erzeugt hat wird eine neue Zeile eingefügt. Solange nicht bezahlt wurde ist <em>isBilled=0</em>."""
__tablename__ = "message"
--- a/tests/db.py Thu Mar 01 21:00:31 2012 +0100
+++ b/tests/db.py Fri Mar 02 02:50:17 2012 +0100
@@ -84,12 +84,32 @@
with self.session() as session:
o=Offer(name="s2", provider="sipgate", route="basic", typ="sms")
+ session.add(o)
o=Offer(name="s3", provider="sipgate", route="basic", typ="sms2")
session.add(o)
with self.session() as session:
self.assertEqual(session.typs.order_by(Offer.typ).all(),[('sms',),('sms2',)])
+ def testOfferRoutes(self):
+ with self.session() as session:
+ o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")
+ session.add(o)
+
+ with self.session() as session:
+ self.assertEqual([o.name for o in Offer.routes(session, "sms").order_by(Offer.name)],["sipgate_basic"])
+
+ with self.session() as session:
+ o=Offer(name="s2", provider="sipgate", route="basic", typ="sms")
+ session.add(o)
+ o=Offer(name="s3", provider="sipgate", route="basic", typ="sms2")
+ session.add(o)
+
+ with self.session() as session:
+ self.assertEqual([o.name for o in Offer.routes(session, "sms2").order_by(Offer.name)],["s3"])
+ self.assertEqual([o.name for o in Offer.routes(session, "sms").order_by(Offer.name)],["s2","sipgate_basic"])
+
+
class Has_RightTests(DBTestCase):
'''test User.has_right'''
def testSimple(self):