← Back to branch summary

~ubuntu-branches/ubuntu/lucid/paramiko/lucid

« back to all changes in this revision

Viewing changes to tests/test_client.py

  • Committer: Bazaar Package Importer
  • Author(s): Adeodato Simó
  • Date: 2008-01-24 13:54:18 UTC
  • mfrom: (1.1.4 upstream)
  • Revision ID: james.westby@ubuntu.com-20080124135418-npxfwu271vsfsa3c
Tags: 1.7.2-0.1
http://bugs.debian.org/415060
* Non-maintainer upload to DELAYED/14-day. (¹)
* New upstream release. (Closes: #415060)
* Drop the patch introduced in 1.6.4-1.1, as it's part of 1.7.2.

  (¹) Counting since the initial 1.7.1-0.1 upload in Jan 13th.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/test_client.py
1
 
# Copyright (C) 2003-2005 Robey Pointer <robey@lag.net>
 
1
# Copyright (C) 2003-2007  Robey Pointer <robey@lag.net>
2
2
#
3
3
# This file is part of paramiko.
4
4
#
22
22
 
23
23
import socket
24
24
import threading
 
25
import time
25
26
import unittest
 
27
import weakref
 
28
from binascii import hexlify
 
29
 
26
30
import paramiko
27
31
 
28
32
 
38
42
            return paramiko.AUTH_SUCCESSFUL
39
43
        return paramiko.AUTH_FAILED
40
44
 
 
45
    def check_auth_publickey(self, username, key):
 
46
        if (key.get_name() == 'ssh-dss') and (hexlify(key.get_fingerprint()) == '4478f0b9a23cc5182009ff755bc1d26c'):
 
47
            return paramiko.AUTH_SUCCESSFUL
 
48
        return paramiko.AUTH_FAILED
 
49
    
41
50
    def check_channel_request(self, kind, chanid):
42
51
        return paramiko.OPEN_SUCCEEDED
43
52
 
59
68
        thread.start()
60
69
 
61
70
    def tearDown(self):
62
 
        self.tc.close()
 
71
        if hasattr(self, 'tc'):
 
72
            self.tc.close()
63
73
        self.ts.close()
64
74
        self.socks.close()
65
75
        self.sockl.close()
105
115
        stdin.close()
106
116
        stdout.close()
107
117
        stderr.close()
108
 
 
109
 
    def test_2_auto_add_policy(self):
 
118
    
 
119
    def test_2_client_dsa(self):
 
120
        """
 
121
        verify that SSHClient works with a DSA key.
 
122
        """
 
123
        host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
 
124
        public_host_key = paramiko.RSAKey(data=str(host_key))
 
125
        
 
126
        self.tc = paramiko.SSHClient()
 
127
        self.tc.get_host_keys().add(self.addr, 'ssh-rsa', public_host_key)
 
128
        self.tc.connect(self.addr, self.port, username='slowdive', key_filename='tests/test_dss.key')
 
129
 
 
130
        self.event.wait(1.0)
 
131
        self.assert_(self.event.isSet())
 
132
        self.assert_(self.ts.is_active())
 
133
        self.assertEquals('slowdive', self.ts.get_username())
 
134
        self.assertEquals(True, self.ts.is_authenticated())
 
135
 
 
136
        stdin, stdout, stderr = self.tc.exec_command('yes')
 
137
        schan = self.ts.accept(1.0)
 
138
 
 
139
        schan.send('Hello there.\n')
 
140
        schan.send_stderr('This is on stderr.\n')
 
141
        schan.close()
 
142
 
 
143
        self.assertEquals('Hello there.\n', stdout.readline())
 
144
        self.assertEquals('', stdout.readline())
 
145
        self.assertEquals('This is on stderr.\n', stderr.readline())
 
146
        self.assertEquals('', stderr.readline())
 
147
        
 
148
        stdin.close()
 
149
        stdout.close()
 
150
        stderr.close()
 
151
 
 
152
    def test_3_auto_add_policy(self):
110
153
        """
111
154
        verify that SSHClient's AutoAddPolicy works.
112
155
        """
125
168
        self.assertEquals(True, self.ts.is_authenticated())
126
169
        self.assertEquals(1, len(self.tc.get_host_keys()))
127
170
        self.assertEquals(public_host_key, self.tc.get_host_keys()[self.addr]['ssh-rsa'])
 
171
 
 
172
    def test_4_cleanup(self):
 
173
        """
 
174
        verify that when an SSHClient is collected, its transport (and the
 
175
        transport's packetizer) is closed.
 
176
        """
 
177
        host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
 
178
        public_host_key = paramiko.RSAKey(data=str(host_key))
 
179
        
 
180
        self.tc = paramiko.SSHClient()
 
181
        self.tc.set_missing_host_key_policy(paramiko.AutoAddPolicy())
 
182
        self.assertEquals(0, len(self.tc.get_host_keys()))
 
183
        self.tc.connect(self.addr, self.port, username='slowdive', password='pygmalion')
 
184
 
 
185
        self.event.wait(1.0)
 
186
        self.assert_(self.event.isSet())
 
187
        self.assert_(self.ts.is_active())
 
188
        
 
189
        p = weakref.ref(self.tc._transport.packetizer)
 
190
        self.assert_(p() is not None)
 
191
        del self.tc
 
192
        # hrm, sometimes p isn't cleared right away.  why is that?
 
193
        st = time.time()
 
194
        while (time.time() - st < 5.0) and (p() is not None):
 
195
            time.sleep(0.1)
 
196
        self.assert_(p() is None)
 
197