mirror of
https://github.com/hoshikawa2/compass_backoffice.git
synced 2026-07-09 22:04:20 +00:00
390 lines
15 KiB
Python
390 lines
15 KiB
Python
"""Unit tests for SiebelClient."""
|
|
|
|
import base64
|
|
import pytest
|
|
import httpx
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from src.components.clients.siebel_client import SiebelClient
|
|
from src.components.clients.exceptions.siebel_exceptions import (
|
|
SiebelClientError,
|
|
SiebelHttpError,
|
|
SiebelConnectionError,
|
|
SiebelTimeoutError,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_SAMPLE_PAYLOAD = {
|
|
"channel": "Anatel",
|
|
"socialSecNo": "06252533106",
|
|
"assetId": "62981152324",
|
|
"serviceRequest": {
|
|
"type": "0119",
|
|
"userId": "SADMIN",
|
|
"reason1": "Processo Interno",
|
|
"reason2": "Turbina",
|
|
"reason3": "Internalização",
|
|
"status": "Encaminhado",
|
|
"notes": "Teste",
|
|
},
|
|
"interaction": {
|
|
"source": "Cliente",
|
|
"requestFlag": False,
|
|
"directionContact": "FROM-CLIENT",
|
|
"status": "OPENED",
|
|
},
|
|
}
|
|
|
|
_SAMPLE_RESPONSE = {"interactionProtocol": "2026000099999"}
|
|
|
|
|
|
def _make_client(**kwargs) -> SiebelClient:
|
|
"""Instantiate SiebelClient with test defaults using patched settings."""
|
|
with patch("src.components.clients.siebel_client.settings") as mock_settings:
|
|
mock_settings.SIEBEL_API_HOST = "https://fake-siebel.example.com"
|
|
mock_settings.SIEBEL_API_TIMEOUT = 30
|
|
mock_settings.VERIFY_SSL = False
|
|
mock_settings.SIEBEL_API_USERNAME = "testuser"
|
|
mock_settings.SIEBEL_API_PASSWORD = "testpass"
|
|
mock_settings.SIEBEL_API_CLIENT_ID = "test-client-id"
|
|
mock_settings.LOG_LEVEL = "INFO"
|
|
mock_settings.LOG_FORMAT = "text"
|
|
return SiebelClient(**kwargs)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Initialization
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.unit
|
|
def test_client_uses_settings_defaults():
|
|
"""When no args are given, client falls back to settings values."""
|
|
with patch("src.components.clients.siebel_client.settings") as mock_settings:
|
|
mock_settings.SIEBEL_API_HOST = "https://default.example.com"
|
|
mock_settings.SIEBEL_API_TIMEOUT = 60
|
|
mock_settings.VERIFY_SSL = True
|
|
mock_settings.SIEBEL_API_USERNAME = "user"
|
|
mock_settings.SIEBEL_API_PASSWORD = "pass"
|
|
mock_settings.SIEBEL_API_CLIENT_ID = "cid"
|
|
mock_settings.LOG_LEVEL = "INFO"
|
|
mock_settings.LOG_FORMAT = "text"
|
|
client = SiebelClient()
|
|
|
|
assert client.base_url == "https://default.example.com"
|
|
assert client.timeout == 60
|
|
assert client.verify_ssl is True
|
|
assert client.username == "user"
|
|
assert client.password == "pass"
|
|
assert client.client_id == "cid"
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_client_explicit_args_override_settings():
|
|
"""Explicit constructor arguments must take precedence over settings."""
|
|
client = _make_client(
|
|
base_url="https://override.example.com",
|
|
timeout=10,
|
|
verify_ssl=True,
|
|
username="override_user",
|
|
password="override_pass",
|
|
client_id="override_cid",
|
|
)
|
|
assert client.base_url == "https://override.example.com"
|
|
assert client.timeout == 10
|
|
assert client.verify_ssl is True
|
|
assert client.username == "override_user"
|
|
assert client.password == "override_pass"
|
|
assert client.client_id == "override_cid"
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_client_accepts_verify_ssl_false_explicitly():
|
|
"""verify_ssl=False must not be overridden by settings (falsy-value bug guard)."""
|
|
with patch("src.components.clients.siebel_client.settings") as mock_settings:
|
|
mock_settings.SIEBEL_API_HOST = "https://x.com"
|
|
mock_settings.SIEBEL_API_TIMEOUT = 30
|
|
mock_settings.VERIFY_SSL = True # settings say True
|
|
mock_settings.SIEBEL_API_USERNAME = "u"
|
|
mock_settings.SIEBEL_API_PASSWORD = "p"
|
|
mock_settings.SIEBEL_API_CLIENT_ID = "c"
|
|
mock_settings.LOG_LEVEL = "INFO"
|
|
mock_settings.LOG_FORMAT = "text"
|
|
client = SiebelClient(verify_ssl=False) # explicit False must win
|
|
|
|
assert client.verify_ssl is False
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_client_accepts_timeout_zero_explicitly():
|
|
"""timeout=0 must not be overridden by settings (falsy-value bug guard)."""
|
|
with patch("src.components.clients.siebel_client.settings") as mock_settings:
|
|
mock_settings.SIEBEL_API_HOST = "https://x.com"
|
|
mock_settings.SIEBEL_API_TIMEOUT = 30 # settings say 30
|
|
mock_settings.VERIFY_SSL = False
|
|
mock_settings.SIEBEL_API_USERNAME = "u"
|
|
mock_settings.SIEBEL_API_PASSWORD = "p"
|
|
mock_settings.SIEBEL_API_CLIENT_ID = "c"
|
|
mock_settings.LOG_LEVEL = "INFO"
|
|
mock_settings.LOG_FORMAT = "text"
|
|
client = SiebelClient(timeout=0) # explicit 0 must win
|
|
|
|
assert client.timeout == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _get_auth_header
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.unit
|
|
def test_get_auth_header_returns_basic_header():
|
|
"""_get_auth_header must return a valid Basic Auth header."""
|
|
client = _make_client(username="user", password="pass")
|
|
header = client._get_auth_header()
|
|
|
|
expected = "Basic " + base64.b64encode(b"user:pass").decode()
|
|
assert header == expected
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_get_auth_header_returns_empty_when_no_credentials():
|
|
"""_get_auth_header must return '' when username and password are absent from settings."""
|
|
with patch("src.components.clients.siebel_client.settings") as mock_settings:
|
|
mock_settings.SIEBEL_API_HOST = "https://fake-siebel.example.com"
|
|
mock_settings.SIEBEL_API_TIMEOUT = 30
|
|
mock_settings.VERIFY_SSL = False
|
|
mock_settings.SIEBEL_API_USERNAME = ""
|
|
mock_settings.SIEBEL_API_PASSWORD = ""
|
|
mock_settings.SIEBEL_API_CLIENT_ID = ""
|
|
mock_settings.LOG_LEVEL = "INFO"
|
|
mock_settings.LOG_FORMAT = "text"
|
|
client = SiebelClient()
|
|
|
|
assert client._get_auth_header() == ""
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_get_auth_header_returns_empty_when_password_missing():
|
|
"""_get_auth_header must return '' when only password is absent from settings."""
|
|
with patch("src.components.clients.siebel_client.settings") as mock_settings:
|
|
mock_settings.SIEBEL_API_HOST = "https://fake-siebel.example.com"
|
|
mock_settings.SIEBEL_API_TIMEOUT = 30
|
|
mock_settings.VERIFY_SSL = False
|
|
mock_settings.SIEBEL_API_USERNAME = "user"
|
|
mock_settings.SIEBEL_API_PASSWORD = ""
|
|
mock_settings.SIEBEL_API_CLIENT_ID = ""
|
|
mock_settings.LOG_LEVEL = "INFO"
|
|
mock_settings.LOG_FORMAT = "text"
|
|
client = SiebelClient()
|
|
|
|
assert client._get_auth_header() == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# open_service_request — happy path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_open_service_request_returns_json_response():
|
|
"""On HTTP 200, open_service_request must return the parsed JSON body."""
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_response.json.return_value = _SAMPLE_RESPONSE
|
|
|
|
mock_http_client = AsyncMock()
|
|
mock_http_client.__aenter__ = AsyncMock(return_value=mock_http_client)
|
|
mock_http_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_http_client.post = AsyncMock(return_value=mock_response)
|
|
|
|
client = _make_client()
|
|
with patch("src.components.clients.siebel_client.traced_async_client", return_value=mock_http_client):
|
|
result = await client.open_service_request(payload=_SAMPLE_PAYLOAD)
|
|
|
|
assert result == _SAMPLE_RESPONSE
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_open_service_request_sends_authorization_header():
|
|
"""Request must include the Authorization header when credentials are set."""
|
|
captured_headers = {}
|
|
|
|
async def fake_post(url, json, headers):
|
|
captured_headers.update(headers)
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_response.json.return_value = _SAMPLE_RESPONSE
|
|
return mock_response
|
|
|
|
mock_http_client = AsyncMock()
|
|
mock_http_client.__aenter__ = AsyncMock(return_value=mock_http_client)
|
|
mock_http_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_http_client.post = fake_post
|
|
|
|
client = _make_client(username="user", password="pass", client_id="cid")
|
|
with patch("src.components.clients.siebel_client.traced_async_client", return_value=mock_http_client):
|
|
await client.open_service_request(payload=_SAMPLE_PAYLOAD)
|
|
|
|
assert "Authorization" in captured_headers
|
|
assert captured_headers["Authorization"].startswith("Basic ")
|
|
assert captured_headers["clientId"] == "cid"
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_open_service_request_no_auth_header_when_no_credentials():
|
|
"""When settings have no credentials, Authorization header must not be sent."""
|
|
captured_headers = {}
|
|
|
|
async def fake_post(url, json, headers):
|
|
captured_headers.update(headers)
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_response.json.return_value = _SAMPLE_RESPONSE
|
|
return mock_response
|
|
|
|
mock_http_client = AsyncMock()
|
|
mock_http_client.__aenter__ = AsyncMock(return_value=mock_http_client)
|
|
mock_http_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_http_client.post = fake_post
|
|
|
|
with patch("src.components.clients.siebel_client.settings") as mock_settings:
|
|
mock_settings.SIEBEL_API_HOST = "https://fake-siebel.example.com"
|
|
mock_settings.SIEBEL_API_TIMEOUT = 30
|
|
mock_settings.VERIFY_SSL = False
|
|
mock_settings.SIEBEL_API_USERNAME = ""
|
|
mock_settings.SIEBEL_API_PASSWORD = ""
|
|
mock_settings.SIEBEL_API_CLIENT_ID = ""
|
|
mock_settings.LOG_LEVEL = "INFO"
|
|
mock_settings.LOG_FORMAT = "text"
|
|
client = SiebelClient()
|
|
|
|
with patch("src.components.clients.siebel_client.traced_async_client", return_value=mock_http_client):
|
|
await client.open_service_request(payload=_SAMPLE_PAYLOAD)
|
|
|
|
assert "Authorization" not in captured_headers
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# open_service_request — error handling
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_http_status_error(status_code: int, body: str) -> httpx.HTTPStatusError:
|
|
"""Build a realistic httpx.HTTPStatusError for testing."""
|
|
response = MagicMock(spec=httpx.Response)
|
|
response.status_code = status_code
|
|
response.text = body
|
|
request = MagicMock(spec=httpx.Request)
|
|
return httpx.HTTPStatusError(message=body, request=request, response=response)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_open_service_request_raises_siebel_http_error_on_4xx():
|
|
"""HTTPStatusError from httpx must be wrapped into SiebelHttpError."""
|
|
mock_http_client = AsyncMock()
|
|
mock_http_client.__aenter__ = AsyncMock(return_value=mock_http_client)
|
|
mock_http_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_http_client.post = AsyncMock(
|
|
side_effect=_make_http_status_error(400, "Bad Request")
|
|
)
|
|
|
|
client = _make_client()
|
|
with patch("src.components.clients.siebel_client.traced_async_client", return_value=mock_http_client):
|
|
with pytest.raises(SiebelHttpError) as exc_info:
|
|
await client.open_service_request(payload=_SAMPLE_PAYLOAD)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "400" in str(exc_info.value)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_open_service_request_raises_siebel_http_error_on_5xx():
|
|
"""5xx responses must also raise SiebelHttpError."""
|
|
mock_http_client = AsyncMock()
|
|
mock_http_client.__aenter__ = AsyncMock(return_value=mock_http_client)
|
|
mock_http_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_http_client.post = AsyncMock(
|
|
side_effect=_make_http_status_error(550, "Generic Error")
|
|
)
|
|
|
|
client = _make_client()
|
|
with patch("src.components.clients.siebel_client.traced_async_client", return_value=mock_http_client):
|
|
with pytest.raises(SiebelHttpError) as exc_info:
|
|
await client.open_service_request(payload=_SAMPLE_PAYLOAD)
|
|
|
|
assert exc_info.value.status_code == 550
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_open_service_request_raises_siebel_http_error_is_subclass_of_base():
|
|
"""SiebelHttpError must be catchable via SiebelClientError."""
|
|
mock_http_client = AsyncMock()
|
|
mock_http_client.__aenter__ = AsyncMock(return_value=mock_http_client)
|
|
mock_http_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_http_client.post = AsyncMock(
|
|
side_effect=_make_http_status_error(500, "Server Error")
|
|
)
|
|
|
|
client = _make_client()
|
|
with patch("src.components.clients.siebel_client.traced_async_client", return_value=mock_http_client):
|
|
with pytest.raises(SiebelClientError):
|
|
await client.open_service_request(payload=_SAMPLE_PAYLOAD)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_open_service_request_raises_siebel_connection_error():
|
|
"""httpx.ConnectError must be wrapped into SiebelConnectionError."""
|
|
mock_http_client = AsyncMock()
|
|
mock_http_client.__aenter__ = AsyncMock(return_value=mock_http_client)
|
|
mock_http_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_http_client.post = AsyncMock(
|
|
side_effect=httpx.ConnectError("Connection refused")
|
|
)
|
|
|
|
client = _make_client()
|
|
with patch("src.components.clients.siebel_client.traced_async_client", return_value=mock_http_client):
|
|
with pytest.raises(SiebelConnectionError):
|
|
await client.open_service_request(payload=_SAMPLE_PAYLOAD)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_open_service_request_raises_siebel_timeout_error():
|
|
"""httpx.TimeoutException must be wrapped into SiebelTimeoutError."""
|
|
mock_http_client = AsyncMock()
|
|
mock_http_client.__aenter__ = AsyncMock(return_value=mock_http_client)
|
|
mock_http_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_http_client.post = AsyncMock(
|
|
side_effect=httpx.TimeoutException("Request timed out")
|
|
)
|
|
|
|
client = _make_client()
|
|
with patch("src.components.clients.siebel_client.traced_async_client", return_value=mock_http_client):
|
|
with pytest.raises(SiebelTimeoutError):
|
|
await client.open_service_request(payload=_SAMPLE_PAYLOAD)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_open_service_request_raises_siebel_client_error_on_unexpected():
|
|
"""Unexpected exceptions must be wrapped into the base SiebelClientError."""
|
|
mock_http_client = AsyncMock()
|
|
mock_http_client.__aenter__ = AsyncMock(return_value=mock_http_client)
|
|
mock_http_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_http_client.post = AsyncMock(
|
|
side_effect=RuntimeError("Something totally unexpected")
|
|
)
|
|
|
|
client = _make_client()
|
|
with patch("src.components.clients.siebel_client.traced_async_client", return_value=mock_http_client):
|
|
with pytest.raises(SiebelClientError):
|
|
await client.open_service_request(payload=_SAMPLE_PAYLOAD)
|