2
Tests for CouchDBDocumentStore module.
4
# This file is part of MAUS: http://micewww.pp.rl.ac.uk:8080/projects/maus
6
# MAUS is free software: you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation, either version 3 of the License, or
9
# (at your option) any later version.
11
# MAUS is distributed in the hope that it will be useful,
12
# but WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14
# GNU General Public License for more details.
16
# You should have received a copy of the GNU General Public License
17
# along with MAUS. If not, see <http://www.gnu.org/licenses/>.
19
# pylint: disable=C0103
22
from socket import gaierror
25
from CouchDBDocumentStore import CouchDBDocumentStore
27
class CouchDBDocumentStoreTestCase(unittest.TestCase): # pylint: disable=R0904, C0301
29
Test class for CouchDBDocumentStore module.
34
Create CouchDBDocumentStore with test-specific database.
35
@param self Object reference.
37
self._database_url = "http://localhost:5984"
38
# Ping server to validate it's available.
39
server = couchdb.Server(self._database_url)
42
except:# pylint: disable=W0702
43
unittest.TestCase.skipTest(self,
44
"CouchDB server is not accessible")
45
self._database_name = self.__class__.__name__.lower()
46
# Create data store and connect.
47
self._data_store = CouchDBDocumentStore()
49
"couchdb_url":self._database_url,
50
"couchdb_database_name":self._database_name}
51
self._data_store.connect(parameters)
55
Delete test-specific database.
56
@param self Object reference.
58
server = couchdb.Server(self._database_url)
59
if self._database_name in server:
60
server.delete(self._database_name)
62
def test_connect(self):
64
Test default invocation of connect in setUp has created
65
a database on the server with the expected name.
66
@param self Object reference.
68
server = couchdb.Server(self._database_url)
69
self.assertTrue(self._database_name in server,
70
"Database has not been created on the server")
72
def test_connect_no_parameters(self):
74
Test connect with no parameters throws a KeyError as a database
76
@param self Object reference.
79
self._data_store.connect({})
83
def test_connect_bad_url(self):
85
Test connect with a bad URL throws a gaierror as expected.
86
@param self Object reference.
88
parameters = {"couchdb_url":"non-existant"}
90
self._data_store.connect(parameters)
94
def test_connect_no_database_name(self):
96
Test connect with no parameters throws a KeyError as a database
98
@param self Object reference.
100
parameters = {"couchdb_url":self._database_url}
102
self._data_store.connect(parameters)
106
def test_connect_existing_database_name(self):
108
Test connect with an existing database name.
109
@param self Object reference.
112
"couchdb_url":self._database_url,
113
"couchdb_database_name":self._database_name}
114
self._data_store.connect(parameters)
116
def test_empty_data_store(self):
118
Test get, delete, clear, len, ids on an empty data store.
119
@param self Object reference.
121
self.assertEquals(0, len(self._data_store),
123
self.assertEquals(0, len(self._data_store.ids()),
124
"Unexpected number of IDs")
125
self.assertEquals(None, self._data_store.get("ID"),
126
"Expected document to be None")
127
# Expect no exceptions.
128
self._data_store.delete("ID")
129
# Expect no exceptions.
130
self._data_store.clear()
132
def test_put_get(self):
134
Test put and get, and also ids and len.
135
@param self Object reference.
138
doc1 = {'a1':'b1', 'c1':'d1'}
139
self._data_store.put("ID1", doc1)
140
doc2 = {'a2':'b2', 'c2':'d2'}
141
self._data_store.put("ID2", doc2)
143
self.assertEquals(2, len(self._data_store),
145
ids = self._data_store.ids()
146
self.assertEquals(2, len(ids), "Unexpected number of IDs")
147
self.assertTrue("ID1" in ids, "ID1 was not in ids")
148
self.assertTrue("ID2" in ids, "ID2 was not in ids")
149
self.assertEquals(doc1, self._data_store.get("ID1"),
150
"Unexpected document for ID1")
151
self.assertEquals(doc2, self._data_store.get("ID2"),
152
"Unexpected document for ID2")
154
def test_put_put(self):
156
Test put then another with the same ID.
157
@param self Object reference.
160
doc = {'a1':'b1', 'c1':'d1'}
161
self._data_store.put("ID", doc)
162
self.assertEquals(doc, self._data_store.get("ID"),
163
"Unexpected document for ID")
165
nudoc = {'a2':'b2', 'c2':'d2'}
166
self._data_store.put("ID", nudoc)
168
self.assertEquals(nudoc, self._data_store.get("ID"),
169
"Unexpected document for ID")
171
def test_delete(self):
174
@param self Object reference.
176
self._data_store.put("ID1", {'a1':'b1', 'c1':'d1'})
177
self._data_store.put("ID2", {'a2':'b2', 'c2':'d2'})
179
self._data_store.delete("ID1")
180
self.assertEquals(1, len(self._data_store),
182
ids = self._data_store.ids()
183
self.assertEquals(1, len(ids), "Unexpected number of IDs")
184
self.assertTrue(not "ID1" in ids, "ID1 was not in ids")
186
self._data_store.delete("ID2")
187
self.assertEquals(0, len(self._data_store),
189
ids = self._data_store.ids()
190
self.assertEquals(0, len(ids), "Unexpected number of IDs")
192
def test_clear(self):
195
@param self Object reference.
197
self._data_store.put("ID1", {'a1':'b1', 'c1':'d1'})
198
self._data_store.put("ID2", {'a2':'b2', 'c2':'d2'})
199
self._data_store.clear()
200
self.assertEquals(0, len(self._data_store),
202
self.assertEquals(0, len(self._data_store.ids()),
203
"Unexpected number of IDs")
205
if __name__ == '__main__':
added
removed
tests/py_unit/test_CouchDBDocumentStore.py
