基于Python Pytest的ETL测试问题求助：命令行传参与循环执行异常

Hey there! Let's work through the two problems you're facing with pytest for ETL testing. I'll break down each issue with concrete code examples and explanations.

1. Passing Command-Line Arguments to pytest #

To pass your ETL job name into the pytest script, you'll use pytest's built-in hooks and fixtures. Here's how to set it up:

Step 1: Configure Command-Line Options (conftest.py) #

Create a conftest.py file in your test directory (pytest automatically picks this up) to define the command-line argument:

import pytest

def pytest_addoption(parser):
    parser.addoption(
        "--job-name",
        action="store",
        required=True,
        help="Name of the ETL job to validate"
    )

@pytest.fixture(scope="session")
def etl_job_name(request):
    # Fetch the job name from command-line arguments
    return request.config.getoption("--job-name")

Step 2: Use the Argument in Your Tests #

In your test script, inject the etl_job_name fixture to access the passed job name:

def test_sample(etl_job_name):
    print(f"Testing ETL job: {etl_job_name}")
    # Your test logic here

How to Run #

Execute pytest with the job name argument:

pytest your_test_script.py --job-name=your_etl_job

2. Testing All Four Tables (Fixing the Loop Exit Issue) #

The problem with your manual for loop is that pytest identifies test functions during the collection phase (before tests run). Dynamically defining test functions inside a loop often fails because pytest doesn't detect them properly. Instead, use pytest's @pytest.mark.parametrize to generate test cases for each table.

Solution Code #

import pytest

# Define the four tables you need to test
TARGET_TABLES = ["table_a", "table_b", "table_c", "table_d"]

@pytest.mark.parametrize("table_name", TARGET_TABLES)
def test_etl_table_load(etl_job_name, table_name):
    """Test that the ETL job loads data correctly into each target table."""
    # Your ETL load test logic goes here
    print(f"Validating ETL job {etl_job_name} for table: {table_name}")
    
    # Example assertions (replace with your actual checks):
    # 1. Verify the table exists in the target database
    # 2. Check that row count meets expectations
    # 3. Validate schema/column correctness
    assert table_name in TARGET_TABLES, f"Invalid table name: {table_name}"

Why This Works #

• @pytest.mark.parametrize tells pytest to create a separate test case for each value in TARGET_TABLES.
• Each table gets its own test run, and pytest will report results for all four tables individually.
• No more loop exits—pytest handles the iteration natively.

Putting It All Together #

When you run the command below, pytest will:

1. Accept the --job-name argument
2. Run the test_etl_table_load function four times (once per table)

pytest test_etl_load.py --job-name=my_daily_etl_job

You'll see output that shows all four tests executing, with pass/fail status for each table separately.

内容的提问来源于stack exchange，提问作者Aritra Sen