Driver

Simultaneous Start-up

Required files:

• test_plan.py
• suites.py

test_plan.py

#!/usr/bin/env python
"""
This example is to demonstrate driver scheduling following dependencies.
"""

import sys

from testplan import test_plan
from testplan.common.utils.context import context
from testplan.report.testing.styles import Style, StyleEnum
from testplan.testing.multitest import MultiTest
from testplan.testing.multitest.driver.tcp import TCPServer, TCPClient

from suites import MultiTCPSuite

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="TCPConnections",
    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``
    """
    server_1 = TCPServer(name="server_1")
    client_1 = TCPClient(
        name="client_1",
        host=context("server_1", "{{host}}"),
        port=context("server_1", "{{port}}"),
    )
    server_2 = TCPServer(name="server_2")
    client_2 = TCPClient(
        name="client_2",
        host=context("server_2", "{{host}}"),
        port=context("server_2", "{{port}}"),
    )
    client_3 = TCPClient(
        name="client_3",
        host=context("server_2", "{{host}}"),
        port=context("server_2", "{{port}}"),
    )

    # If driver A is a dependency for driver B to start, we put driver A in the key
    # of dependencies dictionary and driver B as its corresponding value, so
    # visually driver A appears before driver B.

    # Here server_1 and server_2 will be started simutaneously to reduce the
    # overall test running time while not violating the dependencies.

    plan.add(
        MultiTest(
            name="MultiTCPDrivers",
            suites=[MultiTCPSuite()],
            environment=[
                server_1,
                server_2,
                client_1,
                client_2,
                client_3,
            ],
            dependencies={
                server_1: client_1,
                server_2: (client_2, client_3),
            },
        )
    )


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

suites.py

"""Tests TCP communication among multiple servers and multiple clients."""

from testplan.testing.multitest import testsuite, testcase


@testsuite
class MultiTCPSuite:
    """TCP tests for multiple servers and multiple clients."""

    @testcase
    def test_send_and_receive_msg(self, env, result):
        """
        Client sends a message, server received and responds back.
        """
        env.client_1.send_text("zhengshan xiaozhong")
        env.server_1.accept_connection()
        result.equal("zhengshan xiaozhong", env.server_1.receive_text())
        env.server_1.send_text("good black tea")
        result.equal("good black tea", env.client_1.receive_text())

        env.client_2.send_text("whisky")
        conn_2 = env.server_2.accept_connection()
        env.client_3.send_text("soda water")
        conn_3 = env.server_2.accept_connection()
        result.equal("whisky", env.server_2.receive_text(conn_idx=conn_2))
        result.equal("soda water", env.server_2.receive_text(conn_idx=conn_3))
        env.server_2.send_text("good highball", conn_idx=conn_2)
        env.server_2.send_text("good highball", conn_idx=conn_3)
        result.equal("good highball", env.client_2.receive_text())
        result.equal("good highball", env.client_3.receive_text())

Environment Builder

Required files:

• test_plan.py
• suites.py
• env_builder.py

test_plan.py

#!/usr/bin/env python
"""
This example demonstrates usage of callable object to construct environment,
intial_context and dependencies for a multitest at runtime.
"""

import sys

from testplan import test_plan
from testplan.testing.multitest import MultiTest
from env_builder import EnvBuilder
from suites import TestOneClient, TestTwoClients


@test_plan(name="EnvBuilderExample")
def main(plan):

    env_builder1 = EnvBuilder("One Client", ["client1", "server1"])
    env_builder2 = EnvBuilder("Two Clients", ["client1", "client2", "server1"])
    plan.add(
        MultiTest(
            name="TestOneClient",
            suites=[TestOneClient()],
            environment=env_builder1.build_env,
            dependencies=env_builder1.build_deps,
            initial_context=env_builder1.init_ctx,
        )
    )

    plan.add(
        MultiTest(
            name="TestTwoClients",
            suites=[TestTwoClients()],
            environment=env_builder2.build_env,
            dependencies=env_builder2.build_deps,
            initial_context=env_builder2.init_ctx,
        )
    )


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

suites.py

from testplan.testing.multitest import testsuite, testcase


@testsuite
class TestOneClient:
    def setup(self, env, result):
        result.log(f"Testing with [{env.env_name}] env")
        env.server1.accept_connection()

    @testcase
    def test_send_and_receive_msg(self, env, result):
        env.client1.send_text("hi server")
        result.equal("hi server", env.server1.receive_text())

        env.server1.send_text("hi client")
        result.equal("hi client", env.client1.receive_text())


@testsuite
class TestTwoClients:
    def setup(self, env, result):
        result.log(f"Testing with [{env.env_name}] env")

        env.client1.connect()
        self.conn1 = env.server1.accept_connection()

        env.client2.connect()
        self.conn2 = env.server1.accept_connection()

    @testcase
    def test_send_and_receive_msg(self, env, result):
        env.client1.send_text("hi server from client1")
        env.client2.send_text("hi server from client2")

        result.equal(
            "hi server from client1",
            env.server1.receive_text(conn_idx=self.conn1),
        )
        result.equal(
            "hi server from client2",
            env.server1.receive_text(conn_idx=self.conn2),
        )

        env.server1.send_text("hi client1", conn_idx=self.conn1)
        env.server1.send_text("hi client2", conn_idx=self.conn2)

        result.equal("hi client1", env.client1.receive_text())
        result.equal("hi client2", env.client2.receive_text())

env_builder.py

"""
This demonstrates one possible way of implementing of EnvBuilder class.
"""

from testplan.common.utils.context import context
from testplan.testing.multitest.driver.tcp import TCPClient, TCPServer


class EnvBuilder:
    def __init__(self, env_name: str, drivers: list):
        """
        :param env_name: name of this env builder
        :param drivers: list of drivers to be created
        """
        self.env_name = env_name
        self.drivers = drivers
        self.client_auto_connect = False if len(self.drivers) == 3 else True
        self._client1 = None
        self._client2 = None
        self._server1 = None

    def build_env(self):
        return [getattr(self, driver_name) for driver_name in self.drivers]

    def init_ctx(self):
        return {"env_name": self.env_name}

    def build_deps(self):
        if len(self.drivers) == 2:
            return {self.server1: self.client1}
        elif len(self.drivers) == 3:
            return {self.server1: (self.client1, self.client2)}

    @property
    def client1(self):
        if not self._client1:
            self._client1 = TCPClient(
                name="client1",
                host=context("server1", "{{host}}"),
                port=context("server1", "{{port}}"),
                connect_at_start=self.client_auto_connect,
            )
        return self._client1

    @property
    def client2(self):
        if not self._client2:
            self._client2 = TCPClient(
                name="client2",
                host=context("server1", "{{host}}"),
                port=context("server1", "{{port}}"),
                connect_at_start=self.client_auto_connect,
            )
        return self._client2

    @property
    def server1(self):
        if not self._server1:
            self._server1 = TCPServer(name="server1")
        return self._server1