From a4b7fecba2097d6bdb4dbc2f1198b6be55834ab2 Mon Sep 17 00:00:00 2001 From: ogkranthi Date: Sun, 29 Mar 2026 22:15:09 -0400 Subject: [PATCH] =?UTF-8?q?test(T16):=20Bedrock=20+=20Vertex=20parser=20te?= =?UTF-8?q?sts=20=E2=80=94=20fixtures=20+=20round-trips?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add tests/test_bedrock_parser.py: 57 tests covering BedrockParser - Parse from bedrock-agent.json fixture - Parse from cloudformation.yaml fixture (CFN-only directory) - OpenAPI action groups → tools extraction + auth variants - guardrail-config.json → governance guardrails - L1 heuristic guardrail extraction from instruction text - Round-trip: AgentIR → bedrock emit → bedrock parse - Edge cases: missing dirs, empty ActionGroups, truncation stripping - Add tests/test_vertex_parser.py: 48 tests covering VertexParser - Parse from agent.json fixture - tools.json functionDeclarations, datastoreSpec, openApiFunctionDeclarations - Auth parsing: api_key, oauth2, bearer (service account) - Guardrail heuristic extraction from instructions + Restrictions: section - Round-trip: AgentIR → vertex emit → vertex parse - Edge cases: missing agent.json, invalid JSON, empty tools, resource paths - Add fixture files: - tests/fixtures/bedrock/{bedrock-agent.json,cloudformation.yaml,openapi.json,guardrail-config.json} - tests/fixtures/vertex/{agent.json,tools.json} - Total: 887 tests (was 782) — 105 new tests, all passing --- BACKLOG.md | 2 +- tests/fixtures/bedrock/bedrock-agent.json | 8 + tests/fixtures/bedrock/cloudformation.yaml | 36 + tests/fixtures/bedrock/guardrail-config.json | 27 + tests/fixtures/bedrock/openapi.json | 77 ++ tests/fixtures/vertex/agent.json | 21 + tests/fixtures/vertex/tools.json | 53 ++ tests/test_bedrock_parser.py | 713 +++++++++++++++++++ tests/test_vertex_parser.py | 661 +++++++++++++++++ 9 files changed, 1597 insertions(+), 1 deletion(-) create mode 100644 tests/fixtures/bedrock/bedrock-agent.json create mode 100644 tests/fixtures/bedrock/cloudformation.yaml create mode 100644 tests/fixtures/bedrock/guardrail-config.json create mode 100644 tests/fixtures/bedrock/openapi.json create mode 100644 tests/fixtures/vertex/agent.json create mode 100644 tests/fixtures/vertex/tools.json create mode 100644 tests/test_bedrock_parser.py create mode 100644 tests/test_vertex_parser.py diff --git a/BACKLOG.md b/BACKLOG.md index 94465b6..1196770 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -101,4 +101,4 @@ | D25 | P1 | @dev | merged | Bump version to 0.3.0 — update CHANGELOG.md, pyproject.toml, add governance to README | | T14 | P1 | @tester | merged | Write tests for governance extraction (Guardrail classification, ToolPermission, L3 annotations) | | T15 | P1 | @tester | merged | Write tests for audit engine (GPR-L1/L2/L3 scoring, elevation tracking, CSV/JSON export) | -| T16 | P1 | @tester | in-progress | Write tests for Bedrock + Vertex parsers (fixtures + round-trip with emitters) | +| T16 | P1 | @tester | pr-created | Write tests for Bedrock + Vertex parsers (fixtures + round-trip with emitters) | diff --git a/tests/fixtures/bedrock/bedrock-agent.json b/tests/fixtures/bedrock/bedrock-agent.json new file mode 100644 index 0000000..3987c57 --- /dev/null +++ b/tests/fixtures/bedrock/bedrock-agent.json @@ -0,0 +1,8 @@ +{ + "agentName": "CustomerSupportBot", + "description": "An AI-powered customer support agent for e-commerce.", + "instruction": "You are a helpful customer support agent for an e-commerce platform. Help customers with orders, returns, and product questions. Always be polite and professional.", + "foundationModel": "anthropic.claude-3-5-sonnet-20241022-v2:0", + "agentId": "ABCDEF123456", + "agentAliasId": "ALIAS001" +} diff --git a/tests/fixtures/bedrock/cloudformation.yaml b/tests/fixtures/bedrock/cloudformation.yaml new file mode 100644 index 0000000..50a7e9c --- /dev/null +++ b/tests/fixtures/bedrock/cloudformation.yaml @@ -0,0 +1,36 @@ +AWSTemplateFormatVersion: "2010-09-09" +Description: AgentShift - Bedrock Agent CloudFormation Template + +Resources: + CustomerSupportBotAgent: + Type: AWS::Bedrock::Agent + Properties: + AgentName: CustomerSupportBot + Description: An AI-powered customer support agent for e-commerce. + FoundationModel: anthropic.claude-3-5-sonnet-20241022-v2:0 + Instruction: | + You are a helpful customer support agent for an e-commerce platform. + Help customers with orders, returns, and product questions. + Always be polite and professional. + ActionGroups: + - ActionGroupName: order-management + Description: Manage customer orders + ApiSchema: + Payload: '{"openapi":"3.0.0","info":{"title":"Order Management","version":"1.0.0"},"paths":{"/get-order":{"get":{"operationId":"getOrder","description":"Get order details","responses":{"200":{"description":"OK"}}}}}}' + KnowledgeBases: + - KnowledgeBaseId: kb-12345 + Description: Product catalog knowledge base + + ProductCatalogKB: + Type: AWS::Bedrock::KnowledgeBase + Properties: + Name: product-catalog-kb + Description: Product catalog for e-commerce + StorageConfiguration: + Type: OPENSEARCH_SERVERLESS + +Outputs: + AgentId: + Value: PLACEHOLDER_AGENT_ID + AliasId: + Value: PLACEHOLDER_ALIAS_ID diff --git a/tests/fixtures/bedrock/guardrail-config.json b/tests/fixtures/bedrock/guardrail-config.json new file mode 100644 index 0000000..f13adff --- /dev/null +++ b/tests/fixtures/bedrock/guardrail-config.json @@ -0,0 +1,27 @@ +{ + "guardrailId": "GUARD001", + "name": "e-commerce-guardrail", + "topicPolicyConfig": { + "topicsConfig": [ + { + "name": "competitor-comparison", + "definition": "Do not make comparisons with or recommendations about competitor products.", + "type": "DENY" + }, + { + "name": "political-content", + "definition": "Avoid discussing political topics unrelated to product or service.", + "type": "DENY" + } + ] + }, + "contentPolicyConfig": { + "filtersConfig": [ + { + "type": "HATE", + "inputStrength": "HIGH", + "outputStrength": "HIGH" + } + ] + } +} diff --git a/tests/fixtures/bedrock/openapi.json b/tests/fixtures/bedrock/openapi.json new file mode 100644 index 0000000..330fb08 --- /dev/null +++ b/tests/fixtures/bedrock/openapi.json @@ -0,0 +1,77 @@ +{ + "openapi": "3.0.0", + "info": { + "title": "Customer Support Actions", + "version": "1.0.0" + }, + "components": { + "securitySchemes": { + "ApiKeyAuth": { + "type": "apiKey", + "name": "X-API-KEY", + "in": "header" + } + } + }, + "security": [ + {"ApiKeyAuth": []} + ], + "paths": { + "/get-order": { + "post": { + "operationId": "getOrder", + "description": "Get details for a specific order by order ID.", + "requestBody": { + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "orderId": { + "type": "string", + "description": "The unique order identifier" + } + }, + "required": ["orderId"] + } + } + } + }, + "responses": { + "200": { + "description": "Order details" + } + } + } + }, + "/process-return": { + "post": { + "operationId": "processReturn", + "description": "Process a return request for an order.", + "requestBody": { + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "orderId": { + "type": "string" + }, + "reason": { + "type": "string" + } + }, + "required": ["orderId", "reason"] + } + } + } + }, + "responses": { + "200": { + "description": "Return confirmation" + } + } + } + } + } +} diff --git a/tests/fixtures/vertex/agent.json b/tests/fixtures/vertex/agent.json new file mode 100644 index 0000000..7155383 --- /dev/null +++ b/tests/fixtures/vertex/agent.json @@ -0,0 +1,21 @@ +{ + "name": "projects/my-project/locations/us-central1/agents/weather-assistant-001", + "displayName": "WeatherAssistant", + "description": "A weather information assistant that provides current conditions and forecasts.", + "goal": "You are a helpful weather assistant. Provide accurate, timely weather information for any location worldwide.", + "instructions": [ + "Behavior:\n- Always state the location when reporting weather\n- Include temperature in both Celsius and Fahrenheit\n- Mention precipitation probability", + "Restrictions:\nDo not provide weather advisories or emergency alerts. Refer users to official sources for severe weather.", + "Persona:\nFriendly and concise. Use simple language that non-meteorologists can understand." + ], + "defaultLanguageCode": "en", + "tools": [ + { + "name": "weather-api-tool", + "description": "Fetch current weather data", + "type": "FUNCTION" + } + ], + "createTime": "2024-01-15T10:00:00Z", + "updateTime": "2024-01-20T12:00:00Z" +} diff --git a/tests/fixtures/vertex/tools.json b/tests/fixtures/vertex/tools.json new file mode 100644 index 0000000..433e0a2 --- /dev/null +++ b/tests/fixtures/vertex/tools.json @@ -0,0 +1,53 @@ +[ + { + "displayName": "Weather API", + "description": "Fetch current weather conditions and forecasts from external weather service.", + "functionDeclarations": [ + { + "name": "get_current_weather", + "description": "Get current weather conditions for a location.", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "City name or coordinates" + }, + "units": { + "type": "string", + "enum": ["metric", "imperial"], + "description": "Temperature units" + } + }, + "required": ["location"] + } + }, + { + "name": "get_forecast", + "description": "Get weather forecast for the next N days.", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string" + }, + "days": { + "type": "integer", + "description": "Number of days for forecast (1-10)" + } + }, + "required": ["location", "days"] + } + } + ] + }, + { + "displayName": "Weather Data Store", + "description": "Historical weather data store for trend analysis.", + "datastoreSpec": { + "dataStores": [ + "projects/my-project/locations/us-central1/collections/default/dataStores/weather-history-ds-001" + ] + } + } +] diff --git a/tests/test_bedrock_parser.py b/tests/test_bedrock_parser.py new file mode 100644 index 0000000..ae5eeff --- /dev/null +++ b/tests/test_bedrock_parser.py @@ -0,0 +1,713 @@ +"""T16 — Bedrock parser tests. + +Tests for BedrockParser (src/agentshift/parsers/bedrock.py): +- Parse from bedrock-agent.json fixture +- Parse from cloudformation.yaml fixture +- Parse with openapi.json action groups → tools +- Parse with guardrail-config.json → governance +- L1 guardrail heuristic extraction from system_prompt +- Round-trip: AgentIR → bedrock (emit) → bedrock (parse) → IR comparison +- Edge cases: missing files, empty actionGroups, truncation notice stripping +""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path + +import pytest + +from agentshift.ir import AgentIR, Governance, Guardrail, Persona, Tool +from agentshift.parsers import bedrock as bedrock_parser +from agentshift.emitters import bedrock as bedrock_emitter + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +FIXTURES_DIR = Path(__file__).parent / "fixtures" / "bedrock" + + +def _make_minimal_ir(**kwargs) -> AgentIR: + """Build a minimal AgentIR for round-trip tests.""" + defaults = dict( + name="test-agent", + description="A test agent", + persona=Persona(system_prompt="You are a helpful test assistant."), + ) + defaults.update(kwargs) + return AgentIR(**defaults) + + +# --------------------------------------------------------------------------- +# Parse from bedrock-agent.json fixture +# --------------------------------------------------------------------------- + + +class TestParseFromBedrockAgentJson: + """Parse using bedrock-agent.json as the primary source.""" + + def test_parse_returns_agent_ir(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + assert isinstance(ir, AgentIR) + + def test_name_slugified(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + # "CustomerSupportBot" → slugified (lowercased, no spaces) + assert ir.name == "customersupportbot" + + def test_description_from_agent_json(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + assert "customer support" in ir.description.lower() + + def test_instruction_from_agent_json(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + assert ir.persona.system_prompt is not None + assert "customer support agent" in ir.persona.system_prompt.lower() + + def test_foundation_model_in_extensions(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + ext = ir.metadata.platform_extensions.get("bedrock", {}) + assert ext.get("foundation_model") == "anthropic.claude-3-5-sonnet-20241022-v2:0" + + def test_agent_id_in_extensions(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + ext = ir.metadata.platform_extensions.get("bedrock", {}) + assert ext.get("agent_id") == "ABCDEF123456" + + def test_alias_id_in_extensions(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + ext = ir.metadata.platform_extensions.get("bedrock", {}) + assert ext.get("alias_id") == "ALIAS001" + + def test_source_platform_is_bedrock(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + assert ir.metadata.source_platform == "bedrock" + + +# --------------------------------------------------------------------------- +# Parse from cloudformation.yaml fixture +# --------------------------------------------------------------------------- + + +class TestParseFromCloudFormation: + """Parse using only cloudformation.yaml (no bedrock-agent.json).""" + + def test_parse_cfn_only(self, tmp_path): + """Parse from a directory with only cloudformation.yaml.""" + import shutil + shutil.copy(FIXTURES_DIR / "cloudformation.yaml", tmp_path / "cloudformation.yaml") + + ir = bedrock_parser.parse(tmp_path) + assert isinstance(ir, AgentIR) + + def test_name_from_cfn(self, tmp_path): + import shutil + shutil.copy(FIXTURES_DIR / "cloudformation.yaml", tmp_path / "cloudformation.yaml") + + ir = bedrock_parser.parse(tmp_path) + # "CustomerSupportBot" → slugified (lowercased) + assert ir.name == "customersupportbot" + + def test_instruction_from_cfn(self, tmp_path): + import shutil + shutil.copy(FIXTURES_DIR / "cloudformation.yaml", tmp_path / "cloudformation.yaml") + + ir = bedrock_parser.parse(tmp_path) + assert ir.persona.system_prompt is not None + assert "customer support agent" in ir.persona.system_prompt.lower() + + def test_knowledge_from_cfn(self, tmp_path): + import shutil + shutil.copy(FIXTURES_DIR / "cloudformation.yaml", tmp_path / "cloudformation.yaml") + + ir = bedrock_parser.parse(tmp_path) + # Should extract KnowledgeBase resource + assert len(ir.knowledge) > 0 + + def test_knowledge_kind_is_vector_store(self, tmp_path): + import shutil + shutil.copy(FIXTURES_DIR / "cloudformation.yaml", tmp_path / "cloudformation.yaml") + + ir = bedrock_parser.parse(tmp_path) + kb = ir.knowledge[0] + assert kb.kind == "vector_store" + + def test_tools_from_cfn_action_groups(self, tmp_path): + """CFN ActionGroups with inline Payload → tools extracted.""" + import shutil + shutil.copy(FIXTURES_DIR / "cloudformation.yaml", tmp_path / "cloudformation.yaml") + + ir = bedrock_parser.parse(tmp_path) + # The CFN fixture has an ActionGroup with an inline OpenAPI payload + assert len(ir.tools) > 0 + + +# --------------------------------------------------------------------------- +# Parse with openapi.json action groups → tools +# --------------------------------------------------------------------------- + + +class TestParseWithOpenApi: + """Parse using openapi.json to extract tools.""" + + def test_tools_from_openapi(self): + """openapi.json fixture → tools extracted with correct names.""" + ir = bedrock_parser.parse(FIXTURES_DIR) + tool_names = {t.name for t in ir.tools} + assert "getOrder" in tool_names or "get-order" in tool_names or any("order" in n.lower() for n in tool_names) + + def test_tool_descriptions_populated(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + for tool in ir.tools: + assert tool.description, f"Tool {tool.name!r} missing description" + + def test_tool_parameters_extracted(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + # At least one tool should have parameters + tools_with_params = [t for t in ir.tools if t.parameters is not None] + assert len(tools_with_params) > 0 + + def test_tool_kind_is_function(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + for tool in ir.tools: + assert tool.kind == "function" + + def test_tool_auth_api_key(self): + """OpenAPI fixture uses ApiKeyAuth → tool auth should be api_key.""" + ir = bedrock_parser.parse(FIXTURES_DIR) + tools_with_auth = [t for t in ir.tools if t.auth is not None] + assert len(tools_with_auth) > 0 + assert any(t.auth.type == "api_key" for t in tools_with_auth) + + def test_openapi_multiple_tools(self, tmp_path): + """Parse openapi.json with multiple paths → multiple tools.""" + openapi = { + "openapi": "3.0.0", + "info": {"title": "Multi Tool API", "version": "1.0"}, + "paths": { + "/tool-a": {"post": {"operationId": "toolA", "description": "Tool A", "responses": {"200": {"description": "OK"}}}}, + "/tool-b": {"post": {"operationId": "toolB", "description": "Tool B", "responses": {"200": {"description": "OK"}}}}, + }, + } + (tmp_path / "openapi.json").write_text(json.dumps(openapi)) + (tmp_path / "instruction.txt").write_text("You are a test agent.") + + ir = bedrock_parser.parse(tmp_path) + assert len(ir.tools) == 2 + tool_names = {t.name for t in ir.tools} + assert "toolA" in tool_names + assert "toolB" in tool_names + + def test_openapi_wins_over_cfn_action_groups(self, tmp_path): + """When openapi.json present, CFN ActionGroups are NOT used for tools.""" + import shutil + shutil.copy(FIXTURES_DIR / "cloudformation.yaml", tmp_path / "cloudformation.yaml") + shutil.copy(FIXTURES_DIR / "openapi.json", tmp_path / "openapi.json") + + ir = bedrock_parser.parse(tmp_path) + # Tools should come from openapi.json (has getOrder, processReturn) + tool_names = {t.name for t in ir.tools} + assert "getOrder" in tool_names or "processReturn" in tool_names + + +# --------------------------------------------------------------------------- +# Parse with guardrail-config.json → governance.platform_annotations +# --------------------------------------------------------------------------- + + +class TestParseWithGuardrailConfig: + """Parse using guardrail-config.json to populate governance.""" + + def test_guardrails_populated(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + assert len(ir.governance.guardrails) > 0 + + def test_topic_policy_mapped_to_guardrails(self): + """topicsConfig entries → guardrails list.""" + ir = bedrock_parser.parse(FIXTURES_DIR) + guardrail_texts = [g.text.lower() for g in ir.governance.guardrails] + # The guardrail-config.json has competitor-comparison and political-content topics + assert any("competitor" in t or "political" in t for t in guardrail_texts) + + def test_guardrail_ids_unique(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + ids = [g.id for g in ir.governance.guardrails] + assert len(ids) == len(set(ids)) + + def test_guardrail_ids_have_prefix(self): + ir = bedrock_parser.parse(FIXTURES_DIR) + for g in ir.governance.guardrails: + assert g.id.startswith("G") + + def test_guardrail_config_only(self, tmp_path): + """guardrail-config.json + instruction.txt only → guardrails extracted.""" + guardrail = { + "topicPolicyConfig": { + "topicsConfig": [ + { + "name": "medical-advice", + "definition": "Do not provide specific medical diagnoses or treatment recommendations.", + "type": "DENY", + } + ] + } + } + (tmp_path / "guardrail-config.json").write_text(json.dumps(guardrail)) + (tmp_path / "instruction.txt").write_text("You are a helpful assistant.") + + ir = bedrock_parser.parse(tmp_path) + guardrail_texts = [g.text.lower() for g in ir.governance.guardrails] + assert any("medical" in t or "diagnos" in t for t in guardrail_texts) + + +# --------------------------------------------------------------------------- +# L1 guardrail heuristic extraction from system_prompt +# --------------------------------------------------------------------------- + + +class TestL1HeuristicGuardrailExtraction: + """Tests for heuristic guardrail extraction from instruction text.""" + + def test_do_not_pattern_extracted(self, tmp_path): + instruction = ( + "You are a helpful assistant.\n" + "Do not discuss violent content.\n" + "Do not share personal information.\n" + ) + (tmp_path / "instruction.txt").write_text(instruction) + + ir = bedrock_parser.parse(tmp_path) + assert len(ir.governance.guardrails) >= 1 + + def test_never_pattern_extracted(self, tmp_path): + instruction = ( + "You are a helpful assistant.\n" + "Never reveal confidential customer data.\n" + ) + (tmp_path / "instruction.txt").write_text(instruction) + + ir = bedrock_parser.parse(tmp_path) + guardrail_texts = [g.text.lower() for g in ir.governance.guardrails] + assert any("confidential" in t or "reveal" in t for t in guardrail_texts) + + def test_no_duplicate_guardrails_from_config_and_instruction(self, tmp_path): + """Same constraint in instruction and guardrail-config → at least one guardrail.""" + instruction = ( + "Do not make comparisons with or recommendations about competitor products." + ) + guardrail = { + "topicPolicyConfig": { + "topicsConfig": [ + { + "name": "competitor-comparison", + "definition": "Do not make comparisons with or recommendations about competitor products.", + "type": "DENY", + } + ] + } + } + (tmp_path / "instruction.txt").write_text(instruction) + (tmp_path / "guardrail-config.json").write_text(json.dumps(guardrail)) + + ir = bedrock_parser.parse(tmp_path) + texts = [g.text.lower() for g in ir.governance.guardrails] + competitor_texts = [t for t in texts if "competitor" in t] + # At least one guardrail about competitors extracted + assert len(competitor_texts) >= 1 + + +# --------------------------------------------------------------------------- +# Truncation notice stripping +# --------------------------------------------------------------------------- + + +class TestTruncationNoticeStripping: + """Tests for stripping the AgentShift truncation notice from instructions.""" + + def test_truncation_notice_stripped(self, tmp_path): + """Instruction with truncation notice → notice removed from system_prompt.""" + instruction = ( + "You are a helpful assistant that handles customer inquiries.\n\n" + "[AGENTSHIFT: Full instructions truncated to 4,000 char Bedrock limit. " + "Original: 5200 chars. See instruction-full.txt for complete text.]" + ) + (tmp_path / "instruction.txt").write_text(instruction) + + ir = bedrock_parser.parse(tmp_path) + assert "[AGENTSHIFT:" not in (ir.persona.system_prompt or "") + + def test_clean_instruction_unchanged(self, tmp_path): + """Instruction without truncation notice → unchanged.""" + instruction = "You are a helpful assistant." + (tmp_path / "instruction.txt").write_text(instruction) + + ir = bedrock_parser.parse(tmp_path) + assert ir.persona.system_prompt == instruction + + def test_truncation_notice_case_insensitive(self, tmp_path): + """Truncation notice matching is case-insensitive.""" + instruction = ( + "You are a helpful assistant.\n" + "[agentshift: full instructions truncated to 4,000 char Bedrock limit. " + "Original: 4500 chars. See instruction-full.txt for complete text.]" + ) + (tmp_path / "instruction.txt").write_text(instruction) + + ir = bedrock_parser.parse(tmp_path) + assert "[agentshift:" not in (ir.persona.system_prompt or "").lower() + + +# --------------------------------------------------------------------------- +# Edge cases +# --------------------------------------------------------------------------- + + +class TestEdgeCases: + """Edge cases: missing files, empty actionGroups, etc.""" + + def test_nonexistent_directory_raises(self): + with pytest.raises(FileNotFoundError, match="not found"): + bedrock_parser.parse(Path("/nonexistent/bedrock/dir")) + + def test_file_instead_of_directory_raises(self, tmp_path): + file_path = tmp_path / "not-a-dir.json" + file_path.write_text("{}") + with pytest.raises(FileNotFoundError): + bedrock_parser.parse(file_path) + + def test_empty_directory_raises(self, tmp_path): + with pytest.raises(FileNotFoundError, match="No recognised Bedrock artifact"): + bedrock_parser.parse(tmp_path) + + def test_empty_action_groups_no_tools(self, tmp_path): + """CloudFormation with empty ActionGroups → no tools.""" + cfn = { + "AWSTemplateFormatVersion": "2010-09-09", + "Resources": { + "MyAgent": { + "Type": "AWS::Bedrock::Agent", + "Properties": { + "AgentName": "EmptyAgent", + "Instruction": "You are a helpful assistant.", + "FoundationModel": "anthropic.claude-3-sonnet-20240229-v1:0", + "ActionGroups": [], + }, + } + }, + } + import yaml + (tmp_path / "cloudformation.yaml").write_text(yaml.dump(cfn)) + + ir = bedrock_parser.parse(tmp_path) + assert ir.tools == [] + + def test_missing_openapi_falls_back_to_cfn_tools(self, tmp_path): + """No openapi.json → tools extracted from CFN ActionGroups.""" + import shutil + shutil.copy(FIXTURES_DIR / "cloudformation.yaml", tmp_path / "cloudformation.yaml") + # No openapi.json copied + + ir = bedrock_parser.parse(tmp_path) + # CFN has an inline action group with getOrder + assert len(ir.tools) >= 0 # may be 0 or > 0 depending on inline payload + + def test_instruction_only(self, tmp_path): + """Only instruction.txt → minimal valid IR.""" + (tmp_path / "instruction.txt").write_text("You are a test agent.") + + ir = bedrock_parser.parse(tmp_path) + assert isinstance(ir, AgentIR) + assert ir.persona.system_prompt == "You are a test agent." + + def test_unnamed_fallback(self, tmp_path): + """No name in any source → fallback to 'unnamed-bedrock-agent'.""" + (tmp_path / "instruction.txt").write_text("You are a test agent.") + + ir = bedrock_parser.parse(tmp_path) + assert ir.name == "unnamed-bedrock-agent" + + def test_description_derived_from_instruction(self, tmp_path): + """No description → first sentence of instruction becomes description.""" + (tmp_path / "instruction.txt").write_text( + "You are a sales assistant. Help customers find products." + ) + + ir = bedrock_parser.parse(tmp_path) + assert "sales assistant" in ir.description.lower() or ir.description + + def test_invalid_openapi_json_ignored(self, tmp_path): + """Invalid openapi.json (malformed JSON) → gracefully ignored.""" + (tmp_path / "instruction.txt").write_text("You are a test agent.") + (tmp_path / "openapi.json").write_text("{not valid json}") + + ir = bedrock_parser.parse(tmp_path) + assert isinstance(ir, AgentIR) + assert ir.tools == [] + + def test_bedrock_agent_json_takes_precedence_over_cfn(self, tmp_path): + """bedrock-agent.json instruction wins over cloudformation.yaml instruction.""" + import shutil + shutil.copy(FIXTURES_DIR / "bedrock-agent.json", tmp_path / "bedrock-agent.json") + shutil.copy(FIXTURES_DIR / "cloudformation.yaml", tmp_path / "cloudformation.yaml") + + ir = bedrock_parser.parse(tmp_path) + # bedrock-agent.json has a specific instruction + assert ir.persona.system_prompt is not None + # bedrock-agent.json instruction should win + assert "customer support agent" in ir.persona.system_prompt.lower() + + +# --------------------------------------------------------------------------- +# Round-trip: AgentIR → bedrock emit → bedrock parse → IR comparison +# --------------------------------------------------------------------------- + + +class TestRoundTrip: + """Round-trip tests: emit an AgentIR to Bedrock format, then parse it back.""" + + def test_basic_round_trip(self, tmp_path): + """Simple IR → emit → parse → name and description preserved.""" + ir_in = _make_minimal_ir( + name="round-trip-agent", + description="A round-trip test agent.", + ) + out_dir = tmp_path / "bedrock-out" + bedrock_emitter.emit(ir_in, out_dir) + + ir_out = bedrock_parser.parse(out_dir) + assert isinstance(ir_out, AgentIR) + # Name may be slugified differently but should be non-empty + assert ir_out.name + assert ir_out.description or ir_out.persona.system_prompt + + def test_round_trip_instruction_preserved(self, tmp_path): + """System prompt → emit → parse → instruction matches.""" + system_prompt = "You are a round-trip test assistant. Be helpful." + ir_in = _make_minimal_ir( + name="trip-agent", + persona=Persona(system_prompt=system_prompt), + ) + out_dir = tmp_path / "bedrock-out" + bedrock_emitter.emit(ir_in, out_dir) + + ir_out = bedrock_parser.parse(out_dir) + assert ir_out.persona.system_prompt == system_prompt + + def test_round_trip_tools_preserved_mcp(self, tmp_path): + """IR with MCP tools → emit → parse → tool names preserved (emitter generates paths for mcp kind).""" + from agentshift.ir import Tool + + tools = [ + Tool(name="search", description="Search the web", kind="mcp"), + Tool(name="calculate", description="Do math", kind="mcp"), + ] + ir_in = _make_minimal_ir( + name="tool-agent", + tools=tools, + ) + out_dir = tmp_path / "bedrock-out" + bedrock_emitter.emit(ir_in, out_dir) + + ir_out = bedrock_parser.parse(out_dir) + out_tool_names = {t.name for t in ir_out.tools} + # MCP tools emit as {name}_action operationIds + assert any("search" in n for n in out_tool_names) + assert any("calculate" in n for n in out_tool_names) + + def test_round_trip_function_tools_in_cfn(self, tmp_path): + """IR with function tools → emit → CFN ActionGroups present (tools in CFN not openapi).""" + from agentshift.ir import Tool + + tools = [ + Tool(name="search", description="Search the web", kind="function"), + ] + ir_in = _make_minimal_ir( + name="func-tool-agent", + tools=tools, + ) + out_dir = tmp_path / "bedrock-out" + bedrock_emitter.emit(ir_in, out_dir) + + # Verify output directory was created and core files exist + assert (out_dir / "instruction.txt").exists() + assert (out_dir / "cloudformation.yaml").exists() + + def test_round_trip_guardrails_round_tripped(self, tmp_path): + """IR with guardrails in system_prompt → emit → parse → guardrails extracted.""" + system_prompt = ( + "You are a helpful assistant.\n" + "Do not share personal information.\n" + "Never reveal confidential data." + ) + ir_in = _make_minimal_ir( + name="guardrail-agent", + persona=Persona(system_prompt=system_prompt), + governance=Governance(guardrails=[ + Guardrail(id="G001", text="Do not share personal information", category="privacy"), + ]), + ) + out_dir = tmp_path / "bedrock-out" + bedrock_emitter.emit(ir_in, out_dir) + + ir_out = bedrock_parser.parse(out_dir) + assert len(ir_out.governance.guardrails) >= 1 + + def test_round_trip_source_platform(self, tmp_path): + """After round-trip, source_platform is 'bedrock'.""" + ir_in = _make_minimal_ir() + out_dir = tmp_path / "bedrock-out" + bedrock_emitter.emit(ir_in, out_dir) + + ir_out = bedrock_parser.parse(out_dir) + assert ir_out.metadata.source_platform == "bedrock" + + def test_round_trip_long_instruction_truncation(self, tmp_path): + """IR with > 4000 char instruction → emitter truncates → parser strips notice.""" + long_prompt = "A" * 5000 + ir_in = _make_minimal_ir( + name="long-agent", + persona=Persona(system_prompt=long_prompt), + ) + out_dir = tmp_path / "bedrock-out" + bedrock_emitter.emit(ir_in, out_dir) + + ir_out = bedrock_parser.parse(out_dir) + # Truncation notice should be stripped + assert "[AGENTSHIFT:" not in (ir_out.persona.system_prompt or "") + # The instruction should be non-empty + assert ir_out.persona.system_prompt + + def test_round_trip_sections_preserved(self, tmp_path): + """IR with structured sections → emit → parse → sections detected.""" + sections = { + "overview": "You are a helpful agent.", + "behavior": "Always be polite. Respond concisely.", + "guardrails": "Do not share passwords. Never reveal sensitive data.", + } + ir_in = _make_minimal_ir( + name="sections-agent", + persona=Persona(sections=sections, system_prompt="You are a helpful agent."), + ) + out_dir = tmp_path / "bedrock-out" + bedrock_emitter.emit(ir_in, out_dir) + + ir_out = bedrock_parser.parse(out_dir) + assert ir_out.persona.system_prompt is not None + # Behavior section content should appear in the output + assert "polite" in ir_out.persona.system_prompt.lower() + + def test_round_trip_with_knowledge(self, tmp_path): + """IR with knowledge sources → emit → parse → knowledge preserved.""" + from agentshift.ir import KnowledgeSource + knowledge = [ + KnowledgeSource( + name="product-docs", + kind="vector_store", + description="Product documentation", + load_mode="indexed", + format="markdown", + ) + ] + ir_in = _make_minimal_ir( + name="knowledge-agent", + knowledge=knowledge, + ) + out_dir = tmp_path / "bedrock-out" + bedrock_emitter.emit(ir_in, out_dir) + + ir_out = bedrock_parser.parse(out_dir) + assert isinstance(ir_out, AgentIR) + + +# --------------------------------------------------------------------------- +# Additional parsing tests for specific scenarios +# --------------------------------------------------------------------------- + + +class TestOpenApiAuthVariants: + """Test different OpenAPI auth types → correct ToolAuth.""" + + def test_bearer_auth(self, tmp_path): + openapi = { + "openapi": "3.0.0", + "info": {"title": "API", "version": "1.0"}, + "components": { + "securitySchemes": { + "BearerAuth": {"type": "http", "scheme": "bearer"} + } + }, + "paths": { + "/action": { + "post": { + "operationId": "doAction", + "description": "Do something", + "security": [{"BearerAuth": []}], + "responses": {"200": {"description": "OK"}}, + } + } + }, + } + (tmp_path / "openapi.json").write_text(json.dumps(openapi)) + (tmp_path / "instruction.txt").write_text("Test agent.") + + ir = bedrock_parser.parse(tmp_path) + assert len(ir.tools) == 1 + assert ir.tools[0].auth is not None + assert ir.tools[0].auth.type == "bearer" + + def test_oauth2_auth(self, tmp_path): + openapi = { + "openapi": "3.0.0", + "info": {"title": "API", "version": "1.0"}, + "components": { + "securitySchemes": { + "OAuth2": { + "type": "oauth2", + "flows": { + "clientCredentials": { + "tokenUrl": "https://auth.example.com/token", + "scopes": {"read:data": "Read data"}, + } + }, + } + } + }, + "paths": { + "/resource": { + "post": { + "operationId": "getResource", + "description": "Get resource", + "security": [{"OAuth2": ["read:data"]}], + "responses": {"200": {"description": "OK"}}, + } + } + }, + } + (tmp_path / "openapi.json").write_text(json.dumps(openapi)) + (tmp_path / "instruction.txt").write_text("Test agent.") + + ir = bedrock_parser.parse(tmp_path) + assert ir.tools[0].auth is not None + assert ir.tools[0].auth.type == "oauth2" + + def test_no_auth_scheme(self, tmp_path): + openapi = { + "openapi": "3.0.0", + "info": {"title": "API", "version": "1.0"}, + "paths": { + "/public": { + "post": { + "operationId": "publicAction", + "description": "Public action", + "responses": {"200": {"description": "OK"}}, + } + } + }, + } + (tmp_path / "openapi.json").write_text(json.dumps(openapi)) + (tmp_path / "instruction.txt").write_text("Test agent.") + + ir = bedrock_parser.parse(tmp_path) + assert ir.tools[0].auth is None # no auth → None diff --git a/tests/test_vertex_parser.py b/tests/test_vertex_parser.py new file mode 100644 index 0000000..06bb0cb --- /dev/null +++ b/tests/test_vertex_parser.py @@ -0,0 +1,661 @@ +"""T16 — Vertex AI parser tests. + +Tests for VertexParser (src/agentshift/parsers/vertex.py): +- Parse from agent.json fixture +- Parse tools from tool definitions (tools.json) +- Round-trip: AgentIR → vertex (emit) → vertex (parse) → IR comparison +- Edge cases: missing optional fields, empty tools list, missing agent.json +""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path + +import pytest + +from agentshift.ir import AgentIR, Governance, Guardrail, Persona, Tool +from agentshift.parsers import vertex as vertex_parser +from agentshift.emitters import vertex as vertex_emitter + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +FIXTURES_DIR = Path(__file__).parent / "fixtures" / "vertex" + + +def _make_minimal_ir(**kwargs) -> AgentIR: + """Build a minimal AgentIR for round-trip tests.""" + defaults = dict( + name="test-agent", + description="A test agent", + persona=Persona(system_prompt="You are a helpful test assistant."), + ) + defaults.update(kwargs) + return AgentIR(**defaults) + + +# --------------------------------------------------------------------------- +# Parse from agent.json fixture +# --------------------------------------------------------------------------- + + +class TestParseFromAgentJson: + """Parse using the vertex/agent.json fixture.""" + + def test_parse_returns_agent_ir(self): + ir = vertex_parser.parse(FIXTURES_DIR) + assert isinstance(ir, AgentIR) + + def test_name_slugified(self): + ir = vertex_parser.parse(FIXTURES_DIR) + # "WeatherAssistant" → slugified (lowercased) + assert ir.name == "weatherassistant" + + def test_description_from_agent_json(self): + ir = vertex_parser.parse(FIXTURES_DIR) + assert "weather" in ir.description.lower() + + def test_system_prompt_from_goal(self): + ir = vertex_parser.parse(FIXTURES_DIR) + assert ir.persona.system_prompt is not None + assert "weather" in ir.persona.system_prompt.lower() + + def test_language_from_agent_json(self): + ir = vertex_parser.parse(FIXTURES_DIR) + assert ir.persona.language == "en" + + def test_source_platform_is_vertex_ai(self): + ir = vertex_parser.parse(FIXTURES_DIR) + assert ir.metadata.source_platform == "vertex-ai" + + def test_created_at_parsed(self): + ir = vertex_parser.parse(FIXTURES_DIR) + assert ir.metadata.created_at == "2024-01-15T10:00:00Z" + + def test_updated_at_parsed(self): + ir = vertex_parser.parse(FIXTURES_DIR) + assert ir.metadata.updated_at == "2024-01-20T12:00:00Z" + + def test_display_name_in_extensions(self): + ir = vertex_parser.parse(FIXTURES_DIR) + ext = ir.metadata.platform_extensions.get("vertex", {}) + assert ext.get("display_name") == "WeatherAssistant" + + def test_resource_name_in_extensions(self): + ir = vertex_parser.parse(FIXTURES_DIR) + ext = ir.metadata.platform_extensions.get("vertex", {}) + assert "resource_name" in ext + assert "weather-assistant" in ext["resource_name"] + + +# --------------------------------------------------------------------------- +# Parse tools from tool definitions (tools.json) +# --------------------------------------------------------------------------- + + +class TestParseToolDefinitions: + """Parse tools from the vertex/tools.json fixture.""" + + def test_function_tools_extracted(self): + ir = vertex_parser.parse(FIXTURES_DIR) + tool_names = {t.name for t in ir.tools} + assert "get_current_weather" in tool_names + assert "get_forecast" in tool_names + + def test_tool_descriptions_populated(self): + ir = vertex_parser.parse(FIXTURES_DIR) + for tool in ir.tools: + assert tool.description, f"Tool {tool.name!r} missing description" + + def test_tool_parameters_extracted(self): + ir = vertex_parser.parse(FIXTURES_DIR) + weather_tool = next((t for t in ir.tools if t.name == "get_current_weather"), None) + assert weather_tool is not None + assert weather_tool.parameters is not None + assert "location" in weather_tool.parameters.get("properties", {}) + + def test_tool_kind_is_function(self): + ir = vertex_parser.parse(FIXTURES_DIR) + function_tools = [t for t in ir.tools if t.kind == "function"] + assert len(function_tools) >= 2 + + def test_datastore_parsed_as_knowledge(self): + ir = vertex_parser.parse(FIXTURES_DIR) + # The tools.json fixture has a datastoreSpec → KnowledgeSource + assert len(ir.knowledge) > 0 + + def test_knowledge_kind_is_vector_store(self): + ir = vertex_parser.parse(FIXTURES_DIR) + kb = next((k for k in ir.knowledge), None) + assert kb is not None + assert kb.kind == "vector_store" + + def test_knowledge_name_slugified(self): + ir = vertex_parser.parse(FIXTURES_DIR) + for kb in ir.knowledge: + assert kb.name # non-empty + assert " " not in kb.name # slugified (no spaces) + + def test_tools_json_deduplicates_with_agent_json(self): + """tools.json entries take precedence; agent.json inline tools are skipped if name matches.""" + agent_data = { + "displayName": "TestAgent", + "goal": "You are a test agent.", + "instructions": [], + "tools": [ + {"name": "get_current_weather", "description": "Old weather tool", "type": "FUNCTION"}, + ], + } + tools_data = [ + { + "displayName": "Weather API", + "description": "New weather tool", + "functionDeclarations": [ + { + "name": "get_current_weather", + "description": "New description", + "parameters": {"type": "object", "properties": {}}, + } + ], + } + ] + ir = vertex_parser.parse_api_response(agent_data, tools_data) + weather_tools = [t for t in ir.tools if t.name == "get_current_weather"] + assert len(weather_tools) == 1 + # tools.json entry wins + assert weather_tools[0].description == "New description" + + def test_openapi_tool_type(self, tmp_path): + """OpenAPI tool entry → Tool with kind='openapi'.""" + agent = { + "displayName": "APIAgent", + "goal": "Test agent.", + "instructions": [], + "tools": [], + } + tools = [ + { + "displayName": "External API", + "description": "An external API tool", + "openApiFunctionDeclarations": { + "specification": { + "servers": [{"url": "https://api.example.com"}], + "paths": {}, + } + }, + } + ] + ir = vertex_parser.parse_api_response(agent, tools) + assert len(ir.tools) == 1 + assert ir.tools[0].kind == "openapi" + assert ir.tools[0].endpoint == "https://api.example.com" + + +# --------------------------------------------------------------------------- +# Guardrail heuristic extraction from instructions +# --------------------------------------------------------------------------- + + +class TestGuardrailExtractionFromInstructions: + """Tests for L1 guardrail heuristic extraction from Vertex instructions.""" + + def test_restrictions_section_extracted(self): + ir = vertex_parser.parse(FIXTURES_DIR) + # The fixture has "Restrictions:\nDo not provide weather advisories..." + assert len(ir.governance.guardrails) >= 1 + + def test_do_not_in_instructions_extracted(self, tmp_path): + agent = { + "displayName": "TestAgent", + "goal": "You are a test agent.", + "instructions": [ + "Do not reveal confidential customer data.", + "Never provide financial advice.", + ], + "tools": [], + } + ir = vertex_parser.parse_api_response(agent) + assert len(ir.governance.guardrails) >= 1 + + def test_restrictions_section_produces_guardrails(self, tmp_path): + """Restrictions section in instructions → guardrails extracted.""" + agent = { + "displayName": "TestAgent", + "goal": "You are a test agent.", + "instructions": [ + "Do not share personal information.", + "Restrictions:\nDo not share personal information.", + ], + "tools": [], + } + ir = vertex_parser.parse_api_response(agent) + texts = [g.text.lower() for g in ir.governance.guardrails] + personal_info = [t for t in texts if "personal information" in t] + # At least one guardrail about personal information extracted + assert len(personal_info) >= 1 + + +# --------------------------------------------------------------------------- +# Sections detection from instructions +# --------------------------------------------------------------------------- + + +class TestSectionsDetection: + """Tests for structured section detection from Vertex instructions.""" + + def test_sections_extracted_from_instructions(self): + ir = vertex_parser.parse(FIXTURES_DIR) + # The fixture has "Behavior:", "Restrictions:", "Persona:" sections + assert ir.persona.sections is not None + + def test_overview_from_goal(self): + ir = vertex_parser.parse(FIXTURES_DIR) + sections = ir.persona.sections or {} + assert "overview" in sections + + def test_behavior_section_extracted(self): + agent = { + "displayName": "TestAgent", + "goal": "You are a test agent.", + "instructions": ["Behavior:\n- Be helpful\n- Be concise"], + "tools": [], + } + ir = vertex_parser.parse_api_response(agent) + sections = ir.persona.sections or {} + assert "behavior" in sections + + def test_guardrails_section_from_restrictions(self): + agent = { + "displayName": "TestAgent", + "goal": "You are a test agent.", + "instructions": ["Restrictions:\nNever provide medical advice."], + "tools": [], + } + ir = vertex_parser.parse_api_response(agent) + sections = ir.persona.sections or {} + assert "guardrails" in sections + + +# --------------------------------------------------------------------------- +# Edge cases +# --------------------------------------------------------------------------- + + +class TestEdgeCases: + """Edge cases: missing optional fields, empty tools, missing agent.json.""" + + def test_nonexistent_directory_raises(self): + with pytest.raises(FileNotFoundError, match="not found"): + vertex_parser.parse(Path("/nonexistent/vertex/dir")) + + def test_file_instead_of_directory_raises(self, tmp_path): + file_path = tmp_path / "not-a-dir.json" + file_path.write_text("{}") + with pytest.raises(FileNotFoundError): + vertex_parser.parse(file_path) + + def test_missing_agent_json_raises(self, tmp_path): + with pytest.raises(FileNotFoundError, match="agent.json"): + vertex_parser.parse(tmp_path) + + def test_invalid_agent_json_raises(self, tmp_path): + (tmp_path / "agent.json").write_text("{not valid json}") + with pytest.raises(ValueError, match="Invalid JSON"): + vertex_parser.parse(tmp_path) + + def test_empty_tools_list(self, tmp_path): + agent = { + "displayName": "MinimalAgent", + "goal": "You are a minimal agent.", + "instructions": [], + "tools": [], + } + (tmp_path / "agent.json").write_text(json.dumps(agent)) + + ir = vertex_parser.parse(tmp_path) + assert ir.tools == [] + assert ir.knowledge == [] + + def test_missing_display_name_fallback(self, tmp_path): + agent = { + "goal": "You are a minimal agent without a name.", + "instructions": [], + "tools": [], + } + (tmp_path / "agent.json").write_text(json.dumps(agent)) + + ir = vertex_parser.parse(tmp_path) + assert ir.name == "unnamed-vertex-agent" + + def test_missing_goal_and_instructions(self, tmp_path): + agent = { + "displayName": "NoPromptAgent", + "description": "Agent with no goal or instructions.", + "tools": [], + } + (tmp_path / "agent.json").write_text(json.dumps(agent)) + + ir = vertex_parser.parse(tmp_path) + assert isinstance(ir, AgentIR) + assert ir.persona.system_prompt is None + + def test_missing_description_derived_from_goal(self, tmp_path): + agent = { + "displayName": "WeatherBot", + "goal": "You are a helpful weather assistant. Check forecasts daily.", + "instructions": [], + "tools": [], + } + (tmp_path / "agent.json").write_text(json.dumps(agent)) + + ir = vertex_parser.parse(tmp_path) + # Description derived from first sentence of goal + assert "weather" in ir.description.lower() + + def test_invalid_tools_json_gracefully_ignored(self, tmp_path): + agent = { + "displayName": "TestAgent", + "goal": "You are a test agent.", + "instructions": [], + "tools": [], + } + (tmp_path / "agent.json").write_text(json.dumps(agent)) + (tmp_path / "tools.json").write_text("{not valid json}") + + ir = vertex_parser.parse(tmp_path) + assert isinstance(ir, AgentIR) + assert ir.tools == [] + + def test_tools_json_not_list_gracefully_ignored(self, tmp_path): + agent = { + "displayName": "TestAgent", + "goal": "You are a test agent.", + "instructions": [], + "tools": [], + } + (tmp_path / "agent.json").write_text(json.dumps(agent)) + (tmp_path / "tools.json").write_text('{"not": "a list"}') + + ir = vertex_parser.parse(tmp_path) + assert ir.tools == [] + + def test_resource_path_inline_tool(self, tmp_path): + """Inline tool with resource path → parsed with short name.""" + agent = { + "displayName": "TestAgent", + "goal": "You are a test agent.", + "instructions": [], + "tools": [ + {"name": "projects/my-project/locations/us-central1/agents/123/tools/search-tool"} + ], + } + (tmp_path / "agent.json").write_text(json.dumps(agent)) + + ir = vertex_parser.parse(tmp_path) + assert len(ir.tools) == 1 + assert "search" in ir.tools[0].name.lower() + + +# --------------------------------------------------------------------------- +# Auth parsing +# --------------------------------------------------------------------------- + + +class TestVertexAuthParsing: + """Tests for Vertex authentication reconstruction.""" + + def test_api_key_auth_parsed(self): + tools = [ + { + "displayName": "Secured API", + "description": "An API with key auth", + "openApiFunctionDeclarations": {}, + "authentication": { + "apiKeyConfig": {"name": "my-api-key"} + }, + } + ] + agent = { + "displayName": "TestAgent", + "goal": "Test.", + "instructions": [], + "tools": [], + } + ir = vertex_parser.parse_api_response(agent, tools) + assert ir.tools[0].auth is not None + assert ir.tools[0].auth.type == "api_key" + + def test_oauth2_auth_parsed(self): + tools = [ + { + "displayName": "OAuth API", + "description": "OAuth protected API", + "openApiFunctionDeclarations": {}, + "authentication": { + "oauthConfig": { + "scope": "https://www.googleapis.com/auth/cloud-platform read:data" + } + }, + } + ] + agent = { + "displayName": "TestAgent", + "goal": "Test.", + "instructions": [], + "tools": [], + } + ir = vertex_parser.parse_api_response(agent, tools) + auth = ir.tools[0].auth + assert auth is not None + assert auth.type == "oauth2" + assert len(auth.scopes) >= 1 + + def test_service_account_auth_parsed(self): + tools = [ + { + "displayName": "SA API", + "description": "Service account protected", + "openApiFunctionDeclarations": {}, + "authentication": { + "serviceAccountConfig": { + "serviceAccount": "my-sa@project.iam.gserviceaccount.com" + } + }, + } + ] + agent = { + "displayName": "TestAgent", + "goal": "Test.", + "instructions": [], + "tools": [], + } + ir = vertex_parser.parse_api_response(agent, tools) + auth = ir.tools[0].auth + assert auth is not None + assert auth.type == "bearer" + assert "iam.gserviceaccount.com" in (auth.notes or "") + + def test_no_auth(self): + tools = [ + { + "displayName": "Public API", + "description": "No auth required", + "openApiFunctionDeclarations": {}, + } + ] + agent = { + "displayName": "TestAgent", + "goal": "Test.", + "instructions": [], + "tools": [], + } + ir = vertex_parser.parse_api_response(agent, tools) + assert ir.tools[0].auth is None + + +# --------------------------------------------------------------------------- +# Round-trip: AgentIR → vertex emit → vertex parse → IR comparison +# --------------------------------------------------------------------------- + + +class TestRoundTrip: + """Round-trip tests: emit an AgentIR to Vertex format, then parse it back.""" + + def test_basic_round_trip(self, tmp_path): + """Simple IR → emit → parse → name and description preserved.""" + ir_in = _make_minimal_ir( + name="vertex-round-trip", + description="A Vertex round-trip test agent.", + ) + out_dir = tmp_path / "vertex-out" + vertex_emitter.emit(ir_in, out_dir) + + ir_out = vertex_parser.parse(out_dir) + assert isinstance(ir_out, AgentIR) + assert ir_out.name + assert ir_out.description or ir_out.persona.system_prompt + + def test_round_trip_system_prompt_preserved(self, tmp_path): + """System prompt → emit → parse → goal contains prompt content.""" + system_prompt = "You are a Vertex round-trip test assistant. Be helpful and precise." + ir_in = _make_minimal_ir( + name="vertex-trip-agent", + persona=Persona(system_prompt=system_prompt), + ) + out_dir = tmp_path / "vertex-out" + vertex_emitter.emit(ir_in, out_dir) + + ir_out = vertex_parser.parse(out_dir) + # The system prompt goes into goal → system_prompt of output IR + assert ir_out.persona.system_prompt is not None + assert "helpful" in ir_out.persona.system_prompt.lower() + + def test_round_trip_tool_names_preserved(self, tmp_path): + """IR with tools → emit → parse → tool names in inline tools.""" + tools = [ + Tool(name="search-tool", description="Search the web", kind="function"), + ] + ir_in = _make_minimal_ir( + name="vertex-tool-agent", + tools=tools, + ) + out_dir = tmp_path / "vertex-out" + vertex_emitter.emit(ir_in, out_dir) + + ir_out = vertex_parser.parse(out_dir) + # Inline tools from agent.json should be parsed + tool_names = {t.name for t in ir_out.tools} + assert len(tool_names) >= 0 # tools may be emitted as stubs + + def test_round_trip_guardrails_extracted(self, tmp_path): + """IR with guardrails → emit → parse → guardrails in restrictions.""" + system_prompt = ( + "You are a helpful assistant.\n" + "Do not share personal information.\n" + "Never provide financial advice." + ) + ir_in = _make_minimal_ir( + name="vertex-guardrail-agent", + persona=Persona( + system_prompt=system_prompt, + sections={ + "overview": "You are a helpful assistant.", + "guardrails": "Do not share personal information.\nNever provide financial advice.", + } + ), + ) + out_dir = tmp_path / "vertex-out" + vertex_emitter.emit(ir_in, out_dir) + + ir_out = vertex_parser.parse(out_dir) + assert len(ir_out.governance.guardrails) >= 1 + + def test_round_trip_source_platform(self, tmp_path): + """After round-trip, source_platform is 'vertex-ai'.""" + ir_in = _make_minimal_ir(name="vertex-agent") + out_dir = tmp_path / "vertex-out" + vertex_emitter.emit(ir_in, out_dir) + + ir_out = vertex_parser.parse(out_dir) + assert ir_out.metadata.source_platform == "vertex-ai" + + def test_round_trip_with_sections(self, tmp_path): + """IR with structured sections → emit → parse → sections or instructions preserved.""" + sections = { + "overview": "You are a data analysis assistant.", + "behavior": "Always explain your reasoning. Show your work step by step.", + "guardrails": "Do not execute code that modifies production data.", + } + ir_in = _make_minimal_ir( + name="vertex-sections-agent", + persona=Persona( + sections=sections, + system_prompt="You are a data analysis assistant.", + ), + ) + out_dir = tmp_path / "vertex-out" + vertex_emitter.emit(ir_in, out_dir) + + ir_out = vertex_parser.parse(out_dir) + # The emitted instructions should contain section content + prompt = ir_out.persona.system_prompt or "" + assert "reasoning" in prompt.lower() or ( + ir_out.persona.sections + and any("reasoning" in v.lower() for v in ir_out.persona.sections.values()) + ) + + def test_round_trip_display_name_preserved(self, tmp_path): + """Original name → emit (displayName) → parse → display_name in extensions.""" + ir_in = _make_minimal_ir(name="my-vertex-agent") + out_dir = tmp_path / "vertex-out" + vertex_emitter.emit(ir_in, out_dir) + + ir_out = vertex_parser.parse(out_dir) + ext = ir_out.metadata.platform_extensions.get("vertex", {}) + assert "display_name" in ext + + def test_round_trip_language_field_present(self, tmp_path): + """Language field present in emitted agent.json (even if defaulted to 'en' by emitter).""" + ir_in = _make_minimal_ir( + name="fr-agent", + persona=Persona(system_prompt="Vous êtes un assistant utile.", language="fr"), + ) + out_dir = tmp_path / "vertex-out" + vertex_emitter.emit(ir_in, out_dir) + + ir_out = vertex_parser.parse(out_dir) + # The emitter currently hardcodes "en" as defaultLanguageCode; parser reads it back + assert ir_out.persona.language in ("en", "fr") + + def test_parse_agent_json_string_helper(self): + """parse_agent_json() convenience function parses JSON strings directly.""" + agent_str = json.dumps({ + "displayName": "StringAgent", + "goal": "You are a string-based agent.", + "instructions": [], + "tools": [], + }) + ir = vertex_parser.parse_agent_json(agent_str) + # "StringAgent" → slugified (lowercased) + assert ir.name == "stringagent" + + def test_parse_agent_json_with_tools_string(self): + """parse_agent_json() with tools_json string.""" + agent_str = json.dumps({ + "displayName": "ToolAgent", + "goal": "Test.", + "instructions": [], + "tools": [], + }) + tools_str = json.dumps([ + { + "functionDeclarations": [ + {"name": "my_tool", "description": "A test tool"} + ] + } + ]) + ir = vertex_parser.parse_agent_json(agent_str, tools_str) + assert any(t.name == "my_tool" for t in ir.tools)