unit_test_http_server.py

#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
#
# ** header v3.0
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2018 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
# Copyright (C) 2019-2020 Timm Fitschen (t.fitschen@indiscale.com)
# Copyright (C) 2019-2020 IndiScale GmbH (info@indiscale.com)
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
# ** end header
#
"""unit_test_http_server

For running and logging the unit test suite of webcaosdb.
"""
import sys
import os
from datetime import datetime
from http.server import SimpleHTTPRequestHandler, HTTPServer

os.chdir(sys.argv[4])
counter = 0
class UnitTestsHandler(SimpleHTTPRequestHandler):
    """UnitTestsHandler

    Handles GET requests and return the source files of webcaosdb or the test
    suite and POST requests for the logger.
    """

    # do_POST is compliant with the BaseHTTPRequestHandler API
    # pylint: disable=invalid-name
    def do_POST(self):
        """do_POST

        Handle POST requests to the `log` and `done` resources.
        """
        path = self.path[1:]
        if path in ["done", "log"]:
            getattr(self, path)()
            #return HTTP status: 204 No Content
            self.send_response(204)
        else:
            #return HTTP status: 404 Not Found
            self.send_response(404)

    def send_response(self, code, message=None):
        """send_response

        Set Cache-Control, Pragma and Expires headers. Then call send_response
        of BaseHTTPRequestHandler.
        """
        super().send_response(code=code, message=message)
        if self.command == "GET":
            self.send_header("Cache-Control", "no-cache, no-store, must-revalidate")
            self.send_header("Pragma", "no-cache")
            self.send_header("Expires", "0")

    def done(self):
        """done

        Shut down the server with an exit code which depends on the success of
        the unit test suite - If the tests succeeded the server exists with 0,
        otherwise with 1.

        If the server has the ignore_done flag set this method immediately returns.
        """
        if self.server.ignore_done:
            return
        #pylint: disable=protected-access
        post_data = self.log()
        if "SUCCESS" in post_data:
            self.server._shutdown("[SUCCESS]", 0)
        else:
            self.server._shutdown("[ERROR]", 1)

    def log(self):
        """log

        Print the body of a request.
        """
        global counter
        counter += 1
        content_length = int(self.headers['Content-Length'])
        post_data = self.rfile.read(content_length).decode("utf-8")
        with open("qunit.log", "a") as logfile:
            logfile.write("[LOG {}] ".format(counter) + post_data + "\n")
        return post_data


class UnitTestHTTPServer(HTTPServer):
    """UnitTestHTTPServer

    A HTTP server which handles the request to the webcaosdb sources and the
    tests and servers as a logger when POST request are send to the `log`
    resource.

    ignore_done: If set to true the server does not terminate on receiving a "/Done".
    """

    def __init__(self, server_address, timeout, ignore_done):
        super(UnitTestHTTPServer, self).__init__(server_address,
                                                 UnitTestsHandler)
        self.timeout = timeout
        self.ignore_done = ignore_done
        self._keep_running = True
        self._exit_message = None
        self._exit_code = None

    def handle_timeout(self):
        """handle_timeout as defined in socketserver.BaseServer
        """
        self._shutdown("[TIMEOUT]", 2)

    def _shutdown(self, message, code):
        self._keep_running = False
        self._exit_message = "[{}] {}".format(datetime.now(), message)
        with open(".server_done", "w") as logfile:
            logfile.write("{}".format(code))
        self._exit_code = code

    def start(self):
        """start

        Start the server and handle request until the `done` resource is being
        called or a timeout occurs.
        """
        print(("starting UnitTestHTTPServer on {address} with {t}s "
               "timeout.").format(address=self.server_address, t=self.timeout))
        self._keep_running = True
        while self._keep_running:
            self.handle_request()

        print(self._exit_message)
        os._exit(self._exit_code)#pylint: disable=protected-access


UnitTestHTTPServer(server_address=('127.0.0.1', int(sys.argv[1])),
                   timeout=float(sys.argv[2]), ignore_done=(sys.argv[3] == "True")).start()