2995
2995
# storage to observe the effects of concurrent changes
2996
2996
self.st1 = config.TransportIniFileStore(self.transport, 'foo.conf')
2997
2997
self.st2 = config.TransportIniFileStore(self.transport, 'foo.conf')
3000
self.warnings.append(args[0] % args[1:])
3001
self.overrideAttr(trace, 'warning', warning)
2999
3003
def has_store(self, store):
3000
3004
store_basename = urlutils.relative_url(self.transport.external_url(),
3020
3024
# Changes don't propagate magically
3021
3025
self.assertEquals(None, s1.get('baz'))
3022
3026
s2.store.save_changes()
3027
self.assertEquals('quux', s2.get('baz'))
3023
3028
# Changes are acquired when saving
3024
3029
self.assertEquals('bar', s2.get('foo'))
3026
# concurrent update on the same option should warn about the lost update
3030
# Since there is no overlap, no warnings are emitted
3031
self.assertLength(0, self.warnings)
3033
def test_concurrent_update_modified(self):
3034
s1 = self.get_stack(self.st1)
3035
s2 = self.get_stack(self.st2)
3036
s1.set('foo', 'bar')
3037
s2.set('foo', 'baz')
3040
s2.store.save_changes()
3041
self.assertEquals('baz', s2.get('foo'))
3042
# But the user get a warning
3043
self.assertLength(1, self.warnings)
3044
warning = self.warnings[0]
3045
self.assertStartsWith(warning, 'Option foo in section None')
3046
self.assertEndsWith(warning, 'was changed from <CREATED> to bar.'
3047
' The baz value will be saved.')
3049
def test_concurrent_deletion(self):
3050
self.st1._load_from_string('foo=bar')
3052
s1 = self.get_stack(self.st1)
3053
s2 = self.get_stack(self.st2)
3056
s1.store.save_changes()
3058
self.assertLength(0, self.warnings)
3059
s2.store.save_changes()
3061
self.assertLength(1, self.warnings)
3062
warning = self.warnings[0]
3063
self.assertStartsWith(warning, 'Option foo in section None')
3064
self.assertEndsWith(warning, 'was changed from bar to <CREATED>.'
3065
' The <DELETED> value will be saved.')
3029
3068
class TestQuotingIniFileStore(tests.TestCaseWithTransport):