1:02 p.m.
Create a specific file (test_objectstore.py) to handle all the Objectstore
tests.
Signed-off-by: Aline Manera <alinefm(a)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