--- a/iro/tests/install.py Fri Mar 23 16:16:56 2012 +0100
+++ b/iro/tests/install.py Fri Mar 23 16:18:12 2012 +0100
@@ -1,7 +1,8 @@
#from mock import patch, Mock
from twisted.trial import unittest
-from twisted.python import log
from sqlalchemy import create_engine
+from sets import Set
+import io
import os
from iro import install
@@ -93,6 +94,70 @@
self.assertEqual(len(self.log.e),1)
self.assertEqual(self.log.e[0]['message'], ("iro.conf exists and will not be overwritten.",))
- def testCheck(self):
- pass
- testCheck.todo = "to implement"
+ def testGetAllRoutesNone(self):
+ config.main.dburl=md.dburl2
+ install.createDatabase()
+ self.assertEqual(install.getAllRoutes([]),{"orphand":Set(),"added":Set()})
+
+ def testGetAllRoutesOrphand(self):
+ config.main.dburl=md.dburl2
+ install.createDatabase()
+ self.engine = create_engine(md.dburl2)
+ with WithSession(self.engine,True) as session:
+ session.add(Offer(provider="p",route="r",typ="t",name="test"))
+ session.add(Offer(provider="p",route="r2",typ="t",name="test2"))
+ self.assertEqual(install.getAllRoutes([]),{"orphand":Set(["test","test2"]),"added":Set()})
+
+ def testGetAllRoutesAdded(self):
+ config.main.dburl=md.dburl2
+ install.createDatabase()
+ self.engine = create_engine(md.dburl2)
+ class TestProvider(Provider):
+ def __init__(self, name):
+ Provider.__init__(self, name, {"r":["1","2","3"]})
+ providers["test"]=TestProvider
+ sample_config = """[p]
+typ= test
+test= foo
+"""
+ config.configParser.readfp(io.BytesIO(sample_config))
+ self.assertEqual(install.getAllRoutes(["p"]),{"orphand":Set(),"added":Set(["p_r_1","p_r_2","p_r_3"])})
+
+ #test writing
+ self.assertEqual(install.getAllRoutes(["p"],True),{"orphand":Set(),"added":Set(["p_r_1","p_r_2","p_r_3"])})
+ with WithSession(self.engine,True) as session:
+ o = Set([i[0] for i in session.query(Offer.name).all()])
+ self.assertEqual(o,Set(["p_r_1","p_r_2","p_r_3"]))
+
+ #nothing to do anymore
+ self.assertEqual(install.getAllRoutes(["p"]),{"orphand":Set(),"added":Set()})
+
+ def testGetAllRoutesOaA(self):
+ config.main.dburl=md.dburl2
+ install.createDatabase()
+ self.engine = create_engine(md.dburl2)
+ class TestProvider(Provider):
+ def __init__(self, name):
+ Provider.__init__(self, name, {"r":["1","2","3"]})
+ providers["test"]=TestProvider
+ sample_config = """[p]
+typ= test
+test= foo
+"""
+ config.configParser.readfp(io.BytesIO(sample_config))
+
+ with WithSession(self.engine,True) as session:
+ session.add(Offer(provider="q",route="r",typ="t",name="test"))
+ session.add(Offer(provider="q",route="r2",typ="t",name="test2"))
+ session.add(Offer(provider="p",route="1",typ="r",name="foo"))
+ self.assertEqual(install.getAllRoutes(["p"]),{"orphand":Set(["test","test2"]),"added":Set(["p_r_2","p_r_3"])})
+
+ #test writing
+ self.assertEqual(install.getAllRoutes(["p"],True),{"orphand":Set(["test","test2"]),"added":Set(["p_r_2","p_r_3"])})
+ with WithSession(self.engine,True) as session:
+ o = Set([i[0] for i in session.query(Offer.name).all()])
+ self.assertEqual(o,Set(["test","test2","foo","p_r_2","p_r_3"]))
+
+ self.assertEqual(install.getAllRoutes(["p"]),{"orphand":Set(["test","test2"]),"added":Set()})
+
+