Add black configuration and format code

Author: hramezani

.pre-commit-config.yaml

@@ -0,0 +1,6 @@
+repos:
+-   repo: https://github.com/psf/black
+    rev: 22.3.0
+    hooks:
+    -   id: black
+        args: [--safe, --quiet]

README.rst

@@ -6,6 +6,9 @@
     :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
+
 ==================
 pytest-localserver
 ==================

pyproject.toml

@@ -4,3 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [tool.setuptools_scm]
 write_to = "pytest_localserver/_version.py"
+
+[tool.black]
+target-version = ['py35']
+line-length = 120

pytest_localserver/http.py

@@ -20,14 +20,12 @@ 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):
+    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)
+        super().__init__(name=self.__class__, target=self._server.serve_forever)
 
     def __del__(self):
         self.stop()
@@ -38,8 +36,8 @@ def stop(self):
     @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)
+        proto = "http" if self._server.ssl_context is None else "https"
+        return "%s://%s:%i" % (proto, host, port)
 
 
 class Chunked(enum.Enum):
@@ -54,7 +52,7 @@ def __bool__(self):
 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'
+    return "{:x}".format(len(chunk)).encode(charset) + b"\r\n" + chunk + b"\r\n"
 
 
 class ContentServer(WSGIServer):
@@ -78,9 +76,9 @@ class ContentServer(WSGIServer):
 
     """
 
-    def __init__(self, host='127.0.0.1', port=0, ssl_context=None):
+    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.content, self.code = ("", 204)  # HTTP 204: No Content
         self.headers = {}
         self.show_post_vars = False
         self.compress = None
@@ -93,25 +91,27 @@ def __call__(self, environ, start_response):
         """
         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):
+        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 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'
+            charset = "utf-8"
             if isinstance(content, (str, bytes)):
-                content = (_encode_chunk(content, charset), '0\r\n\r\n')
+                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'])
+                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()
@@ -152,28 +152,27 @@ def serve_content(self, content, code=200, headers=None, chunked=Chunked.NO):
             self.headers = Headers(headers)
 
 
-if __name__ == '__main__':  # pragma: no cover
+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')
+    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')
+        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...')
+        print("\rstopping...")
     server.stop()

pytest_localserver/https.py

@@ -8,11 +8,10 @@
 from pytest_localserver.http import ContentServer
 
 #: default server certificate
-DEFAULT_CERTIFICATE = os.path.join(
-    os.path.abspath(os.path.dirname(__file__)), 'server.pem')
+DEFAULT_CERTIFICATE = os.path.join(os.path.abspath(os.path.dirname(__file__)), "server.pem")
 
 
-class SecureContentServer (ContentServer):
+class SecureContentServer(ContentServer):
 
     """
     Small test server which works just like :class:`http.Server` over HTTP::
@@ -113,8 +112,7 @@ class SecureContentServer (ContentServer):
     .. _pyOpenSSH: https://launchpad.net/pyopenssl
     """
 
-    def __init__(self, host='localhost', port=0,
-                 key=DEFAULT_CERTIFICATE, cert=DEFAULT_CERTIFICATE):
+    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.
@@ -123,31 +121,30 @@ def __init__(self, host='localhost', port=0,
         super().__init__(host, port, ssl_context=(key, cert))
 
 
-if __name__ == '__main__':  # pragma: no cover
+if __name__ == "__main__":  # pragma: no cover
 
     import sys
     import time
 
-    print('Using certificate %s.' % DEFAULT_CERTIFICATE)
+    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')
+    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')
+        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...')
+        print("\rstopping...")
     server.stop()

pytest_localserver/plugin.py

@@ -48,6 +48,7 @@ def test_retrieve_some_content(httpserver):
 
     """
     from pytest_localserver import http
+
     server = http.ContentServer()
     server.start()
     request.addfinalizer(server.stop)
@@ -61,6 +62,7 @@ def httpsserver(request):
     SSL encryption.
     """
     from pytest_localserver import https
+
     server = https.SecureContentServer()
     server.start()
     request.addfinalizer(server.stop)
@@ -76,6 +78,7 @@ def smtpserver(request):
     * ``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)

pytest_localserver/smtp.py

@@ -30,7 +30,7 @@ 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'
+        return "250 OK"
 
 
 class Server(aiosmtpd.controller.Controller):
@@ -56,7 +56,7 @@ class Server(aiosmtpd.controller.Controller):
 
     """
 
-    def __init__(self, host='localhost', port=0):
+    def __init__(self, host="localhost", port=0):
         try:
             super().__init__(Handler(), hostname=host, port=port, server_hostname=host)
         except TypeError:
@@ -79,8 +79,8 @@ def _set_server_socket_attributes(self):
         # 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')
+        if hasattr(self, "addr"):
+            assert hasattr(self, "port")
             return
 
         self.addr = self.server.sockets[0].getsockname()[:2]
@@ -111,7 +111,8 @@ def accepting(self):
             return self.loop.is_running()
 
     # for aiosmtpd <1.4
-    if not hasattr(aiosmtpd.controller.Controller, '_trigger_server'):
+    if not hasattr(aiosmtpd.controller.Controller, "_trigger_server"):
+
         def start(self):
             super().start()
             self._set_server_socket_attributes()
@@ -149,24 +150,25 @@ def __del__(self):
             self.stop()
 
     def __repr__(self):  # pragma: no cover
-        return '<smtp.Server %s:%s>' % self.addr
+        return "<smtp.Server %s:%s>" % self.addr
+
 
 if __name__ == "__main__":  # pragma: no cover
     import time
 
     server = Server()
     server.start()
 
-    print('SMTP server is running on %s:%i' % server.addr)
-    print('Type <Ctrl-C> to stop')
+    print("SMTP server is running on %s:%i" % server.addr)
+    print("Type <Ctrl-C> to stop")
 
     try:
 
         try:
             while True:
                 time.sleep(1)
         finally:
-            print('\rstopping...')
+            print("\rstopping...")
             server.stop()
 
     except KeyboardInterrupt:

runtests.py

@@ -2663,28 +2663,30 @@
 import zlib
 import imp
 
+
 class DictImporter:
     def __init__(self, sources):
         self.sources = sources
 
     def find_module(self, fullname, path=None):
         if fullname in self.sources:
             return self
-        if fullname + '.__init__' in self.sources:
+        if fullname + ".__init__" in self.sources:
             return self
         return None
 
     def load_module(self, fullname):
         # print "load_module:",  fullname
         from types import ModuleType
+
         try:
             s = self.sources[fullname]
             is_pkg = False
         except KeyError:
-            s = self.sources[fullname + '.__init__']
+            s = self.sources[fullname + ".__init__"]
             is_pkg = True
 
-        co = compile(s, fullname, 'exec')
+        co = compile(s, fullname, "exec")
         module = sys.modules.setdefault(fullname, ModuleType(fullname))
         module.__file__ = "{}/{}".format(__file__, fullname)
         module.__loader__ = self
@@ -2697,13 +2699,15 @@ def load_module(self, fullname):
     def get_source(self, name):
         res = self.sources.get(name)
         if res is None:
-            res = self.sources.get(name + '.__init__')
+            res = self.sources.get(name + ".__init__")
         return res
 
+
 if __name__ == "__main__":
     exec("def do_exec(co, loc): exec(co, loc)\n")
     import pickle
-    sources = sources.encode("ascii") # ensure bytes
+
+    sources = sources.encode("ascii")  # ensure bytes
     sources = pickle.loads(zlib.decompress(base64.decodebytes(sources)))
 
     importer = DictImporter(sources)