1from couchstore import CouchStore, CouchStoreException, DocumentInfo, SizedBuf
2import os
3import struct
4import unittest
5
6
7def removeIfExists(path):
8    try:
9        os.remove(path)
10    except OSError as x:
11        pass
12
13
14class NonexistentCouchStoreTest (unittest.TestCase):
15    def testNonexistent(self):
16        removeIfExists("/tmp/nonexistent.couch")
17        self.assertRaises(OSError, CouchStore, "/tmp/nonexistent.couch")
18
19
20class CouchStoreTest (unittest.TestCase):
21    def setUp(self):
22        removeIfExists("/tmp/test.couch")
23        self.store = CouchStore("/tmp/test.couch", 'c')
24
25    def tearDown(self):
26        self.store.close()
27        os.remove("/tmp/test.couch")
28
29    def testBasicSave(self):
30        sequence = self.store.save("foo", "value of foo")
31        self.assertEqual(sequence, 1)
32        self.store.commit()
33        value = self.store.get("foo")
34        self.assertEqual(value, "value of foo")
35
36    def testMissingKey(self):
37        self.assertRaises(KeyError, self.store.get, "nonexistent")
38
39    def testBadKey(self):
40        self.assertRaises(TypeError, self.store.get, 0)
41        self.assertRaises(TypeError, self.store.get, None)
42        self.assertRaises(TypeError, self.store.get, [123])
43
44    def testInfo(self):
45        value = "value"
46        sequence = self.store.save("foo", value)
47        self.assertEqual(sequence, 1)
48        info = self.store.getInfo("foo")
49        self.assertEqual(info.id, "foo")
50        self.assertEqual(info.sequence, sequence)
51        self.assertFalse(info.deleted)
52        #self.assertEqual(info.size, len(value))   #FIXME: Not currently equal, due to bug in CouchStore itself
53        self.assertEqual(info.getContents(), value)
54
55    def testInfoBySequence(self):
56        value = "value"
57        sequence = self.store.save("foo", value)
58        self.assertEqual(sequence, 1)
59        info = self.store.getInfoBySequence(sequence)
60        self.assertEqual(info.id, "foo")
61        self.assertEqual(info.sequence, sequence)
62        self.assertFalse(info.deleted)
63        #self.assertEqual(info.size, len(value))   #FIXME: Not currently equal, due to bug in CouchStore itself
64        self.assertEqual(info.getContents(), value)
65
66    def testMissingSequence(self):
67        self.store.save("foo", "value")
68        self.assertRaises(KeyError, self.store.getInfoBySequence, 99999)
69        self.assertRaises(TypeError, self.store.getInfoBySequence, "huh")
70
71    def testNoContents(self):
72        info = DocumentInfo("howdy")
73        self.assertRaises(Exception, info.getContents)
74
75    def testMetadata(self):
76        info = DocumentInfo("meta")
77        info.revSequence = 23
78        info.revMeta = "fancy metadata here"
79        info.contentType = DocumentInfo.INVALID_JSON
80        self.store[info] = "the regular non-meta data"
81
82        gotInfo = self.store.getInfo("meta")
83        self.assertEquals(gotInfo.id, "meta")
84        self.assertEquals(gotInfo.revSequence, info.revSequence)
85        self.assertEquals(gotInfo.revMeta, info.revMeta)
86        self.assertEquals(gotInfo.contentType, info.contentType)
87        self.assertFalse(gotInfo.compressed)
88
89    def testMetadataSave(self):
90        info = DocumentInfo("meta")
91        info.revSequence = 23
92        info.revMeta = "fancy metadata here"
93        info.contentType = DocumentInfo.INVALID_JSON
94        self.store[info] = "the regular non-meta data"
95
96        self.store.commit()
97        self.store.close()
98        self.store = CouchStore("/tmp/test.couch", 'r')
99
100        gotInfo = self.store.getInfo("meta")
101        self.assertEquals(gotInfo.id, "meta")
102        self.assertEquals(gotInfo.revSequence, info.revSequence)
103        self.assertEquals(gotInfo.revMeta, info.revMeta)
104        self.assertEquals(gotInfo.contentType, info.contentType)
105        self.assertFalse(gotInfo.compressed)
106
107    def testCompression(self):
108        value = "this value is text and text is valued"
109        self.store.save("key", value, CouchStore.COMPRESS)
110        self.assertEqual(self.store.get("key", CouchStore.DECOMPRESS), value)
111        info = self.store.getInfo("key")
112        self.assertTrue(info.compressed)
113
114    def expectedKey(self, i):
115        return "key_%2d" % (i+1)
116    def expectedValue(self, i):
117        return "Hi there! I'm value #%d!" % (i+1)
118    def addDocs(self, n):
119        for i in xrange(n):
120            self.store.save(self.expectedKey(i), self.expectedValue(i))
121    def addBulkDocs(self, n):
122        ids = [self.expectedKey(i) for i in xrange(n)]
123        datas = [self.expectedValue(i) for i in xrange(n)]
124        self.store.saveMultiple(ids, datas)
125
126    def testMultipleDocs(self):
127        self.addDocs(1000)
128        for i in xrange(1000):
129            self.assertEqual(self.store[self.expectedKey(i)], self.expectedValue(i))
130
131        info = self.store.getDbInfo()
132        self.assertEquals(info.filename, "/tmp/test.couch")
133        self.assertEquals(info.last_sequence, 1000)
134        self.assertEquals(info.doc_count, 1000)
135        self.assertEquals(info.deleted_count, 0)
136
137    def testBulkDocs(self):
138        self.addBulkDocs(1000)
139        for i in xrange(1000):
140            self.assertEqual(self.store[self.expectedKey(i)], self.expectedValue(i))
141
142    def testDelete(self):
143        self.store["key"] = "value"
144        del self.store["key"]
145        self.assertRaises(KeyError, self.store.get, "key")
146        info = self.store.getInfo("key")
147        self.assertTrue(info.deleted)
148        self.assertEqual(info.id, "key")
149
150        info = self.store.getDbInfo()
151        self.assertEquals(info.last_sequence, 2)
152        self.assertEquals(info.doc_count, 0)
153        self.assertEquals(info.deleted_count, 1)
154
155    def testChangesSince(self):
156        self.addDocs(50)
157        changes = self.store.changesSince(0)
158        self.assertEqual(len(changes), 50)
159        for i in xrange(50):
160            self.assertEqual(changes[i].id, self.expectedKey(i))
161
162    def testForAllDocs(self):
163        self.addDocs(50)
164        docCount = [0]
165        def checkDoc(docInfo):
166            self.assertEquals(docInfo.id, self.expectedKey(docCount[0]))
167            docCount[0] += 1
168        self.store.forEachDoc(None, None, checkDoc)
169        self.assertEqual(docCount[0], 50)
170
171    def testForSomeDocs(self):
172        self.addDocs(50)
173        docCount = [0]
174        def checkDoc(docInfo):
175            self.assertEquals(docInfo.id, self.expectedKey(docCount[0]))
176            docCount[0] += 1
177
178        self.store.forEachDoc(None, self.expectedKey(10), checkDoc)
179        self.assertEqual(docCount[0], 11)
180
181        docCount = [10]
182        self.store.forEachDoc(self.expectedKey(10), None, checkDoc)
183        self.assertEqual(docCount[0], 50)
184
185        docCount = [10]
186        self.store.forEachDoc(self.expectedKey(10), self.expectedKey(20), checkDoc)
187        self.assertEqual(docCount[0], 21)
188
189    def testLocalDocs(self):
190        locals = self.store.localDocs
191        self.assertRaises(KeyError, locals.__getitem__, "hello")
192        locals["hello"] = "goodbye"
193        self.assertEqual(locals["hello"], "goodbye")
194        locals["hello"] = "bonjour"
195        self.assertEqual(locals["hello"], "bonjour")
196        del locals["hello"]
197        self.assertRaises(KeyError, locals.__getitem__, "hello")
198
199    def testSizedBuf(self):
200        # Converting Python strings to/from SizedBufs is tricky enough (when the strings might
201        # contain null bytes) that it's worth a unit test of its own.
202        data = "foooooobarrrr"
203        buf = SizedBuf(data)
204        self.assertEqual(buf.size, len(data))
205        self.assertEqual(str(buf), data)
206        # Now try some binary data with nul bytes in it:
207        data = "foo\000bar"
208        buf = SizedBuf(data)
209        self.assertEqual(buf.size, len(data))
210        self.assertEqual(str(buf), data)
211
212    def testBinaryMeta(self):
213        # Make sure binary data, as produced by Python's struct module, works in revMeta.
214        packed = struct.pack(">QII", 0, 1, 2)
215        d = DocumentInfo("bin")
216        d.revMeta = packed
217        self.store[d] = "value"
218
219        doc_info = self.store.getInfo("bin")
220        self.assertEqual(doc_info.revMeta, packed)
221        i1, i2, i3 = struct.unpack(">QII", doc_info.revMeta)
222        self.assertEqual(i1, 0)
223        self.assertEqual(i2, 1)
224        self.assertEqual(i3, 2)
225
226    def testMultipleMeta(self):
227        k = []
228        v = []
229        for i in range(1000):
230            d = DocumentInfo(str(i))
231            d.revMeta = "hello-%s" % (i)
232            k.append(d)
233            v.append("world-%s" % (i))
234        self.store.saveMultiple(k, v)
235        self.store.commit()
236        self.store.close()
237        self.store = CouchStore("/tmp/test.couch", 'r')
238        for doc_info in self.store.changesSince(0):
239            i = int(doc_info.id)
240            self.assertEqual(doc_info.revMeta, "hello-%s" % (i))
241            doc_contents = doc_info.getContents()
242            self.assertEqual(doc_contents, "world-%s" % (i))
243
244    def testMultipleMetaStruct(self):
245        k = []
246        v = []
247        for i in range(1000):
248            d = DocumentInfo(str(i))
249            d.revMeta = struct.pack(">QII", i*3, i*2, i)
250            k.append(d)
251            v.append("world-%s" % (i))
252        self.store.saveMultiple(k, v)
253        self.store.commit()
254        self.store.close()
255        self.store = CouchStore("/tmp/test.couch", 'r')
256        for doc_info in self.store.changesSince(0):
257            i = int(doc_info.id)
258            i3, i2, i1 = struct.unpack(">QII", doc_info.revMeta)
259            self.assertEqual(i3, i*3)
260            self.assertEqual(i2, i*2)
261            self.assertEqual(i1, i*1)
262            doc_contents = doc_info.getContents()
263            self.assertEqual(doc_contents, "world-%s" % (doc_info.id))
264
265
266if __name__ == '__main__':
267    unittest.main()
268