Rietveld Code Review Tool
Help | Bug tracker | Discussion group | Source code

Unified Diff: cms/bin/test_server.py

Issue 29912588: Issue 7019 - [CMS] Refactor `test_server.py` (Closed)
Patch Set: Addressed commments from Patch Set #1. Objectified server handler. Added test cases. Created Oct. 18, 2018, 1:41 p.m.
Use n/p to move between diff chunks; N/P to move between comments.
Jump to:
View side-by-side diff with in-line comments
Download patch
« no previous file with comments | « no previous file | tests/conftest.py » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: cms/bin/test_server.py
diff --git a/cms/bin/test_server.py b/cms/bin/test_server.py
old mode 100644
new mode 100755
index 3255cd977b8af1a4fad89f6a324f5487fba2c074..11656f6dd215be75e9062217b2cf09ba44b3c101
--- a/cms/bin/test_server.py
+++ b/cms/bin/test_server.py
@@ -13,20 +13,17 @@
# You should have received a copy of the GNU General Public License
# along with Adblock Plus. If not, see <http://www.gnu.org/licenses/>.
-import mimetypes
+from __future__ import print_function
+
import os
-import sys
-import argparse
+import mimetypes
+from argparse import ArgumentParser
import jinja2
+from cms.converters import converters
from cms.utils import process_page
from cms.sources import create_source
-from cms.converters import converters
-
-source = None
-address = None
-port = None
UNICODE_ENCODING = 'utf-8'
@@ -47,127 +44,288 @@ ERROR_TEMPLATE = '''
</body>
</html>'''
-# Initilize the mimetypes modules manually for consistent behavior,
+# Initialize the mimetypes modules manually for consistent behavior,
# ignoring local files and Windows Registry.
mimetypes.init([])
-def get_page(path):
- path = path.strip('/')
- if path == '':
- path = source.read_config().get('general', 'defaultlocale')
- if '/' in path:
- locale, page = path.split('/', 1)
- else:
- locale, page = path, ''
-
- default_page = source.read_config().get('general', 'defaultpage')
- alternative_page = '/'.join([page, default_page]).lstrip('/')
+class DynamicServerHandler:
+ """General-purpose WSGI server handler that generates pages on request.
+
+ Parameters
+ ----------
+ host: str
+ The host where the server will run.
+ port: int
+ The TCP port the server will run on.
+ source_dir: str
+ The path to the website contents.
+
+ """
+
+ def __init__(self, host, port, source_dir):
+ self.host = host
+ self.port = port
+ self.source = create_source(source_dir)
+ self.full_url = 'http://{0}:{1}'.format(host, port)
+
+ def _get_data(self, path):
+ """Read the data corresponding to a website path.
+
+ Parameters
+ ----------
+ path: str
+ The path to the page to get the data for.
+
+ Returns
+ -------
+ str/ bytes
+ The data corresponding to the path we're trying to access.
+
+ """
+ if self.source.has_static(path):
+ return self.source.read_static(path)
+
+ page, data = self._get_page(path)
+
+ if page and self._has_conflicts(page):
+ raise Exception('The requested page conflicts with another page.')
+
+ return data
+
+ def _get_page(self, path):
+ """Construct a page and return its contents.
+
+ Parameters
+ ----------
+ path: str
+ The path of the page we want to construct.
+
+ Returns
+ -------
+ (str, str)
+ With the following format:
+ <page_name, page_contents>
+
+ """
+ path = path.strip('/')
+ if path == '':
+ locale, page = self.source.read_config().get('general',
+ 'defaultlocale'), ''
+ elif '/' in path:
+ locale, page = path.split('/', 1)
+ else:
+ locale, page = path, ''
+
+ default_page = self.source.read_config().get('general', 'defaultpage')
+ possible_pages = [page, '/'.join([page, default_page]).lstrip('/')]
+
+ for page_format in converters.iterkeys():
+ for p in possible_pages:
+ if self.source.has_page(p, page_format):
+ return p, process_page(self.source, locale, p, page_format,
+ self.full_url)
+
+ if self.source.has_localizable_file(locale, page):
+ return page, self.source.read_localizable_file(locale, page)
+
+ return None, None
+
+ def _has_conflicts(self, page):
+ """Check if a page has conflicts.
+
+ A page has conflicts if there are other pages with the same name.
+ Parameters
+ ----------
+ page: str
+ The path of the page we're checking for conflicts.
+
+ Returns
+ -------
+ bool
+ True - if the page has conflicts
+ False - otherwise
+
+ """
+ pages = [p for p, _ in self.source.list_pages()]
+ pages.extend(self.source.list_localizable_files())
+
+ if pages.count(page) > 1:
+ return True
+ if any(p.startswith(page + '/') or page.startswith(p + '/') for p in
+ pages):
+ return True
+ return False
+
+ def get_error_page(self, start_response, status, **kw):
+ """Create and display an error page.
+
+ Parameters
+ ----------
+ start_response: function
+ It will be called before constructing the error page, to setup
+ things like the status of the response and the headers.
+ status: str
+ The status of the response we're sending the error page with.
+ Needs to have the following format: "<status_code>
+ <status_message>"
+ kw: dict
+ Any additional arguments that will be passed onto the `stream`
+ method
+ of a `jinja2 Template`.
+
+ Returns
+ -------
+ generator
+ of utf8 strings - fragments of the corresponding error HTML
+ template.
+
+ """
+ env = jinja2.Environment(autoescape=True)
+ page_template = env.from_string(ERROR_TEMPLATE)
+ mime = 'text/html; encoding={}'.format(UNICODE_ENCODING)
+
+ start_response(status, [('Content-Type', mime)])
+
+ for fragment in page_template.stream(status=status, **kw):
+ yield fragment.encode(UNICODE_ENCODING)
+
+ def __call__(self, environ, start_response):
+ """Execute the handler, according with the WSGI standards.
+
+ Parameters
+ ---------
+ environ: dict
+ The environment under which the page is requested.
+ The requested page must be under the `PATH_INFO` key.
+ start_response: function
+ Used to initiate a response. Must take two arguments, in this
+ order:
+ - Response status, in the format "<code> <message>".
+ - Response headers, as a list of tuples.
+
+ Returns
+ -------
+ list
+ With the data for a specific page.
+
+ """
+ path = environ.get('PATH_INFO')
+
+ data = self._get_data(path)
+
+ if data is None:
+ return self.get_error_page(start_response, '404 Not Found',
+ uri=path)
+
+ mime = mimetypes.guess_type(path)[0] or 'text/html'
+
+ if isinstance(data, unicode):
+ data = data.encode(UNICODE_ENCODING)
+ mime = '{0}; charset={1}'.format(mime, UNICODE_ENCODING)
+
+ start_response('200 OK', [('Content-Type', mime)])
+ return [data]
+
+
+def get_handler():
+ """Set the arguments required to run the script.
+
+ Returns
+ -------
+ TestServerHandler
+ Initialised with the script parameters.
+
+ """
+ parser = ArgumentParser(description='CMS development server created to '
+ 'test pages locally and on-the-fly')
+
+ parser.add_argument('path', default=os.curdir, nargs='?',
+ help='Path to the website we intend to run. If not '
+ 'provided, defaults, to the current directory.')
+ parser.add_argument('--host', default='localhost',
+ help='Address of the host the server will listen on. '
+ 'Defaults to "localhost".')
+ parser.add_argument('--port', default=5000, type=int,
+ help='TCP port the server will listen on. Default '
+ '5000.')
- for format in converters.iterkeys():
- for p in (page, alternative_page):
- if source.has_page(p, format):
- site_url = 'http://{}:{}'.format(address, port)
- return (p, process_page(source, locale, p, format, site_url))
- if source.has_localizable_file(locale, page):
- return (page, source.read_localizable_file(locale, page))
+ args = parser.parse_args()
- return (None, None)
+ return DynamicServerHandler(args.host, args.port, args.path)
-def has_conflicting_pages(page):
- pages = [p for p, _ in source.list_pages()]
- pages.extend(source.list_localizable_files())
+def run_werkzeug_server(handler, **kw):
+ """Set up a server that uses `werkzeug`.
- if pages.count(page) > 1:
- return True
- if any(p.startswith(page + '/') or page.startswith(p + '/') for p in pages):
- return True
- return False
+ Parameters
+ ----------
+ handler: DynamicServerHandler
+ Defines the parameters and methods required to handle requests.
+ Raises
+ ------
+ ImportError
+ If the package `werkzeug` is not installed
-def get_data(path):
- if source.has_static(path):
- return source.read_static(path)
+ """
+ from werkzeug.serving import run_simple
- page, data = get_page(path)
- if page and has_conflicting_pages(page):
- raise Exception('The requested page conflicts with another page')
- return data
+ def run(*args, **kwargs):
+ # The werkzeug logger must be configured before the
+ # root logger. Also we must prevent it from propagating
+ # messages, otherwise messages are logged twice.
+ import logging
+ logger = logging.getLogger('werkzeug')
+ logger.propagate = False
+ logger.setLevel(logging.INFO)
+ logger.addHandler(logging.StreamHandler())
+ run_simple(threaded=True, *args, **kwargs)
-def show_error(start_response, status, **kwargs):
- env = jinja2.Environment(autoescape=True)
- template = env.from_string(ERROR_TEMPLATE)
- mime = 'text/html; encoding=%s' % UNICODE_ENCODING
- start_response(status, [('Content-Type', mime)])
- for fragment in template.stream(status=status, **kwargs):
- yield fragment.encode(UNICODE_ENCODING)
+ run(handler.host, handler.port, handler, **kw)
-def handler(environ, start_response):
- path = environ.get('PATH_INFO')
+def run_builtins_server(handler, **kw):
+ """Configure a server that only uses builtin packages.
- data = get_data(path)
- if data is None:
- return show_error(start_response, '404 Not Found', uri=path)
+ Parameters
+ ----------
+ handler: DynamicServerHandler
+ Defines the parameters and methods required to handle requests.
- mime = mimetypes.guess_type(path)[0] or 'text/html'
+ """
+ from SocketServer import ThreadingMixIn
+ from wsgiref.simple_server import WSGIServer, make_server
- if isinstance(data, unicode):
- data = data.encode(UNICODE_ENCODING)
- mime = '%s; charset=%s' % (mime, UNICODE_ENCODING)
+ class ThreadedWSGIServer(ThreadingMixIn, WSGIServer):
+ daemon_threads = True
- start_response('200 OK', [('Content-Type', mime)])
- return [data]
+ def run(host, port, app, **kwargs):
+ def wrapper(environ, start_response):
+ try:
+ return app(environ, start_response)
+ except Exception as e:
+ return handler.get_error_page(
+ start_response, '500 Internal Server Error',
+ uri=environ.get('PATH_INFO'), error=e,
+ )
+ server = make_server(host, port, wrapper, ThreadedWSGIServer)
+ print(' * Running on {0}:{1}'.format(*server.server_address))
+ server.serve_forever()
-if __name__ == '__main__':
+ run(handler.host, handler.port, handler, **kw)
- parser = argparse.ArgumentParser(description='CMS development server created to test pages locally and on-the-fly')
- parser.add_argument('path', nargs='?', default=os.curdir)
- parser.add_argument('-a', '--address', default='localhost', help='Address of the interface the server will listen on')
- parser.add_argument('-p', '--port', type=int, default=5000, help='TCP port the server will listen on')
- args = parser.parse_args()
- source = create_source(args.path)
- address = args.address
- port = args.port
+def main():
+ handler = get_handler()
try:
- from werkzeug.serving import ThreadedWSGIServer, run_simple
-
- # see https://github.com/mitsuhiko/werkzeug/pull/770
- ThreadedWSGIServer.daemon_threads = True
-
- def run(*args, **kwargs):
- # The werkzeug logger must be configured before the
- # root logger. Also we must prevent it from propagating
- # messages, otherwise messages are logged twice.
- import logging
- logger = logging.getLogger('werkzeug')
- logger.propagate = False
- logger.setLevel(logging.INFO)
- logger.addHandler(logging.StreamHandler())
-
- run_simple(threaded=True, *args, **kwargs)
+ run_werkzeug_server(handler, use_reloader=True, use_debugger=True)
except ImportError:
- from SocketServer import ThreadingMixIn
- from wsgiref.simple_server import WSGIServer, make_server
+ run_builtins_server(handler)
- class ThreadedWSGIServer(ThreadingMixIn, WSGIServer):
- daemon_threads = True
- def run(host, port, app, **kwargs):
- def wrapper(environ, start_response):
- try:
- return app(environ, start_response)
- except Exception as e:
- return show_error(start_response, '500 Internal Server Error',
- uri=environ.get('PATH_INFO'), error=e)
-
- server = make_server(host, port, wrapper, ThreadedWSGIServer)
- print ' * Running on http://%s:%i/' % server.server_address
- server.serve_forever()
-
- run(address, port, handler, use_reloader=True, use_debugger=True)
+if __name__ == '__main__':
+ main()
« no previous file with comments | « no previous file | tests/conftest.py » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld