Testing Process Execution

Relevant source files

The following files were used as context for generating this wiki page:

• tests/test_process.py
• meltano/edk/process.py
• meltano/edk/extension.py
• tests/test_meltano_edk.py
• tests/copier_template/test_generated.py
• meltano/edk/models.py

Testing Process Execution within the Meltano Extension Developer Kit (EDK) focuses on validating how extensions interact with external binaries, manage subprocess lifecycles, and handle asynchronous input/output streams. The testing framework ensures that the Invoker utility correctly wraps command-line tools, manages environment variables, and captures process telemetry through logging.

The scope of testing covers unit tests for the Invoker class, integration tests for generated extensions using Copier templates, and functional validation of the ExtensionBase lifecycle hooks. These tests verify that commands are executed with the correct arguments and that the standard output (stdout) and standard error (stderr) are appropriately handled by the extension's logging infrastructure.

Sources: tests/test_process.py, meltano/edk/process.py, tests/test_meltano_edk.py

Subprocess Invocation and Mocking

The core of process execution testing relies on the Invoker class, which abstracts subprocess.run and asyncio.create_subprocess_exec. Testing involves mocking asyncio.subprocess to simulate process behavior without executing real binaries.

Process Mocking Strategy

Tests use pytest fixtures to create factory-generated mocks that simulate process attributes like returncode, stdout, and stderr. The process_mock_factory creates a Mock object where wait and stdin.wait_closed are AsyncMock instances, ensuring compatibility with the asynchronous EDK internals.

@pytest.fixture()
def process_mock_factory():
    def _factory(name: str) -> Mock:
        process_mock = Mock()
        process_mock.name = name
        process_mock.wait = AsyncMock(return_value=0)
        process_mock.returncode = 0
        process_mock.stdin.wait_closed = AsyncMock(return_value=True)
        return process_mock
    return _factory

Sources: tests/test_process.py:11-22

Invoker Execution Logic

The Invoker._exec method is tested by patching asyncio.create_subprocess_exec. The test verifies that the binary path, subcommands, and environment variables are passed correctly to the underlying OS process manager.

The following diagram illustrates the flow of a process execution test:

Extension Lifecycle Testing

Extensions built on the EDK follow a specific lifecycle: pre_invoke -> invoke -> post_invoke. Testing ensures that even when errors occur, the extension handles the failure gracefully or exits with the appropriate status code.

Lifecycle Hook Validation

Testing often uses a CustomExtension subclass to record the sequence of calls. This validates that the pass_through_invoker correctly sequences the lifecycle methods and handles exceptions.

Hook	Purpose	Tested Behavior
pre_invoke	Preparation before execution	Arguments are recorded and passed through correctly
invoke	Main execution logic	Executes the primary command or pass-through
post_invoke	Cleanup after execution	Executes regardless of success unless hard exit occurs

Sources: meltano/edk/extension.py:27-142, tests/test_meltano_edk.py:12-30

Error Handling in Execution

The pass_through_invoker includes try...except blocks for every lifecycle stage. In tests, if a hook fails, the logger is expected to record the exception and the process is expected to call sys.exit(1).

Template and Code Quality Testing

Testing extends to the EDK's project generation capabilities via Copier. This ensures that the generated boilerplate code is technically sound and adheres to project standards.

Copier Template Validation

The test_copier_output fixture generates a full extension in a temporary directory and runs static analysis tools against it:

1. Mypy: Validates type hints in the generated library.
2. Ruff: Checks for linting errors (E, F, W, Q) and enforces formatting rules.

Sources: tests/copier_template/test_generated.py:34-75

Description and Metadata Testing

Extensions must be able to "describe" themselves for Meltano integration. Tests validate that the describe_formatted method produces valid JSON and YAML output that matches the expected models.Describe schema.

def test_describe_formatted_json():
    test = CustomExtension()
    output = test.describe_formatted(DescribeFormat.json)
    parsed = json.loads(output)
    assert parsed["commands"][0]["name"] == "test"

Sources: tests/test_meltano_edk.py:41-61, meltano/edk/models.py:38-41

Asynchronous Stream Handling

A critical component of the Invoker is the ability to stream subprocess output to logs in real-time. This is tested using asyncio.streams.StreamReader.

Stream Logging Logic

The _log_stdio method reads lines from the process until EOF is reached. In tests, side_effect is used on the mock's readline and at_eof methods to simulate a stream of data (e.g., "SCHEMA", "RECORD", "STATE") followed by a completion signal.

process.stdout.at_eof.side_effect = (False, False, False, True)
process.stdout.readline = AsyncMock(
    side_effect=(b"SCHEMA\n", b"RECORD\n", b"STATE\n")
)

Sources: tests/test_process.py:25-30, meltano/edk/process.py:80-90

Summary of Testing Components

Component	File Path	Responsibility
Invoker	meltano/edk/process.py	Manages subprocess.run and asyncio execution
ExtensionBase	meltano/edk/extension.py	Defines lifecycle hooks and formatting
test_process.py	tests/test_process.py	Unit tests for async process execution and mocking
test_generated.py	tests/copier_template/test_generated.py	Integration tests for template generation
models.py	meltano/edk/models.py	Data structures for describing extension capabilities

Sources: meltano/edk/process.py, meltano/edk/extension.py, tests/test_process.py, tests/copier_template/test_generated.py