cas-reference-product/tests/test_service_factory.py at e678dad5e9464e721dbcb43adddbd77542e87faa · Coding-Autopilot-System/cas-reference-product · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
from typing import Any
from unittest.mock import patch

import pytest

from cas_reference_product.config import Settings
from cas_reference_product.workflow import (
    FoundryWorkflowAgentService,
    LocalWorkflowAgentService,
    WorkflowAgentServiceError,
    build_workflow_agent_service,
)


def test_factory_builds_local_service() -> None:
    assert isinstance(build_workflow_agent_service(Settings()), LocalWorkflowAgentService)


def test_factory_builds_foundry_service_when_configured() -> None:
    settings = Settings(
        environment="prod",
        workflow_backend="foundry",
        foundry_project_endpoint="https://example.services.ai.azure.com/api/projects/example",
        foundry_agent_name="cas-reference-agent",
    )
    with patch("cas_reference_product.workflow.FoundryWorkflowAgentService") as service:
        build_workflow_agent_service(settings)
    service.assert_called_once_with(settings)


def test_foundry_service_rejects_invalid_project_endpoint() -> None:
    settings = Settings(
        environment="prod",
        workflow_backend="foundry",
        foundry_project_endpoint="https://example.invalid/project",
        foundry_agent_name="cas-reference-agent",
    )

    with pytest.raises(ValueError, match="valid project endpoint"):
        FoundryWorkflowAgentService(settings)


def test_foundry_service_uses_next_gen_agent_reference(envelope: "Any") -> None:
    settings = Settings(
        environment="prod",
        workflow_backend="foundry",
        foundry_project_endpoint="https://example.services.ai.azure.com/api/projects/example",
        foundry_agent_name="cas-reference-agent",
    )
    with (
        patch("cas_reference_product.workflow.build_credential") as credential,
        patch("cas_reference_product.workflow.AIProjectClient") as project_client,
    ):
        responses = project_client.return_value.get_openai_client.return_value.responses
        responses.create.return_value.output_text = "Foundry result"
        service = FoundryWorkflowAgentService(settings)
        result = service.run(envelope)

    project_client.assert_called_once_with(
        endpoint=settings.foundry_project_endpoint,
        credential=credential.return_value,
    )
    responses.create.assert_called_once_with(
        input=envelope.prompt,
        extra_body={
            "agent_reference": {
                "name": "cas-reference-agent",
                "type": "agent_reference",
            }
        },
    )
    assert result == "Foundry result"


def test_foundry_service_sanitizes_sdk_failure(envelope: "Any") -> None:
    settings = Settings(
        environment="prod",
        workflow_backend="foundry",
        foundry_project_endpoint="https://example.services.ai.azure.com/api/projects/example",
        foundry_agent_name="cas-reference-agent",
    )
    with (
        patch("cas_reference_product.workflow.build_credential"),
        patch("cas_reference_product.workflow.AIProjectClient") as project_client,
    ):
        project_client.return_value.get_openai_client.return_value.responses.create.side_effect = (
            RuntimeError("sensitive provider detail")
        )
        service = FoundryWorkflowAgentService(settings)

        with pytest.raises(WorkflowAgentServiceError, match="Foundry workflow invocation failed"):
            service.run(envelope)