diff options
author | alexv-smirnov <[email protected]> | 2023-12-01 12:02:50 +0300 |
---|---|---|
committer | alexv-smirnov <[email protected]> | 2023-12-01 13:28:10 +0300 |
commit | 0e578a4c44d4abd539d9838347b9ebafaca41dfb (patch) | |
tree | a0c1969c37f818c830ebeff9c077eacf30be6ef8 /contrib/python/pytest-localserver/py3 | |
parent | 84f2d3d4cc985e63217cff149bd2e6d67ae6fe22 (diff) |
Change "ya.make"
Diffstat (limited to 'contrib/python/pytest-localserver/py3')
11 files changed, 1044 insertions, 0 deletions
diff --git a/contrib/python/pytest-localserver/py3/.dist-info/METADATA b/contrib/python/pytest-localserver/py3/.dist-info/METADATA new file mode 100644 index 00000000000..42c4db02b3f --- /dev/null +++ b/contrib/python/pytest-localserver/py3/.dist-info/METADATA @@ -0,0 +1,300 @@ +Metadata-Version: 2.1 +Name: pytest-localserver +Version: 0.8.1 +Summary: pytest plugin to test server connections locally. +Home-page: https://github.com/pytest-dev/pytest-localserver +Author: Sebastian Rahlf +Author-email: [email protected] +Maintainer: David Zaslavsky +Maintainer-email: [email protected] +License: MIT License +Keywords: pytest server localhost http smtp +Classifier: Framework :: Pytest +Classifier: Operating System :: OS Independent +Classifier: Development Status :: 4 - Beta +Classifier: Intended Audience :: Developers +Classifier: License :: OSI Approved :: MIT License +Classifier: Programming Language :: Python :: 3 +Classifier: Programming Language :: Python :: 3 :: Only +Classifier: Programming Language :: Python :: 3.5 +Classifier: Programming Language :: Python :: 3.6 +Classifier: Programming Language :: Python :: 3.7 +Classifier: Programming Language :: Python :: 3.8 +Classifier: Programming Language :: Python :: 3.9 +Classifier: Programming Language :: Python :: 3.10 +Classifier: Programming Language :: Python :: 3.11 +Classifier: Programming Language :: Python :: 3.12 +Classifier: Topic :: Software Development :: Testing +Requires-Python: >=3.5 +License-File: LICENSE +License-File: AUTHORS +Requires-Dist: werkzeug >=0.10 +Provides-Extra: smtp +Requires-Dist: aiosmtpd ; extra == 'smtp' + +.. image:: https://img.shields.io/pypi/v/pytest-localserver.svg?style=flat + :alt: PyPI Version + :target: https://pypi.python.org/pypi/pytest-localserver + +.. image:: https://img.shields.io/pypi/pyversions/pytest-localserver.svg + :alt: Supported Python versions + :target: https://pypi.python.org/pypi/pytest-localserver + +.. image:: https://img.shields.io/badge/code%20style-black-000000.svg + :target: https://github.com/psf/black + +.. image:: https://results.pre-commit.ci/badge/github/pytest-dev/pytest-localserver/master.svg + :target: https://results.pre-commit.ci/latest/github/pytest-dev/pytest-localserver/master + :alt: pre-commit.ci status + +================== +pytest-localserver +================== + +pytest-localserver is a plugin for the `pytest`_ testing framework which enables +you to test server connections locally. + +Sometimes `monkeypatching`_ ``urllib2.urlopen()`` just does not cut it, for +instance if you work with ``urllib2.Request``, define your own openers/handlers +or work with ``httplib``. In these cases it may come in handy to have an HTTP +server running locally which behaves just like the real thing [1]_. Well, look +no further! + +Quickstart +========== + +Let's say you have a function to scrape HTML which only required to be pointed +at a URL :: + + import requests + def scrape(url): + html = requests.get(url).text + # some parsing happens here + # ... + return result + +You want to test this function in its entirety without having to rely on a +remote server whose content you cannot control, neither do you want to waste +time setting up a complex mechanism to mock or patch the underlying Python +modules dealing with the actual HTTP request (of which there are more than one +BTW). So what do you do? + +You simply use pytest's `funcargs feature`_ and simulate an entire server +locally! :: + + def test_retrieve_some_content(httpserver): + httpserver.serve_content(open('cached-content.html').read()) + assert scrape(httpserver.url) == 'Found it!' + +What happened here is that for the duration of your tests an HTTP server is +started on a random port on localhost which will serve the content you tell it +to and behaves just like the real thing. + +The added bonus is that you can test whether your code behaves gracefully if +there is a network problem:: + + def test_content_retrieval_fails_graciously(httpserver): + httpserver.serve_content('File not found!', 404) + pytest.raises(ContentNotFoundException, scrape, httpserver.url) + +The same thing works for SMTP servers, too:: + + def test_sending_some_message(smtpserver): + mailer = MyMailer(host=smtpserver.addr[0], port=smtpserver.addr[1]) + mailer.send(to='[email protected]', from_='[email protected]', + subject='MyMailer v1.0', body='Check out my mailer!') + assert len(smtpserver.outbox)==1 + +Here an SMTP server is started which accepts e-mails being sent to it. The +nice feature here is that you can actually check if the message was received +and what was sent by looking into the smtpserver's ``outbox``. + +It is really that easy! + +Available funcargs +================== + +Here is a short overview of the available funcargs. For more details I suggest +poking around in the code itself. + +``httpserver`` + provides a threaded HTTP server instance running on localhost. It has the + following attributes: + + * ``code`` - HTTP response code (int) + * ``content`` - content of next response (str, bytes, or iterable of either) + * ``headers`` - response headers (dict) + * ``chunked`` - whether to chunk-encode the response (enumeration) + + Once these attributes are set, all subsequent requests will be answered with + these values until they are changed or the server is stopped. A more + convenient way to change these is :: + + httpserver.serve_content(content=None, code=200, headers=None, chunked=pytest_localserver.http.Chunked.NO) + + The ``chunked`` attribute or parameter can be set to + + * ``Chunked.YES``, telling the server to always apply chunk encoding + * ``Chunked.NO``, telling the server to never apply chunk encoding + * ``Chunked.AUTO``, telling the server to apply chunk encoding only if + the ``Transfer-Encoding`` header includes ``chunked`` + + If chunk encoding is applied, each str or bytes in ``content`` becomes one + chunk in the response. + + The server address can be found in property + + * ``url`` + + which is the string representation of tuple ``server_address`` (host as str, + port as int). + + If you want to check which form fields have been POSTed, Try :: + + httpserver.serve_content(..., show_post_vars=True) + + which will display them as parsable text. + + If you need to inspect the requests sent to the server, a list of all + received requests can be found in property + + * ``requests`` + + which is a list of ``werkzeug.wrappers.Request`` objects. + +``httpsserver`` + is the same as ``httpserver`` only with SSL encryption. + +``smtpserver`` + provides a threaded SMTP server, with an API similar to ``smtpd.SMTPServer``, + (the deprecated class from the Python standard library) running on localhost. + It has the following attributes: + + * ``addr`` - server address as tuple (host as str, port as int) + * ``outbox`` - list of ``email.message.Message`` instances received. + +Using your a WSGI application as test server +============================================ + +As of version 0.3 you can now use a `WSGI application`_ to run on the test +server :: + + from pytest_localserver.http import WSGIServer + + def simple_app(environ, start_response): + """Simplest possible WSGI application""" + status = '200 OK' + response_headers = [('Content-type', 'text/plain')] + start_response(status, response_headers) + return ['Hello world!\n'] + + @pytest.fixture + def testserver(request): + """Defines the testserver funcarg""" + server = WSGIServer(application=simple_app) + server.start() + request.addfinalizer(server.stop) + return server + + def test_retrieve_some_content(testserver): + assert scrape(testserver.url) == 'Hello world!\n' + +Have a look at the following page for more information on WSGI: +http://wsgi.readthedocs.org/en/latest/learn.html + +Download and Installation +========================= + +You can install the plugin by running :: + + pip install pytest-localserver + +Alternatively, get the latest stable version from `PyPI`_ or the latest +`bleeding-edge`_ from Github. + +License and Credits +=================== + +This plugin is released under the MIT license. You can find the full text of +the license in the LICENSE file. + +Copyright (C) 2010-2022 Sebastian Rahlf and others (see AUTHORS). + +Some parts of this package is based on ideas or code from other people: + +- I borrowed some implementation ideas for the httpserver from `linkchecker`_. +- The implementation for the SMTP server is based on the `Mailsink recipe`_ by + Adam Feuer, Matt Branthwaite and Troy Frever. +- The HTTPS implementation is based on work by `Sebastien Martini`_. + +Thanks guys! + +Development and future plans +============================ + +Feel free to clone the repository and add your own changes. Pull requests are +always welcome!:: + + git clone https://github.com/pytest-dev/pytest-localserver + +If you find any bugs, please file a `report`_. + +Test can be run with tox. + +I already have a couple of ideas for future versions: + +* support for FTP, SSH (maybe base all on twisted?) +* making the SMTP outbox as convenient to use as ``django.core.mail.outbox`` +* add your own here! + +Preparing a release +------------------- + +For package maintainers, here is how we release a new version: + +#. Ensure that the ``CHANGES`` file is up to date with the latest changes. +#. Make sure that all tests pass on the version you want to release. +#. Use the `new release form on Github`_ (or some other equivalent method) to + create a new release, following the pattern of previous releases. + + * Each release has to be based on a tag. You can either create the tag first + (e.g. using ``git tag``) and then make a release from that tag, or you can + have Github create the tag as part of the process of making a release; + either way works. + * The tag name **must** be the `PEP 440`_-compliant version number prefixed + by ``v``, making sure to include at least three version number components + (e.g. ``v0.6.0``). + * The "Auto-generate release notes" button will be useful in summarizing + the changes since the last release. + +#. Using either the `release workflows page`_ or the link in the email you + received about a "Deployment review", go to the workflow run created for + the new release and click "Review deployments", then either approve or reject + the two deployments, one to Test PyPI and one to real PyPI. (It should not be + necessary to reject a deployment unless something really weird happens.) + Once the deployment is approved, Github will automatically upload the files. + +---- + +.. [1] The idea for this project was born when I needed to check that `a piece + of software`_ behaved itself when receiving HTTP error codes 404 and 500. + Having unsuccessfully tried to mock a server, I stumbled across + `linkchecker`_ which uses a the same idea to test its internals. + +.. _monkeypatching: http://pytest.org/latest/monkeypatch.html +.. _pytest: http://pytest.org/ +.. _funcargs feature: http://pytest.org/latest/funcargs.html +.. _linkchecker: http://linkchecker.sourceforge.net/ +.. _WSGI application: http://www.python.org/dev/peps/pep-0333/ +.. _PyPI: http://pypi.python.org/pypi/pytest-localserver/ +.. _bleeding-edge: https://github.com/pytest-dev/pytest-localserver +.. _report: https://github.com/pytest-dev/pytest-localserver/issues/ +.. _tox: http://testrun.org/tox/ +.. _a piece of software: http://pypi.python.org/pypi/python-amazon-product-api/ +.. _Mailsink recipe: http://code.activestate.com/recipes/440690/ +.. _Sebastien Martini: http://code.activestate.com/recipes/442473/ +.. _PEP 440: https://peps.python.org/pep-0440/ +.. _build: https://pypa-build.readthedocs.io/en/latest/ +.. _twine: https://twine.readthedocs.io/en/stable/ +.. _new release form on Github: https://github.com/pytest-dev/pytest-localserver/releases/new +.. _release workflows page: https://github.com/pytest-dev/pytest-localserver/actions/workflows/release.yml diff --git a/contrib/python/pytest-localserver/py3/.dist-info/entry_points.txt b/contrib/python/pytest-localserver/py3/.dist-info/entry_points.txt new file mode 100644 index 00000000000..72608edbeec --- /dev/null +++ b/contrib/python/pytest-localserver/py3/.dist-info/entry_points.txt @@ -0,0 +1,2 @@ +[pytest11] +localserver = pytest_localserver.plugin diff --git a/contrib/python/pytest-localserver/py3/.dist-info/top_level.txt b/contrib/python/pytest-localserver/py3/.dist-info/top_level.txt new file mode 100644 index 00000000000..cfe3f2cf354 --- /dev/null +++ b/contrib/python/pytest-localserver/py3/.dist-info/top_level.txt @@ -0,0 +1 @@ +pytest_localserver diff --git a/contrib/python/pytest-localserver/py3/pytest_localserver/__init__.py b/contrib/python/pytest-localserver/py3/pytest_localserver/__init__.py new file mode 100644 index 00000000000..482058b8229 --- /dev/null +++ b/contrib/python/pytest-localserver/py3/pytest_localserver/__init__.py @@ -0,0 +1 @@ +from pytest_localserver._version import version as VERSION # noqa diff --git a/contrib/python/pytest-localserver/py3/pytest_localserver/_version.py b/contrib/python/pytest-localserver/py3/pytest_localserver/_version.py new file mode 100644 index 00000000000..044524b1665 --- /dev/null +++ b/contrib/python/pytest-localserver/py3/pytest_localserver/_version.py @@ -0,0 +1,16 @@ +# file generated by setuptools_scm +# don't change, don't track in version control +TYPE_CHECKING = False +if TYPE_CHECKING: + from typing import Tuple, Union + VERSION_TUPLE = Tuple[Union[int, str], ...] +else: + VERSION_TUPLE = object + +version: str +__version__: str +__version_tuple__: VERSION_TUPLE +version_tuple: VERSION_TUPLE + +__version__ = version = '0.8.1' +__version_tuple__ = version_tuple = (0, 8, 1) diff --git a/contrib/python/pytest-localserver/py3/pytest_localserver/http.py b/contrib/python/pytest-localserver/py3/pytest_localserver/http.py new file mode 100644 index 00000000000..0899597f5e8 --- /dev/null +++ b/contrib/python/pytest-localserver/py3/pytest_localserver/http.py @@ -0,0 +1,183 @@ +# Copyright (C) 2010-2013 Sebastian Rahlf and others (see AUTHORS). +# +# This program is release under the MIT license. You can find the full text of +# the license in the LICENSE file. +import enum +import itertools +import json +import sys +import threading + +from werkzeug.datastructures import Headers +from werkzeug.serving import make_server +from werkzeug.wrappers import Request +from werkzeug.wrappers import Response + + +class WSGIServer(threading.Thread): + + """ + HTTP server running a WSGI application in its own thread. + """ + + def __init__(self, host="127.0.0.1", port=0, application=None, **kwargs): + self.app = application + self._server = make_server(host, port, self.app, **kwargs) + self.server_address = self._server.server_address + + super().__init__(name=self.__class__, target=self._server.serve_forever) + + def __del__(self): + self.stop() + + def stop(self): + try: + server = self._server + except AttributeError: + pass + else: + server.shutdown() + + @property + def url(self): + host, port = self.server_address + proto = "http" if self._server.ssl_context is None else "https" + return "%s://%s:%i" % (proto, host, port) + + +class Chunked(enum.Enum): + NO = False + YES = True + AUTO = None + + def __bool__(self): + return bool(self.value) + + +def _encode_chunk(chunk, charset): + if isinstance(chunk, str): + chunk = chunk.encode(charset) + return "{:x}".format(len(chunk)).encode(charset) + b"\r\n" + chunk + b"\r\n" + + +class ContentServer(WSGIServer): + + """ + Small test server which can be taught which content (i.e. string) to serve + with which response code. Try the following snippet for testing API calls:: + + server = ContentServer(port=8080) + server.start() + print 'Test server running at http://%s:%i' % server.server_address + + # any request to http://localhost:8080 will get a 503 response. + server.content = 'Hello World!' + server.code = 503 + + # ... + + # we're done + server.stop() + + """ + + def __init__(self, host="127.0.0.1", port=0, ssl_context=None): + super().__init__(host, port, self, ssl_context=ssl_context) + self.content, self.code = ("", 204) # HTTP 204: No Content + self.headers = {} + self.show_post_vars = False + self.compress = None + self.requests = [] + self.chunked = Chunked.NO + + def __call__(self, environ, start_response): + """ + This is the WSGI application. + """ + request = Request(environ) + self.requests.append(request) + if ( + request.content_type == "application/x-www-form-urlencoded" + and request.method == "POST" + and self.show_post_vars + ): + content = json.dumps(request.form) + else: + content = self.content + + if self.chunked == Chunked.YES or ( + self.chunked == Chunked.AUTO and "chunked" in self.headers.get("Transfer-encoding", "") + ): + # If the code below ever changes to allow setting the charset of + # the Response object, the charset used here should also be changed + # to match. But until that happens, use UTF-8 since it is Werkzeug's + # default. + charset = "utf-8" + if isinstance(content, (str, bytes)): + content = (_encode_chunk(content, charset), "0\r\n\r\n") + else: + content = itertools.chain((_encode_chunk(item, charset) for item in content), ["0\r\n\r\n"]) + + response = Response(response=content, status=self.code) + response.headers.clear() + response.headers.extend(self.headers) + + # FIXME get compression working! + # if self.compress == 'gzip': + # content = gzip.compress(content.encode('utf-8')) + # response.content_encoding = 'gzip' + + return response(environ, start_response) + + def serve_content(self, content, code=200, headers=None, chunked=Chunked.NO): + """ + Serves string content (with specified HTTP error code) as response to + all subsequent request. + + :param content: content to be displayed + :param code: HTTP status code + :param headers: HTTP headers to be returned + :param chunked: whether to apply chunked transfer encoding to the content + """ + if not isinstance(content, (str, bytes, list, tuple)): + # If content is an iterable which is not known to be a string, + # bytes, or sequence, it might be something that can only be iterated + # through once, in which case we need to cache it so it can be reused + # to handle multiple requests. + try: + content = tuple(iter(content)) + except TypeError: + # this probably means that content is not iterable, so just go + # ahead in case it's some type that Response knows how to handle + pass + self.content = content + self.code = code + self.chunked = chunked + if headers: + self.headers = Headers(headers) + + +if __name__ == "__main__": # pragma: no cover + import os.path + import time + + app = ContentServer() + server = WSGIServer(application=app) + server.start() + + print("HTTP server is running at %s" % server.url) + print("Type <Ctrl-C> to stop") + + try: + path = sys.argv[1] + except IndexError: + path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "README.rst") + + app.serve_content(open(path).read(), 302) + + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + print("\rstopping...") + server.stop() diff --git a/contrib/python/pytest-localserver/py3/pytest_localserver/https.py b/contrib/python/pytest-localserver/py3/pytest_localserver/https.py new file mode 100644 index 00000000000..856e222ac5c --- /dev/null +++ b/contrib/python/pytest-localserver/py3/pytest_localserver/https.py @@ -0,0 +1,149 @@ +# Copyright (C) 2010-2013 Sebastian Rahlf and others (see AUTHORS). +# +# This program is release under the MIT license. You can find the full text of +# the license in the LICENSE file. +import os.path + +from pytest_localserver.http import ContentServer + +#: default server certificate +DEFAULT_CERTIFICATE = os.path.join(os.getcwd(), "server.pem") + + +class SecureContentServer(ContentServer): + + """ + Small test server which works just like :class:`http.Server` over HTTP:: + + server = SecureContentServer( + port=8080, key='/srv/my.key', cert='my.certificate') + server.start() + print 'Test server running at %s' % server.url + server.serve_content(open('/path/to/some.file').read()) + # any call to https://localhost:8080 will get the contents of + # /path/to/some.file as a response. + + To avoid *ssl handshake failures* you can import the `pytest-localserver + CA`_ into your browser of choice. + + How to create a self-signed certificate + --------------------------------------- + + If you want to create your own server certificate, you need `OpenSSL`_ + installed on your machine. A self-signed certificate consists of a + certificate and a private key for your server. It can be created with + a command like this, using OpenSSL 1.1.1:: + + openssl req \ + -x509 \ + -newkey rsa:4096 \ + -sha256 \ + -days 3650 \ + -nodes \ + -keyout server.pem \ + -out server.pem \ + -subj "/CN=127.0.0.1/O=pytest-localserver/OU=Testing Dept." \ + -addext "subjectAltName=DNS:localhost" + + Note that both key and certificate are in a single file now named + ``server.pem``. + + How to create your own Certificate Authority + -------------------------------------------- + + Generate a server key and request for signing (csr). Make sure that the + common name (CN) is your IP address/domain name (e.g. ``localhost``). :: + + openssl genpkey \ + -algorithm RSA \ + -pkeyopt rsa_keygen_bits:4096 \ + -out server.key + openssl req \ + -new \ + -addext "subjectAltName=DNS:localhost" \ + -key server.key \ + -out server.csr + + Generate your own CA. Make sure that this time the CN is *not* your IP + address/domain name (e.g. ``localhost CA``). :: + + openssl genpkey \ + -algorithm RSA \ + -pkeyopt rsa_keygen_bits:4096 \ + -aes256 \ + -out ca.key + openssl req \ + -new \ + -x509 \ + -key ca.key \ + -out ca.crt + + Sign the certificate signing request (csr) with the self-created CA that + you made earlier. Note that OpenSSL does not copy the subjectAltName field + from the request (csr), so you have to provide it again as a file. If you + issue subsequent certificates and your browser already knows about previous + ones simply increment the serial number. :: + + echo "subjectAltName=DNS:localhost" >server-extensions.txt + openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key \ + -set_serial 01 -extfile server-extensions.txt -out server.crt + + Create a single file for both key and certificate:: + + cat server.key server.crt > server.pem + + Now you only need to import ``ca.crt`` as a CA in your browser. + + Want to know more? + ------------------ + + This information was compiled from the following sources, which you might + find helpful if you want to dig deeper into `pyOpenSSH`_, certificates and + CAs: + + - http://code.activestate.com/recipes/442473/ + - http://www.tc.umn.edu/~brams006/selfsign.html + - + + A more advanced tutorial can be found `here`_. + + .. _pytest-localserver CA: https://raw.githubusercontent.com/pytest-dev/pytest-localserver/master/pytest_localserver/ca.crt # noqa: E501 + .. _pyOpenSSH: https://launchpad.net/pyopenssl + """ + + def __init__(self, host="localhost", port=0, key=DEFAULT_CERTIFICATE, cert=DEFAULT_CERTIFICATE): + """ + :param key: location of file containing the server private key. + :param cert: location of file containing server certificate. + """ + + super().__init__(host, port, ssl_context=(key, cert)) + + +if __name__ == "__main__": # pragma: no cover + + import sys + import time + + print("Using certificate %s." % DEFAULT_CERTIFICATE) + + server = SecureContentServer() + server.start() + server.logging = True + + print("HTTPS server is running at %s" % server.url) + print("Type <Ctrl-C> to stop") + + try: + path = sys.argv[1] + except IndexError: + path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "README.rst") + + server.serve_content(open(path).read(), 302) + + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + print("\rstopping...") + server.stop() diff --git a/contrib/python/pytest-localserver/py3/pytest_localserver/plugin.py b/contrib/python/pytest-localserver/py3/pytest_localserver/plugin.py new file mode 100644 index 00000000000..1e6ad2f1720 --- /dev/null +++ b/contrib/python/pytest-localserver/py3/pytest_localserver/plugin.py @@ -0,0 +1,90 @@ +# Copyright (C) 2011 Sebastian Rahlf <basti at redtoad dot de> +# +# This program is release under the MIT license. You can find the full text of +# the license in the LICENSE file. +import os +import pkgutil + +import pytest + + +def httpserver(request): + """The returned ``httpserver`` provides a threaded HTTP server instance + running on a randomly assigned port on localhost. It can be taught which + content (i.e. string) to serve with which response code and comes with + following attributes: + + * ``code`` - HTTP response code (int) + * ``content`` - content of next response (str) + * ``headers`` - response headers (dict) + + Once these attribute are set, all subsequent requests will be answered with + these values until they are changed or the server is stopped. A more + convenient way to change these is :: + + httpserver.serve_content( + content='My content', code=200, + headers={'content-type': 'text/plain'}) + + The server address can be found in property + + * ``url`` + + which is the string representation of tuple ``server_address`` (host as + str, port as int). + + Example:: + + import requests + def scrape(url): + html = requests.get(url).text + # some parsing happens here + # ... + return result + + def test_retrieve_some_content(httpserver): + httpserver.serve_content(open('cached-content.html').read()) + assert scrape(httpserver.url) == 'Found it!' + + """ + from pytest_localserver import http + + server = http.ContentServer() + server.start() + request.addfinalizer(server.stop) + return server + + +def httpsserver(request): + """The returned ``httpsserver`` (note the additional S!) provides a + threaded HTTP server instance similar to funcarg ``httpserver`` but with + SSL encryption. + """ + from pytest_localserver import https + try: + with open(https.DEFAULT_CERTIFICATE, 'wb') as f: + f.write(pkgutil.get_data('pytest_localserver', 'server.pem')) + server = https.SecureContentServer() + server.start() + request.addfinalizer(server.stop) + yield server + finally: + os.remove(https.DEFAULT_CERTIFICATE) + + +def smtpserver(request): + """The returned ``smtpserver`` provides a threaded instance of + ``smtpd.SMTPServer`` running on localhost. It has the following + attributes: + + * ``addr`` - server address as tuple (host as str, port as int) + """ + from pytest_localserver import smtp + + server = smtp.Server() + server.start() + request.addfinalizer(server.stop) + return server diff --git a/contrib/python/pytest-localserver/py3/pytest_localserver/server.pem b/contrib/python/pytest-localserver/py3/pytest_localserver/server.pem new file mode 100644 index 00000000000..4f7f1ed3227 --- /dev/null +++ b/contrib/python/pytest-localserver/py3/pytest_localserver/server.pem @@ -0,0 +1,84 @@ +-----BEGIN PRIVATE KEY----- +MIIJRAIBADANBgkqhkiG9w0BAQEFAASCCS4wggkqAgEAAoICAQC31RlRDMN6eGpQ +zZaqtgk7/c5h98PoPNRpFFoUuuWGdf5PlHv4fMym7Zmz2ljx1DqutKhIIUKqS1vh +xd5zMFpheOUTVPVfQc5evgTIm1GF0rMSSaQSFPVOX3nNXGmF1/Jq9YWTc/rb2ns/ +s+Ip1zSKBqDsdRbrkvpSa7cyCkxYcuYtYo5jRa930Fbn4cNj+aA3dxGXd4bLLfnR +BpRA0V5SzBv93MtOK9kngQwQhjBJC/L/acHPO5dzQISBhM9NTSCAH4zm0SlTiExK +DhBdExSbdjAjnJ3k82hNFLUqY1JAm4yVlvwD3vNY4hkf/gWzuQeJIhzK8kE4A+dD +8BZzdHroK9xnnpmSlS7/P0raQd3VZPc8swEDyw9MrdA5UU95b07sUVs0LM0vWhi+ +rwNAJHfiQ77twc0bP7niyy/Kg+UYf7m0i/nyvJFKq75rHOfZvmsPNs+gOdJff+yy +4vv9pmImj2nulgOgrGrzc4ICnx3GpoKmGFDq/p+hqk99P92dFHmwd7c2bYHQNpC9 +BJh8VzrVuyndX2mL5P+/LfmEi8tI06Imykzqtk/UODLJks7ZIrJfYlYmm7aVdrvO +1U2s10AfloCX/ZVO7u3k4lH7Stj+/C8Ap+5Cm4Q46sZGO0Z5b808p4ETcoAI/AAl +OwpHAMi9ueLqJ7J0ykCDl/LrTyNqrQIDAQABAoICACObXRn71Okl5cHc8HAEbml2 +UcFcElAraCEqVgBp6wdOV4HmitSop6M1pm3VvyCoMO2iBG5kMtt1WUiz4NCC7x6u +IgDKlfRrdKOZPqf0nafEFfdW2DbAZHtXtun2GmJYX5YkFElpT4/CE9lU6FueWYja +m9TxIQ1kHKRWRNemcv820iq8SkQkPUaBzjN/4S6+LTBRGdEyz6MPNrIsCg87/n8f +FdToLWDo0Vj7f/C7bSLY86pRO77+Fem293N23AhnBgKLGemjXdPWNKCrdLPyfC1Y +iR58uYCdPPihKC4bqtTkzCg1ZH8DcjMnKCKwOz6CelkviFAu+D73UpYwLMkUKLH3 +p3meFBwa0oEzUUof+W9J5HPnVX6nGR2V4fXkejcJoOBHUaSsuRFiPS4XJMj++DI7 +uiMOt7QljqCKirmCp8tVQ5raT9zwFgNCsR3+gemD1KC3zlXixGs1DyI4x2YwTgKU +c16vnh9fGS9zq/drxqbeMvVbyVZF98LjJfgPxcmyEAXVH46Rs3/KSr6ve6MpRk9G +3vLd7BVfEXoGA1Sha7PRg9OaKBgODfkDRsyZJqqkqHurE4P+8NQZ3mhzdGa4Prj9 +er5BrE3gmvagtQUJf0n+E6HRHGCFoq4i+jOeBw8qiwxgWV0ITfhneQDJF8JvBrzJ +IByC9fVUYB4R4wESRoOBAoIBAQDgJz1etpf/47PwM5A5bdqJEEMaT6d0ycI7QtFQ +1L3PgdRmu1ag2pCYogPQx1zrkMUe4qX0K1h372toRUz0RvoQFYF6qVLJNK/hOd+O +GQ/Rw3XuCvCCs+6QbeQNqUjhLeYf3+TH4IEbIvukIACnDlHvAhtu+IQR7sVwCXda +Slu1zW0ya7Pa/pPnEOQpA9D758/GjcZpe44hCBq+EnrV40Q3jHXEVtsDexq1ubzz +BZEVLr4iwVrEjELZo4pbT+wQx2waRFqTVej5RaQadnSMCdRC0LCTa+t7hfuzN+KN +DBoSUeOlcQ88TyEGvcZXo0jAyDBdN5HC38ujZlkqHHCZVEGxAoIBAQDR81WeHMYW +/vtUhrP3BaJMj3RL/Vmpujac/i9IjdxrP2bi9mweunkZBH9UHNPcJp2b/+uAdSJO +aQRzghCM+DmuOIuu4rB9FU6qpXGhcag126iu328eSYS1sJg5CVGs9ZhaxKk5xbro +1cV0uUS6Gxl2z1Kpsb2dy/zhPTSwf6nrKXYwfrM65+EURz0fniKGfgxs6+p+uTVS +kkLMe2nusJ1KLGrXqfJfa25sQKo1zaRFHLDd0/pgchijvVkhXDY3A7913i+xbQZu +KIfbGp0pH4XFUJn1AR4XqPpE+wmHiLeqEmFJ5xcDl4q3j2dGnO3mHUYYNFOxh1nt +1MCDCCbKJVu9AoIBAQCOUtv4o19nrqC1x0ev7zxvAtBYiHL/CIw3LHnTJQFQHFNM +125tu9lL0LMzgSJSwB0pOye8HTmTDYXZMwdloxtr0vvfcluKPdXe3+w+QVN2EPF0 +L6X+l1jGg7/lnLMVpxsS6gpNjxLqtA+ralZ/u+vyIhhhIZJaAI2EUb5iqgwJJ2JK +PXB5gGNQt7zm/fFXwRyAKcztdPINryOrw/gSjrblvl2YSL3PO/79m+2JMOOp24AG +eVa0rYpUvi4/REPTc4wEMZqBKm8+tyU3WDcwI52Ovwsez8s5Jx1l8fn7LM/xCeXN +SjguRt/lc+HYC2lKXtG2nm4Cmi6mlXnP7zbfZExBAoIBAQC1xAkU+W5ajGjFllWK +gJsx02TpQS92bVxI8Ru4ofD5/QszZgrXU7Px/93I0ahuShRb8eZO8ZpA7lTHOAzi +Lymo9xWf1Gzd7iuMO+4zyrXJ4yGYPKL0QswdjQVNJA9NQdekhezIsrKOUD1CQAAL +a9jQ7s9vUQ2L5wZJbvcF85EFooDLnXXIguZv6vk1PXBApjJVvq3nBqvuj+g7JoHg +/5E9nVTm4CCRke4o1JdIO4CDwUIy2wpCo6VHZXAcHLxnRtxkzHbYEj7l8jska1cz +OjJTUOPppQ0LiOUcAYcPi0MPgBgwplxbZMDZCNNt5AFnH2MHI45t/XPTH0WIa+9B +RbS1AoIBAQCqZri7tm9ngZfvKNNvdVTgBcKukDFek4f7ar0bOkISALtNrn1xXIID +1ggELNy9afTmzPlttqMVQIxSTL3p7LIkZzTsuK0uthbsyzXxLHw2m+oHgaYut7he +j2v7qTmaw7rgpTiORTDg00+5HDtdMmp3Km4aurNasPA80i8Z2ElI20i50LlQ4K5Q +lIqpHR4fwrBr4SLStzvBo9UK1YYQ94FyKd7xou3uXLLTlY3G8rD6jjKJE2Gg8Ga/ +gGzbRCZWH6AOk1iO/CmOPH6AdFn5axXTx+uAML1Lr2VQ+azrYZCtIhKmW/kuQPQg +apeiobcSY1vsX7eM8mQkM8TxrDLyNjtl +-----END PRIVATE KEY----- +-----BEGIN CERTIFICATE----- +MIIFiTCCA3GgAwIBAgIUUpWEFJm0PzrYTkhLe05yIBhBMuowDQYJKoZIhvcNAQEL +BQAwSTESMBAGA1UEAwwJMTI3LjAuMC4xMRswGQYDVQQKDBJweXRlc3QtbG9jYWxz +ZXJ2ZXIxFjAUBgNVBAsMDVRlc3RpbmcgRGVwdC4wHhcNMjEwOTE0MDU1NzAxWhcN +MzEwOTEyMDU1NzAxWjBJMRIwEAYDVQQDDAkxMjcuMC4wLjExGzAZBgNVBAoMEnB5 +dGVzdC1sb2NhbHNlcnZlcjEWMBQGA1UECwwNVGVzdGluZyBEZXB0LjCCAiIwDQYJ +KoZIhvcNAQEBBQADggIPADCCAgoCggIBALfVGVEMw3p4alDNlqq2CTv9zmH3w+g8 +1GkUWhS65YZ1/k+Ue/h8zKbtmbPaWPHUOq60qEghQqpLW+HF3nMwWmF45RNU9V9B +zl6+BMibUYXSsxJJpBIU9U5fec1caYXX8mr1hZNz+tvaez+z4inXNIoGoOx1FuuS ++lJrtzIKTFhy5i1ijmNFr3fQVufhw2P5oDd3EZd3hsst+dEGlEDRXlLMG/3cy04r +2SeBDBCGMEkL8v9pwc87l3NAhIGEz01NIIAfjObRKVOITEoOEF0TFJt2MCOcneTz +aE0UtSpjUkCbjJWW/APe81jiGR/+BbO5B4kiHMryQTgD50PwFnN0eugr3GeemZKV +Lv8/StpB3dVk9zyzAQPLD0yt0DlRT3lvTuxRWzQszS9aGL6vA0Akd+JDvu3BzRs/ +ueLLL8qD5Rh/ubSL+fK8kUqrvmsc59m+aw82z6A50l9/7LLi+/2mYiaPae6WA6Cs +avNzggKfHcamgqYYUOr+n6GqT30/3Z0UebB3tzZtgdA2kL0EmHxXOtW7Kd1faYvk +/78t+YSLy0jToibKTOq2T9Q4MsmSztkisl9iViabtpV2u87VTazXQB+WgJf9lU7u +7eTiUftK2P78LwCn7kKbhDjqxkY7RnlvzTyngRNygAj8ACU7CkcAyL254uonsnTK +QIOX8utPI2qtAgMBAAGjaTBnMB0GA1UdDgQWBBRzl1iPBK4XZwChdNhdPHfjAb/z +BzAfBgNVHSMEGDAWgBRzl1iPBK4XZwChdNhdPHfjAb/zBzAPBgNVHRMBAf8EBTAD +AQH/MBQGA1UdEQQNMAuCCWxvY2FsaG9zdDANBgkqhkiG9w0BAQsFAAOCAgEATk+Q +t6psMrtGeFcZKYdmSFqW3SZUba4l76PzvHRf8nMcB1eFuZ4mCdiv0NgcQkE8c9T+ +i/J4wEmJ+mf1033MP1vQmrGqnaYBsVHNBTaTsP+gLg6Z7AGPvPaL2fwmWWNwTT0O +1352bdz9ORacKSXW3Pq0Vi1pTMho0kAya3VQpl2paqz8qSUG7ijyGQ46VXjgqNZ1 +P5lv+6CWa3AwEQo6Edv1x+HLesRWVqVAkxxhlaGOPQm1cDlpnI4rxuYIMlsb5cNZ +XTAIxw6Es1eqlPcZ96EoGXyIrG7Ej6Yb9447PrC1ulMnIu74cWLY25eu+oVr7Nvk +Gjp2I7qbVjz9Ful0o0M9Wps4RzCgrpO4WeirCK/jFIUpmXJdn7V4mX0h2ako+dal +vczg+bAd4ZedJWHTiqJs9lVMh4/YD7Ck6n+iAZ8Jusq6OhyTY43/Nyp2zQbwQmYv +y3V6JVX+vY4Cq8pR1i8x5FBHnOCMPoT4sbOjKuoFWVi9wH1d65Q1JOo6/0eYzfwJ +nuGUJza7+aCxYNlqxtqX0ItM670ClxB7fuWUpKh5WHrHD2dqBhYwtXOl9yBHrFOJ +O8toKk3PmtlMqVZ8QXmgSqEy7wkfxhjJLgi2AQsqeA6nDrCLtr2pWdqDWoUfxY8r +r5rc71nFLay/H2CbOYELI+20VFMp8GF3kOZbkRA= +-----END CERTIFICATE----- diff --git a/contrib/python/pytest-localserver/py3/pytest_localserver/smtp.py b/contrib/python/pytest-localserver/py3/pytest_localserver/smtp.py new file mode 100644 index 00000000000..82dbc394b98 --- /dev/null +++ b/contrib/python/pytest-localserver/py3/pytest_localserver/smtp.py @@ -0,0 +1,177 @@ +# Copyright (C) 2011 Sebastian Rahlf <basti at redtoad dot de> +# with some ideas from http://code.activestate.com/recipes/440690/ +# SmtpMailsink Copyright 2005 Aviarc Corporation +# Written by Adam Feuer, Matt Branthwaite, and Troy Frever +# which is Licensed under the PSF License +import email + +import aiosmtpd.controller + + +class MessageDetails: + def __init__(self, peer, mailfrom, rcpttos, *, mail_options=None, rcpt_options=None): + self.peer = peer + self.mailfrom = mailfrom + self.rcpttos = rcpttos + if mail_options: + self.mail_options = mail_options + if rcpt_options: + self.rcpt_options = rcpt_options + + +class Handler: + def __init__(self): + self.outbox = [] + + async def handle_DATA(self, server, session, envelope): + message = email.message_from_bytes(envelope.content) + message.details = MessageDetails(session.peer, envelope.mail_from, envelope.rcpt_tos) + self.outbox.append(message) + return "250 OK" + + +class Server(aiosmtpd.controller.Controller): + + """ + Small SMTP test server. + + This is little more than a wrapper around aiosmtpd.controller.Controller + which offers a slightly different interface for backward compatibility with + earlier versions of pytest-localserver. You can just as well use a standard + Controller and pass it a Handler instance. + + Here is how to use this class for sending an email, if you really need to:: + + server = Server(port=8080) + server.start() + print 'SMTP server is running on %s:%i' % server.addr + + # any e-mail sent to localhost:8080 will end up in server.outbox + # ... + + server.stop() + + """ + + def __init__(self, host="localhost", port=0): + try: + super().__init__(Handler(), hostname=host, port=port, server_hostname=host) + except TypeError: + # for aiosmtpd <1.3 + super().__init__(Handler(), hostname=host, port=port) + + @property + def outbox(self): + return self.handler.outbox + + def _set_server_socket_attributes(self): + """ + Set the addr and port attributes on this Server instance, if they're not + already set. + """ + + # I split this out into its own method to allow running this code in + # aiosmtpd <1.4, which doesn't have the _trigger_server() method on + # the Controller class. If I put it directly in _trigger_server(), it + # would fail when calling super()._trigger_server(). In the future, when + # we can safely require aiosmtpd >=1.4, this method can be inlined + # directly into _trigger_server(). + if hasattr(self, "addr"): + assert hasattr(self, "port") + return + + self.addr = self.server.sockets[0].getsockname()[:2] + + # Work around a bug/missing feature in aiosmtpd (https://github.com/aio-libs/aiosmtpd/issues/276) + if self.port == 0: + self.port = self.addr[1] + assert self.port != 0 + + def _trigger_server(self): + self._set_server_socket_attributes() + super()._trigger_server() + + def is_alive(self): + return self._thread is not None and self._thread.is_alive() + + @property + def accepting(self): + try: + return self.server.is_serving() + except AttributeError: + # asyncio.base_events.Server.is_serving() only exists in Python 3.6 + # and up. For Python 3.5, asyncio.base_events.BaseEventLoop.is_running() + # is a close approximation; it should mostly return the same value + # except for brief periods when the server is starting up or shutting + # down. Once we drop support for Python 3.5, this branch becomes + # unnecessary. + return self.loop.is_running() + + # for aiosmtpd <1.4 + if not hasattr(aiosmtpd.controller.Controller, "_trigger_server"): + + def start(self): + super().start() + self._set_server_socket_attributes() + + def stop(self, timeout=None): + """ + Stops test server. + :param timeout: When the timeout argument is present and not None, it + should be a floating point number specifying a timeout for the + operation in seconds (or fractions thereof). + """ + + # This mostly copies the implementation from Controller.stop(), with two + # differences: + # - It removes the assertion that the thread exists, allowing stop() to + # be called more than once safely + # - It passes the timeout argument to Thread.join() + if self.loop.is_running(): + try: + self.loop.call_soon_threadsafe(self.cancel_tasks) + except AttributeError: + # for aiosmtpd < 1.4.3 + self.loop.call_soon_threadsafe(self._stop) + if self._thread is not None: + self._thread.join(timeout) + self._thread = None + self._thread_exception = None + self._factory_invoked = None + self.server_coro = None + self.server = None + self.smtpd = None + + def __del__(self): + # This is just for backward compatibility, to preserve the behavior that + # the server is stopped when this object is finalized. But it seems + # sketchy to rely on this to stop the server. Typically, the server + # should be stopped "manually", before it gets deleted. + if self.is_alive(): + self.stop() + + def __repr__(self): # pragma: no cover + return "<smtp.Server %s:%s>" % self.addr + + +def main(): + import time + + server = Server() + server.start() + + print("SMTP server is running on %s:%i" % server.addr) + print("Type <Ctrl-C> to stop") + + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + pass + finally: + print("\rstopping...") + server.stop() + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/contrib/python/pytest-localserver/py3/ya.make b/contrib/python/pytest-localserver/py3/ya.make new file mode 100644 index 00000000000..0a1b91f6c3e --- /dev/null +++ b/contrib/python/pytest-localserver/py3/ya.make @@ -0,0 +1,41 @@ +# Generated by devtools/yamaker (pypi). + +PY3_LIBRARY() + +VERSION(0.8.1) + +LICENSE(MIT) + +PEERDIR( + contrib/python/Werkzeug +) + +NO_LINT() + +NO_CHECK_IMPORTS( + pytest_localserver.smtp +) + +PY_SRCS( + TOP_LEVEL + pytest_localserver/__init__.py + pytest_localserver/_version.py + pytest_localserver/http.py + pytest_localserver/https.py + pytest_localserver/plugin.py + pytest_localserver/smtp.py +) + +RESOURCE_FILES( + PREFIX contrib/python/pytest-localserver/py3/ + .dist-info/METADATA + .dist-info/entry_points.txt + .dist-info/top_level.txt + pytest_localserver/server.pem +) + +END() + +RECURSE_FOR_TESTS( + tests +) |