1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
import logging
import os
import threading
import pytest
from paramiko import RSAKey, SFTPServer, SFTP, Transport
from .loop import LoopSocket
from .stub_sftp import StubServer, StubSFTPServer
from .util import _support
# TODO: not a huge fan of conftest.py files, see if we can move these somewhere
# 'nicer'.
# Perform logging by default; pytest will capture and thus hide it normally,
# presenting it on error/failure.
# Also make sure to set up timestamping for more sanity when debugging.
logging.basicConfig(
level=logging.DEBUG,
format="[%(relativeCreated)s]\t%(levelname)s:%(name)s:%(message)s",
datefmt="%H:%M:%S",
)
def make_sftp_folder(client):
"""
Create some non-existing, new folder on the given SFTP connection.
"""
path = os.environ.get('TEST_FOLDER', 'temp-testing000')
# TODO: this is disgusting and old, replace with something smarter/simpler
for i in range(1000):
path = path[:-3] + '%03d' % i
try:
client.mkdir(path)
return path
except (IOError, OSError):
pass
# TODO: apply at module or session level
# TODO: roll in SFTP folder setup and teardown?
# NOTE: This is defined here for use by both SFTP (normal & 'big') suites.
@pytest.fixture(scope='session')
def sftp():
"""
Set up an in-memory SFTP server, returning its corresponding SFTPClient.
"""
# Sockets & transports
socks = LoopSocket()
sockc = LoopSocket()
sockc.link(socks)
tc = Transport(sockc)
ts = Transport(socks)
# Auth
host_key = RSAKey.from_private_key_file(_support('test_rsa.key'))
ts.add_server_key(host_key)
# Server & client setup
event = threading.Event()
server = StubServer()
ts.set_subsystem_handler('sftp', SFTPServer, StubSFTPServer)
ts.start_server(event, server)
tc.connect(username='slowdive', password='pygmalion')
event.wait(1.0)
client = SFTP.from_transport(tc)
# Work in 'remote' folder setup (as it wants to use the client)
# TODO: how cleanest to make this available to tests? Doing it this way is
# marginally less bad than the previous 'global'-using setup, but not by
# much?
client.FOLDER = make_sftp_folder(client)
# Yield client to caller
yield client
# Clean up
client.rmdir(client.FOLDER)
|
