1from couchstore import CouchStore, CouchStoreException, DocumentInfo
2import os
3import unittest
4
5def removeIfExists(path):
6    try:
7        os.remove(path)
8    except OSError as x:
9        pass
10
11class NonexistentCouchStoreTest (unittest.TestCase):
12    def testNonexistent(self):
13        removeIfExists("/tmp/nonexistent.couch")
14        self.assertRaises(OSError, CouchStore, "/tmp/nonexistent.couch")
15
16class CouchStoreTest (unittest.TestCase):
17    def setUp(self):
18        removeIfExists("/tmp/test.couch")
19        self.store = CouchStore("/tmp/test.couch", 'c')
20
21    def tearDown(self):
22        self.store.close()
23        os.remove("/tmp/test.couch")
24
25    def testBasicSave(self):
26        sequence = self.store.save("foo", "value of foo")
27        self.assertEqual(sequence, 1)
28        self.store.commit()
29        value = self.store.get("foo")
30        self.assertEqual(value, "value of foo")
31
32    def testMissingKey(self):
33        self.assertRaises(KeyError, self.store.get, "nonexistent")
34
35    def testBadKey(self):
36        self.assertRaises(TypeError, self.store.get, 0)
37        self.assertRaises(TypeError, self.store.get, None)
38        self.assertRaises(TypeError, self.store.get, [123])
39
40    def testInfo(self):
41        value = "value"
42        sequence = self.store.save("foo", value)
43        self.assertEqual(sequence, 1)
44        info = self.store.getInfo("foo")
45        self.assertEqual(info.id, "foo")
46        self.assertEqual(info.sequence, sequence)
47        self.assertFalse(info.deleted)
48        #self.assertEqual(info.size, len(value))   #FIXME: Not currently equal, due to bug in CouchStore itself
49        self.assertEqual(info.getContents(), value)
50
51    def testInfoBySequence(self):
52        value = "value"
53        sequence = self.store.save("foo", value)
54        self.assertEqual(sequence, 1)
55        info = self.store.getInfoBySequence(sequence)
56        self.assertEqual(info.id, "foo")
57        self.assertEqual(info.sequence, sequence)
58        self.assertFalse(info.deleted)
59        #self.assertEqual(info.size, len(value))   #FIXME: Not currently equal, due to bug in CouchStore itself
60        self.assertEqual(info.getContents(), value)
61
62    def testMissingSequence(self):
63        self.store.save("foo", "value")
64        self.assertRaises(KeyError, self.store.getInfoBySequence, 99999)
65        self.assertRaises(TypeError, self.store.getInfoBySequence, "huh")
66
67    def testNoContents(self):
68        info = DocumentInfo("howdy")
69        self.assertRaises(Exception, info.getContents)
70
71    def testMetadata(self):
72        info = DocumentInfo("meta")
73        info.revSequence = 23
74        info.revMeta = "fancy metadata here"
75        info.contentType = DocumentInfo.INVALID_JSON
76        self.store[info] = "the regular non-meta data"
77
78        gotInfo = self.store.getInfo("meta")
79        self.assertEquals(gotInfo.id, "meta")
80        self.assertEquals(gotInfo.revSequence, info.revSequence)
81        self.assertEquals(gotInfo.revMeta, info.revMeta)
82        self.assertEquals(gotInfo.contentType, info.contentType)
83        self.assertFalse(gotInfo.compressed)
84
85    def testCompression(self):
86        value = "this value is text and text is valued"
87        self.store.save("key", value, CouchStore.COMPRESS)
88        self.assertEqual(self.store.get("key", CouchStore.DECOMPRESS), value)
89        info = self.store.getInfo("key")
90        self.assertTrue(info.compressed)
91
92    def expectedKey(self, i):
93        return "key_%d" % (i+1)
94    def expectedValue(self, i):
95        return "Hi there! I'm value #%d!" % (i+1)
96    def addDocs(self, n):
97        for i in xrange(n):
98            self.store.save(self.expectedKey(i), self.expectedValue(i))
99    def addBulkDocs(self, n):
100        ids = [self.expectedKey(i) for i in xrange(n)]
101        datas = [self.expectedValue(i) for i in xrange(n)]
102        self.store.saveMultiple(ids, datas)
103
104    def testMultipleDocs(self):
105        self.addDocs(1000)
106        for i in xrange(1000):
107            self.assertEqual(self.store[self.expectedKey(i)], self.expectedValue(i))
108
109    def testBulkDocs(self):
110        self.addBulkDocs(1000)
111        for i in xrange(1000):
112            self.assertEqual(self.store[self.expectedKey(i)], self.expectedValue(i))
113
114    def testDelete(self):
115        self.store["key"] = "value"
116        del self.store["key"]
117        self.assertRaises(KeyError, self.store.get, "key")
118        info = self.store.getInfo("key")
119        self.assertTrue(info.deleted)
120        self.assertEqual(info.id, "key")
121
122    def testChangesSince(self):
123        self.addDocs(50)
124        changes = self.store.changesSince(0)
125        self.assertEqual(len(changes), 50)
126        for i in xrange(50):
127            self.assertEqual(changes[i].id, self.expectedKey(i))
128
129    def testLocalDocs(self):
130        locals = self.store.localDocs
131        self.assertRaises(KeyError, locals.__getitem__, "hello")
132        locals["hello"] = "goodbye"
133        self.assertEqual(locals["hello"], "goodbye")
134        locals["hello"] = "bonjour"
135        self.assertEqual(locals["hello"], "bonjour")
136        del locals["hello"]
137        self.assertRaises(KeyError, locals.__getitem__, "hello")
138
139if __name__ == '__main__':
140    unittest.main()
141