← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_sftp.py

[merge] up-to-date against bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp.py
19
19
import threading
20
20
import unittest
21
21
 
22
 
from bzrlib.selftest import TestCaseInTempDir
23
 
from bzrlib.selftest.testtransport import TestTransportMixIn
 
22
from bzrlib.tests import TestCaseInTempDir
 
23
from bzrlib.tests.test_transport import TestTransportMixIn
24
24
 
25
25
try:
26
26
    import paramiko
54
54
        threading.Thread.__init__(self)
55
55
        self._callback = callback
56
56
        self._socket = socket.socket()
 
57
        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
58
        self._socket.bind(('localhost', 0))
57
59
        self._socket.listen(1)
58
60
        self.port = self._socket.getsockname()[1]
59
61
        self.stop_event = threading.Event()
91
93
        TestCaseInTempDir.setUp(self)
92
94
        self._root = self.test_dir
93
95
 
 
96
    def delayed_setup(self):
 
97
        # some tests are just stubs that call setUp and then immediately call
 
98
        # tearDwon.  so don't create the port listener until get_transport is
 
99
        # called and we know we're in an actual test.
94
100
        self._listener = SingleListener(self._run_server)
95
101
        self._listener.setDaemon(True)
96
102
        self._listener.start()        
97
103
        self._sftp_url = 'sftp://foo:bar@localhost:%d/' % (self._listener.port,)
98
104
        
99
105
    def tearDown(self):
100
 
        self._listener.stop()
 
106
        try:
 
107
            self._listener.stop()
 
108
        except AttributeError:
 
109
            pass
101
110
        TestCaseInTempDir.tearDown(self)
102
111
 
103
112
        
104
113
class SFTPTransportTest (TestCaseWithSFTPServer, TestTransportMixIn):
105
114
    readonly = False
 
115
    setup = True
106
116
 
107
117
    def get_transport(self):
 
118
        if self.setup:
 
119
            self.delayed_setup()
 
120
            self.setup = False
108
121
        from bzrlib.transport.sftp import SFTPTransport
109
122
        url = self._sftp_url
110
123
        return SFTPTransport(url)
111
124
 
 
125
    def test_sftp_locks(self):
 
126
        from bzrlib.errors import LockError
 
127
        t = self.get_transport()
 
128
 
 
129
        l = t.lock_write('bogus')
 
130
        self.failUnlessExists('bogus.write-lock')
 
131
 
 
132
        # Don't wait for the lock, locking an already locked
 
133
        # file should raise an assert
 
134
        self.assertRaises(LockError, t.lock_write, 'bogus')
 
135
 
 
136
        l.unlock()
 
137
        self.failIf(os.path.lexists('bogus.write-lock'))
 
138
 
 
139
        open('something.write-lock', 'wb').write('fake lock\n')
 
140
        self.assertRaises(LockError, t.lock_write, 'something')
 
141
        os.remove('something.write-lock')
 
142
 
 
143
        l = t.lock_write('something')
 
144
 
 
145
        l2 = t.lock_write('bogus')
 
146
 
 
147
        l.unlock()
 
148
        l2.unlock()
 
149
 
 
150
 
 
151
class FakeSFTPTransport (object):
 
152
    _sftp = object()
 
153
fake = FakeSFTPTransport()
 
154
 
 
155
 
 
156
class SFTPNonServerTest (unittest.TestCase):
 
157
    def test_parse_url(self):
 
158
        from bzrlib.transport.sftp import SFTPTransport
 
159
        s = SFTPTransport('sftp://simple.example.com/%2fhome/source', clone_from=fake)
 
160
        self.assertEquals(s._host, 'simple.example.com')
 
161
        self.assertEquals(s._port, 22)
 
162
        self.assertEquals(s._path, '/home/source')
 
163
        self.assert_(s._password is None)
 
164
        
 
165
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/relative', clone_from=fake)
 
166
        self.assertEquals(s._host, 'example.com')
 
167
        self.assertEquals(s._port, 2222)
 
168
        self.assertEquals(s._username, 'robey')
 
169
        self.assertEquals(s._password, 'h@t')
 
170
        self.assertEquals(s._path, 'relative')
 
171
        
 
172
 
 
173
class SFTPBranchTest(TestCaseWithSFTPServer):
 
174
    """Test some stuff when accessing a bzr Branch over sftp"""
 
175
 
 
176
    def test_lock_file(self):
 
177
        """Make sure that a Branch accessed over sftp tries to lock itself."""
 
178
        from bzrlib.branch import Branch
 
179
 
 
180
        self.delayed_setup()
 
181
        b = Branch.initialize(self._sftp_url)
 
182
        self.failUnlessExists('.bzr/')
 
183
        self.failUnlessExists('.bzr/branch-format')
 
184
        self.failUnlessExists('.bzr/branch-lock')
 
185
 
 
186
        self.failIf(os.path.lexists('.bzr/branch-lock.write-lock'))
 
187
        b.lock_write()
 
188
        self.failUnlessExists('.bzr/branch-lock.write-lock')
 
189
        b.unlock()
 
190
        self.failIf(os.path.lexists('.bzr/branch-lock.write-lock'))
 
191
 
 
192
 
112
193
if not paramiko_loaded:
 
194
    # TODO: Skip these
113
195
    del SFTPTransportTest
 
196
    del SFTPNonServerTest
 
197
    del SFTPBranchTest