feat(resources): implement virtual resource type base class for abstract/computed resources #10605

Merged
HAL9000 merged 5 commits from feat/v3.6.0-virtual-resource-types into master 2026-06-04 08:42:50 +00:00
5 changed files with 1025 additions and 0 deletions
+1
View File
@@ -6,6 +6,7 @@ Changed `wf10_batch.robot` to be less likely to create files, and
`plan_generation_graph.robot` to give more test answers.
## [Unreleased]
- **Virtual Resource Type Base Class** (#8610): Implemented `VirtualResource` base class with two example concrete implementations (`MetricResource`, `APIEndpointResource`) for abstract/computed resources that are derived rather than mapped to physical files. Virtual resources are computed on demand via a `compute_fn` callable. Includes Behave BDD scenarios in `features/resource_virtual_types.feature` exercising construction, computation, name validation, kwargs passthrough, exception handling, string representation, and subclassing. Resource names are validated against `^[a-zA-Z][a-zA-Z0-9_-]*$` (must start with a letter; alphanumeric, hyphens, and underscores otherwise).
- **test(e2e): restore complete M2 acceptance test** (#11191): Restored the truncated M2 full actor compiler and LLM integration e2e acceptance test to its complete 10-step form. Added dynamic LLM provider selection via `Resolve LLM Actor` (falls back to Anthropic when OpenAI is unavailable or quota-exhausted), replacing hardcoded `gpt-4` / `openai/gpt-4` references in the actor config and action YAML. Added explicit return-code validation (`Should Be Equal As Integers ${r_actor.rc} 0`) for the actor registration step.
- **docs(a2a): ACP to A2A migration guide** (#10230): Added migration guide documenting how to upgrade from the ACP module to the A2A module introduced in v3.6.0, including symbol renames, field renames, operation-name mappings, and YAML configuration updates.
- **Plan Prompt JSON Timing Field** (#9353): `agents plan prompt --format json` now
+126
View File
@@ -0,0 +1,126 @@
Feature: Virtual Resource Type Base Class and Implementations
As a developer
I want to use virtual resources to represent computed/abstract resources
So that I can work with non-physical resources like metrics and API endpoints
Background:
Given I have imported the virtual resource module
Scenario: Create a basic VirtualResource with valid parameters
When I create a VirtualResource with:
| name | value |
| name | test-metric |
| description | A test metric resource |
| compute_fn | lambda: 42.0 |
Then the VirtualResource should have:
| name | value |
| name | test-metric |
| description | A test metric resource |
And the VirtualResource should be callable
Scenario: Compute a virtual resource value on demand
Given I have a VirtualResource named "cpu-usage" with compute_fn returning 45.2
When I compute the virtual resource
Then the computed value should be 45.2
Scenario: Reject invalid resource names
When I try to create a VirtualResource with invalid name "123-invalid"
Then the virtual resource creation should raise ValueError with message containing "invalid characters"
Scenario: Reject empty resource names
When I try to create a VirtualResource with an empty name
Then the virtual resource creation should raise ValueError with message containing "non-empty string"
Scenario: Reject non-callable compute functions
When I try to create a VirtualResource with non-callable compute_fn "not-a-function"
Then the virtual resource creation should raise TypeError with message containing "callable"
Scenario: Create a MetricResource with unit
When I create a MetricResource with:
| name | value |
| name | memory-usage |
| description | Current memory usage |
| compute_fn | lambda: 78.5 |
| unit | percent |
Then the MetricResource should have:
| name | value |
| name | memory-usage |
| description | Current memory usage |
| unit | percent |
Scenario: Compute a MetricResource value
Given I have a MetricResource named "disk-usage" with compute_fn returning 62.3 and unit "percent"
When I compute the metric resource
Then the computed value should be 62.3
Scenario: Create an APIEndpointResource with URL
When I create an APIEndpointResource with:
| name | value |
| name | api-data |
| description | Data from external API |
| compute_fn | lambda: {"status": "ok"} |
| endpoint_url | https://api.example.com/data |
Then the APIEndpointResource should have:
| name | value |
| name | api-data |
| description | Data from external API |
| endpoint_url | https://api.example.com/data |
Scenario: Compute an APIEndpointResource value
Given I have an APIEndpointResource named "api-status" with compute_fn returning {"status": "ok", "code": 200}
When I compute the API endpoint resource
Then the API endpoint computed value is a dict with keys "status,code"
And the API endpoint value["status"] is "ok"
And the API endpoint value["code"] is "200"
Scenario: Pass kwargs to compute function
Given I have a VirtualResource with compute_fn that accepts kwargs
When I compute the virtual resource with kwargs {"multiplier": 2}
Then the computed value should reflect the passed kwargs
Scenario: Virtual resource string representation
Given I have a VirtualResource named "test-resource" with description "Test resource"
When I get the string representation of the virtual resource
Then the string representation should contain "VirtualResource"
And the string representation should contain "test-resource"
And the string representation should contain "Test resource"
Scenario: MetricResource string representation
Given I have a MetricResource named "cpu-usage" with unit "percent"
When I get the string representation of the virtual resource
Then the string representation should contain "MetricResource"
And the string representation should contain "cpu-usage"
And the string representation should contain "percent"
Scenario: APIEndpointResource string representation
Given I have an APIEndpointResource named "api-data" with endpoint_url "https://api.example.com"
When I get the string representation of the virtual resource
Then the string representation should contain "APIEndpointResource"
And the string representation should contain "api-data"
And the string representation should contain "https://api.example.com"
Scenario: Virtual resource with complex compute function
Given I have a VirtualResource with a complex compute function
When I compute the virtual resource
Then the complex computed value should be correct
Scenario: Multiple virtual resources can coexist
Given I have created 3 different virtual resources
When I compute all of them
Then each should return its expected value
And they should not interfere with each other
Scenario: Virtual resource compute function exception handling
Given I have a VirtualResource with compute_fn that raises an exception
When I compute the virtual resource
Then it should raise the exception from compute_fn
Scenario: Virtual resource with default kwargs
Given I have a VirtualResource with compute_fn that has default kwargs
When I compute without providing kwargs
Then it should use the default values
Scenario: Virtual resource inheritance
Given I have created a custom VirtualResource subclass
When I instantiate and compute it
Then it should work correctly with the custom implementation
@@ -0,0 +1,684 @@
"""Step definitions for resource_virtual_types.feature.
Tests the VirtualResource base class and example implementations
(MetricResource, APIEndpointResource).
"""
from __future__ import annotations
import ast
from typing import Any
from behave import given, then, when
from cleveragents.resource.virtual import (
APIEndpointResource,
MetricResource,
VirtualResource,
)
def _parse_lambda_body(compute_fn_str: str) -> Any:
"""Parse a simple 'lambda: <expr>' string and return the literal value.
Only supports ``lambda: <literal>`` forms where the body is a Python
literal (int, float, str, dict, list, bool, None). This avoids the
security risks of ``eval()`` while still supporting the limited set of
compute-function strings used in Gherkin tables.
Args:
compute_fn_str: A string of the form ``"lambda: <literal>"``.
Returns:
The parsed literal value.
Raises:
ValueError: If the string is not a supported ``lambda: <literal>`` form.
"""
stripped = compute_fn_str.strip()
prefix = "lambda:"
if not stripped.startswith(prefix):
raise ValueError(
f"Unsupported compute_fn format: {compute_fn_str!r}. "
"Only 'lambda: <literal>' forms are supported in Gherkin tables."
)
body = stripped[len(prefix) :].strip()
try:
return ast.literal_eval(body)
except (ValueError, SyntaxError) as exc:
raise ValueError(f"Cannot safely parse lambda body {body!r}: {exc}") from exc
# ────────────────────────────────────────────────────────────
# Background steps
# ────────────────────────────────────────────────────────────
@given("I have imported the virtual resource module")
def step_import_virtual_module(context: object) -> None:
"""Verify the virtual resource module is imported."""
assert VirtualResource is not None
assert MetricResource is not None
assert APIEndpointResource is not None
# ────────────────────────────────────────────────────────────
# VirtualResource creation steps
# ────────────────────────────────────────────────────────────
@when("I create a VirtualResource with:")
def step_create_virtual_resource(context: object) -> None:
"""Create a VirtualResource from table data."""
table_dict = {row["name"]: row["value"] for row in context.table}
name = table_dict.get("name", "test-resource")
description = table_dict.get("description", "Test resource")
compute_fn_str = table_dict.get("compute_fn", "lambda: 42")
literal_value = _parse_lambda_body(compute_fn_str)
compute_fn = lambda: literal_value # noqa: E731
resource = VirtualResource(
name=name,
description=description,
compute_fn=compute_fn,
)
context.virtual_resource = resource
@then("the VirtualResource should have:")
def step_check_virtual_resource_attrs(context: object) -> None:
"""Check VirtualResource attributes."""
resource = context.virtual_resource
for row in context.table:
attr_name = row["name"]
expected_value = row["value"]
actual_value = getattr(resource, attr_name)
assert actual_value == expected_value, (
f"Expected {attr_name}={expected_value!r}, got {actual_value!r}"
)
@then("the VirtualResource should be callable")
def step_virtual_resource_callable(context: object) -> None:
"""Check that VirtualResource has a compute method."""
resource = context.virtual_resource
assert hasattr(resource, "compute"), "VirtualResource should have compute method"
assert callable(resource.compute), "compute should be callable"
# ────────────────────────────────────────────────────────────
# VirtualResource computation steps
# ────────────────────────────────────────────────────────────
Outdated
Review

eval() used here for parsing Gherkin values. This is a security anti-pattern (S307). Replace with ast.literal_eval() or explicit parsing. Multiple occurrences across both step files.

eval() used here for parsing Gherkin values. This is a security anti-pattern (S307). Replace with ast.literal_eval() or explicit parsing. Multiple occurrences across both step files.
@given('I have a VirtualResource named "{name}" with compute_fn returning {value}')
def step_create_virtual_with_value(context: object, name: str, value: str) -> None:
"""Create a VirtualResource that returns a specific value."""
if value.lower() == "true":
computed_value: Any = True
elif value.lower() == "false":
computed_value = False
elif value.isdigit():
computed_value = int(value)
else:
try:
computed_value = float(value)
except ValueError:
computed_value = value
resource = VirtualResource(
name=name,
description=f"Test resource {name}",
compute_fn=lambda: computed_value,
)
context.virtual_resource = resource
@when("I compute the virtual resource")
def step_compute_virtual_resource(context: object) -> None:
"""Compute the virtual resource value.
Exceptions raised by the compute function are captured into
``context.computation_error`` so that downstream ``Then`` steps can
assert on them; without this, an expected-exception scenario would
error at the ``When`` step before reaching the ``Then``.
"""
resource = context.virtual_resource
try:
context.computed_value = resource.compute()
context.computation_error = None
except Exception as exc:
context.computed_value = None
context.computation_error = exc
@then("the computed value should be {expected}")
def step_check_computed_value(context: object, expected: str) -> None:
"""Check the computed value."""
if expected.lower() == "true":
expected_value: Any = True
elif expected.lower() == "false":
expected_value = False
elif expected.isdigit():
expected_value = int(expected)
else:
try:
expected_value = float(expected)
except ValueError:
expected_value = expected
actual = context.computed_value
assert actual == expected_value, f"Expected {expected_value!r}, got {actual!r}"
# ────────────────────────────────────────────────────────────
# Error handling steps
# ────────────────────────────────────────────────────────────
@when('I try to create a VirtualResource with invalid name "{name}"')
def step_create_with_invalid_name(context: object, name: str) -> None:
"""Try to create a VirtualResource with invalid name."""
try:
VirtualResource(
name=name,
description="Test",
compute_fn=lambda: 42,
)
context.creation_error = None
except (ValueError, TypeError) as exc:
context.creation_error = exc
@when("I try to create a VirtualResource with an empty name")
def step_create_with_empty_name(context: object) -> None:
"""Try to create a VirtualResource with an empty name."""
try:
VirtualResource(
name="",
description="Test",
compute_fn=lambda: 42,
)
context.creation_error = None
except (ValueError, TypeError) as exc:
context.creation_error = exc
@when('I try to create a VirtualResource with non-callable compute_fn "{fn}"')
def step_create_with_non_callable(context: object, fn: str) -> None:
"""Try to create a VirtualResource with non-callable compute_fn."""
non_callable: Any = fn
try:
VirtualResource(
name="test",
Outdated
Review

FAIL: # type: ignore comment on this line. Zero tolerance for # type: ignore per CONTRIBUTING.md Type Safety rule. Remove this suppression by adding proper type annotations.

FAIL: # type: ignore comment on this line. Zero tolerance for # type: ignore per CONTRIBUTING.md Type Safety rule. Remove this suppression by adding proper type annotations.
description="Test",
compute_fn=non_callable,
)
context.creation_error = None
except (ValueError, TypeError) as exc:
context.creation_error = exc
@then(
'the virtual resource creation should raise {error_type} with message containing "{message}"'
)
def step_check_error(context: object, error_type: str, message: str) -> None:
"""Check that the expected error was raised."""
error = context.creation_error
assert error is not None, f"Expected {error_type} but no error was raised"
error_class_name = error.__class__.__name__
assert error_class_name == error_type, (
f"Expected {error_type}, got {error_class_name}"
)
assert message.lower() in str(error).lower(), (
f"Error message does not contain '{message}': {error}"
)
# ────────────────────────────────────────────────────────────
# MetricResource steps
# ────────────────────────────────────────────────────────────
@when("I create a MetricResource with:")
def step_create_metric_resource(context: object) -> None:
"""Create a MetricResource from table data."""
table_dict = {row["name"]: row["value"] for row in context.table}
name = table_dict.get("name", "test-metric")
description = table_dict.get("description", "Test metric")
compute_fn_str = table_dict.get("compute_fn", "lambda: 42.0")
unit = table_dict.get("unit", "")
literal_value = _parse_lambda_body(compute_fn_str)
compute_fn = lambda: literal_value # noqa: E731
resource = MetricResource(
name=name,
description=description,
compute_fn=compute_fn,
unit=unit,
)
context.metric_resource = resource
@then("the MetricResource should have:")
def step_check_metric_attrs(context: object) -> None:
"""Check MetricResource attributes."""
resource = context.metric_resource
for row in context.table:
attr_name = row["name"]
expected_value = row["value"]
actual_value = getattr(resource, attr_name)
assert actual_value == expected_value, (
f"Expected {attr_name}={expected_value!r}, got {actual_value!r}"
)
@given(
'I have a MetricResource named "{name}" with compute_fn returning {value} and unit "{unit}"'
)
def step_create_metric_with_value(
context: object, name: str, value: str, unit: str
) -> None:
"""Create a MetricResource that returns a specific value."""
computed_value = float(value)
resource = MetricResource(
name=name,
description=f"Test metric {name}",
compute_fn=lambda: computed_value,
unit=unit,
)
context.metric_resource = resource
@when("I compute the metric resource")
def step_compute_metric_resource(context: object) -> None:
"""Compute the metric resource value."""
resource = context.metric_resource
context.computed_value = resource.compute()
# ────────────────────────────────────────────────────────────
# APIEndpointResource steps
# ────────────────────────────────────────────────────────────
@when("I create an APIEndpointResource with:")
def step_create_api_endpoint_resource(context: object) -> None:
"""Create an APIEndpointResource from table data."""
table_dict = {row["name"]: row["value"] for row in context.table}
name = table_dict.get("name", "test-api")
description = table_dict.get("description", "Test API")
compute_fn_str = table_dict.get("compute_fn", "lambda: {}")
endpoint_url = table_dict.get("endpoint_url", "")
literal_value = _parse_lambda_body(compute_fn_str)
compute_fn = lambda: literal_value # noqa: E731
resource = APIEndpointResource(
name=name,
description=description,
compute_fn=compute_fn,
endpoint_url=endpoint_url,
)
context.api_resource = resource
@then("the APIEndpointResource should have:")
def step_check_api_attrs(context: object) -> None:
"""Check APIEndpointResource attributes."""
resource = context.api_resource
for row in context.table:
attr_name = row["name"]
expected_value = row["value"]
actual_value = getattr(resource, attr_name)
assert actual_value == expected_value, (
f"Expected {attr_name}={expected_value!r}, got {actual_value!r}"
)
@given('I have an APIEndpointResource named "{name}" with compute_fn returning {value}')
def step_create_api_with_value(context: object, name: str, value: str) -> None:
"""Create an APIEndpointResource that returns a specific value."""
computed_value: Any = ast.literal_eval(value)
resource = APIEndpointResource(
name=name,
description=f"Test API {name}",
compute_fn=lambda: computed_value,
)
context.api_resource = resource
@when("I compute the API endpoint resource")
def step_compute_api_resource(context: object) -> None:
"""Compute the API endpoint resource value."""
resource = context.api_resource
context.computed_value = resource.compute()
@then('the API endpoint computed value is a dict with keys "{keys}"')
def step_api_check_dict_keys(context: object, keys: str) -> None:
"""Check that computed value is a dict with expected keys."""
value = context.computed_value
assert isinstance(value, dict), f"Expected dict, got {type(value).__name__}"
expected_keys = [k.strip() for k in keys.strip('"').split(",")]
for key in expected_keys:
assert key in value, f"Expected key '{key}' not in dict: {value.keys()}"
@then('the API endpoint value["{key}"] is "{expected}"')
def step_api_check_dict_value(context: object, key: str, expected: str) -> None:
"""Check a specific value in the computed dict."""
value = context.computed_value
assert isinstance(value, dict), f"Expected dict, got {type(value).__name__}"
if expected.lower() == "true":
expected_value: Any = True
elif expected.lower() == "false":
expected_value = False
elif expected.isdigit():
expected_value = int(expected)
else:
try:
expected_value = float(expected)
except ValueError:
expected_value = expected
actual = value.get(key)
assert actual == expected_value, (
f"Expected {key}={expected_value!r}, got {actual!r}"
)
# ────────────────────────────────────────────────────────────
# Kwargs handling steps
# ────────────────────────────────────────────────────────────
@given("I have a VirtualResource with compute_fn that accepts kwargs")
def step_create_virtual_with_kwargs(context: object) -> None:
"""Create a VirtualResource with a compute_fn that accepts kwargs."""
def compute_with_kwargs(multiplier: int = 1) -> int:
return 42 * multiplier
resource = VirtualResource(
name="test-kwargs",
description="Test resource with kwargs",
compute_fn=compute_with_kwargs,
)
context.virtual_resource = resource
@when("I compute the virtual resource with kwargs {kwargs_str}")
def step_compute_with_kwargs(context: object, kwargs_str: str) -> None:
"""Compute the virtual resource with kwargs."""
kwargs: dict[str, Any] = ast.literal_eval(kwargs_str)
resource = context.virtual_resource
context.computed_value = resource.compute(**kwargs)
@then("the computed value should reflect the passed kwargs")
def step_check_kwargs_reflected(context: object) -> None:
"""Check that kwargs were properly used."""
assert context.computed_value == 84, f"Expected 84, got {context.computed_value}"
# ────────────────────────────────────────────────────────────
# String representation steps
# ────────────────────────────────────────────────────────────
@given('I have a VirtualResource named "{name}" with description "{description}"')
def step_create_virtual_for_repr(context: object, name: str, description: str) -> None:
"""Create a VirtualResource for string representation testing."""
resource = VirtualResource(
name=name,
description=description,
compute_fn=lambda: 42,
)
context.virtual_resource = resource
@when("I get the string representation of the virtual resource")
def step_get_string_repr(context: object) -> None:
"""Get the string representation of the resource."""
resource = (
getattr(context, "virtual_resource", None)
or getattr(context, "metric_resource", None)
or getattr(context, "api_resource", None)
)
context.string_repr = repr(resource)
@then('the string representation should contain "{text}"')
def step_check_repr_contains(context: object, text: str) -> None:
"""Check that string representation contains text."""
assert text in context.string_repr, (
f"Expected '{text}' in repr: {context.string_repr}"
)
@given('I have a MetricResource named "{name}" with unit "{unit}"')
def step_create_metric_for_repr(context: object, name: str, unit: str) -> None:
"""Create a MetricResource for string representation testing."""
resource = MetricResource(
name=name,
description="Test metric",
compute_fn=lambda: 42.0,
unit=unit,
)
context.metric_resource = resource
@given('I have an APIEndpointResource named "{name}" with endpoint_url "{url}"')
def step_create_api_for_repr(context: object, name: str, url: str) -> None:
"""Create an APIEndpointResource for string representation testing."""
resource = APIEndpointResource(
name=name,
description="Test API",
compute_fn=lambda: {},
endpoint_url=url,
)
context.api_resource = resource
# ────────────────────────────────────────────────────────────
# Complex compute function steps
# ────────────────────────────────────────────────────────────
@given("I have a VirtualResource with a complex compute function")
def step_create_complex_virtual(context: object) -> None:
"""Create a VirtualResource with a complex compute function."""
def complex_compute() -> dict[str, Any]:
return {
"result": 42,
"nested": {"value": "test"},
"list": [1, 2, 3],
}
resource = VirtualResource(
name="complex-resource",
description="Complex resource",
compute_fn=complex_compute,
)
context.virtual_resource = resource
@then("the complex computed value should be correct")
def step_check_complex_value(context: object) -> None:
"""Check that complex computed value is correct."""
value = context.computed_value
assert isinstance(value, dict), f"Expected dict, got {type(value).__name__}"
assert value["result"] == 42
assert value["nested"]["value"] == "test"
assert value["list"] == [1, 2, 3]
# ────────────────────────────────────────────────────────────
# Multiple resources steps
# ────────────────────────────────────────────────────────────
@given("I have created {count:d} different virtual resources")
def step_create_multiple_resources(context: object, count: int) -> None:
"""Create multiple different virtual resources."""
resources = []
for i in range(count):
resource = VirtualResource(
name=f"resource-{i}",
description=f"Resource {i}",
compute_fn=lambda i=i: i * 10,
)
resources.append(resource)
context.resources = resources
@when("I compute all of them")
def step_compute_all_resources(context: object) -> None:
"""Compute all resources."""
context.computed_values = [r.compute() for r in context.resources]
@then("each should return its expected value")
def step_check_each_value(context: object) -> None:
"""Check that each resource returned its expected value."""
for i, value in enumerate(context.computed_values):
expected = i * 10
assert value == expected, f"Resource {i}: expected {expected}, got {value}"
@then("they should not interfere with each other")
def step_check_no_interference(context: object) -> None:
"""Check that resources don't interfere with each other."""
assert len(context.computed_values) == len(context.resources)
# ────────────────────────────────────────────────────────────
# Exception handling steps
# ────────────────────────────────────────────────────────────
@given("I have a VirtualResource with compute_fn that raises an exception")
def step_create_virtual_with_exception(context: object) -> None:
"""Create a VirtualResource with compute_fn that raises an exception."""
def compute_with_error() -> None:
raise RuntimeError("Test error")
resource = VirtualResource(
name="error-resource",
description="Resource that raises",
compute_fn=compute_with_error,
)
context.virtual_resource = resource
@then("it should raise the exception from compute_fn")
def step_check_exception_raised(context: object) -> None:
"""Check that the exception was raised."""
try:
context.virtual_resource.compute()
raise AssertionError("Expected RuntimeError but no exception was raised")
except RuntimeError as exc:
assert "Test error" in str(exc)
# ────────────────────────────────────────────────────────────
# Default kwargs steps
# ────────────────────────────────────────────────────────────
@given("I have a VirtualResource with compute_fn that has default kwargs")
def step_create_virtual_with_defaults(context: object) -> None:
"""Create a VirtualResource with default kwargs."""
def compute_with_defaults(value: int = 100) -> int:
return value
resource = VirtualResource(
name="default-resource",
description="Resource with defaults",
compute_fn=compute_with_defaults,
)
context.virtual_resource = resource
@when("I compute without providing kwargs")
def step_compute_without_kwargs(context: object) -> None:
"""Compute without providing kwargs."""
context.computed_value = context.virtual_resource.compute()
@then("it should use the default values")
def step_check_defaults_used(context: object) -> None:
"""Check that default values were used."""
assert context.computed_value == 100, (
f"Expected 100 (default), got {context.computed_value}"
)
# ────────────────────────────────────────────────────────────
# Inheritance steps
# ────────────────────────────────────────────────────────────
@given("I have created a custom VirtualResource subclass")
def step_create_custom_subclass(context: object) -> None:
"""Create a custom VirtualResource subclass."""
class CustomResource(VirtualResource[str]):
"""Custom virtual resource subclass."""
def __init__(self, name: str, description: str) -> None:
super().__init__(
name=name,
description=description,
compute_fn=self._compute,
)
def _compute(self) -> str:
return f"Custom: {self.name}"
resource = CustomResource(
name="custom-resource",
description="Custom implementation",
)
context.custom_resource = resource
@when("I instantiate and compute it")
def step_compute_custom(context: object) -> None:
"""Compute the custom resource."""
context.computed_value = context.custom_resource.compute()
@then("it should work correctly with the custom implementation")
def step_check_custom_works(context: object) -> None:
"""Check that custom implementation works."""
expected = "Custom: custom-resource"
assert context.computed_value == expected, (
f"Expected '{expected}', got '{context.computed_value}'"
)
+8
View File
@@ -13,14 +13,22 @@ from cleveragents.resource.inheritance import (
resolve_inheritance_chain,
validate_chain,
)
from cleveragents.resource.virtual import (
APIEndpointResource,
MetricResource,
VirtualResource,
)
__all__ = [
"MAX_CHAIN_DEPTH",
"APIEndpointResource",
"MetricResource",
"ResourceTypeCircularInheritanceError",
"ResourceTypeInheritanceDepthError",
"ResourceTypeParentNotFoundError",
"ResourceTypeParentRemovalError",
"TypeRegistryMap",
"VirtualResource",
"find_subtypes",
"is_subtype_of",
"resolve_fields",
+206
View File
@@ -0,0 +1,206 @@
"""Virtual resource type base class and implementations.
Virtual resources represent abstract or computed resources that don't map to
physical files. Examples include computed metrics, API endpoints, or derived data.
This module provides:
- :class:`VirtualResource` — base class for all virtual resources
- :class:`MetricResource` — example: computed metrics
- :class:`APIEndpointResource` — example: API endpoints
Based on:
- docs/specification.md — Virtual Resource Types
- Issue #8610 — Virtual Resource Type Base Class Implementation
"""
from __future__ import annotations
import logging
import re
from collections.abc import Callable
from typing import Any
logger = logging.getLogger(__name__)
_NAME_PATTERN = re.compile(r"^[a-zA-Z][a-zA-Z0-9_-]*$")
class VirtualResource[T]:
"""Base class for virtual (computed/abstract) resources.
Virtual resources are abstract or computed resources that don't map to
physical files. They can be computed on demand via a compute function.
Attributes:
name: Unique identifier for this virtual resource.
description: Human-readable description of the resource.
compute_fn: Callable that computes the resource value on demand.
"""
def __init__(
self,
name: str,
description: str,
compute_fn: Callable[..., T],
) -> None:
"""Initialize a virtual resource.
Args:
name: Unique identifier for this virtual resource.
Must be a valid resource name (alphanumeric, hyphens, underscores).
description: Human-readable description of what this resource computes.
compute_fn: Callable that computes the resource value on demand.
Should accept no required arguments (all args should have defaults).
Raises:
ValueError: If name is empty or invalid.
TypeError: If compute_fn is not callable.
"""
if not name or not isinstance(name, str):
raise ValueError(
Outdated
Review

BLOCKING: Name validation logic has a bug causing unit test failures.

The current validation:

if not name.replace("-", "").replace("_", "").isalnum():
    raise ValueError(...)

Stripping hyphens and underscores before calling .isalnum() means "123-invalid" becomes "123invalid" which passes the check. The BDD test Scenario: Reject invalid resource names passes "123-invalid" and expects ValueError with "invalid characters", but no error is raised. This is causing the unit_tests CI failure.

How to fix: Use a regex that validates the full name pattern without stripping special characters:

import re
if not re.match(r'^[a-zA-Z][a-zA-Z0-9_-]*$', name):
    raise ValueError(
        f"Resource name '{name}' contains invalid characters. "
        "Names must start with a letter and use only alphanumeric characters, hyphens, and underscores."
    )

This correctly rejects "123-invalid" (starts with a digit), aligning with the test expectation.


Automated by CleverAgents Bot
Supervisor: PR Review | Agent: pr-review-worker

**BLOCKING: Name validation logic has a bug causing unit test failures.** The current validation: ```python if not name.replace("-", "").replace("_", "").isalnum(): raise ValueError(...) ``` Stripping hyphens and underscores before calling `.isalnum()` means `"123-invalid"` becomes `"123invalid"` which passes the check. The BDD test `Scenario: Reject invalid resource names` passes `"123-invalid"` and expects `ValueError` with `"invalid characters"`, but no error is raised. This is causing the `unit_tests` CI failure. **How to fix:** Use a regex that validates the full name pattern without stripping special characters: ```python import re if not re.match(r'^[a-zA-Z][a-zA-Z0-9_-]*$', name): raise ValueError( f"Resource name '{name}' contains invalid characters. " "Names must start with a letter and use only alphanumeric characters, hyphens, and underscores." ) ``` This correctly rejects `"123-invalid"` (starts with a digit), aligning with the test expectation. --- Automated by CleverAgents Bot Supervisor: PR Review | Agent: pr-review-worker
f"Resource name must be a non-empty string, got {type(name).__name__}"
)
if not _NAME_PATTERN.match(name):
raise ValueError(
f"Resource name '{name}' contains invalid characters. "
"Names must start with a letter and contain only "
"alphanumeric characters, hyphens, and underscores."
)
if not callable(compute_fn):
raise TypeError(
f"compute_fn must be callable, got {type(compute_fn).__name__}"
)
self.name: str = name
self.description: str = description
self.compute_fn: Callable[..., T] = compute_fn
def compute(self, **kwargs: Any) -> T:
"""Compute the resource value on demand.
Args:
**kwargs: Optional keyword arguments to pass to the compute function.
Returns:
The computed resource value.
Raises:
Exception: Any exception raised by the compute function.
"""
logger.debug(
"Computing virtual resource",
extra={
"resource_name": self.name,
"kwargs_keys": list(kwargs.keys()),
},
)
return self.compute_fn(**kwargs)
def __repr__(self) -> str:
"""Return string representation of the virtual resource."""
return (
f"{self.__class__.__name__}("
f"name={self.name!r}, "
f"description={self.description!r})"
)
class MetricResource(VirtualResource[float]):
"""Virtual resource that computes a metric value.
Example:
>>> def compute_cpu_usage() -> float:
... # Compute CPU usage percentage
... return 45.2
>>> metric = MetricResource(
... name="cpu-usage",
... description="Current CPU usage percentage",
... compute_fn=compute_cpu_usage,
... )
>>> value = metric.compute()
>>> print(f"CPU Usage: {value}%")
CPU Usage: 45.2%
"""
def __init__(
self,
name: str,
description: str,
compute_fn: Callable[..., float],
unit: str = "",
) -> None:
"""Initialize a metric resource.
Args:
name: Unique identifier for this metric.
description: Human-readable description of the metric.
compute_fn: Callable that computes the metric value (returns float).
unit: Optional unit of measurement (e.g., "percent", "bytes", "ms").
"""
super().__init__(name, description, compute_fn)
self.unit: str = unit
def __repr__(self) -> str:
"""Return string representation of the metric resource."""
unit_str = f", unit={self.unit!r}" if self.unit else ""
return (
f"MetricResource("
f"name={self.name!r}, "
f"description={self.description!r}{unit_str})"
)
class APIEndpointResource(VirtualResource[dict[str, Any]]):
"""Virtual resource that represents an API endpoint.
Example:
>>> def fetch_api_data() -> dict[str, Any]:
... # Fetch data from an API
... return {"status": "ok", "data": [1, 2, 3]}
>>> endpoint = APIEndpointResource(
... name="api-data",
... description="Data from external API",
... compute_fn=fetch_api_data,
... endpoint_url="https://api.example.com/data",
... )
>>> data = endpoint.compute()
>>> print(data["status"])
ok
"""
def __init__(
self,
name: str,
description: str,
compute_fn: Callable[..., dict[str, Any]],
endpoint_url: str = "",
) -> None:
"""Initialize an API endpoint resource.
Args:
name: Unique identifier for this endpoint.
description: Human-readable description of the endpoint.
compute_fn: Callable that fetches data from the endpoint
(returns dict).
endpoint_url: Optional URL of the API endpoint.
"""
super().__init__(name, description, compute_fn)
self.endpoint_url: str = endpoint_url
def __repr__(self) -> str:
"""Return string representation of the API endpoint resource."""
url_str = f", endpoint_url={self.endpoint_url!r}" if self.endpoint_url else ""
return (
f"APIEndpointResource("
f"name={self.name!r}, "
f"description={self.description!r}{url_str})"
)
__all__ = [
"APIEndpointResource",
"MetricResource",
"VirtualResource",
]