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

Driver Connection

Required files:

• test_plan.py

• custom_connection.py

• writer.py

• reader.py

• drivers.py

test_plan.py

#!/usr/bin/env python
"""
This example is to showcase the driver connection functionality.
"""

import os
import re
import sys

from drivers import (
    CustomTCPClient,
    WritingDriver,
    ReadingDriver,
    UnconnectedDriver,
)

from testplan import test_plan
from testplan.common.utils.context import context
from testplan.testing.multitest import MultiTest
from testplan.testing.multitest.driver.tcp import TCPServer


def environment():
    """
    MultiTest environment that will be made available within the testcases.
    """
    server = TCPServer(name="server")
    client = CustomTCPClient(
        name="client",
        host=context("server", "{{host}}"),
        port=context("server", "{{port}}"),
    )

    writer_name = "WritingDriver"
    writer_regexps = [
        re.compile(r".*Writing to file: (?P<file_path>.*)"),
    ]
    writer = WritingDriver(
        name=writer_name,
        pre_args=[sys.executable],
        binary=os.path.join(
            os.path.dirname(os.path.abspath(__file__)), "writer.py"
        ),
        log_regexps=writer_regexps,
    )
    reader_regexps = [
        re.compile(r".*Reading from file: (?P<file_path>.*)"),
    ]
    reader = ReadingDriver(
        name="ReadingDriver",
        pre_args=[sys.executable],
        binary=os.path.join(
            os.path.dirname(os.path.abspath(__file__)), "reader.py"
        ),
        args=[context(writer_name, "{{file_path}}")],
        log_regexps=reader_regexps,
    )

    unconnected = UnconnectedDriver("unconnected_driver")

    return [server, client, writer, reader, unconnected]


@test_plan(name="DriverConnectionExample", driver_info=True)
def main(plan):
    """
    Testplan decorated main function to add and execute MultiTests.

    :return: Testplan result object.
    :rtype:  ``testplan.base.TestplanResult``
    """
    test = MultiTest(
        name="DriverConnectionExample",
        suites=[],
        environment=environment(),
    )
    plan.add(test)


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

custom_connection.py

"""
Example for creating new custom connections
"""

from typing import List
from testplan.testing.multitest.driver.connection import (
    BaseConnectionInfo,
    BaseDriverConnectionGroup,
    BaseConnectionExtractor,
    Direction,
)


class CustomConnectionInfo(BaseConnectionInfo):
    @property
    def connection_rep(self):
        # this is the value by which drivers will be matched
        # by default, this is f"self.protocol}//{self.identifier}
        return f"custom://{self.identifier}"

    def promote_to_connection(self):
        return CustomDriverConnectionGroup.from_connection_info(self)


class CustomDriverConnectionGroup(BaseDriverConnectionGroup):
    @classmethod
    def from_connection_info(
        cls, driver_connection_info: CustomConnectionInfo
    ):
        # Add any custom logic here
        # For example, to add a dummy driver
        conn = super(CustomDriverConnectionGroup, cls).from_connection_info(
            driver_connection_info
        )
        conn.in_drivers["Dummy Driver"].add("")
        return conn

    def add_driver_if_in_connection(
        self, driver_name: str, driver_connection_info: CustomConnectionInfo
    ):
        # Define logic on how to match drivers here
        # Default behavior for predefined connections is to match based on the connection_rep
        if self.connection_rep == driver_connection_info.connection_rep:
            # Append the identifier
            # For port based connection, ports are the identifier
            # For file based connections, read/write are the identifier
            self.out_drivers[driver_name].add("Read")
            return True
        return False


class CustomConnectionExtractor(BaseConnectionExtractor):
    def extract_connection(self, driver) -> List[CustomConnectionInfo]:
        return [
            CustomConnectionInfo(
                protocol="custom",
                identifier=driver.name,
                direction=Direction.CONNECTING,
            )
        ]

writer.py

"""
Sample python script that writes to a file
"""

import os
import sys
import logging

logging.basicConfig(stream=sys.stdout, format="%(message)s")


class Writer:
    def __init__(self):
        self._logger = logging.getLogger()
        self._logger.setLevel(logging.INFO)
        self.file = os.path.join(os.getcwd(), "test.txt")

    def loop(self):
        with open(self.file, "w"):
            self._logger.info("Writing to file: %s", self.file)
            while True:
                continue


if __name__ == "__main__":
    writer = Writer()
    writer.loop()

reader.py

"""
Sample python script that reads from a file
"""

import sys
import logging

logging.basicConfig(stream=sys.stdout, format="%(message)s")


class Reader:
    def __init__(self, file):
        self._logger = logging.getLogger()
        self._logger.setLevel(logging.INFO)
        self.file = file

    def loop(self):
        with open(self.file, "r"):
            self._logger.info("Reading from file: %s", self.file)
            while True:
                continue


if __name__ == "__main__":
    sys.stderr.flush()
    _, file_path = sys.argv
    reader = Reader(file_path)
    reader.loop()

drivers.py

from testplan.testing.multitest.driver.base import Driver
from testplan.testing.multitest.driver.app import App
from testplan.testing.multitest.driver.tcp import TCPClient
from testplan.testing.multitest.driver.connection import (
    Direction,
    Protocol,
    ConnectionExtractor,
)

from custom_connection import CustomConnectionExtractor


class CustomTCPClient(TCPClient):
    # override EXTRACTORS
    EXTRACTORS = [
        ConnectionExtractor(Protocol.TCP, Direction.CONNECTING),
        CustomConnectionExtractor(),
    ]


class WritingDriver(App):
    """
    Inherits the generic ``testplan.testing.multitest.driver.app.App`` driver
    and expose file path read from log extracts.
    """

    def __init__(self, **options):
        super(WritingDriver, self).__init__(**options)
        self.file_path = None

    def post_start(self):
        """
        Store file_path to be made available in its context
        so that reading driver can connect to it.
        """
        super(WritingDriver, self).post_start()
        self.file_path = self.extracts["file_path"]


class ReadingDriver(App):
    """
    Inherits the generic ``testplan.testing.multitest.driver.app.App`` driver
    """

    pass


class UnconnectedDriver(Driver):
    pass