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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
|
import logging
import os
import shutil
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
from icecream import ic, install as install_ic
install_ic()
ic.configureOutput(includeContext=True)
# Perform logging by default; pytest will capture and thus hide it normally,
# presenting it on error/failure. (But also allow turning it off when doing
# very pinpoint debugging - e.g. using breakpoints, so you don't want output
# hiding enabled, but also don't want all the logging to gum up the terminal.)
if not os.environ.get("DISABLE_LOGGING", False):
logging.basicConfig(
level=logging.DEBUG,
# Also make sure to set up timestamping for more sanity when debugging.
format="[%(relativeCreated)s]\t%(levelname)s:%(name)s:%(message)s",
datefmt="%H:%M:%S",
)
def make_sftp_folder():
"""
Ensure expected target temp folder exists on the remote end.
Will clean it out if it already exists.
"""
# TODO: go back to using the sftp functionality itself for folder setup so
# we can test against live SFTP servers again someday. (Not clear if anyone
# is/was using the old capability for such, though...)
# TODO: something that would play nicer with concurrent testing (but
# probably e.g. using thread ID or UUIDs or something; not the "count up
# until you find one not used!" crap from before...)
# TODO: if we want to lock ourselves even harder into localhost-only
# testing (probably not?) could use tempdir modules for this for improved
# safety. Then again...why would someone have such a folder???
path = os.environ.get("TEST_FOLDER", "paramiko-test-target")
# Forcibly nuke this directory locally, since at the moment, the below
# fixtures only ever run with a locally scoped stub test server.
shutil.rmtree(path, ignore_errors=True)
# Then create it anew, again locally, for the same reason.
os.mkdir(path)
return path
@pytest.fixture # (scope='session')
def sftp_server():
"""
Set up an in-memory SFTP server thread. Yields the client Transport/socket.
The resulting client Transport (along with all the server components) will
be the same object throughout the test session; the `sftp` fixture then
creates new higher level client objects wrapped around the client
Transport, as necessary.
"""
# 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 setup
event = threading.Event()
server = StubServer()
ts.set_subsystem_handler("sftp", SFTPServer, StubSFTPServer)
ts.start_server(event, server)
# Wait (so client has time to connect? Not sure. Old.)
event.wait(1.0)
# Make & yield connection.
tc.connect(username="slowdive", password="pygmalion")
yield tc
# TODO: any need for shutdown? Why didn't old suite do so? Or was that the
# point of the "join all threads from threading module" crap in test.py?
@pytest.fixture
def sftp(sftp_server):
"""
Yield an SFTP client connected to the global in-session SFTP server thread.
"""
# Client setup
client = SFTP.from_transport(sftp_server)
# 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()
# Yield client to caller
yield client
# Clean up - as in make_sftp_folder, we assume local-only exec for now.
shutil.rmtree(client.FOLDER, ignore_errors=True)
|
