[Kimchi-devel] [PATCH 1/2] Object store tests

[Aline Manera]

Create a specific file (test_objectstore.py) to handle all the Objectstore
tests.

Signed-off-by: Aline Manera <alinefm at linux.vnet.ibm.com>
---
 tests/test_model.py       | 54 --------------------------
 tests/test_objectstore.py | 96 +++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 96 insertions(+), 54 deletions(-)
 create mode 100644 tests/test_objectstore.py

diff --git a/tests/test_model.py b/tests/test_model.py
index 63bd721..2a57507 100644
--- a/tests/test_model.py
+++ b/tests/test_model.py
@@ -706,60 +706,6 @@ class ModelTests(unittest.TestCase):
         for t in threads:
             t.join()
 
-    def test_object_store(self):
-        store = kimchi.objectstore.ObjectStore(self.tmp_store)
-
-        with store as session:
-            # Test create
-            session.store('fǒǒ', 'těst1', {'α': 1})
-            session.store('fǒǒ', 'těst2', {'β': 2})
-
-            # Test list
-            items = session.get_list('fǒǒ')
-            self.assertTrue(u'těst1' in items)
-            self.assertTrue(u'těst2' in items)
-
-            # Test get
-            item = session.get('fǒǒ', 'těst1')
-            self.assertEquals(1, item[u'α'])
-
-            # Test delete
-            session.delete('fǒǒ', 'těst2')
-            self.assertEquals(1, len(session.get_list('fǒǒ')))
-
-            # Test get non-existent item
-
-            self.assertRaises(NotFoundError, session.get,
-                              'α', 'β')
-
-            # Test delete non-existent item
-            self.assertRaises(NotFoundError, session.delete,
-                              'fǒǒ', 'těst2')
-
-            # Test refresh existing item
-            session.store('fǒǒ', 'těst1', {'α': 2})
-            item = session.get('fǒǒ', 'těst1')
-            self.assertEquals(2, item[u'α'])
-
-    def test_object_store_threaded(self):
-        def worker(ident):
-            with store as session:
-                session.store('foo', ident, {})
-
-        store = kimchi.objectstore.ObjectStore(self.tmp_store)
-
-        threads = []
-        for i in xrange(50):
-            t = threading.Thread(target=worker, args=(i,))
-            t.setDaemon(True)
-            t.start()
-            threads.append(t)
-        for t in threads:
-            t.join()
-        with store as session:
-            self.assertEquals(50, len(session.get_list('foo')))
-            self.assertEquals(10, len(store._connections.keys()))
-
     def test_get_interfaces(self):
         inst = model.Model('test:///default',
                            objstore_loc=self.tmp_store)
diff --git a/tests/test_objectstore.py b/tests/test_objectstore.py
new file mode 100644
index 0000000..f62217e
--- /dev/null
+++ b/tests/test_objectstore.py
@@ -0,0 +1,96 @@
+# -*- coding: utf-8 -*-
+#
+# Project Kimchi
+#
+# Copyright IBM, Corp. 2015
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
+
+import os
+import tempfile
+import threading
+import unittest
+
+from kimchi.exception import NotFoundError
+from kimchi import objectstore
+
+tmpfile = None
+
+
+def setUpModule():
+    global tmpfile
+    tmpfile = tempfile.mktemp()
+
+
+def tearDownModule():
+    os.unlink(tmpfile)
+
+
+class ObjectStoreTests(unittest.TestCase):
+    def test_objectstore(self):
+        store = objectstore.ObjectStore(tmpfile)
+
+        with store as session:
+            # Test create
+            session.store('fǒǒ', 'těst1', {'α': 1})
+            session.store('fǒǒ', 'těst2', {'β': 2})
+
+            # Test list
+            items = session.get_list('fǒǒ')
+            self.assertTrue(u'těst1' in items)
+            self.assertTrue(u'těst2' in items)
+
+            # Test get
+            item = session.get('fǒǒ', 'těst1')
+            self.assertEquals(1, item[u'α'])
+
+            # Test delete
+            session.delete('fǒǒ', 'těst2')
+            self.assertEquals(1, len(session.get_list('fǒǒ')))
+
+            # Test get non-existent item
+
+            self.assertRaises(NotFoundError, session.get,
+                              'α', 'β')
+
+            # Test delete non-existent item
+            self.assertRaises(NotFoundError, session.delete,
+                              'fǒǒ', 'těst2')
+
+            # Test refresh existing item
+            session.store('fǒǒ', 'těst1', {'α': 2})
+            item = session.get('fǒǒ', 'těst1')
+            self.assertEquals(2, item[u'α'])
+
+    def test_object_store_threaded(self):
+        def worker(ident):
+            with store as session:
+                session.store('foo', ident, {})
+
+        store = objectstore.ObjectStore(tmpfile)
+
+        threads = []
+        for i in xrange(50):
+            t = threading.Thread(target=worker, args=(i,))
+            t.setDaemon(True)
+            t.start()
+            threads.append(t)
+
+        for t in threads:
+            t.join()
+
+        with store as session:
+            self.assertEquals(50, len(session.get_list('foo')))
+            self.assertEquals(10, len(store._connections.keys()))
-- 
2.1.0