99% Coverage for celery.datastructures

celery/datastructures.py

@@ -128,14 +128,16 @@ class ConfigurationView(AttributeDictMixin):
         return repr(dict(iter(self)))
 
     def __iter__(self):
-        return chain(*[d.iteritems() for d in (self.__dict__["changes"],
-                                               self.__dict__["defaults"])])
+        # defaults must be first in the stream, so values in
+        # in changes takes precedence.
+        return chain(*[d.iteritems() for d in (self.__dict__["defaults"],
+                                               self.__dict__["changes"])])
 
     def iteritems(self):
         return iter(self)
 
-    def iter(self):
-        return tuple(iter(self))
+    def items(self):
+        return tuple(self.iteritems())
 
 
 class PositionQueue(UserList):

celery/tests/test_datastructures.py

@@ -4,7 +4,71 @@ from Queue import Queue
 
 from celery.datastructures import PositionQueue, ExceptionInfo, LocalCache
 from celery.datastructures import LimitedSet, SharedCounter, consume_queue
-from celery.datastructures import AttributeDict
+from celery.datastructures import AttributeDict, DictAttribute
+from celery.datastructures import ConfigurationView
+
+
+class Object(object):
+    pass
+
+
+class test_DictAttribute(unittest.TestCase):
+
+    def test_get_set(self):
+        x = DictAttribute(Object())
+        x["foo"] = "The quick brown fox"
+        self.assertEqual(x["foo"], "The quick brown fox")
+        self.assertEqual(x["foo"], x.obj.foo)
+        self.assertEqual(x.get("foo"), "The quick brown fox")
+        self.assertIsNone(x.get("bar"))
+        self.assertRaises(KeyError, x.__getitem__, "bar")
+
+    def test_setdefault(self):
+        x = DictAttribute(Object())
+        self.assertEqual(x.setdefault("foo", "NEW"), "NEW")
+        self.assertEqual(x.setdefault("foo", "XYZ"), "NEW")
+
+    def test_contains(self):
+        x = DictAttribute(Object())
+        x["foo"] = 1
+        self.assertIn("foo", x)
+        self.assertNotIn("bar", x)
+
+    def test_iteritems(self):
+        obj = Object()
+        obj.attr1 = 1
+        x = DictAttribute(obj)
+        x["attr2"] = 2
+        self.assertDictEqual(dict(x.iteritems()),
+                             dict(attr1=1, attr2=2))
+
+
+class test_ConfigurationView(unittest.TestCase):
+
+    def setUp(self):
+        self.view = ConfigurationView({"changed_key": 1,
+                                       "both": 2},
+                                      {"default_key": 1,
+                                       "both": 1})
+
+    def test_setdefault(self):
+        self.assertEqual(self.view.setdefault("both", 36), 2)
+        self.assertEqual(self.view.setdefault("new", 36), 36)
+
+    def test_contains(self):
+        self.assertIn("changed_key", self.view)
+        self.assertIn("default_key", self.view)
+        self.assertNotIn("new", self.view)
+
+    def test_repr(self):
+        self.assertIn("changed_key", repr(self.view))
+        self.assertIn("default_key", repr(self.view))
+
+    def test_iter(self):
+        expected = {"changed_key": 1,
+                    "default_key": 1,
+                    "both": 2}
+        self.assertDictEqual(dict(self.view.items()), expected)
 
 
 class test_PositionQueue(unittest.TestCase):