Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
B
BTrees
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
Analytics
Analytics
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Commits
Issue Boards
Open sidebar
Kirill Smelkov
BTrees
Commits
25cbfca4
Commit
25cbfca4
authored
Nov 16, 2012
by
Tres Seaver
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Moar coverage.
parent
07e15bdf
Changes
2
Show whitespace changes
Inline
Side-by-side
Showing
2 changed files
with
213 additions
and
63 deletions
+213
-63
BTrees/_base.py
BTrees/_base.py
+9
-4
BTrees/tests/test__base.py
BTrees/tests/test__base.py
+204
-59
No files found.
BTrees/_base.py
View file @
25cbfca4
...
...
@@ -375,8 +375,8 @@ class Bucket(_BucketBase):
while
i_old
.
active
and
i_com
.
active
and
i_new
.
active
:
cmpOC
=
cmp
(
i_old
.
key
,
i_com
.
key
)
cmpON
=
cmp
(
i_old
.
key
,
i_new
.
key
)
if
cmpOC
==
0
:
if
cmpON
==
0
:
if
cmpOC
==
0
:
if
cmpON
==
0
:
if
i_com
.
value
==
i_old
.
value
:
result
[
i_old
.
key
]
=
i_new
.
value
elif
i_new
.
value
==
i_old
.
value
:
...
...
@@ -413,7 +413,7 @@ class Bucket(_BucketBase):
raise
merge_error
(
3
)
else
:
# both keys changed
cmpCN
=
cmp
(
i_com
.
key
,
i_new
.
key
)
if
cmpCN
==
0
:
if
cmpCN
==
0
:
# dueling insert
raise
merge_error
(
4
)
if
cmpOC
>
0
:
# insert committed
if
cmpCN
>
0
:
# insert i_new first
...
...
@@ -468,6 +468,9 @@ class Bucket(_BucketBase):
if
len
(
result
.
_keys
)
==
0
:
# If the output bucket is empty, conflict resolution doesn't have
# enough info to unlink it from its containing BTree correctly.
#
# XXX TS, 2012-11-16: I don't think this is possible
#
raise
merge_error
(
10
)
result
.
_next
=
b_old
.
_next
...
...
@@ -656,7 +659,9 @@ class Set(_BucketBase):
if
len
(
result
.
_keys
)
==
0
:
# If the output bucket is empty, conflict resolution doesn't have
# enough info to unlink it from its containing BTree correctly.
assert
0
,
"CAN'T GET HERE"
#
# XXX TS, 2012-11-16: I don't think this is possible
#
raise
merge_error
(
10
)
result
.
_next
=
b_old
.
_next
...
...
BTrees/tests/test__base.py
View file @
25cbfca4
...
...
@@ -423,24 +423,26 @@ class BucketTests(unittest.TestCase):
return
Bucket
def
_makeOne
(
self
):
return
self
.
_getTargetClass
()()
class
_Bucket
(
self
.
_getTargetClass
()):
def
_to_key
(
self
,
x
):
return
x
def
_to_value
(
self
,
x
):
return
x
return
_Bucket
()
def
test_ctor_defaults
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
self
.
assertEqual
(
bucket
.
_keys
,
[])
self
.
assertEqual
(
bucket
.
_values
,
[])
def
test_setdefault_miss
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
self
.
assertEqual
(
bucket
.
setdefault
(
'a'
,
'b'
),
'b'
)
self
.
assertEqual
(
bucket
.
_keys
,
[
'a'
])
self
.
assertEqual
(
bucket
.
_values
,
[
'b'
])
def
test_setdefault_hit
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
bucket
.
_keys
.
append
(
'a'
)
bucket
.
_values
.
append
(
'b'
)
self
.
assertEqual
(
bucket
.
setdefault
(
'a'
,
'b'
),
'b'
)
...
...
@@ -449,17 +451,14 @@ class BucketTests(unittest.TestCase):
def
test_pop_miss_no_default
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
self
.
assertRaises
(
KeyError
,
bucket
.
pop
,
'nonesuch'
)
def
test_pop_miss_w_default
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
self
.
assertEqual
(
bucket
.
pop
(
'nonesuch'
,
'b'
),
'b'
)
def
test_pop_hit
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
bucket
.
_keys
.
append
(
'a'
)
bucket
.
_values
.
append
(
'b'
)
self
.
assertEqual
(
bucket
.
pop
(
'a'
),
'b'
)
...
...
@@ -468,14 +467,12 @@ class BucketTests(unittest.TestCase):
def
test_update_value_w_iteritems
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
bucket
.
update
({
'a'
:
'b'
})
self
.
assertEqual
(
bucket
.
_keys
,
[
'a'
])
self
.
assertEqual
(
bucket
.
_values
,
[
'b'
])
def
test_update_value_w_items
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
class
Foo
(
object
):
def
items
(
self
):
return
[(
'a'
,
'b'
)]
...
...
@@ -485,7 +482,6 @@ class BucketTests(unittest.TestCase):
def
test_update_value_w_invalid_items
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
class
Foo
(
object
):
def
items
(
self
):
return
(
'a'
,
'b'
,
'c'
)
...
...
@@ -493,41 +489,35 @@ class BucketTests(unittest.TestCase):
def
test_update_sequence
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
bucket
.
update
([(
'a'
,
'b'
)])
self
.
assertEqual
(
bucket
.
_keys
,
[
'a'
])
self
.
assertEqual
(
bucket
.
_values
,
[
'b'
])
def
test___setitem___incomparable
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
def
_should_error
():
bucket
[
object
()]
=
'b'
self
.
assertRaises
(
TypeError
,
_should_error
)
def
test___setitem___comparable
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
bucket
[
'a'
]
=
'b'
self
.
assertEqual
(
bucket
[
'a'
],
'b'
)
def
test___setitem___replace
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
bucket
[
'a'
]
=
'b'
bucket
[
'a'
]
=
'c'
self
.
assertEqual
(
bucket
[
'a'
],
'c'
)
def
test___delitem___miss
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
def
_should_error
():
del
bucket
[
'nonesuch'
]
self
.
assertRaises
(
KeyError
,
_should_error
)
def
test___delitem___hit
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
bucket
.
_keys
.
append
(
'a'
)
bucket
.
_values
.
append
(
'b'
)
del
bucket
[
'a'
]
...
...
@@ -536,7 +526,6 @@ class BucketTests(unittest.TestCase):
def
test_clear_filled
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
bucket
[
'a'
]
=
'b'
bucket
[
'c'
]
=
'd'
bucket
.
clear
()
...
...
@@ -551,31 +540,26 @@ class BucketTests(unittest.TestCase):
def
test_get_miss_no_default
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
self
.
assertEqual
(
bucket
.
get
(
'nonesuch'
),
None
)
def
test_get_miss_w_default
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
self
.
assertEqual
(
bucket
.
get
(
'nonesuch'
,
'b'
),
'b'
)
def
test_get_hit
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
bucket
.
_keys
.
append
(
'a'
)
bucket
.
_values
.
append
(
'b'
)
self
.
assertEqual
(
bucket
.
get
(
'a'
),
'b'
)
def
test___getitem___miss
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
def
_should_error
():
return
bucket
[
'nonesuch'
]
self
.
assertRaises
(
KeyError
,
_should_error
)
def
test___getitem___hit
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
bucket
.
_keys
.
append
(
'a'
)
bucket
.
_values
.
append
(
'b'
)
self
.
assertEqual
(
bucket
[
'a'
],
'b'
)
...
...
@@ -594,7 +578,6 @@ class BucketTests(unittest.TestCase):
def
test__split_filled_default_index
(
self
):
bucket
=
self
.
_makeOne
()
next_b
=
bucket
.
_next
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
new_b
=
bucket
.
_split
()
...
...
@@ -611,7 +594,6 @@ class BucketTests(unittest.TestCase):
def
test_keys_filled_no_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
self
.
assertEqual
(
bucket
.
keys
(),
...
...
@@ -619,7 +601,6 @@ class BucketTests(unittest.TestCase):
def
test_keys_filled_w_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
self
.
assertEqual
(
bucket
.
keys
(
min
=
'b'
,
excludemin
=
True
,
...
...
@@ -631,7 +612,6 @@ class BucketTests(unittest.TestCase):
def
test_iterkeys_filled_no_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
self
.
assertEqual
(
list
(
bucket
.
iterkeys
()),
...
...
@@ -639,7 +619,6 @@ class BucketTests(unittest.TestCase):
def
test_iterkeys_filled_w_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
self
.
assertEqual
(
list
(
bucket
.
iterkeys
(
...
...
@@ -652,14 +631,12 @@ class BucketTests(unittest.TestCase):
def
test_values_filled_no_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
self
.
assertEqual
(
bucket
.
values
(),
range
(
6
))
def
test_values_filled_w_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
self
.
assertEqual
(
bucket
.
values
(
min
=
'b'
,
excludemin
=
True
,
...
...
@@ -671,14 +648,12 @@ class BucketTests(unittest.TestCase):
def
test_itervalues_filled_no_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
self
.
assertEqual
(
list
(
bucket
.
itervalues
()),
range
(
6
))
def
test_itervalues_filled_w_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
self
.
assertEqual
(
list
(
bucket
.
itervalues
(
...
...
@@ -691,7 +666,6 @@ class BucketTests(unittest.TestCase):
def
test_items_filled_no_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
EXPECTED
=
[]
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
...
...
@@ -700,7 +674,6 @@ class BucketTests(unittest.TestCase):
def
test_items_filled_w_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
EXPECTED
=
[]
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
...
...
@@ -715,7 +688,6 @@ class BucketTests(unittest.TestCase):
def
test_iteritems_filled_no_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
EXPECTED
=
[]
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
...
...
@@ -724,7 +696,6 @@ class BucketTests(unittest.TestCase):
def
test_iteritems_filled_w_args
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
EXPECTED
=
[]
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
...
...
@@ -744,7 +715,6 @@ class BucketTests(unittest.TestCase):
def
test___getstate___non_empty_no_next
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
EXPECTED
=
()
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
...
...
@@ -754,17 +724,19 @@ class BucketTests(unittest.TestCase):
def
test___getstate___non_empty_w_next
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_next
=
next_b
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
EXPECTED
=
()
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
EXPECTED
+=
(
c
,
i
)
self
.
assertEqual
(
bucket
.
__getstate__
(),
(
EXPECTED
,
next_b
))
def
test___setstate___w_non_tuple
(
self
):
bucket
=
self
.
_makeOne
()
self
.
assertRaises
(
TypeError
,
bucket
.
__setstate__
,
(
None
,))
def
test___setstate___w_empty_no_next
(
self
):
bucket
=
self
.
_makeOne
()
bucket
.
_next
=
next_b
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
for
i
,
c
in
enumerate
(
'abcdef'
):
bucket
[
c
]
=
i
bucket
.
__setstate__
(((),))
...
...
@@ -774,7 +746,6 @@ class BucketTests(unittest.TestCase):
def
test___setstate___w_non_empty_w_next
(
self
):
bucket
=
self
.
_makeOne
()
next_b
=
self
.
_makeOne
()
bucket
.
_to_key
=
lambda
x
:
x
ITEMS
=
()
EXPECTED
=
[]
for
i
,
c
in
enumerate
(
'abcdef'
):
...
...
@@ -787,85 +758,259 @@ class BucketTests(unittest.TestCase):
def
test__p_resolveConflict_x_on_com_next
(
self
):
from
..Interfaces
import
BTreesConflictError
_s
et
=
self
.
_makeOne
()
buck
et
=
self
.
_makeOne
()
N_NEW
=
object
()
s_old
=
((),
None
)
s_com
=
((),
N_NEW
)
s_new
=
((),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
_s
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
buck
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
0
)
def
test__p_resolveConflict_x_on_new_next
(
self
):
from
..Interfaces
import
BTreesConflictError
_s
et
=
self
.
_makeOne
()
buck
et
=
self
.
_makeOne
()
N_NEW
=
object
()
s_old
=
((),
None
)
s_com
=
((),
None
)
s_new
=
((),
N_NEW
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
_s
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
buck
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
0
)
def
test__p_resolveConflict_x_on_com_empty
(
self
):
from
..Interfaces
import
BTreesConflictError
_s
et
=
self
.
_makeOne
()
buck
et
=
self
.
_makeOne
()
s_old
=
((
'a'
,
'b'
,
'c'
,
'd'
),
None
)
s_com
=
((),
None
)
s_new
=
((
'a'
,
'b'
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
_s
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
buck
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
12
)
def
test__p_resolveConflict_x_on_new_empty
(
self
):
from
..Interfaces
import
BTreesConflictError
_s
et
=
self
.
_makeOne
()
buck
et
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
),
None
)
s_com
=
((
'a'
,
0
),
None
)
s_new
=
((),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
_s
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
buck
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
12
)
def
test__p_resolveConflict_x_both_update_same_key
(
self
):
from
..Interfaces
import
BTreesConflictError
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
),
None
)
s_com
=
((
'a'
,
5
,
'b'
,
1
,
'c'
,
2
),
None
)
s_new
=
((
'a'
,
6
,
'd'
,
3
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
bucket
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
1
)
def
test__p_resolveConflict_x_on_del_first_com_x
(
self
):
from
..Interfaces
import
BTreesConflictError
_s
et
=
self
.
_makeOne
()
buck
et
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),
None
)
s_com
=
((
'b'
,
1
),
None
)
s_new
=
((
'a'
,
0
,
'b'
,
1
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
_s
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
buck
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
13
)
def
test__p_resolveConflict_x_on_del_first_new_x
(
self
):
from
..Interfaces
import
BTreesConflictError
_s
et
=
self
.
_makeOne
()
buck
et
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),
None
)
s_com
=
((
'a'
,
0
,
'b'
,
1
),
None
)
s_new
=
((
'b'
,
1
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
_s
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
buck
et
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
13
)
def
test__p_resolveConflict_x_on_del_first_new
(
self
):
from
..Interfaces
import
BTreesConflictError
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
),
None
)
s_com
=
((
'a'
,
1
,
'b'
,
2
,
'c'
,
3
),
None
)
s_new
=
((
'b'
,
4
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
bucket
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
2
)
def
test__p_resolveConflict_x_on_del_first_com
(
self
):
from
..Interfaces
import
BTreesConflictError
_mapping
=
self
.
_makeOne
()
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
),
None
)
s_com
=
((
'b'
,
4
),
None
)
s_new
=
((
'a'
,
1
,
'b'
,
2
,
'c'
,
3
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
_mapping
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
bucket
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
3
)
def
test__p_resolveConflict_x_on_
del_first_new
(
self
):
def
test__p_resolveConflict_x_on_
ins_same_after_del
(
self
):
from
..Interfaces
import
BTreesConflictError
_mapping
=
self
.
_makeOne
()
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
),
None
)
s_com
=
((
'a'
,
1
,
'b'
,
2
,
'c'
,
3
),
None
)
s_new
=
((
'
b'
,
4
),
None
)
s_com
=
((
'a'
,
0
,
'c'
,
2
),
None
)
s_new
=
((
'
a'
,
0
,
'c'
,
2
,
'd'
,
3
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
_mapping
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
2
)
bucket
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
4
)
def
test__p_resolveConflict_x_on_del_same
(
self
):
from
..Interfaces
import
BTreesConflictError
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),
None
)
s_com
=
((
'a'
,
0
,
'c'
,
2
),
None
)
s_new
=
((
'a'
,
0
,
'd'
,
3
,
'e'
,
4
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
bucket
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
5
)
def
test__p_resolveConflict_x_on_append_same
(
self
):
from
..Interfaces
import
BTreesConflictError
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
),
None
)
s_com
=
((
'a'
,
0
,
'b'
,
1
),
None
)
s_new
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
bucket
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
6
)
def
test__p_resolveConflict_x_on_new_deletes_all_com_adds
(
self
):
from
..Interfaces
import
BTreesConflictError
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),
None
)
s_com
=
((
'a'
,
0
,
'd'
,
3
,
'e'
,
4
,
'f'
,
5
),
None
)
s_new
=
((
'a'
,
0
,
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
bucket
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
7
)
def
test__p_resolveConflict_x_on_com_deletes_all_new_adds
(
self
):
from
..Interfaces
import
BTreesConflictError
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),
None
)
s_com
=
((
'a'
,
0
,
),
None
)
s_new
=
((
'a'
,
0
,
'd'
,
3
,
'e'
,
4
,
'f'
,
5
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
bucket
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
8
)
def
test__p_resolveConflict_x_on_com_deletes_all_new_deletes
(
self
):
from
..Interfaces
import
BTreesConflictError
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),
None
)
s_com
=
((
'a'
,
0
,
),
None
)
s_new
=
((
'a'
,
0
,
'b'
,
1
),
None
)
e
=
self
.
assertRaises
(
BTreesConflictError
,
bucket
.
_p_resolveConflict
,
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
e
.
reason
,
9
)
def
test__p_resolveConflict_ok_both_add_new_max
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
),
None
)
s_com
=
((
'a'
,
0
,
'b'
,
1
),
None
)
s_new
=
((
'a'
,
0
,
'c'
,
2
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),))
def
test__p_resolveConflict_ok_com_updates
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
),
None
)
s_com
=
((
'a'
,
5
),
None
)
s_new
=
((
'a'
,
0
,
'd'
,
3
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
5
,
'd'
,
3
),))
def
test__p_resolveConflict_ok_new_updates
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
),
None
)
s_com
=
((
'a'
,
0
,
'd'
,
3
),
None
)
s_new
=
((
'a'
,
5
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
5
,
'd'
,
3
),))
def
test__p_resolveConflict_ok_com_inserts_new_adds
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'c'
,
2
),
None
)
s_com
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),
None
)
s_new
=
((
'a'
,
0
,
'c'
,
2
,
'd'
,
3
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
,
'd'
,
3
),))
def
test__p_resolveConflict_ok_com_adds_new_inserts
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'c'
,
2
),
None
)
s_com
=
((
'a'
,
0
,
'c'
,
2
,
'd'
,
3
),
None
)
s_new
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
,
'd'
,
3
),))
def
test__p_resolveConflict_ok_com_adds_new_deletes
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),
None
)
s_com
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
,
'd'
,
3
),
None
)
s_new
=
((
'a'
,
0
,
'e'
,
4
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
0
,
'd'
,
3
,
'e'
,
4
),))
def
test__p_resolveConflict_ok_com_deletes_new_adds
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
),
None
)
s_com
=
((
'a'
,
0
,
'e'
,
4
),
None
)
s_new
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
,
'd'
,
3
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
0
,
'd'
,
3
,
'e'
,
4
),))
def
test__p_resolveConflict_ok_both_insert_new_lt_com
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'd'
,
3
),
None
)
s_com
=
((
'a'
,
0
,
'c'
,
2
,
'd'
,
3
),
None
)
s_new
=
((
'a'
,
0
,
'b'
,
1
,
'd'
,
3
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
,
'd'
,
3
),))
def
test__p_resolveConflict_ok_both_insert_new_gt_com
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'd'
,
3
),
None
)
s_com
=
((
'a'
,
0
,
'b'
,
1
,
'd'
,
3
),
None
)
s_new
=
((
'a'
,
0
,
'c'
,
2
,
'd'
,
3
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
,
'd'
,
3
),))
def
test__p_resolveConflict_ok_new_insert_then_com_append
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'd'
,
3
),
None
)
s_com
=
((
'a'
,
0
,
'e'
,
4
),
None
)
s_new
=
((
'a'
,
0
,
'b'
,
1
,
'd'
,
3
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
0
,
'b'
,
1
,
'e'
,
4
),))
def
test__p_resolveConflict_ok_com_insert_then_new_append
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'd'
,
3
),
None
)
s_com
=
((
'a'
,
0
,
'b'
,
1
,
'd'
,
3
),
None
)
s_new
=
((
'a'
,
0
,
'e'
,
4
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
0
,
'b'
,
1
,
'e'
,
4
),))
def
test__p_resolveConflict_ok_new_deletes_tail_com_inserts
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
,
'd'
,
3
),
None
)
s_com
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
,
'd'
,
3
),
None
)
s_new
=
((
'a'
,
0
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
0
,
'c'
,
2
),))
def
test__p_resolveConflict_ok_com_deletes_tail_new_inserts
(
self
):
bucket
=
self
.
_makeOne
()
s_old
=
((
'a'
,
0
,
'b'
,
1
,
'd'
,
3
),
None
)
s_com
=
((
'a'
,
0
),
None
)
s_new
=
((
'a'
,
0
,
'b'
,
1
,
'c'
,
2
,
'd'
,
3
),
None
)
result
=
bucket
.
_p_resolveConflict
(
s_old
,
s_com
,
s_new
)
self
.
assertEqual
(
result
,
((
'a'
,
0
,
'c'
,
2
),))
class
SetTests
(
unittest
.
TestCase
):
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment