FM / tests

/usr/share/doc/python3-pycurl/tests

Name	Size	Mode	Actions
certs/	-	0755	rm
ext/	-	0755	rm
fake-curl/	-	0755	rm
fixtures/	-	0755	rm
matrix/	-	0755	rm
tmp/	-	0755	rm
__pycache__/	-	0755	rm
app.py	4088	0644	edit dl rm
appmanager.py	1380	0644	edit dl rm
cadata_test.py	1452	0644	edit dl rm
certinfo_test.py	3468	0644	edit dl rm
close_socket_cb_test.py	2168	0644	edit dl rm
curl_object_test.py	4883	0644	edit dl rm
debug_test.py	2275	0644	edit dl rm
default_write_cb_test.py	3039	0644	edit dl rm
error_constants_test.py	662	0644	edit dl rm
error_test.py	2901	0644	edit dl rm
failonerror_test.py	3509	0644	edit dl rm
ftp_test.py	1537	0644	edit dl rm
global_init_test.py	1063	0644	edit dl rm
header_cb_test.py	1560	0644	edit dl rm
header_test.py	1848	0644	edit dl rm
high_level_curl_test.py	632	0644	edit dl rm
info_constants_test.py	387	0644	edit dl rm
internals_test.py	6674	0644	edit dl rm
matrix.py	6816	0644	edit dl rm
multi_option_constants_test.py	2503	0644	edit dl rm
multi_socket_test.py	3049	0644	edit dl rm
multi_test.py	12313	0644	edit dl rm
multi_timer_test.py	2695	0644	edit dl rm
open_socket_cb_test.py	4761	0644	edit dl rm
option_constants_test.py	16371	0644	edit dl rm
perform_test.py	2179	0644	edit dl rm
procmgr.py	2731	0644	edit dl rm
protocol_constants_test.py	1679	0644	edit dl rm
readdata_test.py	9623	0644	edit dl rm
read_cb_test.py	4340	0644	edit dl rm
relative_url_test.py	551	0644	edit dl rm
reload_test.py	430	0644	edit dl rm
reset_test.py	2699	0644	edit dl rm
resolve_test.py	780	0644	edit dl rm
run-quickstart.sh	768	0755	edit dl rm
run.sh	690	0755	edit dl rm
runwsgi.py	3733	0644	edit dl rm
seek_cb_constants_test.py	833	0644	edit dl rm
setopt_lifecycle_test.py	1797	0644	edit dl rm
setopt_string_test.py	816	0644	edit dl rm
setopt_test.py	3773	0644	edit dl rm
setopt_unicode_test.py	1092	0644	edit dl rm
setup_test.py	10035	0644	edit dl rm
share_test.py	1726	0644	edit dl rm
sockopt_cb_test.py	2453	0644	edit dl rm
unset_range_test.py	1482	0644	edit dl rm
user_agent_string_test.py	820	0644	edit dl rm
util.py	8291	0644	edit dl rm
version_comparison_test.py	514	0644	edit dl rm
version_constants_test.py	2056	0644	edit dl rm
version_test.py	313	0644	edit dl rm
vsftpd.conf	294	0644	edit dl rm
weakref_test.py	443	0644	edit dl rm
write_abort_test.py	1194	0644	edit dl rm
write_cb_bogus_test.py	1381	0644	edit dl rm
write_test.py	10048	0644	edit dl rm
write_to_stringio_test.py	1228	0644	edit dl rm
xferinfo_cb_test.py	2147	0644	edit dl rm
__init__.py	535	0644	edit dl rm

Edit: /usr/share/doc/python3-pycurl/tests/runwsgi.py (3733B)

# Run a WSGI application in a daemon thread

import bottle
import threading
import os.path

from . import util

global_stop = False

class Server(bottle.WSGIRefServer):
    def run(self, handler): # pragma: no cover
        self.srv = self.make_server(handler)
        self.serve()

def make_server(self, handler):
        from wsgiref.simple_server import make_server, WSGIRequestHandler
        if self.quiet:
            base = self.options.get('handler_class', WSGIRequestHandler)
            class QuietHandler(base):
                def log_request(*args, **kw):
                    pass
            self.options['handler_class'] = QuietHandler
        srv = make_server(self.host, self.port, handler, **self.options)
        return srv

def serve(self):
        self.srv.serve_forever(poll_interval=0.1)

# http://www.socouldanyone.com/2014/01/bottle-with-ssl.html
# https://github.com/mfm24/miscpython/blob/master/bottle_ssl.py
class SslServer(Server):
    def run(self, handler): # pragma: no cover
        self.srv = self.make_server(handler)

import ssl
        cert_dir = os.path.join(os.path.dirname(__file__), 'certs')
        self.srv.socket = ssl.wrap_socket(
            self.srv.socket,
            keyfile=os.path.join(cert_dir, 'server.key'),
            certfile=os.path.join(cert_dir, 'server.crt'),
            server_side=True)

self.serve()

def start_bottle_server(app, port, server, **kwargs):
    server_thread = ServerThread(app, port, server, kwargs)
    server_thread.daemon = True
    server_thread.start()

ok = util.wait_for_network_service(('127.0.0.1', port), 0.1, 10)
    if not ok:
        import warnings
        warnings.warn('Server did not start after 1 second')

return server_thread.server

class ServerThread(threading.Thread):
    def __init__(self, app, port, server, server_kwargs):
        threading.Thread.__init__(self)
        self.app = app
        self.port = port
        self.server_kwargs = server_kwargs
        self.server = server(host='127.0.0.1', port=self.port, **self.server_kwargs)

def run(self):
        bottle.run(self.app, server=self.server, quiet=True)

started_servers = {}

def app_runner_setup(*specs):
    '''Returns setup and teardown methods for running a list of WSGI
    applications in a daemon thread.

Each argument is an (app, port) pair.

Return value is a (setup, teardown) function pair.

The setup and teardown functions expect to be called with an argument
    on which server state will be stored.

Example usage with nose:

>>> setup_module, teardown_module = \
        runwsgi.app_runner_setup((app_module.app, 8050))
    '''

def setup(self):
        self.servers = []
        for spec in specs:
            if len(spec) == 2:
                app, port = spec
                kwargs = {}
            else:
                app, port, kwargs = spec
            if port in started_servers:
                assert started_servers[port] == (app, kwargs)
            else:
                server = Server
                if 'server' in kwargs:
                    server = kwargs['server']
                    del kwargs['server']
                elif 'ssl' in kwargs:
                    if kwargs['ssl']:
                        server = SslServer
                    del kwargs['ssl']
                self.servers.append(start_bottle_server(app, port, server, **kwargs))
            started_servers[port] = (app, kwargs)

def teardown(self):
        return
        for server in self.servers:
            # if no tests from module were run, there is no server to shut down
            if hasattr(server, 'srv'):
                server.srv.shutdown()

return [setup, teardown]

# Run a WSGI application in a daemon thread import bottle import threading import os.path from . import util global_stop = False class Server(bottle.WSGIRefServer): def run(self, handler): # pragma: no cover self.srv = self.make_server(handler) self.serve() def make_server(self, handler): from wsgiref.simple_server import make_server, WSGIRequestHandler if self.quiet: base = self.options.get('handler_class', WSGIRequestHandler) class QuietHandler(base): def log_request(*args, **kw): pass self.options['handler_class'] = QuietHandler srv = make_server(self.host, self.port, handler, **self.options) return srv def serve(self): self.srv.serve_forever(poll_interval=0.1) # http://www.socouldanyone.com/2014/01/bottle-with-ssl.html # https://github.com/mfm24/miscpython/blob/master/bottle_ssl.py class SslServer(Server): def run(self, handler): # pragma: no cover self.srv = self.make_server(handler) import ssl cert_dir = os.path.join(os.path.dirname(__file__), 'certs') self.srv.socket = ssl.wrap_socket( self.srv.socket, keyfile=os.path.join(cert_dir, 'server.key'), certfile=os.path.join(cert_dir, 'server.crt'), server_side=True) self.serve() def start_bottle_server(app, port, server, **kwargs): server_thread = ServerThread(app, port, server, kwargs) server_thread.daemon = True server_thread.start() ok = util.wait_for_network_service(('127.0.0.1', port), 0.1, 10) if not ok: import warnings warnings.warn('Server did not start after 1 second') return server_thread.server class ServerThread(threading.Thread): def __init__(self, app, port, server, server_kwargs): threading.Thread.__init__(self) self.app = app self.port = port self.server_kwargs = server_kwargs self.server = server(host='127.0.0.1', port=self.port, **self.server_kwargs) def run(self): bottle.run(self.app, server=self.server, quiet=True) started_servers = {} def app_runner_setup(*specs): '''Returns setup and teardown methods for running a list of WSGI applications in a daemon thread. Each argument is an (app, port) pair. Return value is a (setup, teardown) function pair. The setup and teardown functions expect to be called with an argument on which server state will be stored. Example usage with nose: >>> setup_module, teardown_module = \ runwsgi.app_runner_setup((app_module.app, 8050)) ''' def setup(self): self.servers = [] for spec in specs: if len(spec) == 2: app, port = spec kwargs = {} else: app, port, kwargs = spec if port in started_servers: assert started_servers[port] == (app, kwargs) else: server = Server if 'server' in kwargs: server = kwargs['server'] del kwargs['server'] elif 'ssl' in kwargs: if kwargs['ssl']: server = SslServer del kwargs['ssl'] self.servers.append(start_bottle_server(app, port, server, **kwargs)) started_servers[port] = (app, kwargs) def teardown(self): return for server in self.servers: # if no tests from module were run, there is no server to shut down if hasattr(server, 'srv'): server.srv.shutdown() return [setup, teardown]