♻️ Use pytest parametrization to reduce test duplication
Nobody has claimed this yet.
- 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.pywith backend-parametrized fixture - Migrate common tests from
test_localstack.pytotest_workflows.py - Migrate common tests from
test_aws.pytotest_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_managerhelper method - Consolidate 7+ deploy option tests into single parametrized test
- Keep
test_deploy_default_parametersas smoke test - Verify all CLI tests pass
Phase 3: Unit Test Parametrization
- Parametrize
test_naming.pyspecial character tests - Parametrize
test_naming.pyinvalid name tests - Parametrize
test_models.pyvalidation 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 -voutput - 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
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- 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