HTTP

Required files:

• test_plan.py
• http_basic.py
• http_custom_handler.py
• custom_http_request_handler.py
• test.txt

test_plan.py

#!/usr/bin/env python
"""
This example is to demonstrate HTTP communication test scenarios.
"""

import sys

from testplan import test_plan
from testplan.report.testing.styles import Style, StyleEnum

import http_basic
import http_custom_handler

OUTPUT_STYLE = Style(StyleEnum.ASSERTION_DETAIL, StyleEnum.ASSERTION_DETAIL)


# Hard-coding `pdf_path`, 'stdout_style' and 'pdf_style' so that the
# downloadable example gives meaningful and presentable output.
# NOTE: this programmatic arguments passing approach will cause Testplan
# to ignore any command line arguments related to that functionality.
@test_plan(
    name="HTTPConnections",
    pdf_path="report.pdf",
    stdout_style=OUTPUT_STYLE,
    pdf_style=OUTPUT_STYLE,
)
def main(plan):
    """
    Testplan decorated main function to add and execute 2 MultiTests.

    :return: Testplan result object.
    :rtype:  ``testplan.base.TestplanResult``
    """
    plan.add(http_basic.get_multitest("HTTPBasic"))
    plan.add(http_custom_handler.get_multitest("HTTPCustomHandler"))


if __name__ == "__main__":
    res = main()
    print("Exiting code: {}".format(res.exit_code))
    sys.exit(res.exit_code)

http_basic.py

"""Tests HTTP requests between a server and a client."""
import json

from testplan.common.utils.context import context

from testplan.testing.multitest import MultiTest, testsuite, testcase
from testplan.testing.multitest.driver.http import (
    HTTPServer,
    HTTPClient,
    HTTPResponse,
)


@testsuite
class HTTPTestsuite:
    """HTTP requests between a server and a client."""

    @testcase
    def request_and_response(self, env, result):
        """
        Client makes a request, server received and responds back.
        """
        # The HTTPClient sends a GET request to some section of the API. The
        # HTTPServer will respond with the next message in it's response queue
        # no matter the HTTP method (GET, POST etc.) or the section of the API
        # it has been sent.
        result.log("Client sends GET request")
        env.http_client.get(api="/random/text")

        # Need to do a receive otherwise it will ruin our next testcase
        received_request = env.http_server.receive()
        result.log(f"Server got GET request: {received_request}")

        # Create some JSON.
        json_content = {"this": ["is", "a", "json", "object"]}

        # We then prepare an HTTPResponse. Headers are added as a dictionary and
        # content as a list. For this example we just indicate that the content
        # type is JSON and dump the JSON as a string so it can be sent.
        prepared_response = HTTPResponse(
            headers={"Content-type": "application/json"},
            content=[json.dumps(json_content)],
        )

        # The HTTPServer then responds. Under the hood this adds the response to
        # the HTTPServer's response queue which will be immediately sent as the
        # HTTPClient has already sent a request.
        result.log("Server receives request and sends response")
        env.http_server.respond(prepared_response)

        # The HTTPClient then receives the HTTPServer's response.
        response = env.http_client.receive()

        # We are verifying the JSON sent back is the same as the one sent by the
        # HTTPServer.
        result.dict.match(
            response.json(), json_content, "JSON response from server"
        )

    @testcase
    def post_and_response(self, env, result):
        """
        Client makes a request, server received and responds back.
        """
        # Create some JSON.
        json_content = {"this": ["is", "another", "json", "object"]}

        # The HTTPClient sends a POST request with some data to some section of the API. The
        # HTTPServer will respond with the same message in it's response queue
        # no matter the HTTP method (GET, POST etc.) or the section of the API
        # it has been sent.
        result.log("Client sends POST request")
        env.http_client.post(
            api="/random/text",
            json=json_content,
            headers={"Content-Type": "application/json"},
        )

        # The HTTP Server receives the request
        received_request = env.http_server.receive()
        result.log(f"Server got POST request: {received_request}")

        # We are verifying the JSON sent back is the same as the one sent by the
        # HTTPServer.
        result.dict.match(
            received_request.json, json_content, "JSON sent to the server"
        )

        # We then prepare an HTTPResponse. Headers are added as a dictionary and
        # content as a list. For this example we just indicate that the content
        # type is JSON and dump the JSON as a string so it can be sent.
        prepared_response = HTTPResponse(
            headers={"Content-type": "application/json"},
            content=[json.dumps(json_content)],
        )

        # The HTTPServer then responds. Under the hood this adds the response to
        # the HTTPServer's response queue which will be immediately sent as the
        # HTTPClient has already sent a request.
        result.log("Server receives request and sends response")
        env.http_server.respond(prepared_response)

        # The HTTPClient then receives the HTTPServer's response.
        response = env.http_client.receive()

        # We are verifying the JSON sent back is the same as the one sent by the
        # HTTPServer.
        result.dict.match(
            response.json(), json_content, "JSON response from server"
        )


def get_multitest(name):
    """
    Creates and returns a new MultiTest instance to be added to the plan.
    The environment is a server and a client connecting using the context
    functionality that retrieves host/port of the server after is started.
    """
    test = MultiTest(
        name=name,
        suites=[HTTPTestsuite()],
        environment=[
            HTTPServer(name="http_server"),
            HTTPClient(
                name="http_client",
                host=context("http_server", "{{host}}"),
                port=context("http_server", "{{port}}"),
            ),
        ],
    )
    return test

http_custom_handler.py

"""Tests using a custom HTTP response handler."""
from testplan.common.utils.context import context

from testplan.testing.multitest import MultiTest, testsuite, testcase
from testplan.testing.multitest.driver.http import HTTPServer, HTTPClient

from custom_http_request_handler import CustomHTTPRequestHandler


@testsuite
class HTTPTestsuite:
    """Sending requests to an HTTPServer with a custom HTTP request handler."""

    @testcase
    def general_request(self, env, result):
        """
        Client makes a general request, server responds with text file contents.
        """
        # The HTTPClient sends a GET request to some section of the API.
        result.log("Client sends GET request")
        env.http_client.get(api="/random/text")

        # The HTTPServer will use the CustomHTTPRequestHandler to automatically
        # send the contents of the test.txt file back to every request. The
        # HTTPClient receives the HTTPServer's response.
        response = env.http_client.receive()

        # We are verifying the contents of the test.txt file is what the server
        # has sent back.
        text = open("test.txt").read()
        result.equal(
            response.text, text, "HTTPServer sends file contents in response."
        )

    @testcase
    def post_request(self, env, result):
        """
        Client makes a POST request, server responds with custom POST response.
        """
        result.log("Client sends POST request")
        env.http_client.post(api="/random/text")

        response = env.http_client.receive()

        result.equal(
            response.text,
            "POST response.",
            "HTTPServer sends custom POST response.",
        )


def get_multitest(name):
    """
    Creates and returns a new MultiTest instance to be added to the plan.
    The environment is a server and a client connecting using the context
    functionality that retrieves host/port of the server after is started.
    """
    # The HTTPServer can be passed handler_attributes in a dictionary. These
    # will be accessible in the custom HTTP request handler
    # (see custom_http_request_handler.py).
    attributes = {"text_file": "test.txt"}
    test = MultiTest(
        name=name,
        suites=[HTTPTestsuite()],
        environment=[
            HTTPServer(
                name="http_server",
                request_handler=CustomHTTPRequestHandler,
                handler_attributes=attributes,
            ),
            HTTPClient(
                name="http_client",
                host=context("http_server", "{{host}}"),
                port=context("http_server", "{{port}}"),
            ),
        ],
    )
    return test

custom_http_request_handler.py

"""Custom HTTP request handler."""
from testplan.testing.multitest.driver.http import (
    HTTPRequestHandler,
    HTTPResponse,
)


class CustomHTTPRequestHandler(HTTPRequestHandler):
    """Define a custom request handler."""

    def get_response(self, request):
        """
        Override the get_response method to determine how the server will
        respond to all requests. You must return an HTTPResponse object as the
        _parse_request method expects this.
        """
        text_file = self.server.handler_attributes["text_file"]
        with open(text_file) as input:
            text = input.read()
        response = HTTPResponse(content=[text])
        self.server.log_callback(
            "Creating custom response from {}".format(text_file)
        )
        return response

    def do_POST(self):
        """
        Override individual request methods (e.g. do_POST) to determine how the
        server will respond to individual requests. You will have to create the
        response, then call _send_header and _send_content.
        """
        response = HTTPResponse(content=["POST response."])
        self._send_header(response.status_code, response.headers)
        self._send_content(response.content)

test.txt

This is a test file.
It has multiple lines.