82
82
self.assertFalse(lexists('bogus.write-lock'))
84
open('something.write-lock', 'wb').write('fake lock\n')
84
with open('something.write-lock', 'wb') as f: f.write('fake lock\n')
85
85
self.assertRaises(LockError, t.lock_write, 'something')
86
86
os.remove('something.write-lock')
199
199
b2 = bzrdir.BzrDir.create_branch_and_repo(self.get_url('/b'))
202
self.assertEquals(b2.revision_history(), ['a1'])
202
self.assertEquals(b2.last_revision(), 'a1')
204
open('a/foo', 'wt').write('something new in foo\n')
204
with open('a/foo', 'wt') as f: f.write('something new in foo\n')
205
205
t.commit('new', rev_id='a2')
208
self.assertEquals(b2.revision_history(), ['a1', 'a2'])
208
self.assertEquals(b2.last_revision(), 'a2')
211
211
class SSHVendorConnection(TestCaseWithSFTPServer):
279
279
self.addCleanup(s.close)
280
280
self.bogus_url = 'sftp://%s:%s/' % s.getsockname()
282
def set_vendor(self, vendor):
282
def set_vendor(self, vendor, subprocess_stderr=None):
283
283
from bzrlib.transport import ssh
284
284
self.overrideAttr(ssh._ssh_vendor_manager, '_cached_ssh_vendor', vendor)
285
if subprocess_stderr is not None:
286
self.overrideAttr(ssh.SubprocessVendor, "_stderr_target",
286
289
def test_bad_connection_paramiko(self):
287
290
"""Test that a real connection attempt raises the right error"""
293
296
def test_bad_connection_ssh(self):
294
297
"""None => auto-detect vendor"""
295
self.set_vendor(None)
296
# This is how I would normally test the connection code
297
# it makes it very clear what we are testing.
298
# However, 'ssh' will create stipple on the output, so instead
299
# I'm using run_bzr_subprocess, and parsing the output
301
# t = _mod_transport.get_transport(self.bogus_url)
302
# except errors.ConnectionError:
305
# except errors.NameError, e:
306
# if 'SSHException' in str(e):
307
# raise TestSkipped('Known NameError bug in paramiko 1.6.1')
310
# self.fail('Excepted ConnectionError to be raised')
312
out, err = self.run_bzr_subprocess(['log', self.bogus_url], retcode=3)
313
self.assertEqual('', out)
314
if "NameError: global name 'SSHException'" in err:
315
# We aren't fixing this bug, because it is a bug in
316
# paramiko, but we know about it, so we don't have to
318
raise TestSkipped('Known NameError bug with paramiko-1.6.1')
319
self.assertContainsRe(err, r'bzr: ERROR: Unable to connect to SSH host'
320
r' 127\.0\.0\.1:\d+; ')
298
f = file(os.devnull, "wb")
299
self.addCleanup(f.close)
300
self.set_vendor(None, f)
301
t = _mod_transport.get_transport_from_url(self.bogus_url)
303
self.assertRaises(errors.ConnectionError, t.get, 'foobar')
305
if "global name 'SSHException'" in str(e):
306
self.knownFailure('Known NameError bug in paramiko 1.6.1')
323
310
class SFTPLatencyKnob(TestCaseWithSFTPServer):