test(frontend): e2e tests for library page

[Abhi1992002]

In this PR, I’ve added library page tests.

Changes

I’ve added 9 tests: 8 for normal flows and 1 for checking edge cases.

Test names are something like:

• Library navigation is accessible from the navbar.
• The library page loads successfully.
• Agents are visible, and cards work correctly.
• Pagination works correctly.
• Sorting works correctly.
• Searching works correctly.
• Pagination while searching works correctly.
• Uploading an agent works correctly.
• Edge case: Search edge cases and error handling behave correctly.

Other than that, I’ve added a new utility that uses the build page to help us create users at the start, which we could use to test the library page.

• All tests are passing locally

Checklist 📋

For code changes:

• I have clearly listed my changes in the PR description
• I have made a test plan
• I have tested my changes according to the test plan:
  • All library tests are working locally and on CI perfectly.

[github-actions]

This PR targets the master branch but does not come from dev or a hotfix/* branch.

Automatically setting the base branch to dev.

[netlify]

✅ Deploy Preview for auto-gpt-docs canceled.

Name	Link
🔨 Latest commit	a8a0654
🔍 Latest deploy log	https://app.netlify.com/projects/auto-gpt-docs/deploys/6871f4c0ab194a00088bef3c

[netlify]

✅ Deploy Preview for auto-gpt-docs-dev canceled.

Name	Link
🔨 Latest commit	7410500
🔍 Latest deploy log	https://app.netlify.com/projects/auto-gpt-docs-dev/deploys/6892fee029a73f0008b010d2

[netlify]

✅ Deploy Preview for auto-gpt-docs canceled.

Name	Link
🔨 Latest commit	7410500
🔍 Latest deploy log	https://app.netlify.com/projects/auto-gpt-docs/deploys/6892fee0d48769000841da1e

[qodo-code-review]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 4 🔵🔵🔵🔵⚪

🧪 PR contains tests

🔒 Security concerns Sensitive information exposure: The test files contain hardcoded API keys and secrets (e.g., "test-api-key", "sk-1234567890") which could be accidentally committed to version control. The SecretStr usage in tests may not properly mask these values during debugging or logging.

⚡ Recommended focus areas for review Test Quality The test file contains extensive mock testing but lacks integration with actual database or external services. Some tests use hardcoded credentials and may not reflect real-world usage patterns. """ Tests for creating blocks using the SDK. This test suite verifies that blocks can be created using only SDK imports and that they work correctly without decorators. """ from typing import Any, Optional, Union import pytest from backend.sdk import ( APIKeyCredentials, Block, BlockCategory, BlockCostType, BlockOutput, BlockSchema, CredentialsMetaInput, OAuth2Credentials, ProviderBuilder, SchemaField, SecretStr, ) from ._config import test_api, test_service class TestBasicBlockCreation: """Test creating basic blocks using the SDK.""" @pytest.mark.asyncio async def test_simple_block(self): """Test creating a simple block without any decorators.""" class SimpleBlock(Block): """A simple test block.""" class Input(BlockSchema): text: str = SchemaField(description="Input text") count: int = SchemaField(description="Repeat count", default=1) class Output(BlockSchema): result: str = SchemaField(description="Output result") def __init__(self): super().__init__( id="simple-test-block", description="A simple test block", categories={BlockCategory.TEXT}, input_schema=SimpleBlock.Input, output_schema=SimpleBlock.Output, ) async def run(self, input_data: Input, **kwargs) -> BlockOutput: result = input_data.text * input_data.count yield "result", result # Create and test the block block = SimpleBlock() assert block.id == "simple-test-block" assert BlockCategory.TEXT in block.categories # Test execution outputs = [] async for name, value in block.run( SimpleBlock.Input(text="Hello ", count=3), ): outputs.append((name, value)) assert len(outputs) == 1 assert outputs[0] == ("result", "Hello Hello Hello ") @pytest.mark.asyncio async def test_block_with_credentials(self): """Test creating a block that requires credentials.""" class APIBlock(Block): """A block that requires API credentials.""" class Input(BlockSchema): credentials: CredentialsMetaInput = test_api.credentials_field( description="API credentials for test service", ) query: str = SchemaField(description="API query") class Output(BlockSchema): response: str = SchemaField(description="API response") authenticated: bool = SchemaField(description="Was authenticated") def __init__(self): super().__init__( id="api-test-block", description="Test block with API credentials", categories={BlockCategory.DEVELOPER_TOOLS}, input_schema=APIBlock.Input, output_schema=APIBlock.Output, ) async def run( self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs ) -> BlockOutput: # Simulate API call api_key = credentials.api_key.get_secret_value() authenticated = bool(api_key) yield "response", f"API response for: {input_data.query}" yield "authenticated", authenticated # Create test credentials test_creds = APIKeyCredentials( id="test-creds", provider="test_api", api_key=SecretStr("test-api-key"), title="Test API Key", ) # Create and test the block block = APIBlock() outputs = [] async for name, value in block.run( APIBlock.Input( credentials={ # type: ignore "provider": "test_api", "id": "test-creds", "type": "api_key", }, query="test query", ), credentials=test_creds, ): outputs.append((name, value)) assert len(outputs) == 2 assert outputs[0] == ("response", "API response for: test query") assert outputs[1] == ("authenticated", True) @pytest.mark.asyncio async def test_block_with_multiple_outputs(self): """Test block that yields multiple outputs.""" class MultiOutputBlock(Block): """Block with multiple outputs.""" class Input(BlockSchema): text: str = SchemaField(description="Input text") class Output(BlockSchema): uppercase: str = SchemaField(description="Uppercase version") lowercase: str = SchemaField(description="Lowercase version") length: int = SchemaField(description="Text length") is_empty: bool = SchemaField(description="Is text empty") def __init__(self): super().__init__( id="multi-output-block", description="Block with multiple outputs", categories={BlockCategory.TEXT}, input_schema=MultiOutputBlock.Input, output_schema=MultiOutputBlock.Output, ) async def run(self, input_data: Input, **kwargs) -> BlockOutput: text = input_data.text yield "uppercase", text.upper() yield "lowercase", text.lower() yield "length", len(text) yield "is_empty", len(text) == 0 # Test the block block = MultiOutputBlock() outputs = [] async for name, value in block.run(MultiOutputBlock.Input(text="Hello World")): outputs.append((name, value)) assert len(outputs) == 4 assert ("uppercase", "HELLO WORLD") in outputs assert ("lowercase", "hello world") in outputs assert ("length", 11) in outputs assert ("is_empty", False) in outputs class TestBlockWithProvider: """Test creating blocks associated with providers.""" @pytest.mark.asyncio async def test_block_using_provider(self): """Test block that uses a registered provider.""" class TestServiceBlock(Block): """Block for test service.""" class Input(BlockSchema): credentials: CredentialsMetaInput = test_service.credentials_field( description="Test service credentials", ) action: str = SchemaField(description="Action to perform") class Output(BlockSchema): result: str = SchemaField(description="Action result") provider_name: str = SchemaField(description="Provider used") def __init__(self): super().__init__( id="test-service-block", description="Block using test service provider", categories={BlockCategory.DEVELOPER_TOOLS}, input_schema=TestServiceBlock.Input, output_schema=TestServiceBlock.Output, ) async def run( self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs ) -> BlockOutput: # The provider name should match yield "result", f"Performed: {input_data.action}" yield "provider_name", credentials.provider # Create credentials for our provider creds = APIKeyCredentials( id="test-service-creds", provider="test_service", api_key=SecretStr("test-key"), title="Test Service Key", ) # Test the block block = TestServiceBlock() outputs = {} async for name, value in block.run( TestServiceBlock.Input( credentials={ # type: ignore "provider": "test_service", "id": "test-service-creds", "type": "api_key", }, action="test action", ), credentials=creds, ): outputs[name] = value assert outputs["result"] == "Performed: test action" assert outputs["provider_name"] == "test_service" class TestComplexBlockScenarios: """Test more complex block scenarios.""" @pytest.mark.asyncio async def test_block_with_optional_fields(self): """Test block with optional input fields.""" # Optional is already imported at the module level class OptionalFieldBlock(Block): """Block with optional fields.""" class Input(BlockSchema): required_field: str = SchemaField(description="Required field") optional_field: Optional[str] = SchemaField( description="Optional field", default=None, ) optional_with_default: str = SchemaField( description="Optional with default", default="default value", ) class Output(BlockSchema): has_optional: bool = SchemaField(description="Has optional value") optional_value: Optional[str] = SchemaField( description="Optional value" ) default_value: str = SchemaField(description="Default value") def __init__(self): super().__init__( id="optional-field-block", description="Block with optional fields", categories={BlockCategory.TEXT}, input_schema=OptionalFieldBlock.Input, output_schema=OptionalFieldBlock.Output, ) async def run(self, input_data: Input, **kwargs) -> BlockOutput: yield "has_optional", input_data.optional_field is not None yield "optional_value", input_data.optional_field yield "default_value", input_data.optional_with_default # Test with optional field provided block = OptionalFieldBlock() outputs = {} async for name, value in block.run( OptionalFieldBlock.Input( required_field="test", optional_field="provided", ) ): outputs[name] = value assert outputs["has_optional"] is True assert outputs["optional_value"] == "provided" assert outputs["default_value"] == "default value" # Test without optional field outputs = {} async for name, value in block.run( OptionalFieldBlock.Input( required_field="test", ) ): outputs[name] = value assert outputs["has_optional"] is False assert outputs["optional_value"] is None assert outputs["default_value"] == "default value" @pytest.mark.asyncio async def test_block_with_complex_types(self): """Test block with complex input/output types.""" class ComplexBlock(Block): """Block with complex types.""" class Input(BlockSchema): items: list[str] = SchemaField(description="List of items") mapping: dict[str, int] = SchemaField( description="String to int mapping" ) class Output(BlockSchema): item_count: int = SchemaField(description="Number of items") total_value: int = SchemaField(description="Sum of mapping values") combined: list[str] = SchemaField(description="Combined results") def __init__(self): super().__init__( id="complex-types-block", description="Block with complex types", categories={BlockCategory.DEVELOPER_TOOLS}, input_schema=ComplexBlock.Input, output_schema=ComplexBlock.Output, ) async def run(self, input_data: Input, **kwargs) -> BlockOutput: yield "item_count", len(input_data.items) yield "total_value", sum(input_data.mapping.values()) # Combine items with their mapping values combined = [] for item in input_data.items: value = input_data.mapping.get(item, 0) combined.append(f"{item}: {value}") yield "combined", combined # Test the block block = ComplexBlock() outputs = {} async for name, value in block.run( ComplexBlock.Input( items=["apple", "banana", "orange"], mapping={"apple": 5, "banana": 3, "orange": 4}, ) ): outputs[name] = value assert outputs["item_count"] == 3 assert outputs["total_value"] == 12 assert outputs["combined"] == ["apple: 5", "banana: 3", "orange: 4"] @pytest.mark.asyncio async def test_block_error_handling(self): """Test block error handling.""" class ErrorHandlingBlock(Block): """Block that demonstrates error handling.""" class Input(BlockSchema): value: int = SchemaField(description="Input value") should_error: bool = SchemaField( description="Whether to trigger an error", default=False, ) class Output(BlockSchema): result: int = SchemaField(description="Result") error_message: Optional[str] = SchemaField( description="Error if any", default=None ) def __init__(self): super().__init__( id="error-handling-block", description="Block with error handling", categories={BlockCategory.DEVELOPER_TOOLS}, input_schema=ErrorHandlingBlock.Input, output_schema=ErrorHandlingBlock.Output, ) async def run(self, input_data: Input, **kwargs) -> BlockOutput: if input_data.should_error: raise ValueError("Intentional error triggered") if input_data.value < 0: yield "error_message", "Value must be non-negative" yield "result", 0 else: yield "result", input_data.value * 2 yield "error_message", None # Test normal operation block = ErrorHandlingBlock() outputs = {} async for name, value in block.run( ErrorHandlingBlock.Input(value=5, should_error=False) ): outputs[name] = value assert outputs["result"] == 10 assert outputs["error_message"] is None # Test with negative value outputs = {} async for name, value in block.run( ErrorHandlingBlock.Input(value=-5, should_error=False) ): outputs[name] = value assert outputs["result"] == 0 assert outputs["error_message"] == "Value must be non-negative" # Test with error with pytest.raises(ValueError, match="Intentional error triggered"): async for _ in block.run( ErrorHandlingBlock.Input(value=5, should_error=True) ): pass class TestAuthenticationVariants: """Test complex authentication scenarios including OAuth, API keys, and scopes.""" @pytest.mark.asyncio async def test_oauth_block_with_scopes(self): """Test creating a block that uses OAuth2 with scopes.""" from backend.sdk import OAuth2Credentials, ProviderBuilder # Create a test OAuth provider with scopes # For testing, we don't need an actual OAuth handler # In real usage, you would provide a proper OAuth handler class oauth_provider = ( ProviderBuilder("test_oauth_provider") .with_api_key("TEST_OAUTH_API", "Test OAuth API") .with_base_cost(5, BlockCostType.RUN) .build() ) class OAuthScopedBlock(Block): """Block requiring OAuth2 with specific scopes.""" class Input(BlockSchema): credentials: CredentialsMetaInput = oauth_provider.credentials_field( description="OAuth2 credentials with scopes", scopes=["read:user", "write:data"], ) resource: str = SchemaField(description="Resource to access") class Output(BlockSchema): data: str = SchemaField(description="Retrieved data") scopes_used: list[str] = SchemaField( description="Scopes that were used" ) token_info: dict[str, Any] = SchemaField( description="Token information" ) def __init__(self): super().__init__( id="oauth-scoped-block", description="Test OAuth2 with scopes", categories={BlockCategory.DEVELOPER_TOOLS}, input_schema=OAuthScopedBlock.Input, output_schema=OAuthScopedBlock.Output, ) async def run( self, input_data: Input, *, credentials: OAuth2Credentials, **kwargs ) -> BlockOutput: # Simulate OAuth API call with scopes token = credentials.access_token.get_secret_value() yield "data", f"OAuth data for {input_data.resource}" yield "scopes_used", credentials.scopes or [] yield "token_info", { "has_token": bool(token), "has_refresh": credentials.refresh_token is not None, "provider": credentials.provider, "expires_at": credentials.access_token_expires_at, } # Create test OAuth credentials test_oauth_creds = OAuth2Credentials( id="test-oauth-creds", provider="test_oauth_provider", access_token=SecretStr("test-access-token"), refresh_token=SecretStr("test-refresh-token"), scopes=["read:user", "write:data"], title="Test OAuth Credentials", ) # Test the block block = OAuthScopedBlock() outputs = {} async for name, value in block.run( OAuthScopedBlock.Input( credentials={ # type: ignore "provider": "test_oauth_provider", "id": "test-oauth-creds", "type": "oauth2", }, resource="user/profile", ), credentials=test_oauth_creds, ): outputs[name] = value assert outputs["data"] == "OAuth data for user/profile" assert set(outputs["scopes_used"]) == {"read:user", "write:data"} assert outputs["token_info"]["has_token"] is True assert outputs["token_info"]["expires_at"] is None assert outputs["token_info"]["has_refresh"] is True @pytest.mark.asyncio async def test_mixed_auth_block(self): """Test block that supports both OAuth2 and API key authentication.""" # No need to import these again, already imported at top # Create provider supporting both auth types # Create provider supporting API key auth # In real usage, you would add OAuth support with .with_oauth() mixed_provider = ( ProviderBuilder("mixed_auth_provider") .with_api_key("MIXED_API_KEY", "Mixed Provider API Key") .with_base_cost(8, BlockCostType.RUN) .build() ) class MixedAuthBlock(Block): """Block supporting multiple authentication methods.""" class Input(BlockSchema): credentials: CredentialsMetaInput = mixed_provider.credentials_field( description="API key or OAuth2 credentials", supported_credential_types=["api_key", "oauth2"], ) operation: str = SchemaField(description="Operation to perform") class Output(BlockSchema): result: str = SchemaField(description="Operation result") auth_type: str = SchemaField(description="Authentication type used") auth_details: dict[str, Any] = SchemaField(description="Auth details") def __init__(self): super().__init__( id="mixed-auth-block", description="Block supporting OAuth2 and API key", categories={BlockCategory.DEVELOPER_TOOLS}, input_schema=MixedAuthBlock.Input, output_schema=MixedAuthBlock.Output, ) async def run( self, input_data: Input, *, credentials: Union[APIKeyCredentials, OAuth2Credentials], **kwargs, ) -> BlockOutput: # Handle different credential types if isinstance(credentials, APIKeyCredentials): auth_type = "api_key" auth_details = { "has_key": bool(credentials.api_key.get_secret_value()), "key_prefix": credentials.api_key.get_secret_value()[:5] + "...", } elif isinstance(credentials, OAuth2Credentials): auth_type = "oauth2" auth_details = { "has_token": bool(credentials.access_token.get_secret_value()), "scopes": credentials.scopes or [], } else: auth_type = "unknown" auth_details = {} yield "result", f"Performed {input_data.operation} with {auth_type}" yield "auth_type", auth_type yield "auth_details", auth_details # Test with API key api_creds = APIKeyCredentials( id="mixed-api-creds", provider="mixed_auth_provider", api_key=SecretStr("sk-1234567890"), title="Mixed API Key", ) block = MixedAuthBlock() outputs = {} async for name, value in block.run( MixedAuthBlock.Input( credentials={ # type: ignore "provider": "mixed_auth_provider", "id": "mixed-api-creds", "type": "api_key", }, operation="fetch_data", ), credentials=api_creds, ): outputs[name] = value assert outputs["auth_type"] == "api_key" assert outputs["result"] == "Performed fetch_data with api_key" assert outputs["auth_details"]["key_prefix"] == "sk-12..." # Test with OAuth2 oauth_creds = OAuth2Credentials( id="mixed-oauth-creds", provider="mixed_auth_provider", access_token=SecretStr("oauth-token-123"), scopes=["full_access"], title="Mixed OAuth", ) outputs = {} async for name, value in block.run( MixedAuthBlock.Input( credentials={ # type: ignore "provider": "mixed_auth_provider", "id": "mixed-oauth-creds", "type": "oauth2", }, operation="update_data", ), credentials=oauth_creds, ): outputs[name] = value assert outputs["auth_type"] == "oauth2" assert outputs["result"] == "Performed update_data with oauth2" assert outputs["auth_details"]["scopes"] == ["full_access"] @pytest.mark.asyncio async def test_multiple_credentials_block(self): """Test block requiring multiple different credentials.""" from backend.sdk import ProviderBuilder # Create multiple providers primary_provider = ( ProviderBuilder("primary_service") .with_api_key("PRIMARY_API_KEY", "Primary Service Key") .build() ) # For testing purposes, using API key instead of OAuth handler secondary_provider = ( ProviderBuilder("secondary_service") .with_api_key("SECONDARY_API_KEY", "Secondary Service Key") .build() ) class MultiCredentialBlock(Block): """Block requiring credentials from multiple services.""" class Input(BlockSchema): primary_credentials: CredentialsMetaInput = ( primary_provider.credentials_field( description="Primary service API key" ) ) secondary_credentials: CredentialsMetaInput = ( secondary_provider.credentials_field( description="Secondary service OAuth" ) ) merge_data: bool = SchemaField( description="Whether to merge data from both services", default=True, ) class Output(BlockSchema): primary_data: str = SchemaField(description="Data from primary service") secondary_data: str = SchemaField( description="Data from secondary service" ) merged_result: Optional[str] = SchemaField( description="Merged data if requested" ) def __init__(self): super().__init__( id="multi-credential-block", description="Block using multiple credentials", categories={BlockCategory.DEVELOPER_TOOLS}, input_schema=MultiCredentialBlock.Input, output_schema=MultiCredentialBlock.Output, ) async def run( self, input_data: Input, *, primary_credentials: APIKeyCredentials, secondary_credentials: OAuth2Credentials, **kwargs, ) -> BlockOutput: # Simulate fetching data with primary API key primary_data = f"Primary data using {primary_credentials.provider}" yield "primary_data", primary_data # Simulate fetching data with secondary OAuth secondary_data = f"Secondary data with {len(secondary_credentials.scopes or [])} scopes" yield "secondary_data", secondary_data # Merge if requested if input_data.merge_data: merged = f"{primary_data} + {secondary_data}" yield "merged_result", merged else: yield "merged_result", None # Create test credentials primary_creds = APIKeyCredentials( id="primary-creds", provider="primary_service", api_key=SecretStr("primary-key-123"), title="Primary Key", ) secondary_creds = OAuth2Credentials( id="secondary-creds", provider="secondary_service", access_token=SecretStr("secondary-token"), scopes=["read", "write"], title="Secondary OAuth", ) # Test the block block = MultiCredentialBlock() outputs = {} # Note: In real usage, the framework would inject the correct credentials # based on the field names. Here we simulate that behavior. async for name, value in block.run( MultiCredentialBlock.Input( primary_credentials={ # type: ignore "provider": "primary_service", "id": "primary-creds", "type": "api_key", }, secondary_credentials={ # type: ignore "provider": "secondary_service", "id": "secondary-creds", "type": "oauth2", }, merge_data=True, ), primary_credentials=primary_creds, secondary_credentials=secondary_creds, ): outputs[name] = value assert outputs["primary_data"] == "Primary data using primary_service" assert outputs["secondary_data"] == "Secondary data with 2 scopes" assert "Primary data" in outputs["merged_result"] assert "Secondary data" in outputs["merged_result"] @pytest.mark.asyncio async def test_oauth_scope_validation(self): """Test OAuth scope validation and handling.""" from backend.sdk import OAuth2Credentials, ProviderBuilder # Provider with specific required scopes # For testing OAuth scope validation scoped_provider = ( ProviderBuilder("scoped_oauth_service") .with_api_key("SCOPED_OAUTH_KEY", "Scoped OAuth Service") .build() ) class ScopeValidationBlock(Block): """Block that validates OAuth scopes.""" class Input(BlockSchema): credentials: CredentialsMetaInput = scoped_provider.credentials_field( description="OAuth credentials with specific scopes", scopes=["user:read", "user:write"], # Required scopes ) require_admin: bool = SchemaField( description="Whether admin scopes are required", default=False, ) class Output(BlockSchema): allowed_operations: list[str] = SchemaField( description="Operations allowed with current scopes" ) missing_scopes: list[str] = SchemaField( description="Scopes that are missing for full access" ) has_required_scopes: bool = SchemaField( description="Whether all required scopes are present" ) def __init__(self): super().__init__( id="scope-validation-block", description="Block that validates OAuth scopes", categories={BlockCategory.DEVELOPER_TOOLS}, input_schema=ScopeValidationBlock.Input, output_schema=ScopeValidationBlock.Output, ) async def run( self, input_data: Input, *, credentials: OAuth2Credentials, **kwargs ) -> BlockOutput: current_scopes = set(credentials.scopes or []) required_scopes = {"user:read", "user:write"} if input_data.require_admin: required_scopes.update({"admin:read", "admin:write"}) # Determine allowed operations based on scopes allowed_ops = [] if "user:read" in current_scopes: allowed_ops.append("read_user_data") if "user:write" in current_scopes: allowed_ops.append("update_user_data") if "admin:read" in current_scopes: allowed_ops.append("read_admin_data") if "admin:write" in current_scopes: allowed_ops.append("update_admin_data") missing = list(required_scopes - current_scopes) has_required = len(missing) == 0 yield "allowed_operations", allowed_ops yield "missing_scopes", missing yield "has_required_scopes", has_required # Test with partial scopes partial_creds = OAuth2Credentials( id="partial-oauth", provider="scoped_oauth_service", access_token=SecretStr("partial-token"), scopes=["user:read"], # Only one of the required scopes title="Partial OAuth", ) block = ScopeValidationBlock() outputs = {} async for name, value in block.run( ScopeValidationBlock.Input( credentials={ # type: ignore "provider": "scoped_oauth_service", "id": "partial-oauth", "type": "oauth2", }, require_admin=False, ), credentials=partial_creds, ): outputs[name] = value assert outputs["allowed_operations"] == ["read_user_data"] assert "user:write" in outputs["missing_scopes"] assert outputs["has_required_scopes"] is False # Test with all required scopes full_creds = OAuth2Credentials( id="full-oauth", provider="scoped_oauth_service", access_token=SecretStr("full-token"), scopes=["user:read", "user:write", "admin:read"], title="Full OAuth", ) outputs = {} async for name, value in block.run( ScopeValidationBlock.Input( credentials={ # type: ignore "provider": "scoped_oauth_service", "id": "full-oauth", "type": "oauth2", }, require_admin=False, ), credentials=full_creds, ): outputs[name] = value assert set(outputs["allowed_operations"]) == { "read_user_data", "update_user_data", "read_admin_data", } assert outputs["missing_scopes"] == [] assert outputs["has_required_scopes"] is True if __name__ == "__main__": pytest.main([__file__, "-v"]) Error Handling Generic exception handling catches all exceptions and yields error messages without proper logging or specific error types. This could mask important debugging information and make troubleshooting difficult. try: response = await Requests().post(url, headers=headers, json=payload) data = response.json() yield "webset_id", data.get("id", "") yield "status", data.get("status", "") yield "external_id", data.get("externalId") yield "created_at", data.get("createdAt", "") except Exception as e: yield "error", str(e) yield "webset_id", "" yield "status", "" yield "created_at", "" Performance Risk The new get_block_error_stats function uses raw SQL with aggregation but lacks proper indexing considerations and could be slow on large datasets. The HAVING clause with COUNT(*) >= 10 is hardcoded. async def get_block_error_stats( start_time: datetime, end_time: datetime ) -> list[BlockErrorStats]: """Get block execution stats using efficient SQL aggregation.""" query_template = """ SELECT n."agentBlockId" as block_id, COUNT(*) as total_executions, SUM(CASE WHEN ne."executionStatus" = 'FAILED' THEN 1 ELSE 0 END) as failed_executions FROM {schema_prefix}"AgentNodeExecution" ne JOIN {schema_prefix}"AgentNode" n ON ne."agentNodeId" = n.id WHERE ne."addedTime" >= $1::timestamp AND ne."addedTime" <= $2::timestamp GROUP BY n."agentBlockId" HAVING COUNT(*) >= 10 """ result = await query_raw_with_schema(query_template, start_time, end_time) # Convert to typed data structures return [ BlockErrorStats( block_id=row["block_id"], total_executions=int(row["total_executions"]), failed_executions=int(row["failed_executions"]), ) for row in result ]

[deepsource-io]

Here's the code health analysis summary for commits 3fe88b6..7410500. View details on DeepSource ↗.

Analysis Summary

Analyzer	Status	Summary	Link
JavaScript	✅ Success	❗ 22 occurences introduced 🎯 3 occurences resolved	View Check ↗
Python	✅ Success	❗ 9 occurences introduced 🎯 8 occurences resolved	View Check ↗

---

💡 If you’re a repository administrator, you can configure the quality gates from the settings.

[AutoGPT-Agent]

Thank you for adding these comprehensive e2e tests for the library page! This is valuable work that will help ensure the library functionality remains stable. However, there are a few issues that need to be addressed before this PR can be merged:

Title Format

The PR title has a typo ('frontent' should be 'frontend') and should use the singular 'test' instead of 'tests' to follow the conventional commit format. Please update to:

test(frontend): e2e tests for library page

Missing Checklist

Your PR is missing the required checklist. Since you're adding substantial new test code, please include the complete checklist from the PR template, with appropriate items checked off.

Potentially Out of Scope Changes

I noticed a couple of changes that appear unrelated to adding e2e tests:

1. In client.ts and helpers.ts, you're changing return types from response.text() to response.json(). This seems like a functional change rather than a test-only change. If this is required for the tests to work, please explain the connection in your PR description.

2. You've modified the CI workflow file to run only your specific test file. This change should be temporary for development only and should be reverted before merging.

Data-testid Additions

You've added several data-testid attributes to components. This is good for testing, but ensure these IDs follow any project conventions for test IDs.

Once you've addressed these items, this PR will be ready for another review. The test implementation itself looks thorough and well-structured!

[AutoGPT-Agent]

Thanks for adding these comprehensive e2e tests for the library page! The tests look thorough and well-structured, covering many important aspects of the library functionality.

A few observations:

1. The PR title has a typo: "tests(frontent)" should be "test(frontend)" - both to fix the spelling and to use the singular "test" for conventional commit format

2. The changes to the API client (changing from .text() to .json() in two places) are small but functional changes. These seem necessary for the tests to work correctly, but it would be good to briefly mention these in the description under the Changes section

3. The test names and organization look good, and I appreciate the detailed comments explaining what each test is checking

4. I see that you've added a new utility for creating users at the start, which seems like a useful addition for other tests

5. In the CI workflow, I noticed you're updating platform-frontend-ci.yml to only run the library.spec.ts test. Is this intended to be committed, or is it just for your local testing?

Overall, this is a great addition that will improve test coverage of the library page. Once you've addressed the PR title and clarified the intended changes to the workflow, this should be ready to merge.

[github-actions]

This pull request has conflicts with the base branch, please resolve those so we can evaluate the pull request.

[github-actions]

Conflicts have been resolved! 🎉 A maintainer will review the pull request shortly.