/usr/lib/python2.7/dist-packages/fs/tests/test_expose.py is in python-fs 0.4.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 | """
fs.tests.test_expose: testcases for fs.expose and associated FS classes
"""
import unittest
import sys
import os, os.path
import socket
import threading
import time
from fs.tests import FSTestCases, ThreadingTestCases
from fs.tempfs import TempFS
from fs.osfs import OSFS
from fs.memoryfs import MemoryFS
from fs.path import *
from fs.errors import *
from fs import rpcfs
from fs.expose.xmlrpc import RPCFSServer
class TestRPCFS(unittest.TestCase, FSTestCases, ThreadingTestCases):
def makeServer(self,fs,addr):
return RPCFSServer(fs,addr,logRequests=False)
def startServer(self):
port = 3000
self.temp_fs = TempFS()
self.server = None
self.serve_more_requests = True
self.server_thread = threading.Thread(target=self.runServer)
self.server_thread.setDaemon(True)
self.start_event = threading.Event()
self.end_event = threading.Event()
self.server_thread.start()
self.start_event.wait()
def runServer(self):
"""Run the server, swallowing shutdown-related execptions."""
port = 3000
while not self.server:
try:
self.server = self.makeServer(self.temp_fs,("127.0.0.1",port))
except socket.error, e:
if e.args[1] == "Address already in use":
port += 1
else:
raise
self.server_addr = ("127.0.0.1", port)
self.server.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# if sys.platform != "win32":
# try:
# self.server.socket.settimeout(1)
# except socket.error:
# pass
#
self.start_event.set()
try:
#self.server.serve_forever()
while self.serve_more_requests:
self.server.handle_request()
except Exception, e:
pass
self.end_event.set()
def setUp(self):
self.startServer()
self.fs = rpcfs.RPCFS("http://%s:%d" % self.server_addr)
def tearDown(self):
self.serve_more_requests = False
#self.server.socket.close()
# self.server.socket.shutdown(socket.SHUT_RDWR)
# self.server.socket.close()
# self.temp_fs.close()
#self.server_thread.join()
#self.end_event.wait()
#return
try:
self.bump()
self.server.server_close()
except Exception:
pass
#self.server_thread.join()
self.temp_fs.close()
def bump(self):
host, port = self.server_addr
for res in socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM):
af, socktype, proto, cn, sa = res
sock = None
try:
sock = socket.socket(af, socktype, proto)
sock.settimeout(.1)
sock.connect(sa)
sock.send("\n")
except socket.error, e:
pass
finally:
if sock is not None:
sock.close()
from fs import sftpfs
from fs.expose.sftp import BaseSFTPServer
class TestSFTPFS(TestRPCFS):
def makeServer(self,fs,addr):
return BaseSFTPServer(addr,fs)
def setUp(self):
self.startServer()
self.fs = sftpfs.SFTPFS(self.server_addr, no_auth=True)
#def runServer(self):
# self.server.serve_forever()
#
#def tearDown(self):
# self.server.shutdown()
# self.server_thread.join()
# self.temp_fs.close()
def bump(self):
# paramiko doesn't like being bumped, just wait for it to timeout.
# TODO: do this using a paramiko.Transport() connection
pass
try:
from fs.expose import fuse
except ImportError:
pass
else:
from fs.osfs import OSFS
class TestFUSE(unittest.TestCase,FSTestCases,ThreadingTestCases):
def setUp(self):
self.temp_fs = TempFS()
self.temp_fs.makedir("root")
self.temp_fs.makedir("mount")
self.mounted_fs = self.temp_fs.opendir("root")
self.mount_point = self.temp_fs.getsyspath("mount")
self.fs = OSFS(self.temp_fs.getsyspath("mount"))
self.mount_proc = fuse.mount(self.mounted_fs,self.mount_point)
def tearDown(self):
self.mount_proc.unmount()
try:
self.temp_fs.close()
except OSError:
# Sometimes FUSE hangs onto the mountpoint if mount_proc is
# forcibly killed. Shell out to fusermount to make sure.
fuse.unmount(self.mount_point)
self.temp_fs.close()
def check(self,p):
return self.mounted_fs.exists(p)
from fs.expose import dokan
if dokan.is_available:
from fs.osfs import OSFS
class DokanTestCases(FSTestCases):
"""Specialised testcases for filesystems exposed via Dokan.
This modifies some of the standard tests to work around apparent
bugs in the current Dokan implementation.
"""
def test_remove(self):
self.fs.createfile("a.txt")
self.assertTrue(self.check("a.txt"))
self.fs.remove("a.txt")
self.assertFalse(self.check("a.txt"))
self.assertRaises(ResourceNotFoundError,self.fs.remove,"a.txt")
self.fs.makedir("dir1")
# This appears to be a bug in Dokan - DeleteFile will happily
# delete an empty directory.
#self.assertRaises(ResourceInvalidError,self.fs.remove,"dir1")
self.fs.createfile("/dir1/a.txt")
self.assertTrue(self.check("dir1/a.txt"))
self.fs.remove("dir1/a.txt")
self.assertFalse(self.check("/dir1/a.txt"))
def test_open_on_directory(self):
# Dokan seems quite happy to ask me to open a directory and
# then treat it like a file.
pass
def test_settimes(self):
# Setting the times does actually work, but there's some sort
# of caching effect which prevents them from being read back
# out. Disabling the test for now.
pass
def test_safety_wrapper(self):
rawfs = MemoryFS()
safefs = dokan.Win32SafetyFS(rawfs)
rawfs.setcontents("autoRun.inf","evilcodeevilcode")
self.assertTrue(safefs.exists("_autoRun.inf"))
self.assertTrue("autoRun.inf" not in safefs.listdir("/"))
safefs.setcontents("file:stream","test")
self.assertFalse(rawfs.exists("file:stream"))
self.assertTrue(rawfs.exists("file__colon__stream"))
self.assertTrue("file:stream" in safefs.listdir("/"))
class TestDokan(unittest.TestCase,DokanTestCases,ThreadingTestCases):
def setUp(self):
self.temp_fs = TempFS()
self.drive = "K"
while os.path.exists(self.drive+":\\") and self.drive <= "Z":
self.drive = chr(ord(self.drive) + 1)
if self.drive > "Z":
raise RuntimeError("no free drive letters")
fs_to_mount = OSFS(self.temp_fs.getsyspath("/"))
self.mount_proc = dokan.mount(fs_to_mount,self.drive)#,flags=dokan.DOKAN_OPTION_DEBUG|dokan.DOKAN_OPTION_STDERR,numthreads=1)
self.fs = OSFS(self.mount_proc.path)
def tearDown(self):
self.mount_proc.unmount()
for _ in xrange(10):
try:
if self.mount_proc.poll() is None:
self.mount_proc.terminate()
except EnvironmentError:
time.sleep(0.1)
else:
break
else:
if self.mount_proc.poll() is None:
self.mount_proc.terminate()
self.temp_fs.close()
if __name__ == '__main__':
unittest.main()
|