Database

Sqlite3

Required files:

• test_plan.py

test_plan.py

#!/usr/bin/env python
"""
Demostrates Sqlite3 driver usage from within the testcases.
"""

import sys
import os
import tempfile

from testplan.testing.multitest import MultiTest
from testplan.testing.multitest.driver.sqlite import Sqlite3

from testplan import test_plan
from testplan.testing.multitest import testsuite, testcase
from testplan.report.testing.styles import Style, StyleEnum


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


@testsuite
class DBQueries:
    """Suite that contains testcases that perform db queries."""

    def setup(self, env, result):
        """
        Setup method that will be executed before all testcases. It is
        used to create the tables that the testcases require.
        """
        # Create a table called 'users'.
        with env.db.commit_at_exit():
            env.db.execute("""DROP TABLE IF EXISTS users""")
            env.db.execute(
                """CREATE TABLE users(id INTEGER PRIMARY KEY, name TEXT,
                           phone TEXT, email TEXT unique, password TEXT)"""
            )

        # Fill the table with data.
        with env.db.commit_at_exit():
            items = [
                {
                    "name": "John",
                    "phone": "123456",
                    "email": "john@email",
                    "password": "abc123",
                },
                {
                    "name": "Mary",
                    "phone": "234567",
                    "email": "mary@email",
                    "password": "qwe234",
                },
            ]
            env.db.executemany(
                """INSERT INTO users(name, phone, email, password)
                  VALUES(:name,:phone, :email, :password)""",
                items,
            )
        result.log(
            'Database file of driver "{}" created at "{}"'.format(
                env.db.cfg.name, env.db.db_path
            )
        )

    @testcase
    def sample_match_query(self, env, result):
        """Table match example after fetching two table columns."""
        table = env.db.fetch_table("users", columns=["name", "email"])
        result.table.log(table, description="Two columns.")

        expected_table = [
            ["name", "email"],
            ["John", "john@email"],
            ["Mary", "mary@email"],
        ]

        # Match the table fetched against the expected.
        result.table.match(actual=table, expected=expected_table)

    @testcase
    def sample_column_query(self, env, result):
        """Table column content assertion after fetching the whole table."""
        table = env.db.fetch_table("users")
        result.table.log(table, description="Whole table.")

        # Checks that the column 'name' of the table contain one of the
        # expected values.
        result.table.column_contain(
            values=["John", "Mary"],
            table=env.db.fetch_table("users"),
            column="name",
        )

    def teardown(self, env):
        env.db.execute("DROP TABLE IF EXISTS users")


# 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="Sqlite3Example",
    stdout_style=OUTPUT_STYLE,
    pdf_style=OUTPUT_STYLE,
    pdf_path="report.pdf",
)
def main(plan):
    """
    Testplan decorated main function to add and execute MultiTests.

    :return: Testplan result object.
    :rtype:  ``testplan.base.TestplanResult``
    """
    db_file_name = "mydb"
    my_db_path = os.path.join(tempfile.gettempdir(), db_file_name)
    plan.add(
        MultiTest(
            name="Sqlite3Test",
            suites=[DBQueries()],
            environment=[Sqlite3(name="db", db_path=my_db_path)],
        )
    )


if __name__ == "__main__":
    sys.exit(not main())

Zookeeper

Required files:

• test_plan.py
• zoo_template.cfg

test_plan.py

#!/usr/bin/env python
"""
Demostrates Zookeeper driver usage from within the testcases.
"""

import os
import sys

try:
    from kazoo.client import KazooClient
except ImportError:
    print("Cannot import kazoo!")
    exit()

from testplan import test_plan
from testplan.testing.multitest import MultiTest
from testplan.testing.multitest.driver.zookeeper import (
    ZookeeperStandalone,
    ZK_SERVER,
)
from testplan.testing.multitest import testsuite, testcase
from testplan.report.testing.styles import Style, StyleEnum


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


@testsuite
class ZookeeperTest:
    """Suite that contains testcases that perform zookeeper operation."""

    def setup(self, env, result):
        """
        Setup method that will be executed before all testcases. It is
        used to ensure the path that the testcases require.
        """
        zk = KazooClient(hosts="127.0.0.1:{}".format(env.zk.port))
        zk.start()
        zk.ensure_path("/testplan")

    @testcase
    def get_node(self, env, result):
        """Get and log node information example."""
        zk = KazooClient(hosts="127.0.0.1:{}".format(env.zk.port))
        zk.start()
        node = zk.get("/")
        result.log(node)

    @testcase
    def create_node(self, env, result):
        """Create node example."""
        zk = KazooClient(hosts="127.0.0.1:{}".format(env.zk.port))
        zk.start()
        test_value = b"testplan"
        zk.create("/testplan/test", test_value)
        data, _ = zk.get("/testplan/test")
        result.equal(data, test_value)


# 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="ZookeeperExample",
    stdout_style=OUTPUT_STYLE,
    pdf_style=OUTPUT_STYLE,
    pdf_path="report.pdf",
)
def main(plan):
    """
    Testplan decorated main function to add and execute MultiTests.

    :return: Testplan result object.
    :rtype:  ``testplan.base.TestplanResult``
    """
    cfg_template = os.path.join(
        os.path.dirname(os.path.abspath(__file__)), "zoo_template.cfg"
    )

    plan.add(
        MultiTest(
            name="ZookeeperTest",
            suites=[ZookeeperTest()],
            environment=[
                ZookeeperStandalone(name="zk", cfg_template=cfg_template)
            ],
        )
    )


if __name__ == "__main__":
    if os.path.exists(ZK_SERVER):
        sys.exit(not main())
    else:
        print("Zookeeper doesn't exist in this server.")

zoo_template.cfg

# limits the number of active connections from a host,
# specified by IP address, to a single ZooKeeper server.
maxClientCnxns=100

# The basic time unit in milliseconds used by ZooKeeper.
# It is used to do heartbeats and the minimum session timeout will be twice the tickTime.
tickTime=2000

# Timeouts ZooKeeper uses to limit the length of time the ZooKeeper
# servers in quorum have to connect to a leader.
initLimit=10

# Limits how far out of date a server can be from a leader.
syncLimit=5

# Enable admin server.
admin.enableServer=false

# The localtion to store the in-memory database snapshots and, unless specified otherwise,
# the transaction log of updates to the database.
dataDir={{zkdata_path}}

# The port to listen for client connections.
clientPort={{port}}