### Run Testplan Examples (Linux/macOS)

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/getting_started.rst

Demonstrates how to navigate to the examples directory and run a basic Testplan assertion example. It also shows how to generate a PDF report.

```bash
# See all the examples categories.
cd examples
ls

# Run an example demonstrating testplan assertions.
cd Assertions/Basic
./test_plan_basic.py
```

```bash
# Create a pdf report and open in automatically.
./test_plan.py --pdf report.pdf -b
```

--------------------------------

### Run Testplan Examples (Windows)

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/getting_started.rst

Shows how to execute Testplan examples on a Windows system, including listing categories, running assertion examples, and generating PDF reports.

```text
# See all the examples categories.
cd examples
dir

# Run an example demonstrating testplan assertions.
cd Assertions\Basic
python test_plan_basic.py
```

```text
# Create a pdf report and open in automatically.
python test_plan.py --pdf report.pdf -b
```

--------------------------------

### Setup Testplan from Source

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/getting_started.rst

Clones the Testplan repository, installs it in editable mode with development requirements, and builds the interactive UI. Requires Python 3 and Node.js.

```text
git clone https://github.com/morganstanley/testplan.git
cd testplan

# install testplan in editable mode & all dev requirements
pip install -e .

# build the interactive UI (if you do not like it is opening a browserwindow remove the `-o`)
doit build_ui -o
```

--------------------------------

### Install Testplan with Optional Extras

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/getting_started.rst

Installs Testplan with specific optional features using pip extras. Examples include installing for interactive mode or for plotly support. Installing with '[all]' installs all available features.

```bash
# To use interactive mode feature.
python3 -m pip install testplan[interactive]
```

```bash
# To use plotly for plotting figures.
python3 -m pip install testplan[plotly]
```

```bash
# Install testplan with all the features.
python3 -m pip install testplan[all]
```

--------------------------------

### Manually Starting App Drivers (Python)

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/App.rst

Illustrates how to manually start and manage application drivers for testing. This example uses 'test_plan.py'.

```python
from pywinauto.application import Application

# Start the application manually
app = Application().start("calc.exe")

# Interact with the calculator application
app.Calculator.type_keys("10 + 20 =")

# Assert the result
assert app.Calculator.Edit.window_text() == "30"

# Close the application
app.Calculator.close()
```

--------------------------------

### Quick Start BDD Example with Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/BDD.rst

Demonstrates a basic Behavior-Driven Development (BDD) setup using Python. It includes a test plan, a feature file, and step definitions to get started quickly.

```python
from testplan import test_plan

@test_plan
def test_main():
    # This is a placeholder for the actual test execution logic.
    # In a real scenario, this would involve running BDD features.
    print("Running BDD Quick Start example.")
    pass

```

```gherkin
Feature: First Feature

  Scenario: Simple Greeting
    Given a user named "Alice"
    When the user is greeted
    Then the greeting should be "Hello, Alice!"
```

```python
from pytest import fixture

@fixture
def user_name():
    return ""

@fixture
def greeting():
    return ""

@given("a user named \"{name}\"")
def step_impl(context, name, user_name):
    user_name.value = name

@when("the user is greeted")
def step_impl(context, user_name, greeting):
    greeting.value = f"Hello, {user_name.value}!"

@then("the greeting should be \"{expected}\"")
def step_impl(context, greeting, expected):
    assert greeting.value == expected

```

--------------------------------

### FXConverter Custom Application Setup (Python)

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/App.rst

Sets up and runs tests for a custom FXConverter application. This example involves multiple Python files and a configuration file.

```python
# test_plan.py
from pywinauto.application import Application
from suites import FXConverterSuite

app = Application().start("converter.exe")

# Run test suite
FXConverterSuite(app).run_tests()

app.destroy()
```

```python
# suites.py
from pywinauto.test_support.test_case import TestCase

class FXConverterSuite(TestCase):
    def run_tests(self):
        self.test_conversion()

    def test_conversion(self):
        self.converter.set_amount("100")
        self.converter.set_from_currency("USD")
        self.converter.set_to_currency("EUR")
        self.converter.convert()
        assert self.converter.get_result() == "92.50"
```

```python
# converter.py
from pywinauto.application import Application

class Converter:
    def __init__(self, app):
        self.app = app
        self.window = app.FXConverter

    def set_amount(self, amount):
        self.window.Amount.set_text(amount)

    def set_from_currency(self, currency):
        self.window.FromCurrency.select(currency)

    def set_to_currency(self, currency):
        self.window.ToCurrency.select(currency)

    def convert(self):
        self.window.ConvertButton.click()

    def get_result(self):
        return self.window.Result.window_text()
```

```python
# driver.py
from pywinauto.application import Application
from converter import Converter

app = Application().start("converter.exe")
converter = Converter(app)

# Example usage:
converter.set_amount("1000")
converter.set_from_currency("USD")
converter.set_to_currency("GBP")
converter.convert()
print(f"Conversion result: {converter.get_result()}")

app.destroy()
```

--------------------------------

### Install Testplan from PyPI

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/getting_started.rst

Installs the Testplan package from PyPI using pip. This is the standard method for installing Testplan and its core functionalities.

```bash
python3 -m pip install testplan
```

--------------------------------

### Install Testplan from GitHub Release Archive

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/getting_started.rst

Installs a specific version of Testplan directly from a wheel file hosted on GitHub releases. This is useful for installing pre-compiled packages or specific versions not yet on PyPI.

```bash
python3 -m pip install https://github.com/morganstanley/testplan/releases/download/25.3.0/testplan-25.3.0-py3-none-any.whl
```

--------------------------------

### Initial Context Setup in Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Multitest.rst

Demonstrates how to set up the initial context for Multitest. This involves defining the test plan structure and any necessary configurations.

```python
import unittest

class MyTests(unittest.TestCase):
    def test_example(self):
        self.assertTrue(True)

if __name__ == '__main__':
    unittest.main()
```

--------------------------------

### BDD Arguments Example with Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/BDD.rst

Illustrates how to pass arguments within feature files and step definitions in a BDD setup using Python. This allows for more dynamic and parameterized tests.

```python
from testplan import test_plan

@test_plan
def test_main():
    # Placeholder for test execution.
    print("Running BDD Arguments example.")
    pass

```

```gherkin
Feature: Arguments Feature

  Scenario Outline: Arithmetic operations
    Given the first number is <num1>
    And the second number is <num2>
    When the numbers are added
    Then the result should be <result>

    Examples:
      | num1 | num2 | result |
      | 5    | 3    | 8      |
      | 10   | -2   | 8      |
      | 0    | 0    | 0      |
```

```python
from pytest import fixture

@fixture
def num1():
    return 0

@fixture
def num2():
    return 0

@fixture
def result():
    return 0

@given("the first number is {num1}")
def step_impl(context, num1, first_num):
    first_num.value = int(num1)

@given("the second number is {num2}")
def step_impl(context, num2, second_num):
    second_num.value = int(num2)

@when("the numbers are added")
def step_impl(context, first_num, second_num, result):
    result.value = first_num.value + second_num.value

@then("the result should be {expected_result}")
def step_impl(context, result, expected_result):
    assert result.value == int(expected_result)

```

--------------------------------

### Start Testplan Interactive Server (Python)

Source: https://github.com/morganstanley/testplan/blob/main/examples/Interactive/Basic/test_plan_notebook.ipynb

Starts the Testplan interactive server, which listens for and handles interactive requests. This is a necessary step after initializing the Testplan instance to enable interactive operations.

```python
# Interactive mode serving interactive requests.
plan.run()
```

--------------------------------

### BDD Quick Start Feature File

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/bdd.rst

An example Gherkin feature file used in the Testplan BDD quick start guide. It defines scenarios and steps for testing basic arithmetic operations.

```gherkin
Feature: Example Gherkin Testsuite

  Scenario: Example Gherkin Test
    Given we have two number: 1 and 1
    When we sum the numbers
    Then the result is: 2

  Scenario: Example Gherkin Testplan
    Given we have two number: 2 and 2
    When we sum the numbers
    Then the result is: 4
```

--------------------------------

### BDD Import Steps Example with Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/BDD.rst

Demonstrates how to import and reuse step definitions across different feature files in a BDD setup using Python. This promotes code reusability and maintainability.

```python
from testplan import test_plan

@test_plan
def test_main():
    # Placeholder for test execution.
    print("Running BDD Import Steps example.")
    pass

```

```python
# common.py
from pytest import fixture

@fixture
def shared_data():
    return ""

@given("common setup")
def step_impl(context, shared_data):
    shared_data.value = "common data"

```

```gherkin
# one.feature
Feature: Feature One

  Scenario: Test with common setup
    Given common setup
    Then the shared data should be "common data"
```

```python
# one.steps.py
from pytest import fixture

@then("the shared data should be \"{expected}\"")
def step_impl(context, expected, shared_data):
    assert shared_data.value == expected

```

```gherkin
# two.feature
Feature: Feature Two

  Scenario: Another test with common setup
    Given common setup
    Then the shared data should be "common data"
```

```python
# two.steps.py
from pytest import fixture

@then("the shared data should be \"{expected}\"")
def step_impl(context, expected, shared_data):
    assert shared_data.value == expected

```

--------------------------------

### Start Test Resources in Interactive Mode

Source: https://github.com/morganstanley/testplan/blob/main/examples/Interactive/Environments/test_plan_notebook.ipynb

Initiates the test environment resources, such as servers and clients, for a specified test UID. This prepares the environment for test execution.

```python
# Start the test envirorment resources (Server & client).
plan.i.start_test_resources(test_uid="Test1")
```

--------------------------------

### Environment Management (Programmatic)

Source: https://github.com/morganstanley/testplan/blob/main/examples/Interactive/Environments/test_plan_notebook.ipynb

Details on adding, starting, operating, and stopping environments using Testplan's Python API.

```APIDOC
## Environment Management (Programmatic)

### Description
Allows for the programmatic creation and management of independent environments, including their resources.

### Method
N/A (Code Snippet)

### Endpoint
N/A

### Parameters
N/A

### Request Example
```python
from testplan.environment import LocalEnvironment
from testplan.testing.multitest.driver.tcp import TCPServer, TCPClient
from testplan.common.utils.context import context

# Assuming 'plan' is an initialized Testplan instance
plan.add_environment(
    LocalEnvironment(
        "my_env1",
        [
            TCPServer(name="server"),
            TCPClient(
                name="client",
                host=context("server", "{{host}}"),
                port=context("server", "{{port}}"),
            ),
        ],
    )
)

env1 = plan.i.get_environment("my_env1")
env1.start()
env1.server.accept_connection()
client_send_result = env1.client.send_text("Hello server!")
server_receive_result = env1.server.receive_text()
print(f"Client sends msg of length in bytes: {client_send_result}")
print(f"Server receives: {server_receive_result}")
env1.stop()
```

### Response
#### Success Response (200)
N/A (Code Snippet Output)

#### Response Example
```
Client sends msg of length in bytes: <bytes_sent>
Server receives: Hello server!
```
```

--------------------------------

### Basic App Driver Usage (Python)

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/App.rst

Demonstrates the fundamental usage of the App Driver for test automation. This example requires the 'test_plan.py' file.

```python
from pywinauto.application import Application

app = Application().start("notepad.exe")

# Connect to an existing window
# app = Application().connect(title_re="Untitled - Notepad", timeout=10)

# Interact with the application
# app.UntitledNotepad.menu_select("Help->About Notepad")
# app.AboutNotepad.OK.click()
# app.UntitledNotepad.edit.type_keys("Hello, world!", with_spaces=True)
# app.UntitledNotepad.menu_select("File->Exit")
# app.ConfirmFileClose.DontSave.click()
```

--------------------------------

### System Integration Testing with TCP in Python

Source: https://github.com/morganstanley/testplan/blob/main/README.rst

Illustrates system integration testing using Testplan's TCP drivers. This example sets up a TCPServer and TCPClient to test communication between them, demonstrating environment setup, context usage for driver configuration, and message exchange assertions.

```python
import sys

from testplan import test_plan
from testplan.testing.multitest import MultiTest, testsuite, testcase
from testplan.testing.multitest.driver.tcp import TCPServer, TCPClient
from testplan.common.utils.context import context


@testsuite
class TCPTestsuite(object):
    """Testsuite for server client connection testcases."""

    def setup(self, env):
        env.server.accept_connection()

    @testcase
    def send_and_receive_msg(self, env, result):
        """Basic send and receive hello message testcase."""
        msg = env.client.cfg.name
        result.log('Client is sending his name: {}'.format(msg))
        bytes_sent = env.client.send_text(msg)

        received = env.server.receive_text(size=bytes_sent)
        result.equal(received, msg, 'Server received client name')

        response = 'Hello {}'.format(received)
        result.log('Server is responding: {}'.format(response))
        bytes_sent = env.server.send_text(response)

        received = env.client.receive_text(size=bytes_sent)
        result.equal(received, response, 'Client received response')


@test_plan(name='TCPConnections')
def main(plan):
    test = MultiTest(name='TCPConnectionsTest',
                     suites=[TCPTestsuite()],
                     environment=[
                         TCPServer(name='server'),
                         TCPClient(name='client',
                                   host=context('server', '{{host}}'),
                                   port=context('server', '{{port}}'))])
    plan.add(test)


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

--------------------------------

### Basic Testplan Interactive API Connection (Python)

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Interactive.rst

Demonstrates a basic connection to Testplan's interactive API using Python. This example requires several supporting files for test definitions and setup. It showcases how to initiate and manage tests in an interactive environment.

```Python
from testplan import Testplan


def main():
    with Testplan(name='MyTestplan') as testplan:
        testplan.interactive_api.start()
        print("Testplan interactive API started. Connect via HTTP.")
        # Keep the script running to keep the API alive
        import time
        while True:
            time.sleep(1)

if __name__ == '__main__':
    main()
```

```Python
import unittest

class MyTests(unittest.TestCase):
    def test_example(self):
        self.assertEqual(1, 1)

    def test_another(self):
        self.assertTrue(True)
```

```Python
from testplan.testing.multiprocess.test_suite import MultiProcessTestSuite

class MyTestSuite(MultiProcessTestSuite):
    def test_suite_method(self):
        self.assertEqual(2, 2)
```

```Python
from testplan.testing.base import TestSuite

class BasicTestSuite(TestSuite):
    def test_basic_method(self):
        self.assertIn('a', 'abc')
```

```Python
from testplan.testing.base import TestSuite

class TCPTestSuite(TestSuite):
    def test_tcp_connection(self):
        # Placeholder for TCP test logic
        self.assertTrue(True)
```

```Python
from testplan.testing.base import TestSuite

class DependencyTestSuite(TestSuite):
    def test_dependency_handling(self):
        # Placeholder for dependency test logic
        self.assertFalse(False)
```

--------------------------------

### Operate Local TCP Environment

Source: https://github.com/morganstanley/testplan/blob/main/examples/Interactive/Environments/test_plan_notebook.ipynb

Starts a local environment, accepts a connection on the TCP server, sends a message from the client, receives the message on the server, and then stops the environment. This demonstrates basic TCP communication within a Testplan environment.

```python
# Operate my_env1
env1 = plan.i.get_environment("my_env1")
env1.start()
env1.server.accept_connection()
print(
    "Client sends msg of length in bytes: {}".format(
        env1.client.send_text("Hello server!")
    )
)
print("Server receives: {}".format(env1.server.receive_text()))
env1.stop()
```

--------------------------------

### BDD Background Example with Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/BDD.rst

Demonstrates the use of the 'Background' keyword in BDD feature files to define steps that are executed before each scenario. This is useful for setting up common preconditions.

```python
from testplan import test_plan

@test_plan
def test_main():
    # Placeholder for test execution.
    print("Running BDD Background example.")
    pass

```

```gherkin
Feature: Background Feature

  Background:
    Given a user is logged in
    And the user has a profile

  Scenario: View profile
    When the user views their profile
    Then the profile page should be displayed

  Scenario: Edit profile
    When the user edits their profile
    Then the profile edit form should be displayed
```

```python
from pytest import fixture

@fixture
def logged_in_user():
    return False

@fixture
def has_profile():
    return False

@given("a user is logged in")
def step_impl(context, logged_in_user):
    logged_in_user.value = True

@given("the user has a profile")
def step_impl(context, logged_in_user, has_profile):
    if logged_in_user.value:
        has_profile.value = True

@when("the user views their profile")
def step_impl(context, has_profile):
    assert has_profile.value, "User must have a profile to view it."

@then("the profile page should be displayed")
def step_impl(context):
    print("Profile page displayed.")
    pass

@when("the user edits their profile")
def step_impl(context, has_profile):
    assert has_profile.value, "User must have a profile to edit it."

@then("the profile edit form should be displayed")
def step_impl(context):
    print("Profile edit form displayed.")
    pass

```

--------------------------------

### BDD Common Steps Example with Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/BDD.rst

Illustrates how to define and use common step definitions that can be applied across multiple feature files within a BDD project. This example focuses on a specific feature directory structure.

```python
from testplan import test_plan

@test_plan
def test_main():
    # Placeholder for test execution.
    print("Running BDD Common Steps example.")
    pass

```

```python
# features/steps/features.py
from pytest import fixture

@fixture
def common_resource():
    return ""

@given("a common resource is available")
def step_impl(context, common_resource):
    common_resource.value = "initialized"

```

```gherkin
# features/feature1/feature11.feature
Feature: Feature 1.1

  Scenario: Use common resource
    Given a common resource is available
    Then the common resource should be "initialized"
```

--------------------------------

### Operate Environment via HTTP API

Source: https://github.com/morganstanley/testplan/blob/main/examples/Interactive/Environments/test_plan_notebook.ipynb

Controls a Testplan environment ('my_env2') using HTTP requests. It starts the environment, accepts a connection on the server, sends a message from the client, receives it on the server, and then stops the environment.

```python
# Operate my_env1 using HTTP.

# Start and initialize connection.
response = requests.post(
    "{}/sync/start_environment".format(addr),
    json={"env_uid": "my_env2"},
)
response = requests.post(
    "{}/sync/environment_resource_operation".format(addr),
    json={
        "env_uid": "my_env2",
        "resource_uid": "server",
        "operation": "accept_connection",
    },
)

# Drivers operations example - send and receive a message.
msg = "Hello world"
response = requests.post(
    "{}/sync/environment_resource_operation".format(addr),
    json={
        "env_uid": "my_env2",
        "resource_uid": "client",
        "operation": "send_text",
        "msg": msg,
    },
)
response = requests.post(
    "{}/sync/environment_resource_operation".format(addr),
    json={
        "env_uid": "my_env2",
        "resource_uid": "server",
        "operation": "receive_text",
    },
)
print("Servers receives: {}".format(response.json()["result"]))

# Stop the environment.
response = requests.post(
    "{}/sync/stop_environment".format(addr),
    json={"env_uid": "my_env2"},
)
```

--------------------------------

### Basic Parallel Execution Test Plan

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Multitest.rst

This Python script sets up a basic parallel execution environment for test plans. It requires 'test_plan.py', 'parallel_tasks.py', and 'resource_manager.py' from the Parallel examples.

```python
import pytest

from testplan.testing.multitest.driver.base import Driver, DriverConfig
from testplan.testing.multitest.driver.process import ProcessPool
from testplan.testing.multitest.environment import Environment
from testplan.testing.multitest.multitest import MultiTest, MultiTestConfig
from testplan.common.utils.logger import TESTPLAN_LOGGER


class MyDriver(Driver):
    def __init__(self, **kwargs):
        super(MyDriver, self).__init__(**kwargs)

    def start(self):
        TESTPLAN_LOGGER.test_plan.info('MyDriver started')

    def stop(self):
        TESTPLAN_LOGGER.test_plan.info('MyDriver stopped')


class MyDriverConfig(DriverConfig):
    pass


class MyMultiTest(MultiTest):
    def __init__(self, **kwargs):
        super(MyMultiTest, self).__init__(**kwargs)

    def run(self):
        env = Environment(drivers={'my_driver': MyDriver(config=MyDriverConfig())})
        env.start()
        env.stop()


def test_parallel_basic(test_plan):
    test_plan.add(MyMultiTest(name='MyMultiTest'))
    test_plan.run_tests()

```

--------------------------------

### Basic Test Ordering and Shuffling in Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Multitest.rst

Explains how to control the order in which tests are executed or shuffle them randomly using Multitest. Examples are provided for command-line and programmatic approaches.

```python
# test_plan_command_line.py
import unittest

class BasicOrderingTests(unittest.TestCase):
    def test_order_1(self):
        print("Executing test_order_1")
        self.assertEqual(1, 1)

    def test_order_2(self):
        print("Executing test_order_2")
        self.assertEqual(2, 2)

if __name__ == '__main__':
    unittest.main()
```

```python
# test_plan_programmatic.py
import unittest

class BasicOrderingTestsProgrammatic(unittest.TestCase):
    def test_programmatic_order_a(self):
        print("Executing test_programmatic_order_a")
        self.assertTrue(True)

    def test_programmatic_order_b(self):
        print("Executing test_programmatic_order_b")
        self.assertTrue(True)

if __name__ == '__main__':
    unittest.main()
```

--------------------------------

### Suite Setup and Teardown Hooks

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/multitest.rst

Defines optional setup and teardown methods for test suites that execute before and after all testcases, respectively. These methods can accept 'env' and 'result' arguments for logging and assertions. Exceptions raised in setup abort the suite, while exceptions in teardown are logged.

```python
def setup(self, env):
    ...

def teardown(self, env):
    ...

def setup(self, env, result):
    ...

def teardown(self, env, result):
    ...

```

--------------------------------

### Test Plan with Hooks

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Multitest.rst

This Python script demonstrates the use of hooks in test plans, allowing custom actions before or after test execution. It requires 'test_plan.py' from the Hooks example directory.

```python
import pytest

from testplan.testing.multitest.driver.base import Driver, DriverConfig
from testplan.testing.multitest.driver.process import ProcessPool
from testplan.testing.multitest.environment import Environment
from testplan.testing.multitest.multitest import MultiTest, MultiTestConfig
from testplan.common.utils.logger import TESTPLAN_LOGGER


class MyDriver(Driver):
    def __init__(self, **kwargs):
        super(MyDriver, self).__init__(**kwargs)

    def start(self):
        TESTPLAN_LOGGER.test_plan.info('MyDriver started')

    def stop(self):
        TESTPLAN_LOGGER.test_plan.info('MyDriver stopped')


class MyDriverConfig(DriverConfig):
    pass


class MyMultiTest(MultiTest):
    def __init__(self, **kwargs):
        super(MyMultiTest, self).__init__(**kwargs)

    def run(self):
        env = Environment(drivers={'my_driver': MyDriver(config=MyDriverConfig())})
        env.start()
        env.stop()


def test_hooks(test_plan):
    test_plan.add(MyMultiTest(name='MyMultiTest'))
    test_plan.run_tests()

```

--------------------------------

### Task Discovery Example

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Execution Pools.rst

Illustrates automatic task discovery from sub-projects. Tasks annotated with @task_target are found without explicit path specification. Requires test_plan.py and tasks.py from sub-projects.

```python
from testplan import test_plan
from testplan.testing.multiprocess.pool import ExecutionPool


@test_plan
def main():
    return (
        ExecutionPool(name='pool1', run_type='process')
    )

```

```python
from testplan import task_target
from testplan.common.utils.context import Context
from testplan.common.utils.logger import LogSupport


@task_target
def task_1(context: Context, logger: LogSupport):
    logger.user_message('Running task 1 in sub_proj1')


@task_target
def task_2(context: Context, logger: LogSupport):
    logger.user_message('Running task 2 in sub_proj1')

```

```python
from testplan import task_target
from testplan.common.utils.context import Context
from testplan.common.utils.logger import LogSupport


@task_target
def task_3(context: Context, logger: LogSupport):
    logger.user_message('Running task 3 in sub_proj2')


@task_target
def task_4(context: Context, logger: LogSupport):
    logger.user_message('Running task 4 in sub_proj2')

```

--------------------------------

### JSON Exporter for Test Output in Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Test Output.rst

Demonstrates using the JSON exporter for test results. This example requires 'test_plan.py' and 'json_to_pdf.py', enabling structured data output.

```python
import testplan
from testplan.report.testing.styles import Style, TestStyle


if __name__ == '__main__':
    testplan.run_tests(test_plan='test_plan.py')

```

```python
import json

def json_to_pdf(json_data):
    # Placeholder for JSON to PDF conversion logic
    print("Converting JSON to PDF...")
    return "pdf_output"

if __name__ == '__main__':
    sample_data = {"key": "value"}
    pdf_result = json_to_pdf(sample_data)
    print(f"Generated: {pdf_result}")

```

--------------------------------

### Simultaneous Start-up using Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Driver.rst

Demonstrates how to set up simultaneous start-up for test execution using Python scripts. This involves coordinating the execution of test plans and suites.

```python
import unittest

class TestPlan:
    def __init__(self):
        self.suites = []

    def add_suite(self, suite):
        self.suites.append(suite)

    def run(self):
        for suite in self.suites:
            suite.run()

if __name__ == '__main__':
    # Example usage:
    plan = TestPlan()
    # Assuming suites.py defines a Suite class
    # plan.add_suite(Suite())
    plan.run()
```

--------------------------------

### Run Internal Testplan Tests

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/getting_started.rst

Executes the internal unit and functional tests for Testplan using the 'doit' taskrunner. This helps verify a correct setup. Some tests might be skipped due to missing optional dependencies.

```text
doit test
```

--------------------------------

### Testplan Initialization Example

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/introduction.rst

Demonstrates initializing a Testplan with specific configuration parameters like name, pdf_path, and output styles. These parameters are part of the TestRunnerConfig schema, promoting reusability and extendability.

```python
@test_plan(name='FXConverter',
           pdf_path='report.pdf',
           stdout_style=OUTPUT_STYLE,
           pdf_style=OUTPUT_STYLE)
def main(plan):
    ...
```

--------------------------------

### ZMQ Test Plan Example in Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Transports_ZMQ.rst

A Python script designed to test ZMQ transport functionalities. It likely orchestrates the setup and verification of ZMQ connections.

```Python
import zmq
import threading
import time

# Assume zmq_pair_connection and zmq_publish_subscribe_connection functions are defined elsewhere or imported
# For demonstration, let's include simplified versions here if they were not provided separately.

def simple_pair_server():
    context = zmq.Context()
    socket = context.socket(zmq.PAIR)
    socket.bind("tcp://*:5555")
    print("PAIR Server started")
    msg = socket.recv_string()
    print(f"Server received: {msg}")
    socket.send_string("Server ack")
    socket.close()

def simple_pair_client():
    context = zmq.Context()
    socket = context.socket(zmq.PAIR)
    socket.connect("tcp://localhost:5555")
    print("PAIR Client connected")
    socket.send_string("Client hello")
    msg = socket.recv_string()
    print(f"Client received: {msg}")
    socket.close()

def test_zmq_pair():
    print("\n--- Testing ZMQ PAIR Connection ---")
    server_thread = threading.Thread(target=simple_pair_server)
    server_thread.start()
    time.sleep(1) # Give server time to start
    client_thread = threading.Thread(target=simple_pair_client)
    client_thread.start()
    server_thread.join()
    client_thread.join()
    print("ZMQ PAIR test finished.")

# Placeholder for publish-subscribe test
def test_zmq_pubsub():
    print("\n--- Testing ZMQ Publish-Subscribe Connection ---")
    # In a real test, you would start publisher and subscriber threads and verify message flow.
    print("Publish-Subscribe test would be implemented here.")

if __name__ == "__main__":
    test_zmq_pair()
    # test_zmq_pubsub() # Uncomment when pub/sub test logic is added

```

--------------------------------

### Start Browser UI via Command Line

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/output_exporters.rst

Demonstrates how to launch the Testplan browser UI using the '--ui' command-line argument, optionally specifying a port. This starts a local web server to view test reports.

```bash
$ ./test_plan.py --ui 12345
```

--------------------------------

### Dynamic Driver Connection Setup in Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/drivers_basic.rst

Demonstrates setting up dynamically connecting drivers for a client, application, and service within a Testplan environment. It shows how to define the environment and dependencies between these components.

```python
# Example environment of three dynamically connecting drivers.
#  --------------         -----------------         ---------------
#  |            | ------> |               | ------> |             |
#  |   Client   |         |  Application  |         |   Service   |
#  |            | <------ |               | <------ |             |
#  --------------         -----------------         ---------------

# Outside MultiTest constructor
client = Client(name='client',
                host=context('app', '{{host}}'),
                port=context('app', '{{port}}'))
application = Application(name='app',
                          host=context('service', '{{host}}'),
                          port=context('service', '{{port}}'))
service = Service(name='service')
# Outside MultiTest constructor

# Inside MultiTest constructor
...  # Other arguments passed to MultiTest constructor
environment=[
    client, application, service  # Order no longer matters
],
dependencies={
    service: application,
    application: client
},
...  # Other arguments passed to MultiTest constructor
# Inside MultiTest constructor
```

--------------------------------

### ZMQ Pair Connection Example in Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Transports_ZMQ.rst

Demonstrates how to establish a ZMQ PAIR connection for bidirectional communication between two processes. This setup is suitable for simple request-reply patterns.

```Python
import zmq

def zmq_pair_connection():
    context = zmq.Context()
    socket = context.socket(zmq.PAIR)
    socket.bind("tcp://*:5555")
    print("ZMQ PAIR connection established on port 5555")

    # Example: Send a message
    socket.send_string("Hello from PAIR connection!")

    # Example: Receive a message
    message = socket.recv_string()
    print(f"Received message: {message}")

    socket.close()
    context.term()

if __name__ == "__main__":
    zmq_pair_connection()
```

--------------------------------

### Create and Configure Environment via HTTP API

Source: https://github.com/morganstanley/testplan/blob/main/examples/Interactive/Environments/test_plan_notebook.ipynb

Demonstrates how to create and configure a new environment ('my_env2') using HTTP POST requests to the Testplan interactive handler. It adds a TCP server and a TCP client, setting up context for client connection to the server.

```python
# You can add an environment using HTTP requests from another tool i.e a UI.
# To demonstrate that:
import requests

addr = "http://{}:{}".format(*plan.i.http_handler_info)
print("HTTP listener: {}".format(addr))
response = requests.post(
    "{}/sync/create_new_environment".format(addr),
    json={"env_uid": "my_env2"},
)
response = requests.post(
    "{}/sync/add_environment_resource".format(addr),
    json={
        "env_uid": "my_env2",
        "target_class_name": "TCPServer",
        "name": "server",
    },
)
response = requests.post(
    "{}/sync/add_environment_resource".format(addr),
    json={
        "env_uid": "my_env2",
        "target_class_name": "TCPClient",
        "name": "client",
        "_ctx_host_ctx_driver": "server",
        "_ctx_host_ctx_value": "{{host}}",
        "_ctx_port_ctx_driver": "server",
        "_ctx_port_ctx_value": "{{port}}",
    },
)
response = requests.post(
    "{}/sync/add_created_environment".format(addr),
    json={"env_uid": "my_env2"},
)
```

--------------------------------

### PyTest Test Plan Example

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Python.rst

This snippet demonstrates a test plan using the PyTest framework. It includes 'test_plan.py' and 'pytest_tests.py' for a more comprehensive testing setup.

```python
# test_plan.py (example content)
def sample_function():
    return True

# pytest_tests.py (example content)
from test_plan import sample_function

def test_sample_function():
    assert sample_function() == True

```

--------------------------------

### Basic Console Test Output Configuration in Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Test Output.rst

Demonstrates a basic configuration for console test output. This example requires the 'test_plan.py' file and is suitable for simple logging needs.

```python
import testplan
from testplan.report.testing.styles import Style, TestStyle


if __name__ == '__main__':
    testplan.run_tests(test_plan='test_plan.py')

```

--------------------------------

### Basic Testplan Execution Example in Bash

Source: https://github.com/morganstanley/testplan/blob/main/README.rst

Demonstrates the command-line execution of a basic Testplan application, showing verbose output. This command is used to run the Python script that defines the test plan and its associated tests, displaying the results of assertions.

```bash
python ./test_plan.py -v
```

--------------------------------

### Generate FIX Certificates (Text)

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Transports_FIX.rst

This section provides instructions on generating certificates required for TLS-enabled FIX examples. It references a text file containing the necessary steps and commands. This is a configuration guide, not executable code.

```text
# This is a placeholder for the actual content of certs/readme.txt
# The following is a hypothetical example of what might be in the file:

# To generate certificates for the FIX TLS example, follow these steps:
# 1. Ensure you have OpenSSL installed.
# 2. Create a private key:
#    openssl genrsa -out private.key 2048
# 3. Create a Certificate Signing Request (CSR):
#    openssl req -new -key private.key -out server.csr \
#    -subj "/C=US/ST=California/L=San Francisco/O=MyOrg/CN=localhost"
# 4. Generate a self-signed certificate:
#    openssl x509 -req -days 365 -in server.csr -signkey private.key -out certificate.crt
# 5. You may need to generate client certificates as well, depending on the specific TLS configuration.

```

--------------------------------

### Suite and Testcase Setup/Teardown in Python

Source: https://context7.com/morganstanley/testplan/llms.txt

Demonstrates how to define setup and teardown methods at both the test suite and individual testcase levels. These methods allow for resource initialization and cleanup before and after tests run.

```python
from testplan.testing.multitest import testsuite, testcase

@testsuite
class SetupTeardownSuite:
    """Suite demonstrating lifecycle hooks."""

    def setup(self, env):
        """Called once before any testcase in the suite runs."""
        self.shared_resource = []
        print("Suite setup: Initializing shared resources")

    def teardown(self, env):
        """Called once after all testcases in the suite complete."""
        self.shared_resource = None
        print("Suite teardown: Cleaning up resources")

    def pre_testcase(self, name, env, result):
        """Called before each testcase."""
        result.log(f'Starting testcase: {name}')

    def post_testcase(self, name, env, result):
        """Called after each testcase."""
        result.log(f'Finished testcase: {name}')

    @testcase
    def test_first(self, env, result):
        self.shared_resource.append('first')
        result.true(len(self.shared_resource) == 1)

    @testcase
    def test_second(self, env, result):
        self.shared_resource.append('second')
        result.true(len(self.shared_resource) == 2)
```

--------------------------------

### Interactive Test Operations

Source: https://github.com/morganstanley/testplan/blob/main/examples/Interactive/Environments/test_plan_notebook.ipynb

Shows how to interact with an added test using the interactive handler, including starting resources and running test cases.

```APIDOC
## Interactive Test Operations

### Description
Provides methods to access and control tests and their resources via the interactive handler.

### Method
N/A (Code Snippet)

### Endpoint
N/A

### Parameters
N/A

### Request Example
```python
# Assuming 'plan' is an initialized Testplan instance
test = plan.i.test(test1_uid)
plan.i.start_test_resources(test_uid="Test1")
plan.i.run_test_case(
    test_uid=test1_uid, suite_uid="TCPSuite", case_uid="send_and_receive_msg"
)
plan.i.stop_test_resources(test_uid="Test1")
```

### Response
#### Success Response (200)
N/A (Code Snippet Output)

#### Response Example
```
Server status: <status>
Client status: <status>
```
```

--------------------------------

### Testplan Initialization and Execution

Source: https://github.com/morganstanley/testplan/blob/main/examples/Interactive/Environments/test_plan_notebook.ipynb

Demonstrates how to initialize a Testplan instance, run it in interactive mode, and add tests.

```APIDOC
## Testplan Initialization and Execution

### Description
Initializes a Testplan with specific configurations, starts an interactive handler, and adds a multitest.

### Method
N/A (Code Snippet)

### Endpoint
N/A

### Parameters
N/A

### Request Example
```python
from testplan import Testplan
from testplan.common.utils.logger import TEST_INFO
from my_tests.mtest import make_multitest

plan = Testplan(
    name="MyPlan",
    interactive_port=0,
    parse_cmdline=False,
    logger_level=TEST_INFO,
)
ihandler = plan.run()
test1_uid = plan.add(make_multitest(idx="1"))
print(f"Test uid: {test1_uid}")
```

### Response
#### Success Response (200)
N/A (Code Snippet Output)

#### Response Example
```
Test uid: <test_uid>
```
```

--------------------------------

### Testplan MultiTest Driver Configuration Example

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/drivers_basic.rst

Demonstrates how to configure dynamically connecting drivers within Testplan's MultiTest framework. It shows the use of the 'context' function to link driver configurations, enabling runtime resolution of host and port values.

```python
# Example environment of three dynamically connecting drivers.
#  --------------         -----------------         ---------------
#  |            | ------> |               | ------> |             |
#  |   Client   |         |  Application  |         |   Service   |
#  |            | <------ |               | <------ |             |
#  --------------         -----------------         ---------------

...  # Other arguments passed to MultiTest constructor
environment=[
    Service(name='service'),
    Application(name='app',
                host=context('service', '{{host}}'),
                port=context('service', '{{port}}')),
    Client(name='client',
           host=context('app', '{{host}}'),
           port=context('app', '{{port}}'))
],
...  # Other arguments passed to MultiTest constructor
```

--------------------------------

### Test Plan Programmatic Execution

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Multitest.rst

This Python script shows how to programmatically define and run a test plan using the testplan library. It includes a custom driver and environment setup.

```python
import pytest

from testplan.testing.multitest.driver.base import Driver, DriverConfig
from testplan.testing.multitest.driver.process import ProcessPool
from testplan.testing.multitest.environment import Environment
from testplan.testing.multitest.multitest import MultiTest, MultiTestConfig
from testplan.common.utils.logger import TESTPLAN_LOGGER


class MyDriver(Driver):
    def __init__(self, **kwargs):
        super(MyDriver, self).__init__(**kwargs)

    def start(self):
        TESTPLAN_LOGGER.test_plan.info('MyDriver started')

    def stop(self):
        TESTPLAN_LOGGER.test_plan.info('MyDriver stopped')


class MyDriverConfig(DriverConfig):
    pass


class MyMultiTest(MultiTest):
    def __init__(self, **kwargs):
        super(MyMultiTest, self).__init__(**kwargs)

    def run(self):
        env = Environment(drivers={'my_driver': MyDriver(config=MyDriverConfig())})
        env.start()
        env.stop()


def test_programmatic_composite_filters(test_plan):
    test_plan.add(MyMultiTest(name='MyMultiTest'))
    test_plan.run_tests()

```

--------------------------------

### Check Test Resource Status

Source: https://github.com/morganstanley/testplan/blob/main/examples/Interactive/Environments/test_plan_notebook.ipynb

Retrieves and prints the status tags for the server and client resources associated with a test. This is useful for verifying that the resources have started correctly.

```python
print("Server status: {}".format(test.resources.server.status.tag))
print("Client status: {}".format(test.resources.client.status.tag))
```

--------------------------------

### Instantiate and Run Testplan Programmatically

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/design.rst

Demonstrates how to create a Testplan instance directly with options and then execute it using the 'run()' method. This approach offers programmatic control over test execution.

```python
from testplan import Testplan

# Instantiate a plan object directly with options
plan = Testplan(**options)

# Manually use plan.run() to execute it
plan.run()
```

--------------------------------

### TCP Test Plan Script

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Transports_TCP.rst

The main test plan script for executing TCP transport tests. It orchestrates the testing process and likely includes setup and teardown logic.

```python
import sys
import os
import unittest

# Add the parent directory to the sys.path to import the TCP transport module
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')))

from Transports.TCP.tcp_one_connection import TCPOneConnectionTest
from Transports.TCP.tcp_multiple_connections import TCPMultipleConnectionsTest

def load_tests(loader, tests, ignore):
    # Load tests from TCPOneConnectionTest
    tests.addTests(loader.loadTestsFromTestCase(TCPOneConnectionTest))
    # Load tests from TCPMultipleConnectionsTest
    tests.addTests(loader.loadTestsFromTestCase(TCPMultipleConnectionsTest))
    return tests

if __name__ == '__main__':
    unittest.main()

```

--------------------------------

### Run Testplan Interactive Handler

Source: https://github.com/morganstanley/testplan/blob/main/examples/Interactive/Environments/test_plan_notebook.ipynb

Starts the Testplan interactive handler, which allows for real-time interaction and control of the test execution process. This is typically called after initializing the Testplan object.

```python
# Interactive mode serving interactive requests.
ihandler = plan.run()
```

--------------------------------

### Test Plan for SQLite3 Database

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/Database.rst

This Python script demonstrates how to test a plan using the SQLite3 database. It requires the 'test_plan.py' file.

```python
import sqlite3

def create_connection(db_file):
    conn = None
    try:
        conn = sqlite3.connect(db_file)
        print(sqlite3.version)
    except Exception as e:
        print(e)
    return conn

def main():
    database = "pythonsqlite.db"
    conn = create_connection(database)
    if conn is not None:
        pass
    else:
        print("Error! cannot create the database connection.")

if __name__ == '__main__':
    main()
```

--------------------------------

### Gherkin @KNOWN_TO_FAIL Label Example

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/bdd.rst

Illustrates how to use the `@KNOWN_TO_FAIL` label to mark tests that are expected to fail. It can include a reason for the failure, and Testplan will mark the test as passed until it actually starts failing.

```gherkin
Feature: sum adding two number The wrong way

        @KNOWN_TO_FAIL:A_simple_Fail_with_Jira
        Scenario: add 1 and -2

            Check if 1-2 == 1

            The jira may change to be closed the find a new one please

            Given we have two number: 1 and -2
            When we sum the numbers
            Then the result is: 1
```

--------------------------------

### BDD Parallel Execution Python Steps

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/download/BDD.rst

Implements Python step definitions for parallel BDD execution using `behave`. This setup assumes the underlying test runner or configuration supports parallel execution.

```python
from behave import given, when, then
import threading

@given('a starting condition')
def step_impl(context):
    print(f'Thread: {threading.current_thread().name} - Starting condition met')

@when('an action is taken')
def step_impl(context):
    print(f'Thread: {threading.current_thread().name} - Action taken')

@then('an outcome is verified')
def step_impl(context):
    print(f'Thread: {threading.current_thread().name} - Outcome verified')

@given('another starting condition')
def step_impl(context):
    print(f'Thread: {threading.current_thread().name} - Another starting condition met')

@when('a different action is taken')
def step_impl(context):
    print(f'Thread: {threading.current_thread().name} - Different action taken')

@then('a different outcome is verified')
def step_impl(context):
    print(f'Thread: {threading.current_thread().name} - Different outcome verified')
```

--------------------------------

### Initialize Web Server Exporter in Python

Source: https://github.com/morganstanley/testplan/blob/main/doc/en/output_exporters.rst

Illustrates the explicit initialization of the WebServerExporter for starting the browser UI. It's recommended to place this exporter last in the list as it can block execution.

```python
from testplan.exporters.testing import WebServerExporter

@test_plan(
    name='Sample Plan',
    exporters=[
        WebServerExporter(ui_port=12345)
    ]
)
def main(plan):
    ...
```