Spaces:
Runtime error
Runtime error
| """Stage 7c tests — multi-API executor (mocked HTTP layer).""" | |
| import pytest | |
| from backend.agent.executor import ( | |
| build_payload_for_api, | |
| execute, | |
| execution_to_dict, | |
| ) | |
| from backend.agent.kb import KnowledgeBase | |
| def kb(): | |
| return KnowledgeBase("PO") | |
| def pilot(kb): | |
| return kb.journey("Create Direct Purchase Order") | |
| def test_main_payload_builds_with_user_slots(pilot): | |
| """User-provided slot values land in the right API property.""" | |
| main_apis = pilot.save_apis_by_screen.get(pilot.main_screen_name) or [] | |
| create_api = next(a for a in main_apis if "create" in a["api_file"].lower()) | |
| payload, records = build_payload_for_api( | |
| pilot, create_api, | |
| slots={ | |
| "buyerhdr": "Maria", | |
| "currencycode": "USD", | |
| "potypeenum": "2", | |
| "supplier_code": "SUP-001", | |
| }, | |
| ) | |
| # Look for Buyer / Currency / Type / SupplierCode in the assembled payload | |
| flat = {} | |
| for schema_block in payload.values(): | |
| flat.update(schema_block) | |
| assert flat.get("Buyer") == "Maria", f"Buyer not assembled correctly: {flat.get('Buyer')}" | |
| assert flat.get("Currency") == "USD" or flat.get("Currency") == "USD" | |
| assert flat.get("SupplierCode") in ("SUP-001",), f"SupplierCode: {flat.get('SupplierCode')}" | |
| # Some user-sourced records should appear | |
| user_records = [r for r in records if r.source_kind == "user"] | |
| assert len(user_records) >= 3 | |
| def test_execute_fires_main_only_when_no_subscreens(pilot): | |
| # Use a supplier_code + buyerhdr that exist in master_data_stub.json so the Stage 8 | |
| # master-data validation passes. Values outside the stub correctly produce a mock failure. | |
| result = execute(pilot, slots={"buyerhdr": "Maria", "supplier_code": "SUP-100"}) | |
| assert result.success, result.error_summary | |
| assert len(result.calls) == 1 | |
| assert "Create" in result.calls[0].api_file | |
| assert result.calls[0].response is not None | |
| assert result.calls[0].response.get("po_no", "").startswith("PO/MOCK/") | |
| def test_execute_fires_subscreen_when_subscreen_slot_filled(pilot): | |
| """Filling a slot whose alias binds to SpecifyTerms must trigger that API.""" | |
| # Pick a slot we know binds to SpecifyTerms3.0.json (verified via slot_alias_map) | |
| terms_slot = "payterm" | |
| result = execute(pilot, slots={ | |
| "buyerhdr": "Maria", | |
| terms_slot: "NET30", | |
| }) | |
| api_files = [c.api_file for c in result.calls] | |
| assert any("Specify" in f and "Terms" in f for f in api_files), ( | |
| f"expected SpecifyTerms in fired APIs, got {api_files}" | |
| ) | |
| def test_execute_fires_subscreen_when_subscreen_slot_filled_2(pilot): | |
| """Same as above but with a different slot to make sure we're not coincidentally right.""" | |
| result = execute(pilot, slots={ | |
| "buyerhdr": "Maria", | |
| "groption": "Y", | |
| }) | |
| api_files = [c.api_file for c in result.calls] | |
| assert any("Specify" in f and "Terms" in f for f in api_files), ( | |
| f"expected SpecifyTerms with groption, got {api_files}" | |
| ) | |
| def test_execute_halts_on_injected_error(pilot): | |
| """If we inject a failure on the main call, no sub-screen calls should be fired afterwards.""" | |
| result = execute( | |
| pilot, | |
| slots={"buyerhdr": "M", "payterm": "NET30"}, | |
| inject_error_on=pilot.main_screen_name, | |
| ) | |
| assert not result.success | |
| fired = [c for c in result.calls if c.fired_at is not None] | |
| assert any(c.success is False for c in fired) | |
| assert not any(c.success is True for c in fired) | |
| def test_execution_to_dict_serialises(pilot): | |
| result = execute(pilot, slots={"buyerhdr": "Maria"}) | |
| d = execution_to_dict(result) | |
| assert d["success"] | |
| assert d["call_count"] >= 1 | |
| assert d["calls"][0]["payload"] | |