zeroae / zeroae/zae-limiter

♻️ Use pytest parametrization to reduce test duplication

Open
#178 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

testing
Dominant language
Python
Stars
0
Forks
0
Avg merge
6h 51m
Merged PRs (30d)
104

Description

Summary

The test suite has significant duplication that can be eliminated using pytest.mark.parametrize. This would reduce ~880 lines of test code while improving test clarity and coverage visibility.

Problem

Current Duplication Patterns
Location Issue Duplicate Lines
test_localstack.py + test_aws.py ~90% identical workflow logic ~500
test_cli.py deploy tests 7+ methods testing different flags ~200
StackOptions test classes 3 classes for different configs ~100
Input validation loops for char in [...] patterns ~30
Error code tests Repeated mock setup ~50
Examples of Current Duplication

1. LocalStack vs AWS (biggest impact)

test_localstack.py (713 lines) and test_aws.py (527 lines) contain nearly identical test logic:

# test_localstack.py
async def test_hierarchical_rate_limiting_workflow(self, e2e_limiter):
    parent = await e2e_limiter.create_entity("org-acme")
    child = await e2e_limiter.create_entity("api-key-123", parent_id="org-acme")
    ...

# test_aws.py - SAME LOGIC, different fixture
async def test_complete_aws_workflow(self, aws_limiter):
    await aws_limiter.create_entity("aws-parent")
    await aws_limiter.create_entity("aws-child", parent_id="aws-parent")
    ...

2. CLI Deploy Options

test_cli.py has 7+ nearly identical test methods:

def test_deploy_with_pitr_recovery_days(self, ...):
    result = runner.invoke(cli, ["deploy", "--pitr-recovery-days", "7", ...])
    assert stack_options.pitr_recovery_days == 7

def test_deploy_with_log_retention_days(self, ...):
    result = runner.invoke(cli, ["deploy", "--log-retention-days", "90", ...])
    assert stack_options.log_retention_days == 90

# ... 5 more similar methods

3. Input Validation Loops

# test_naming.py - loop hides individual test cases
def test_special_chars_raise(self):
    for char in ["@", "!", "$", "%", "&", "*", "(", ")", "+"]:
        with pytest.raises(ValidationError):
            validate_name(f"app{char}name")

Proposed Solution

1. Backend-Parametrized E2E Tests (Highest Impact)

Unify LocalStack and AWS tests with a backend parameter:

# tests/e2e/conftest.py

@pytest.fixture(params=["localstack", "aws"])
def e2e_backend(request, localstack_endpoint, unique_name):
    """Parametrized fixture for LocalStack or AWS backend."""
    if request.param == "aws" and not request.config.getoption("--run-aws"):
        pytest.skip("AWS tests require --run-aws")
    
    endpoint_url = localstack_endpoint if request.param == "localstack" else None
    
    stack_options = StackOptions(
        enable_aggregator=True,
        enable_alarms=(request.param == "aws"),  # Only AWS has real alarms
    )
    
    limiter = RateLimiter(
        name=unique_name,
        endpoint_url=endpoint_url,
        region="us-east-1",
        stack_options=stack_options,
    )
    
    async with limiter:
        yield limiter, request.param  # Yield backend name for conditional logic
    
    try:
        await limiter.delete_stack()
    except Exception as e:
        warnings.warn(f"Cleanup failed: {e}", ResourceWarning)


# tests/e2e/test_workflows.py - UNIFIED TESTS

class TestE2EWorkflows:
    """E2E workflows that run on both LocalStack and AWS."""
    
    @pytest.mark.asyncio
    async def test_hierarchical_rate_limiting(self, e2e_backend):
        limiter, backend = e2e_backend
        
        await limiter.create_entity("org-acme", name="ACME Organization")
        await limiter.create_entity("api-key-123", parent_id="org-acme")
        
        children = await limiter.get_children("org-acme")
        assert len(children) == 1
        
        # ... rest of test works on both backends
    
    @pytest.mark.asyncio
    async def test_rate_limit_exceeded(self, e2e_backend):
        limiter, backend = e2e_backend
        # Same test code for both backends
        ...

Running:

# LocalStack only (default)
pytest tests/e2e/test_workflows.py -v

# Both LocalStack and AWS
pytest tests/e2e/test_workflows.py -v --run-aws

# AWS only
pytest tests/e2e/test_workflows.py -v --run-aws -k "aws"
2. CLI Deploy Options Parametrization
# tests/unit/test_cli.py

@pytest.mark.parametrize("cli_args,option_name,expected_value", [
    (["--pitr-recovery-days", "7"], "pitr_recovery_days", 7),
    (["--log-retention-days", "90"], "log_retention_days", 90),
    (["--lambda-timeout", "120"], "lambda_timeout", 120),
    (["--lambda-memory", "512"], "lambda_memory", 512),
    (["--lambda-duration-threshold-pct", "90"], "lambda_duration_threshold_pct", 90),
    (["--no-aggregator"], "enable_aggregator", False),
    (["--no-alarms"], "enable_alarms", False),
    (["--permission-boundary", "MyPolicy"], "permission_boundary", "MyPolicy"),
    (["--role-name-format", "app-{}"], "role_name_format", "app-{}"),
])
@patch("zae_limiter.repository.Repository")
@patch("zae_limiter.cli.StackManager")
def test_deploy_option_sets_stack_options(
    self, mock_stack_manager, mock_repository, runner,
    cli_args, option_name, expected_value
):
    """Verify each CLI option correctly sets StackOptions."""
    # Common mock setup
    mock_instance = self._create_mock_stack_manager(mock_stack_manager)
    mock_repo = self._create_mock_repository(mock_repository)
    
    result = runner.invoke(cli, ["deploy", "--no-aggregator"] + cli_args)
    
    assert result.exit_code == 0
    stack_options = mock_instance.create_stack.call_args[1]["stack_options"]
    assert getattr(stack_options, option_name) == expected_value
3. StackOptions Test Parametrization
# tests/e2e/test_stack_variations.py

@pytest.mark.parametrize("stack_options,description", [
    pytest.param(
        StackOptions(enable_aggregator=False, enable_alarms=False),
        "minimal",
        id="minimal"
    ),
    pytest.param(
        StackOptions(enable_aggregator=True, enable_alarms=False),
        "aggregator-only",
        id="aggregator-only"
    ),
    pytest.param(
        StackOptions(enable_aggregator=True, enable_alarms=True),
        "full",
        id="full"
    ),
    pytest.param(
        StackOptions(enable_aggregator=False, pitr_recovery_days=7),
        "with-pitr",
        id="with-pitr"
    ),
])
class TestStackVariations:
    """Test stack deployment with various configurations."""
    
    @pytest.fixture
    async def limiter_with_options(self, localstack_endpoint, unique_name, stack_options):
        limiter = RateLimiter(
            name=unique_name,
            endpoint_url=localstack_endpoint,
            region="us-east-1",
            stack_options=stack_options,
        )
        async with limiter:
            yield limiter
        await limiter.delete_stack()
    
    async def test_can_create_and_use_entity(self, limiter_with_options):
        entity = await limiter_with_options.create_entity("test-entity")
        assert entity.id == "test-entity"
4. Input Validation Parametrization
# tests/unit/test_naming.py

@pytest.mark.parametrize("invalid_char", ["@", "!", "$", "%", "&", "*", "(", ")", "+"])
def test_special_char_raises_validation_error(self, invalid_char):
    """Each special character should raise ValidationError."""
    with pytest.raises(ValidationError):
        validate_name(f"app{invalid_char}name")


@pytest.mark.parametrize("invalid_name,expected_reason", [
    ("", "cannot be empty"),
    ("rate_limits", "underscore"),
    ("my.app", "period"),
    ("my app", "space"),
    ("123app", "start with a letter"),
    ("-app", "start with a letter"),
    ("a" * 39, "38 character"),
])
def test_invalid_name_with_specific_reason(self, invalid_name, expected_reason):
    """Validation errors should have helpful messages."""
    with pytest.raises(ValidationError) as exc_info:
        validate_name(invalid_name)
    assert expected_reason in exc_info.value.reason.lower()
5. Error Code Parametrization
# tests/unit/test_sync_limiter.py

@pytest.mark.parametrize("error_code", [
    "ServiceUnavailable",
    "ProvisionedThroughputExceededException",
    "InternalServerError",
    "ThrottlingException",
])
def test_block_mode_raises_on_dynamodb_error(self, sync_limiter, monkeypatch, error_code):
    """BLOCK mode should raise RateLimiterUnavailable on DynamoDB errors."""
    async def mock_error(*args, **kwargs):
        raise ClientError(
            {"Error": {"Code": error_code, "Message": "Test error"}},
            "GetItem",
        )
    
    monkeypatch.setattr(sync_limiter._limiter._repository, "get_bucket", mock_error)
    sync_limiter._limiter.on_unavailable = OnUnavailable.BLOCK
    
    with pytest.raises(RateLimiterUnavailable) as exc_info:
        with sync_limiter.acquire(entity_id="test", resource="api", limits=[...]):
            pass
    
    assert error_code in str(exc_info.value.cause)

Tasks

Phase 1: Backend Parametrization (Highest Impact)
  • Create tests/e2e/test_workflows.py with backend-parametrized fixture
  • Migrate common tests from test_localstack.py to test_workflows.py
  • Migrate common tests from test_aws.py to test_workflows.py
  • Keep backend-specific tests in original files (CloudWatch alarms, DLQ checks)
  • Verify both LocalStack and AWS tests pass
  • Delete duplicate code from original files
Phase 2: CLI Parametrization
  • Create _create_mock_stack_manager helper method
  • Consolidate 7+ deploy option tests into single parametrized test
  • Keep test_deploy_default_parameters as smoke test
  • Verify all CLI tests pass
Phase 3: Unit Test Parametrization
  • Parametrize test_naming.py special character tests
  • Parametrize test_naming.py invalid name tests
  • Parametrize test_models.py validation tests where applicable
  • Parametrize error code tests in test_sync_limiter.py
Phase 4: Documentation
  • Update CLAUDE.md with parametrization guidelines
  • Add to .claude/rules/zeroae/testing.md

Acceptance Criteria

  • ~500 lines reduced from e2e tests via backend parametrization
  • ~200 lines reduced from CLI tests via option parametrization
  • All parametrized tests show individual test IDs in pytest -v output
  • Test coverage unchanged or improved
  • Documentation updated with parametrization patterns

Documentation Updates

CLAUDE.md Addition
### Pytest Parametrization

Use `@pytest.mark.parametrize` to reduce duplication:

**When to parametrize:**
- Same test logic with different inputs
- Testing multiple error codes/edge cases
- Backend-agnostic tests (LocalStack vs AWS)
- CLI option variations

**Pattern for backend parametrization:**
```python
@pytest.fixture(params=["localstack", "aws"])
def e2e_backend(request, localstack_endpoint, unique_name):
    if request.param == "aws" and not request.config.getoption("--run-aws"):
        pytest.skip("AWS tests require --run-aws")
    endpoint_url = localstack_endpoint if request.param == "localstack" else None
    ...

Pattern for input validation:

@pytest.mark.parametrize("invalid_input,expected_error", [
    ("", "cannot be empty"),
    ("bad_name", "underscore"),
])
def test_validation_rejects_invalid_input(invalid_input, expected_error):
    with pytest.raises(ValidationError) as exc:
        validate(invalid_input)
    assert expected_error in str(exc.value)

Benefits:

  • Each parameter combination shown as separate test in pytest -v
  • Clear failure messages (e.g., "test_x[aws]" vs "test_x[localstack]")
  • Single source of truth for test logic

### New Rule: `.claude/rules/zeroae/testing.md`

Add to testing conventions:

```markdown
## Parametrization Guidelines

**DO parametrize:**
- Input validation with multiple invalid cases
- Error handling with multiple error codes
- Same workflow on different backends
- CLI options that follow identical patterns

**DON'T parametrize:**
- Tests with significantly different logic per case
- Tests where setup differs substantially
- One-off edge cases (just write separate tests)

**Example - Good parametrization:**
```python
@pytest.mark.parametrize("limit_type", ["per_minute", "per_hour", "per_day"])
def test_limit_factory_sets_correct_period(limit_type):
    ...

Example - Bad parametrization (too different):

# Don't do this - logic is too different per case
@pytest.mark.parametrize("mode", ["create", "update", "delete"])
def test_entity_lifecycle(mode):
    if mode == "create": ...  # Completely different logic
    elif mode == "update": ...  # Completely different logic
    elif mode == "delete": ...  # Completely different logic

## Impact

| Area | Lines Reduced | Benefit |
|------|---------------|---------|
| E2E backend unification | ~500 | Single source of truth, feature parity |
| CLI option tests | ~200 | Easier to add new options |
| Input validation | ~50 | Clear per-case test output |
| Error code tests | ~50 | Easier to add new error codes |
| StackOptions variations | ~80 | Consistent configuration testing |
| **Total** | **~880** | Cleaner, more maintainable tests |

## Related

- Part of #177 (v0.8.0: Test Infrastructure epic)
- Builds on #170 (shared fixtures) - parametrized fixtures benefit from shared base
- Complements #176 (sync/async strategy) - can parametrize sync/async where needed

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start by comparing tests/e2e/test_localstack.py, tests/e2e/test_aws.py, tests/unit/test_cli.py, tests/unit/test_naming.py, and tests/unit/test_sync_limiter.py, then run the relevant pytest commands. Done means the duplicated tests are consolidated across the listed phases, backend-specific coverage remains, verbose output shows parameter IDs, all tests pass, and the requested documentation is updated.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
testing
Issue type
Refactor
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
32/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.