fix(acms): fix BDD step pattern quoting to resolve key mismatch in storage tier tests
CI / lint (pull_request) Successful in 1m7s
CI / quality (pull_request) Successful in 1m16s
CI / typecheck (pull_request) Successful in 1m30s
CI / security (pull_request) Successful in 1m43s
CI / helm (pull_request) Successful in 30s
CI / push-validation (pull_request) Successful in 33s
CI / build (pull_request) Successful in 57s
CI / e2e_tests (pull_request) Successful in 3m44s
CI / integration_tests (pull_request) Successful in 4m47s
CI / unit_tests (pull_request) Failing after 8m18s
CI / docker (pull_request) Has been skipped
CI / coverage (pull_request) Successful in 15m2s
CI / status-check (pull_request) Failing after 3s

Add missing step definitions for cold tier large value storage and
lifecycle TTL expiry checks to improve test coverage.

ISSUES CLOSED: #9580
This commit is contained in:
2026-05-05 05:40:04 +00:00
parent bb2bf1dcf4
commit 6eb877c2fc
2 changed files with 91 additions and 15 deletions
+11
View File
@@ -152,3 +152,14 @@ Feature: ACMS Hot/Warm/Cold Storage Tiers for Context Lifecycle Management
Then the dictionary should contain all metric fields
And the dictionary should include timestamp
And the dictionary should be JSON serializable
Scenario: Lifecycle policy TTL expiry checks
Given I have a storage tier manager with default configuration
When I check if a recent hot tier entry is expired
Then the hot tier entry should not be expired
When I check if an old hot tier entry is expired
Then the hot tier entry should be expired
When I check if a recent warm tier entry is expired
Then the warm tier entry should not be expired
When I check if a recent cold tier entry is expired
Then the cold tier entry should not be expired
+80 -15
View File
@@ -61,39 +61,39 @@ def step_set_cold_unlimited(context: Any) -> None:
assert context.manager.cold is not None
@when("I store a context {key} with value {value} in the hot tier")
@when('I store a context "{key}" with value "{value}" in the hot tier')
def step_store_in_hot(context: Any, key: str, value: str) -> None:
"""Store a context in the hot tier."""
context.manager.put(key, value, tier="hot")
@when("I store a context {key} with value {value} in the warm tier")
@when('I store a context "{key}" with value "{value}" in the warm tier')
def step_store_in_warm(context: Any, key: str, value: str) -> None:
"""Store a context in the warm tier."""
context.manager.put(key, value, tier="warm")
@when("I store a context {key} with value {value} in the cold tier")
@when('I store a context "{key}" with value "{value}" in the cold tier')
def step_store_in_cold(context: Any, key: str, value: str) -> None:
"""Store a context in the cold tier."""
context.manager.put(key, value, tier="cold")
@then("the context {key} should be in the hot tier")
@then('the context "{key}" should be in the hot tier')
def step_verify_in_hot(context: Any, key: str) -> None:
"""Verify context is in hot tier."""
value = context.manager.hot.get(key)
assert value is not None, f"Context {key} not found in hot tier"
@then("the context {key} should be in the warm tier")
@then('the context "{key}" should be in the warm tier')
def step_verify_in_warm(context: Any, key: str) -> None:
"""Verify context is in warm tier."""
value = context.manager.warm.get(key)
assert value is not None, f"Context {key} not found in warm tier"
@then("the context {key} should be in the cold tier")
@then('the context "{key}" should be in the cold tier')
def step_verify_in_cold(context: Any, key: str) -> None:
"""Verify context is in cold tier."""
value = context.manager.cold.get(key)
@@ -130,7 +130,7 @@ def step_retrieve_context(context: Any, key: str) -> None:
context.retrieved_value = context.manager.get(key)
@then("I should get the value {value}")
@then('I should get the value "{value}"')
def step_verify_retrieved_value(context: Any, value: str) -> None:
"""Verify retrieved value."""
assert context.retrieved_value == value
@@ -157,39 +157,39 @@ def step_verify_cold_hits(context: Any, count: int) -> None:
assert hits >= count
@given("I have stored context {key} with value {value} in the hot tier")
@given('I have stored context "{key}" with value "{value}" in the hot tier')
def step_store_context_hot(context: Any, key: str, value: str) -> None:
"""Store a context in hot tier."""
context.manager.put(key, value, tier="hot")
@given("I have stored context {key} with value {value} in the warm tier")
@given('I have stored context "{key}" with value "{value}" in the warm tier')
def step_store_context_warm(context: Any, key: str, value: str) -> None:
"""Store a context in warm tier."""
context.manager.put(key, value, tier="warm")
@given("I have stored context {key} with value {value} in the cold tier")
@given('I have stored context "{key}" with value "{value}" in the cold tier')
def step_store_context_cold(context: Any, key: str, value: str) -> None:
"""Store a context in cold tier."""
context.manager.put(key, value, tier="cold")
@then("the context {key} should not be in the hot tier")
@then('the context "{key}" should not be in the hot tier')
def step_verify_not_in_hot(context: Any, key: str) -> None:
"""Verify context is not in hot tier."""
value = context.manager.hot.get(key)
assert value is None, f"Context {key} should not be in hot tier"
@then("the context {key} should not be in the warm tier")
@then('the context "{key}" should not be in the warm tier')
def step_verify_not_in_warm(context: Any, key: str) -> None:
"""Verify context is not in warm tier."""
value = context.manager.warm.get(key)
assert value is None, f"Context {key} should not be in warm tier"
@then("the context {key} should not be in the cold tier")
@then('the context "{key}" should not be in the cold tier')
def step_verify_not_in_cold(context: Any, key: str) -> None:
"""Verify context is not in cold tier."""
value = context.manager.cold.get(key)
@@ -239,7 +239,7 @@ def step_verify_promotions(context: Any, count: int) -> None:
assert metrics.promotions >= count
@when("I delete context {key}")
@when('I delete context "{key}"')
def step_delete_context(context: Any, key: str) -> None:
"""Delete a context."""
context.manager.delete(key)
@@ -312,7 +312,7 @@ def step_verify_miss_count(context: Any) -> None:
assert misses > 0
@when("I store a context {key} with large value in the hot tier")
@when('I store a context "{key}" with large value in the hot tier')
def step_store_large_hot(context: Any, key: str) -> None:
"""Store large value in hot tier."""
large_value = "x" * 10000
@@ -320,6 +320,14 @@ def step_store_large_hot(context: Any, key: str) -> None:
context.manager.put(key, large_value, tier="hot")
@given('I have stored context "{key}" with large value in the cold tier')
def step_store_large_cold(context: Any, key: str) -> None:
"""Store large value in cold tier."""
large_value = "x" * 10000
context.large_value = large_value
context.manager.put(key, large_value, tier="cold")
@then("I should get the large value back")
def step_verify_large_value(context: Any) -> None:
"""Verify large value retrieved."""
@@ -493,3 +501,60 @@ def step_store_contexts_all_tiers_no_table(context: Any) -> None:
context.manager.put("ctx-metrics-hot", "data-hot", tier="hot")
context.manager.put("ctx-metrics-warm", "data-warm", tier="warm")
context.manager.put("ctx-metrics-cold", "data-cold", tier="cold")
@when("I check if a recent hot tier entry is expired")
def step_check_recent_hot_expired(context: Any) -> None:
"""Check if a recent hot tier entry is expired."""
context.hot_expired = context.manager.engine.is_hot_expired(
time.time(), ttl_seconds=3600
)
@then("the hot tier entry should not be expired")
def step_verify_hot_not_expired(context: Any) -> None:
"""Verify hot tier entry is not expired."""
assert not context.hot_expired
@when("I check if an old hot tier entry is expired")
def step_check_old_hot_expired(context: Any) -> None:
"""Check if an old hot tier entry is expired."""
old_timestamp = time.time() - 7200 # 2 hours ago
context.hot_expired = context.manager.engine.is_hot_expired(
old_timestamp, ttl_seconds=3600
)
@then("the hot tier entry should be expired")
def step_verify_hot_expired(context: Any) -> None:
"""Verify hot tier entry is expired."""
assert context.hot_expired
@when("I check if a recent warm tier entry is expired")
def step_check_recent_warm_expired(context: Any) -> None:
"""Check if a recent warm tier entry is expired."""
context.warm_expired = context.manager.engine.is_warm_expired(
time.time(), ttl_seconds=86400
)
@then("the warm tier entry should not be expired")
def step_verify_warm_not_expired(context: Any) -> None:
"""Verify warm tier entry is not expired."""
assert not context.warm_expired
@when("I check if a recent cold tier entry is expired")
def step_check_recent_cold_expired(context: Any) -> None:
"""Check if a recent cold tier entry is expired."""
context.cold_expired = context.manager.engine.is_cold_expired(
time.time(), ttl_seconds=604800
)
@then("the cold tier entry should not be expired")
def step_verify_cold_not_expired(context: Any) -> None:
"""Verify cold tier entry is not expired."""
assert not context.cold_expired