mirror of
https://github.com/dograh-hq/dograh.git
synced 2026-07-25 12:01:04 +02:00
117 lines
3.4 KiB
Python
117 lines
3.4 KiB
Python
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from api.routes import organization
|
|
from api.services import tool_management
|
|
|
|
|
|
def _credentials(password: str = "agent-secret") -> dict:
|
|
return {
|
|
"ari_endpoint": "https://asterisk.example.com",
|
|
"app_name": "dograh",
|
|
"app_password": "ari-secret",
|
|
"external_pbx": {
|
|
"type": "vicidial",
|
|
"agent_api": {
|
|
"url": "https://vici.example.com/agc/api.php",
|
|
"username": "agent-user",
|
|
"password": password,
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def test_nested_external_pbx_secrets_are_masked_without_mutating_source():
|
|
credentials = _credentials()
|
|
|
|
masked = organization._mask_sensitive("ari", credentials)
|
|
|
|
assert masked["app_password"] != "ari-secret"
|
|
assert masked["external_pbx"]["agent_api"]["password"] != "agent-secret"
|
|
assert credentials["external_pbx"]["agent_api"]["password"] == "agent-secret"
|
|
|
|
|
|
def test_nested_masked_external_pbx_secrets_are_restored_on_update():
|
|
existing = _credentials()
|
|
request = organization._mask_sensitive("ari", existing)
|
|
|
|
organization.preserve_masked_fields("ari", request, existing)
|
|
|
|
assert request["app_password"] == "ari-secret"
|
|
assert request["external_pbx"]["agent_api"]["password"] == "agent-secret"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_feature_allows_unchanged_telephony_configuration(monkeypatch):
|
|
monkeypatch.setattr(
|
|
organization,
|
|
"external_pbx_integrations_enabled",
|
|
AsyncMock(return_value=False),
|
|
)
|
|
existing = _credentials()
|
|
|
|
await organization._enforce_external_pbx_feature(
|
|
7,
|
|
"ari",
|
|
_credentials(),
|
|
existing_credentials=existing,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_feature_rejects_removing_telephony_configuration(monkeypatch):
|
|
monkeypatch.setattr(
|
|
organization,
|
|
"external_pbx_integrations_enabled",
|
|
AsyncMock(return_value=False),
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await organization._enforce_external_pbx_feature(
|
|
7,
|
|
"ari",
|
|
{"external_pbx": None},
|
|
existing_credentials=_credentials(),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_feature_preserves_existing_tool_mapping(monkeypatch):
|
|
monkeypatch.setattr(
|
|
tool_management,
|
|
"external_pbx_integrations_enabled",
|
|
AsyncMock(return_value=False),
|
|
)
|
|
definition = {
|
|
"type": "transfer_call",
|
|
"config": {
|
|
"destination_source": "context_mapping",
|
|
"context_mapping": {
|
|
"context_path": "qualified",
|
|
"routes": [{"context_value": "yes", "destination": "sales"}],
|
|
},
|
|
},
|
|
}
|
|
|
|
await tool_management.validate_external_pbx_tool_definition(
|
|
definition,
|
|
organization_id=7,
|
|
existing_definition=definition,
|
|
)
|
|
|
|
changed = {
|
|
"type": "transfer_call",
|
|
"config": {"destination_source": "static", "destination": "+15555550100"},
|
|
}
|
|
with pytest.raises(tool_management.ToolManagementError) as exc_info:
|
|
await tool_management.validate_external_pbx_tool_definition(
|
|
changed,
|
|
organization_id=7,
|
|
existing_definition=definition,
|
|
)
|
|
|
|
assert exc_info.value.status_code == 403
|