1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
# Copyright 2011 Canonical Ltd.
#
# This file is part of u1db.
#
# u1db is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License version 3
# as published by the Free Software Foundation.
#
# u1db 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 u1db. If not, see <http://www.gnu.org/licenses/>.
"""Test in-memory backend internals."""
from u1db import (
errors,
tests,
)
from u1db.backends import inmemory
simple_doc = '{"key": "value"}'
class TestInMemoryDatabaseInternals(tests.TestCase):
def setUp(self):
super(TestInMemoryDatabaseInternals, self).setUp()
self.db = inmemory.InMemoryDatabase('test')
def test__allocate_doc_rev_from_None(self):
self.assertEqual('test:1', self.db._allocate_doc_rev(None))
def test__allocate_doc_rev_incremental(self):
self.assertEqual('test:2', self.db._allocate_doc_rev('test:1'))
def test__allocate_doc_rev_other(self):
self.assertEqual('replica:1|test:1',
self.db._allocate_doc_rev('replica:1'))
def test__get_replica_uid(self):
self.assertEqual('test', self.db._replica_uid)
class TestInMemoryIndex(tests.TestCase):
def test_has_name_and_definition(self):
idx = inmemory.InMemoryIndex('idx-name', ['key'])
self.assertEqual('idx-name', idx._name)
self.assertEqual(['key'], idx._definition)
def test_evaluate_json(self):
idx = inmemory.InMemoryIndex('idx-name', ['key'])
self.assertEqual(['value'], idx.evaluate_json(simple_doc))
def test_evaluate_json_field_None(self):
idx = inmemory.InMemoryIndex('idx-name', ['missing'])
self.assertEqual([], idx.evaluate_json(simple_doc))
def test_evaluate_json_subfield_None(self):
idx = inmemory.InMemoryIndex('idx-name', ['key', 'missing'])
self.assertEqual([], idx.evaluate_json(simple_doc))
def test_evaluate_multi_index(self):
doc = '{"key": "value", "key2": "value2"}'
idx = inmemory.InMemoryIndex('idx-name', ['key', 'key2'])
self.assertEqual(['value\x01value2'],
idx.evaluate_json(doc))
def test_update_ignores_None(self):
idx = inmemory.InMemoryIndex('idx-name', ['nokey'])
idx.add_json('doc-id', simple_doc)
self.assertEqual({}, idx._values)
def test_update_adds_entry(self):
idx = inmemory.InMemoryIndex('idx-name', ['key'])
idx.add_json('doc-id', simple_doc)
self.assertEqual({'value': ['doc-id']}, idx._values)
def test_remove_json(self):
idx = inmemory.InMemoryIndex('idx-name', ['key'])
idx.add_json('doc-id', simple_doc)
self.assertEqual({'value': ['doc-id']}, idx._values)
idx.remove_json('doc-id', simple_doc)
self.assertEqual({}, idx._values)
def test_remove_json_multiple(self):
idx = inmemory.InMemoryIndex('idx-name', ['key'])
idx.add_json('doc-id', simple_doc)
idx.add_json('doc2-id', simple_doc)
self.assertEqual({'value': ['doc-id', 'doc2-id']}, idx._values)
idx.remove_json('doc-id', simple_doc)
self.assertEqual({'value': ['doc2-id']}, idx._values)
def test_keys(self):
idx = inmemory.InMemoryIndex('idx-name', ['key'])
idx.add_json('doc-id', simple_doc)
self.assertEqual(['value'], idx.keys())
def test_lookup(self):
idx = inmemory.InMemoryIndex('idx-name', ['key'])
idx.add_json('doc-id', simple_doc)
self.assertEqual(['doc-id'], idx.lookup(['value']))
def test_lookup_multi(self):
idx = inmemory.InMemoryIndex('idx-name', ['key'])
idx.add_json('doc-id', simple_doc)
idx.add_json('doc2-id', simple_doc)
self.assertEqual(['doc-id', 'doc2-id'], idx.lookup(['value']))
def test__find_non_wildcards(self):
idx = inmemory.InMemoryIndex('idx-name', ['k1', 'k2', 'k3'])
self.assertEqual(-1, idx._find_non_wildcards(('a', 'b', 'c')))
self.assertEqual(2, idx._find_non_wildcards(('a', 'b', '*')))
self.assertEqual(3, idx._find_non_wildcards(('a', 'b', 'c*')))
self.assertEqual(2, idx._find_non_wildcards(('a', 'b*', '*')))
self.assertEqual(0, idx._find_non_wildcards(('*', '*', '*')))
self.assertEqual(1, idx._find_non_wildcards(('a*', '*', '*')))
self.assertRaises(errors.InvalidValueForIndex,
idx._find_non_wildcards, ('a', 'b'))
self.assertRaises(errors.InvalidValueForIndex,
idx._find_non_wildcards, ('a', 'b', 'c', 'd'))
self.assertRaises(errors.InvalidGlobbing,
idx._find_non_wildcards, ('*', 'b', 'c'))
|
