diff --git a/README.md b/README.md index 3270882d..cc48559e 100644 --- a/README.md +++ b/README.md @@ -429,6 +429,7 @@ gcs: bucket_name: "your-gcs-bucket-name" project_id: "your-gcp-project-id" credentials_path: null # Optional; falls back to GOOGLE_APPLICATION_CREDENTIALS / ADC + signing_service_account_email: null # Optional override for IAM signBlob (GKE Workload Identity) prefix: "audio/" # Same object key layout as S3 # CORS Settings diff --git a/app/api/v1/routes/call_import_evaluations.py b/app/api/v1/routes/call_import_evaluations.py index 53b0a11e..8a5844b7 100644 --- a/app/api/v1/routes/call_import_evaluations.py +++ b/app/api/v1/routes/call_import_evaluations.py @@ -1737,6 +1737,12 @@ async def export_call_import_evaluation_csv( continue # would clobber a real column custom_export.append((name, csv_header)) + if ( + call_import.source_format == "audio" + and "conversation_id" not in standard_export_headers + ): + standard_export_headers.insert(0, "conversation_id") + # Build the metric columns: each parent (if any) gets a value column # and (when capture_rationale=true) a " - LLM Rationale" # column. The per-child boolean columns are intentionally suppressed @@ -1824,6 +1830,8 @@ def _project_rows() -> Iterator[Dict[str, str]]: ) for header in standard_export_headers: value = raw.get(header) + if value is None and header == "conversation_id": + value = source_row.conversation_id row_out[header] = "" if value is None else str(value) for export_header, csv_header in custom_export: value = raw.get(csv_header) diff --git a/app/api/v1/routes/call_import_schemas.py b/app/api/v1/routes/call_import_schemas.py index 86b414b8..2a8877b8 100644 --- a/app/api/v1/routes/call_import_schemas.py +++ b/app/api/v1/routes/call_import_schemas.py @@ -7,9 +7,10 @@ parameters are mapped to source columns. Every schema MUST contain exactly one parameter with -``type='conversation_id'``. At most one parameter each may use -``type='recording_url'``, ``type='recording_date'``, or -``type='transcript'``. Only ``conversation_id`` is forced to +``type='conversation_id'`` and exactly one with +``type='recording_url'``. At most one parameter each may use +``type='recording_date'`` or ``type='transcript'``. +``conversation_id`` and ``recording_url`` are forced to ``is_required=True``. The invariant is enforced here on create + update because it spans the parent (`call_import_schemas`) and the children (`call_import_schema_parameters`) which are written in the same @@ -77,7 +78,11 @@ def _materialize_parameters( rows: List[CallImportSchemaParameter] = [] for idx, param in enumerate(payload_params): is_required = ( - param.type == CallImportParameterType.CONVERSATION_ID + param.type + in ( + CallImportParameterType.CONVERSATION_ID, + CallImportParameterType.RECORDING_URL, + ) or bool(param.is_required) ) rows.append( diff --git a/app/api/v1/routes/call_imports.py b/app/api/v1/routes/call_imports.py index 56daa19d..86cfaa7f 100644 --- a/app/api/v1/routes/call_imports.py +++ b/app/api/v1/routes/call_imports.py @@ -378,6 +378,20 @@ def _coerce_parameter_value( return cell +def _parameter_is_required(param: CallImportSchemaParameter) -> bool: + """Return whether a schema parameter must be mapped on every upload.""" + if param.is_required: + return True + try: + param_type = CallImportParameterType(param.type) + except ValueError: + return False + return param_type in ( + CallImportParameterType.CONVERSATION_ID, + CallImportParameterType.RECORDING_URL, + ) + + def _apply_schema_mapping( fieldnames: List[str], rows_iter: Iterable[Dict[str, str]], @@ -439,7 +453,7 @@ def _apply_schema_mapping( if mapped_header else None ) - if param.is_required and canonical is None: + if _parameter_is_required(param) and canonical is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( @@ -543,7 +557,7 @@ def _apply_schema_mapping( row_idx=idx, param_name=param.name, ) - if param.is_required and coerced is None: + if _parameter_is_required(param) and coerced is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( @@ -2248,6 +2262,37 @@ async def list_call_import_datasets( return [row[0] for row in rows if row[0]] +@router.get( + "/diarisation-prompt-default", + response_model=CallImportDiarisationPromptDefaultResponse, + operation_id="getCallImportDiarisationPromptDefault", +) +async def get_call_import_diarisation_prompt_default( + api_key: str = Depends(get_api_key), + organization_id: UUID = Depends(get_organization_id), +) -> CallImportDiarisationPromptDefaultResponse: + """Return the canonical LLM diariser prompt. + + The Transcribe / Run Evaluation modals call this on open so they + can pre-fill the prompt textarea. Returning the constant from the + backend (rather than hard-coding it in the frontend) keeps the + fallback used by the worker and the placeholder shown in the UI + in lock-step — operators always see the *actual* default they'd + get if they leave the field blank. + + Registered before ``GET /{call_import_id}`` so the static path is + not mistaken for a UUID import id (which would 422). + """ + del api_key, organization_id + from app.workers.tasks.helpers.llm_diarisation import ( + DEFAULT_DIARIZATION_PROMPT, + ) + + return CallImportDiarisationPromptDefaultResponse( + prompt=DEFAULT_DIARIZATION_PROMPT + ) + + @router.patch( "/{call_import_id}", response_model=CallImportResponse, @@ -2983,34 +3028,6 @@ async def bulk_delete_call_import_rows( # --------------------------------------------------------------------------- -@router.get( - "/diarisation-prompt-default", - response_model=CallImportDiarisationPromptDefaultResponse, - operation_id="getCallImportDiarisationPromptDefault", -) -async def get_call_import_diarisation_prompt_default( - api_key: str = Depends(get_api_key), - organization_id: UUID = Depends(get_organization_id), -) -> CallImportDiarisationPromptDefaultResponse: - """Return the canonical LLM diariser prompt. - - The Transcribe / Run Evaluation modals call this on open so they - can pre-fill the prompt textarea. Returning the constant from the - backend (rather than hard-coding it in the frontend) keeps the - fallback used by the worker and the placeholder shown in the UI - in lock-step — operators always see the *actual* default they'd - get if they leave the field blank. - """ - del api_key, organization_id - from app.workers.tasks.helpers.llm_diarisation import ( - DEFAULT_DIARIZATION_PROMPT, - ) - - return CallImportDiarisationPromptDefaultResponse( - prompt=DEFAULT_DIARIZATION_PROMPT - ) - - def _select_rows_for_transcription( db: Session, call_import: CallImport, diff --git a/app/config.py b/app/config.py index da6588b7..4bf5d008 100644 --- a/app/config.py +++ b/app/config.py @@ -56,6 +56,7 @@ class Settings(BaseSettings): GCS_BUCKET_NAME: Optional[str] = None GCS_PROJECT_ID: Optional[str] = None GCS_CREDENTIALS_PATH: Optional[str] = None + GCS_SIGNING_SERVICE_ACCOUNT_EMAIL: Optional[str] = None GCS_PREFIX: str = "audio/" # Azure Blob Storage Configuration @@ -458,6 +459,10 @@ def load_config_from_file(config_path: str) -> None: settings.GCS_PROJECT_ID = gcs_config["project_id"] if "credentials_path" in gcs_config: settings.GCS_CREDENTIALS_PATH = gcs_config["credentials_path"] + if "signing_service_account_email" in gcs_config: + settings.GCS_SIGNING_SERVICE_ACCOUNT_EMAIL = gcs_config[ + "signing_service_account_email" + ] if "prefix" in gcs_config: settings.GCS_PREFIX = gcs_config["prefix"] diff --git a/app/models/schemas.py b/app/models/schemas.py index ace77f74..ae2513e7 100644 --- a/app/models/schemas.py +++ b/app/models/schemas.py @@ -2209,9 +2209,10 @@ class CallImportSchemaParameterBase(BaseModel): "Parameter type. One of conversation_id / recording_url / " "recording_date / transcript / text / number / boolean / " "datetime / url. Exactly one parameter of type " - "'conversation_id' must be present; at most one each of " - "'recording_url', 'recording_date', and 'transcript'. Only " - "'conversation_id' is forced required." + "'conversation_id' and exactly one of type 'recording_url' " + "must be present; at most one each of 'recording_date' and " + "'transcript'. Both conversation_id and recording_url are " + "forced required." ), ) description: Optional[str] = Field( @@ -2223,8 +2224,9 @@ class CallImportSchemaParameterBase(BaseModel): default=False, description=( "When True, the parameter must be mapped to a CSV column on " - "every upload. The ``conversation_id`` parameter is always " - "required and is force-set to True by the server." + "every upload. The ``conversation_id`` and ``recording_url`` " + "parameters are always required and are force-set to True by " + "the server." ), ) @@ -2279,6 +2281,11 @@ def _validate_schema_parameters( "Schema must contain exactly one parameter of type " "'conversation_id'." ) + if rec_url_count != 1: + raise ValueError( + "Schema must contain exactly one parameter of type " + "'recording_url'." + ) if recording_date_count > 1: raise ValueError( "Schema may contain at most one parameter of type " diff --git a/app/services/storage/gcs_service.py b/app/services/storage/gcs_service.py index 57b569e4..06a53550 100644 --- a/app/services/storage/gcs_service.py +++ b/app/services/storage/gcs_service.py @@ -69,6 +69,16 @@ def _resolve_credentials_path() -> Optional[str]: return str(resolved) if resolved.exists() else str(path) +_GCS_SIGNING_UNAVAILABLE_MSG = ( + "GCS signed URLs require signing credentials. Provide a service account JSON " + "(gcs.credentials_path or GOOGLE_APPLICATION_CREDENTIALS), or configure " + "GKE Workload Identity with the IAM Credentials API enabled and grant " + "roles/iam.serviceAccountTokenCreator to the workload service account on itself. " + "Optional: set gcs.signing_service_account_email when the service account email " + "is not detected from ADC." +) + + class GcsService: """Service for managing GCS file storage.""" @@ -129,6 +139,35 @@ def _get_signing_credentials(self): return None + def _get_iam_signing_params(self) -> Optional[Tuple[str, str]]: + """Return (service_account_email, access_token) for IAM signBlob URL signing.""" + if self.gcs_client is None: + return None + + creds = getattr(self.gcs_client, "_credentials", None) + if creds is None: + return None + + sa_email = settings.GCS_SIGNING_SERVICE_ACCOUNT_EMAIL or getattr( + creds, "service_account_email", None + ) or getattr(creds, "signer_email", None) + if not sa_email: + return None + + try: + from google.auth.transport import requests as auth_requests + + auth_request = auth_requests.Request() + if not creds.valid: + creds.refresh(auth_request) + token = creds.token + if token: + return sa_email, token + except Exception: + return None + + return None + def _ensure_initialized(self): """Lazily initialize GCS client if not already initialized.""" if self.gcs_client is not None and self.bucket is not None: @@ -524,20 +563,28 @@ def generate_presigned_url_by_key(self, key: str, expiration: int = 3600) -> str _, NotFound, GoogleCloudError = _get_gcs_exception_types() credentials = self._get_signing_credentials() - if credentials is None: - raise StorageError( - "GCS signed URLs require a service account JSON with a private key. " - "Set gcs.credentials_path or GOOGLE_APPLICATION_CREDENTIALS." - ) + iam_params = None if credentials is not None else self._get_iam_signing_params() + if credentials is None and iam_params is None: + raise StorageError(_GCS_SIGNING_UNAVAILABLE_MSG) try: blob = self.bucket.blob(key) - url = blob.generate_signed_url( - version="v4", - expiration=timedelta(seconds=expiration), - method="GET", - credentials=credentials, - ) + if credentials is not None: + url = blob.generate_signed_url( + version="v4", + expiration=timedelta(seconds=expiration), + method="GET", + credentials=credentials, + ) + else: + sa_email, access_token = iam_params + url = blob.generate_signed_url( + version="v4", + expiration=timedelta(seconds=expiration), + method="GET", + service_account_email=sa_email, + access_token=access_token, + ) return url except GoogleCloudError as e: raise StorageError(f"Failed to generate signed URL: {str(e)}") diff --git a/app/services/telephony/exotel_client.py b/app/services/telephony/exotel_client.py index b8feb31f..f043bcf8 100644 --- a/app/services/telephony/exotel_client.py +++ b/app/services/telephony/exotel_client.py @@ -19,6 +19,16 @@ DEFAULT_TIMEOUT_SECONDS = 60.0 DEFAULT_MAX_RECORDING_BYTES = 50 * 1024 * 1024 # 50 MB +# Recognized Exotel REST API bases. ``sip_domain`` on integrations is also +# used by Plivo for SIP routing; for Exotel we only treat values matching +# these patterns as Calls API hosts — legacy SIP routing hosts are ignored. +_KNOWN_EXOTEL_API_HOSTS = frozenset( + { + "api.exotel.com", + "api.in.exotel.com", + } +) + class ExotelAuthError(Exception): """Exotel rejected the credentials (HTTP 401/403). Not retryable.""" @@ -159,14 +169,87 @@ def download_recording(self, recording_url: str) -> Tuple[bytes, str]: ) +def _hostname_from_api_host_value(value: str) -> str: + """Extract a lowercase hostname from a bare host, host:port, or URL.""" + from urllib.parse import urlparse + + raw = value.strip().rstrip("/") + if "://" in raw: + parsed = urlparse(raw) + host = parsed.hostname or "" + else: + host = raw.split("/", 1)[0].split("@", 1)[-1] + if ":" in host and not host.startswith("["): + host = host.rsplit(":", 1)[0] + return host.lower() + + +def is_exotel_rest_api_host(api_host: Optional[str]) -> bool: + """Return True when ``api_host`` is a known Exotel REST API base. + + SIP routing domains (``sip.*``, customer SIP hosts, etc.) must not be + used for Calls API lookups — those integrations fall back to the + configured default instead. + """ + if not api_host or not api_host.strip(): + return False + host = _hostname_from_api_host_value(api_host) + if host in _KNOWN_EXOTEL_API_HOSTS: + return True + # Future Exotel REST shards that follow api..exotel.com. + return host.startswith("api.") and host.endswith(".exotel.com") + + +def validate_exotel_api_host_for_save(api_host: Optional[str]) -> None: + """Reject non-empty Exotel API host values that are not REST API bases.""" + if not api_host or not str(api_host).strip(): + return + if not is_exotel_rest_api_host(str(api_host)): + raise ValueError( + "Exotel API Host must be a REST API base such as api.exotel.com or " + "api.in.exotel.com. SIP routing domains belong on Plivo integrations, " + "not here." + ) + + +def resolve_exotel_api_base(api_host: Optional[str] = None) -> str: + """Pick the Exotel REST base URL for call-import / telephony clients. + + Priority: per-integration API Host (``sip_domain``) when it matches a + recognized Exotel REST host → ``EXOTEL_API_BASE`` in config → Singapore + default. Unrecognized ``sip_domain`` values (e.g. legacy SIP routing + hosts) are ignored with a warning so call-id lookup does not hit the + wrong service. + """ + if api_host and api_host.strip(): + if is_exotel_rest_api_host(api_host): + host = api_host.strip().rstrip("/") + if not host.startswith(("http://", "https://")): + host = f"https://{host}" + return host + logger.warning( + "Ignoring Exotel integration API host {!r}: not a recognized REST " + "API base (expected api.exotel.com or api.in.exotel.com). " + "Falling back to configured default.", + api_host.strip(), + ) + + configured = getattr(settings, "EXOTEL_API_BASE", None) + if configured and str(configured).strip(): + return str(configured).strip().rstrip("/") + + return DEFAULT_API_BASE + + def build_exotel_client_from_integration( auth_id: str, auth_token: str, account_sid: Optional[str] = None, + api_host: Optional[str] = None, ) -> ExotelClient: - """Helper that reads any optional overrides from settings.""" + """Helper that reads optional API-host and timeout overrides from settings.""" - api_base = getattr(settings, "EXOTEL_API_BASE", None) or DEFAULT_API_BASE + api_base = resolve_exotel_api_base(api_host) timeout = float(getattr(settings, "EXOTEL_HTTP_TIMEOUT_SECONDS", DEFAULT_TIMEOUT_SECONDS)) max_bytes = int( getattr(settings, "EXOTEL_MAX_RECORDING_BYTES", DEFAULT_MAX_RECORDING_BYTES) diff --git a/app/services/telephony/telephony_service.py b/app/services/telephony/telephony_service.py index 91e2a73a..9996dcbd 100644 --- a/app/services/telephony/telephony_service.py +++ b/app/services/telephony/telephony_service.py @@ -82,6 +82,7 @@ def get_provider_client( auth_id=auth_id, auth_token=auth_token, account_sid=integration.voice_app_id, + api_host=integration.sip_domain, ) raise ValueError(f"Unsupported telephony provider: {provider}") @@ -150,6 +151,13 @@ def save_integration( if provider.lower() == "exotel" and not effective_voice_app_id: raise ValueError("voice_app_id (Exotel Account SID) is required for Exotel") + if provider.lower() == "exotel" and "sip_domain" in data: + from app.services.telephony.exotel_client import ( + validate_exotel_api_host_for_save, + ) + + validate_exotel_api_host_for_save(data.get("sip_domain")) + becoming_default = bool(data.get("is_default")) if "is_default" in data else False if integration: diff --git a/app/workers/tasks/process_call_import_row.py b/app/workers/tasks/process_call_import_row.py index 9989759c..b1685192 100644 --- a/app/workers/tasks/process_call_import_row.py +++ b/app/workers/tasks/process_call_import_row.py @@ -39,6 +39,13 @@ _RETRYABLE_COUNTDOWN_SECONDS = 60 +def _use_credentialed_recording_download(call_import, client) -> bool: + """True when CSV recording URLs should be fetched with provider auth.""" + if client is None or not hasattr(client, "download_recording"): + return False + return (call_import.provider or "").lower() == "exotel" + + def _is_direct_url_import(call_import) -> bool: """True only when the batch was explicitly imported without telephony creds. @@ -308,7 +315,10 @@ def process_call_import_row_task(self, row_id: str): if audio_bytes is None and original_csv_url: try: - fetched = download_public_recording(original_csv_url) + if _use_credentialed_recording_download(call_import, client): + fetched = client.download_recording(original_csv_url) + else: + fetched = download_public_recording(original_csv_url) audio_bytes, content_type = fetched used_url = original_csv_url if primary_failure is not None: diff --git a/config.yml.example b/config.yml.example index 5b6587f4..ccc22d0c 100644 --- a/config.yml.example +++ b/config.yml.example @@ -69,6 +69,7 @@ gcs: bucket_name: "your-gcs-bucket-name" project_id: "your-gcp-project-id" credentials_path: null # Optional path to service-account JSON; falls back to GOOGLE_APPLICATION_CREDENTIALS / ADC + signing_service_account_email: null # Optional override for IAM signBlob when ADC does not expose the SA email (GKE Workload Identity) prefix: "audio/" # Prefix for audio files in bucket (same layout as S3) # Azure Blob Storage Configuration (alternative when storage.blob_provider is azure) diff --git a/docs-fumadocs/content/docs/getting-started/cloud-storage.mdx b/docs-fumadocs/content/docs/getting-started/cloud-storage.mdx index 53cf03f7..6aff3d41 100644 --- a/docs-fumadocs/content/docs/getting-started/cloud-storage.mdx +++ b/docs-fumadocs/content/docs/getting-started/cloud-storage.mdx @@ -136,6 +136,7 @@ gcs: bucket_name: "your-gcs-bucket-name" project_id: "your-gcp-project-id" credentials_path: null # Optional path to service-account JSON + signing_service_account_email: null # Optional override for IAM signBlob (GKE Workload Identity) prefix: "audio/" ``` @@ -145,6 +146,7 @@ gcs: | `bucket_name` | Yes | Name of your GCS bucket | | `project_id` | Yes | GCP project ID | | `credentials_path` | No | Path to a service-account JSON key file | +| `signing_service_account_email` | No | Override service account email for signed URL generation when ADC does not expose it | | `prefix` | No | Folder prefix for uploaded files (default: `audio/`) | ### Authentication @@ -155,15 +157,32 @@ EfficientAI resolves GCS credentials in this order: 2. `GOOGLE_APPLICATION_CREDENTIALS` environment variable 3. Application Default Credentials (ADC) on GCE, GKE, or Cloud Run +Uploads and server-side downloads work with ADC alone (including **GKE Workload Identity**). **Signed URLs** for browser playback require either a service account JSON key with a private key, or Workload Identity plus IAM signBlob (see below). + The `google-cloud-storage` client is included in the standard EfficientAI install (`pip install -e .`); no extra Python package step is required for GCS. ### GCP setup 1. **Create a GCS bucket** in your project (Console or `gcloud storage buckets create`). 2. **Create a service account** with object read/write access (e.g., `roles/storage.objectAdmin` on the bucket, or a tighter custom role). -3. **Download a JSON key** and set `credentials_path`, or mount the key and set `GOOGLE_APPLICATION_CREDENTIALS`. +3. **Authenticate the app** using one of: + - **GKE Workload Identity (recommended):** bind your Kubernetes service account to the GCP service account. Enable the [IAM Credentials API](https://cloud.google.com/iam/docs/reference/credentials/rest) and grant the GCP service account `roles/iam.serviceAccountTokenCreator` **on itself** so EfficientAI can sign playback URLs without a JSON key. + - **Service account JSON key:** download a key and set `credentials_path`, or mount the key and set `GOOGLE_APPLICATION_CREDENTIALS`. 4. Set `storage.blob_provider: gcs`, `gcs.enabled: true`, and fill in `bucket_name` and `project_id`. +Example IAM binding for Workload Identity signed URLs: + +```bash +gcloud services enable iamcredentials.googleapis.com + +gcloud iam service-accounts add-iam-policy-binding \ + WORKLOAD_SA@PROJECT.iam.gserviceaccount.com \ + --role=roles/iam.serviceAccountTokenCreator \ + --member="serviceAccount:WORKLOAD_SA@PROJECT.iam.gserviceaccount.com" +``` + +If ADC does not expose the service account email (some federation setups), set `gcs.signing_service_account_email` to the workload GCP service account email. + Object layout under `prefix` matches S3 (organization-scoped paths), so you can migrate between providers without changing application logic. --- @@ -253,6 +272,7 @@ Use **Test Connection** on the Data Sources page, or upload a test file and conf | GCS `Forbidden` | Confirm service account has `storage.objects.*` on the bucket | | GCS bucket not found | Verify `bucket_name`, `project_id`, and that the bucket exists | | GCS auth failure | Set `credentials_path` or `GOOGLE_APPLICATION_CREDENTIALS`; on GCP VMs you can use ADC | +| GCS signed URL failure on GKE | Enable IAM Credentials API; grant `roles/iam.serviceAccountTokenCreator` to the workload SA on itself; optionally set `signing_service_account_email` | | Wrong provider in UI | Ensure `storage.blob_provider` matches the enabled block (`s3`, `gcs`, or `azure`) | | Azure container not found | Verify `container_name` and that the container exists in the storage account | | Azure auth failure | Set `connection_string` or `account_name` + `account_key` | diff --git a/docs-fumadocs/content/docs/reference/configuration.mdx b/docs-fumadocs/content/docs/reference/configuration.mdx index 15e2778a..19adbdb1 100644 --- a/docs-fumadocs/content/docs/reference/configuration.mdx +++ b/docs-fumadocs/content/docs/reference/configuration.mdx @@ -73,6 +73,7 @@ gcs: bucket_name: "your-gcs-bucket-name" project_id: "your-gcp-project-id" credentials_path: null # service-account JSON; or GOOGLE_APPLICATION_CREDENTIALS / ADC + signing_service_account_email: null # optional override for IAM signBlob (GKE Workload Identity) prefix: "audio/" # Azure Blob Storage (optional, used when blob_provider is azure) diff --git a/frontend/src/pages/callImports/CallImportDetail.tsx b/frontend/src/pages/callImports/CallImportDetail.tsx index 75366da0..07c95043 100644 --- a/frontend/src/pages/callImports/CallImportDetail.tsx +++ b/frontend/src/pages/callImports/CallImportDetail.tsx @@ -1814,13 +1814,10 @@ export default function CallImportDetail() { {preImport && data.source_s3_key && ( <> - {data.status === 'uploaded' && } - {data.status === 'mapped' && ( - <> - - - + {(data.status === 'uploaded' || data.status === 'mapped') && ( + )} + {data.status === 'mapped' && } )} diff --git a/frontend/src/pages/callImports/Schemas.tsx b/frontend/src/pages/callImports/Schemas.tsx index d503ca1d..0fb167de 100644 --- a/frontend/src/pages/callImports/Schemas.tsx +++ b/frontend/src/pages/callImports/Schemas.tsx @@ -62,6 +62,20 @@ function makeConversationIdParameter(): EditableParameter { } } +function makeRecordingUrlParameter(): EditableParameter { + return { + key: 'recording_url', + name: 'recording_url', + type: 'recording_url', + description: 'URL of the call recording for each imported row.', + is_required: true, + } +} + +function isSystemRequiredParameter(param: EditableParameter): boolean { + return param.type === 'conversation_id' || param.type === 'recording_url' +} + function parametersFromSchema( parameters: CallImportSchemaParameter[], ): EditableParameter[] { @@ -95,6 +109,9 @@ function validateParameters(params: EditableParameter[]): string | null { if (convCount !== 1) { return 'Exactly one parameter must be of type "conversation_id".' } + if (recordingCount !== 1) { + return 'Exactly one parameter must be of type "recording_url".' + } if (recordingDateCount > 1) { return 'At most one parameter can be of type "recording_date".' } @@ -121,7 +138,7 @@ function SchemaEditor({ open, schema, onClose, onSaved }: SchemaEditorProps) { const [parameters, setParameters] = useState( schema ? parametersFromSchema(schema.parameters) - : [makeConversationIdParameter()], + : [makeConversationIdParameter(), makeRecordingUrlParameter()], ) const [errorMsg, setErrorMsg] = useState(null) @@ -129,6 +146,10 @@ function SchemaEditor({ open, schema, onClose, onSaved }: SchemaEditorProps) { () => parameters.findIndex((p) => p.type === 'conversation_id'), [parameters], ) + const recordingUrlIdx = useMemo( + () => parameters.findIndex((p) => p.type === 'recording_url'), + [parameters], + ) // Reset local state whenever the editor opens against a new target so a // stale parameter list doesn't bleed across edit / new sessions. @@ -139,7 +160,7 @@ function SchemaEditor({ open, schema, onClose, onSaved }: SchemaEditorProps) { setParameters( schema ? parametersFromSchema(schema.parameters) - : [makeConversationIdParameter()], + : [makeConversationIdParameter(), makeRecordingUrlParameter()], ) setErrorMsg(null) } @@ -154,8 +175,7 @@ function SchemaEditor({ open, schema, onClose, onSaved }: SchemaEditorProps) { name: p.name.trim(), type: p.type, description: p.description.trim() || null, - is_required: - p.type === 'conversation_id' ? true : p.is_required, + is_required: isSystemRequiredParameter(p) ? true : p.is_required, })), }), onSuccess: () => { @@ -178,8 +198,7 @@ function SchemaEditor({ open, schema, onClose, onSaved }: SchemaEditorProps) { name: p.name.trim(), type: p.type, description: p.description.trim() || null, - is_required: - p.type === 'conversation_id' ? true : p.is_required, + is_required: isSystemRequiredParameter(p) ? true : p.is_required, })), }) }, @@ -222,7 +241,16 @@ function SchemaEditor({ open, schema, onClose, onSaved }: SchemaEditorProps) { } const removeParameter = (idx: number) => { - setParameters((prev) => prev.filter((_, i) => i !== idx)) + setParameters((prev) => { + const param = prev[idx] + if ( + param?.type === 'conversation_id' || + param?.type === 'recording_url' + ) { + return prev + } + return prev.filter((_, i) => i !== idx) + }) } const moveParameter = (idx: number, direction: 'up' | 'down') => { @@ -230,11 +258,14 @@ function SchemaEditor({ open, schema, onClose, onSaved }: SchemaEditorProps) { const next = [...prev] const target = direction === 'up' ? idx - 1 : idx + 1 if (target < 0 || target >= next.length) return prev - // Keep the conversation_id row pinned at the top — it must always - // be index 0 so the UI can lock it visually. + // Keep the system rows pinned — conversation_id at the top and + // recording_url immediately after it. + const pinnedTypes = new Set(['conversation_id', 'recording_url']) if ( - next[idx].type === 'conversation_id' || - next[target].type === 'conversation_id' + pinnedTypes.has(next[idx].type) || + pinnedTypes.has(next[target].type) || + target <= conversationIdIdx || + (recordingUrlIdx >= 0 && target <= recordingUrlIdx) ) { return prev } @@ -319,17 +350,19 @@ function SchemaEditor({ open, schema, onClose, onSaved }: SchemaEditorProps) {
{parameters.map((p, idx) => { const isConversationId = p.type === 'conversation_id' - const locked = isConversationId - const isSystemRequired = isConversationId + const isRecordingUrl = p.type === 'recording_url' + const nameLocked = isConversationId + const typeLocked = isConversationId || isRecordingUrl + const isSystemRequired = isSystemRequiredParameter(p) return (
- {locked ? ( + {typeLocked ? ( ) : ( <> @@ -339,7 +372,8 @@ function SchemaEditor({ open, schema, onClose, onSaved }: SchemaEditorProps) { disabled={ isSubmitting || idx === 0 || - idx - 1 === conversationIdIdx + idx - 1 === conversationIdIdx || + idx - 1 === recordingUrlIdx } className="leading-none text-[10px] disabled:opacity-30" aria-label="Move up" @@ -351,7 +385,10 @@ function SchemaEditor({ open, schema, onClose, onSaved }: SchemaEditorProps) { type="button" onClick={() => moveParameter(idx, 'down')} disabled={ - isSubmitting || idx === parameters.length - 1 + isSubmitting || + idx === parameters.length - 1 || + idx + 1 === conversationIdIdx || + idx + 1 === recordingUrlIdx } className="leading-none text-[10px] disabled:opacity-30" aria-label="Move down" @@ -368,15 +405,15 @@ function SchemaEditor({ open, schema, onClose, onSaved }: SchemaEditorProps) { onChange={(e) => updateParameter(idx, { name: e.target.value }) } - disabled={locked || isSubmitting} + disabled={nameLocked || isSubmitting} placeholder="parameter_name" className="w-full px-2 py-1.5 text-sm border border-gray-300 rounded focus:ring-2 focus:ring-primary-500 disabled:bg-gray-50" />
- {locked ? ( + {typeLocked ? ( - Conversation ID + {isConversationId ? 'Conversation ID' : 'Recording URL'} ) : ( setTelephonySipDomain(e.target.value)} - placeholder={selectedTelephonyProvider === TelephonyProvider.EXOTEL ? 'Optional: api.exotel.com' : 'Optional: SIP domain'} className="w-full px-3 py-2 border border-gray-300 rounded-lg focus:ring-2 focus:ring-green-500" /> + placeholder={selectedTelephonyProvider === TelephonyProvider.EXOTEL ? 'Optional: api.exotel.com or api.in.exotel.com' : 'Optional: SIP domain'} className="w-full px-3 py-2 border border-gray-300 rounded-lg focus:ring-2 focus:ring-green-500" />

Credentials are encrypted and stored securely. Your browser never displays stored secrets.

diff --git a/tests/conftest.py b/tests/conftest.py index 895dde42..0d400de9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -298,6 +298,24 @@ def update_threshold_defaults(self, *_args, **_kwargs): fake_workers_tasks_pkg.__path__ = [] sys.modules["app.workers.tasks"] = fake_workers_tasks_pkg + # ``app.workers.tasks`` is stubbed with an empty ``__path__`` so Celery + # task modules are not eagerly imported, but several API routes and tests + # still need the real ``helpers`` subpackage (e.g. the diariser default + # prompt endpoint). Register it explicitly so + # ``app.workers.tasks.helpers.llm_diarisation`` resolves normally. + if "app.workers.tasks.helpers" not in sys.modules: + helpers_pkg = types.ModuleType("app.workers.tasks.helpers") + helpers_pkg.__path__ = [ + str( + Path(__file__).resolve().parents[1] + / "app" + / "workers" + / "tasks" + / "helpers" + ) + ] + sys.modules["app.workers.tasks.helpers"] = helpers_pkg + fake_run_prompt_opt_module = sys.modules.get("app.workers.tasks.run_prompt_optimization") if fake_run_prompt_opt_module is None: fake_run_prompt_opt_module = types.ModuleType("app.workers.tasks.run_prompt_optimization") diff --git a/tests/test_api/test_call_import_datasets_tags.py b/tests/test_api/test_call_import_datasets_tags.py index f36ea690..79322b6c 100644 --- a/tests/test_api/test_call_import_datasets_tags.py +++ b/tests/test_api/test_call_import_datasets_tags.py @@ -105,7 +105,7 @@ def upload_schema(db_session, org_id, seed_org): schema_id=schema.id, name=name, type=ptype.value, - is_required=name in {"conversation_id", "recording_date"}, + is_required=name in {"conversation_id", "recording_date", "recording_url"}, ordering=idx, ) ) diff --git a/tests/test_api/test_call_import_diarization_and_eval_llm.py b/tests/test_api/test_call_import_diarization_and_eval_llm.py index ffd32385..8f3fbcc0 100644 --- a/tests/test_api/test_call_import_diarization_and_eval_llm.py +++ b/tests/test_api/test_call_import_diarization_and_eval_llm.py @@ -1239,8 +1239,9 @@ def test_diarisation_prompt_default_endpoint_returns_canonical_constant(): from app.api.v1.routes.call_imports import ( get_call_import_diarisation_prompt_default, ) - from app.workers.tasks.helpers.llm_diarisation import ( - DEFAULT_DIARIZATION_PROMPT, + + llm_diarisation = _resolve_submodule( + "app.workers.tasks.helpers.llm_diarisation" ) response = asyncio.run( @@ -1248,7 +1249,22 @@ def test_diarisation_prompt_default_endpoint_returns_canonical_constant(): api_key="ignored", organization_id=uuid4() ) ) - assert response.prompt == DEFAULT_DIARIZATION_PROMPT + assert response.prompt == llm_diarisation.DEFAULT_DIARIZATION_PROMPT + + +def test_diarisation_prompt_default_http_route_not_shadowed_by_import_id( + authenticated_client, +): + """Static path must be registered before ``GET /{call_import_id}``.""" + expected_prompt = _resolve_submodule( + "app.workers.tasks.helpers.llm_diarisation" + ).DEFAULT_DIARIZATION_PROMPT + + response = authenticated_client.get( + "/api/v1/call-imports/diarisation-prompt-default" + ) + assert response.status_code == 200 + assert response.json()["prompt"] == expected_prompt # --------------------------------------------------------------------------- diff --git a/tests/test_api/test_call_import_evaluations_export.py b/tests/test_api/test_call_import_evaluations_export.py index 6a5a4e5d..7f8d723e 100644 --- a/tests/test_api/test_call_import_evaluations_export.py +++ b/tests/test_api/test_call_import_evaluations_export.py @@ -64,6 +64,47 @@ def _make_call_import(db_session, org_id, *, custom_columns=None): return call_import +def _make_audio_call_import(db_session, org_id): + workspace = _ensure_default_workspace(db_session, org_id) + call_import = CallImport( + id=uuid4(), + organization_id=org_id, + workspace_id=workspace.id, + provider=None, + original_filename="sales call.wav", + source_format="audio", + column_mapping={}, + extra_columns=[], + custom_column_mapping={}, + total_rows=1, + completed_rows=1, + failed_rows=0, + status=CallImportStatus.COMPLETED, + ) + db_session.add(call_import) + db_session.commit() + return call_import + + +def _make_audio_call_import_row( + db_session, call_import, *, conversation_id="sales_call", row_index=0 +): + row = CallImportRow( + id=uuid4(), + call_import_id=call_import.id, + organization_id=call_import.organization_id, + row_index=row_index, + conversation_id=conversation_id, + recording_url=None, + transcript="hello world", + raw_columns={"conversation_id": conversation_id}, + status=CallImportRowStatus.COMPLETED, + ) + db_session.add(row) + db_session.commit() + return row + + def _make_call_import_row(db_session, call_import, row_index=0, raw_columns=None): row = CallImportRow( id=uuid4(), @@ -661,3 +702,63 @@ def test_export_defaults_to_csv_when_format_omitted( assert response.headers["content-type"].startswith("text/csv") disposition = response.headers.get("content-disposition", "") assert ".csv" in disposition + + +def test_export_includes_conversation_id_for_audio_upload( + authenticated_client, db_session, org_id, seed_org +): + """Manual audio uploads have no schema/column_mapping, but each row's + filename-derived conversation_id must still appear in CSV/XLSX exports.""" + import io as io_module + + from openpyxl import load_workbook + + call_import = _make_audio_call_import(db_session, org_id) + source_row = _make_audio_call_import_row( + db_session, call_import, conversation_id="sales_call" + ) + + metric = _make_metric( + db_session, org_id, name="Quality", capture_rationale=False + ) + evaluation = _make_evaluation_with_row( + db_session, + call_import=call_import, + source_row=source_row, + metrics=[metric], + metric_scores={ + str(metric.id): { + "value": 0.9, + "type": "rating", + "metric_name": "Quality", + } + }, + ) + + csv_response = authenticated_client.get( + f"/api/v1/call-imports/{call_import.id}/evaluations/{evaluation.id}/export" + ) + assert csv_response.status_code == 200, csv_response.text + headers, rows = _parse_csv(csv_response.content) + + assert headers[0] == "conversation_id" + assert len(rows) == 1 + assert rows[0]["conversation_id"] == "sales_call" + assert rows[0]["Quality"] == "0.9" + + xlsx_response = authenticated_client.get( + f"/api/v1/call-imports/{call_import.id}/evaluations/{evaluation.id}/export", + params={"format": "xlsx"}, + ) + assert xlsx_response.status_code == 200, xlsx_response.text + + workbook = load_workbook(io_module.BytesIO(xlsx_response.content), read_only=True) + worksheet = workbook.active + rows_iter = worksheet.iter_rows(values_only=True) + header_row = list(next(rows_iter)) + data_rows = [list(r) for r in rows_iter] + + assert header_row[0] == "conversation_id" + assert len(data_rows) == 1 + assert data_rows[0][header_row.index("conversation_id")] == "sales_call" + assert data_rows[0][header_row.index("Quality")] == "0.9" diff --git a/tests/test_api/test_call_import_schemas.py b/tests/test_api/test_call_import_schemas.py index 2c3be484..09e142a8 100644 --- a/tests/test_api/test_call_import_schemas.py +++ b/tests/test_api/test_call_import_schemas.py @@ -53,7 +53,7 @@ def _minimal_payload(name: str = "Standard QA") -> dict: { "name": "recording_url", "type": "recording_url", - "is_required": False, + "is_required": True, }, { "name": "recording_date", @@ -102,14 +102,14 @@ def test_create_schema_happy_path(authenticated_client, db_session, org_id, seed assert len(schema.parameters) == 4 -def test_create_schema_forces_conversation_id_required( +def test_create_schema_forces_system_required_parameters( authenticated_client, db_session, org_id, seed_org ): - """Even if the client sends is_required=False for conversation_id, - the server stamps it back to True (the parameter is mandatory by - definition). recording_date stays optional when the client omits it.""" + """Even if the client sends is_required=False for conversation_id or + recording_url, the server stamps them back to True.""" payload = _minimal_payload() payload["parameters"][0]["is_required"] = False + payload["parameters"][1]["is_required"] = False payload["parameters"][2]["is_required"] = False response = authenticated_client.post("/api/v1/call-import-schemas", json=payload) @@ -118,10 +118,14 @@ def test_create_schema_forces_conversation_id_required( conv_param = next( p for p in body["parameters"] if p["type"] == "conversation_id" ) + rec_param = next( + p for p in body["parameters"] if p["type"] == "recording_url" + ) date_param = next( p for p in body["parameters"] if p["type"] == "recording_date" ) assert conv_param["is_required"] is True + assert rec_param["is_required"] is True assert date_param["is_required"] is False @@ -141,6 +145,18 @@ def test_create_schema_rejects_missing_conversation_id( assert "conversation_id" in response.text.lower() +def test_create_schema_rejects_missing_recording_url( + authenticated_client, db_session, org_id, seed_org +): + payload = _minimal_payload() + payload["parameters"] = [ + p for p in payload["parameters"] if p["type"] != "recording_url" + ] + response = authenticated_client.post("/api/v1/call-import-schemas", json=payload) + assert response.status_code == 422 + assert "recording_url" in response.text.lower() + + def test_create_schema_accepts_missing_recording_date( authenticated_client, db_session, org_id, seed_org ): @@ -322,6 +338,7 @@ def test_update_schema_replaces_parameters( new_params = [ {"name": "conversation_id", "type": "conversation_id", "is_required": True}, + {"name": "recording_url", "type": "recording_url", "is_required": True}, {"name": "recording_date", "type": "recording_date", "is_required": True}, {"name": "agent_name", "type": "text"}, {"name": "latency_ms", "type": "number"}, @@ -334,7 +351,13 @@ def test_update_schema_replaces_parameters( body = response.json() assert body["name"] == "Renamed" names = [p["name"] for p in body["parameters"]] - assert names == ["conversation_id", "recording_date", "agent_name", "latency_ms"] + assert names == [ + "conversation_id", + "recording_url", + "recording_date", + "agent_name", + "latency_ms", + ] # The old parameters were actually deleted, not appended. persisted = ( @@ -342,9 +365,10 @@ def test_update_schema_replaces_parameters( .filter(CallImportSchemaParameter.schema_id == UUID(created["id"])) .all() ) - assert len(persisted) == 4 + assert len(persisted) == 5 assert {p.name for p in persisted} == { "conversation_id", + "recording_url", "recording_date", "agent_name", "latency_ms", @@ -407,13 +431,35 @@ def test_update_schema_accepts_dropping_recording_date( json={ "parameters": [ {"name": "conversation_id", "type": "conversation_id"}, + {"name": "recording_url", "type": "recording_url"}, {"name": "agent_name", "type": "text"}, ] }, ) assert response.status_code == 200, response.text names = [p["name"] for p in response.json()["parameters"]] - assert names == ["conversation_id", "agent_name"] + assert names == ["conversation_id", "recording_url", "agent_name"] + + +def test_update_schema_rejects_dropping_recording_url( + authenticated_client, db_session, org_id, seed_org +): + created = authenticated_client.post( + "/api/v1/call-import-schemas", json=_minimal_payload() + ).json() + + response = authenticated_client.patch( + f"/api/v1/call-import-schemas/{created['id']}", + json={ + "parameters": [ + {"name": "conversation_id", "type": "conversation_id"}, + {"name": "transcript", "type": "transcript"}, + {"name": "agent_name", "type": "text"}, + ] + }, + ) + assert response.status_code == 422 + assert "recording_url" in response.text.lower() # --------------------------------------------------------------------------- diff --git a/tests/test_api/test_call_imports_routes.py b/tests/test_api/test_call_imports_routes.py index 95cbbe7d..150b2dd4 100644 --- a/tests/test_api/test_call_imports_routes.py +++ b/tests/test_api/test_call_imports_routes.py @@ -86,6 +86,7 @@ def _standard_params() -> list[CallImportSchemaParameter]: _param( name="recording_url", type_=CallImportParameterType.RECORDING_URL, + is_required=True, ordering=2, ), _param( @@ -194,27 +195,25 @@ def test_parse_csv_rejects_missing_mapped_required_header(): assert "conversation_id" in exc.value.detail.lower() -def test_parse_csv_allows_optional_param_without_mapping(): +def test_parse_csv_rejects_unmapped_required_recording_url(): csv_text = ( "CallID,Recording Date,Transcript\n" "abc-1,18/05/2026,Hello world\n" ) - # Drop the recording_url mapping entry so it is treated as "not used". mapping = { "conversation_id": "CallID", "recording_date": "Recording Date", "transcript": "Transcript", } - rows = _parse_csv( - _csv_bytes(csv_text), - _standard_params(), - mapping, - _standard_skipped(), - ) - assert len(rows) == 1 - assert rows[0]["conversation_id"] == "abc-1" - assert rows[0]["recording_url"] is None - assert rows[0]["transcript"] == "Hello world" + with pytest.raises(HTTPException) as exc: + _parse_csv( + _csv_bytes(csv_text), + _standard_params(), + mapping, + _standard_skipped(), + ) + assert exc.value.status_code == 400 + assert "recording_url" in exc.value.detail.lower() def test_parse_csv_rejects_row_missing_conversation_id(): @@ -703,7 +702,7 @@ def _seed_schema( schema_id=schema.id, name="recording_url", type=CallImportParameterType.RECORDING_URL.value, - is_required=False, + is_required=True, ordering=2, ), CallImportSchemaParameter( diff --git a/tests/test_services/test_storage/test_gcs_service.py b/tests/test_services/test_storage/test_gcs_service.py index 41002ac8..a2697e1a 100644 --- a/tests/test_services/test_storage/test_gcs_service.py +++ b/tests/test_services/test_storage/test_gcs_service.py @@ -20,6 +20,7 @@ def __init__(self, name: str, bucket): self.updated = datetime.now(UTC) self.time_created = self.updated self._content = b"" + self.last_signed_url_kwargs = {} def upload_from_string(self, data, content_type=None): self._content = data @@ -38,6 +39,7 @@ def delete(self): self.bucket.objects.pop(self.name, None) def generate_signed_url(self, **_kwargs): + self.last_signed_url_kwargs = _kwargs return f"https://storage.googleapis.com/{self.bucket.name}/{self.name}?signed=1" @@ -59,9 +61,12 @@ class _FakeBucket: def __init__(self, name): self.name = name self.objects = {} + self._blobs = {} def blob(self, key): - return _FakeBlob(key, self) + if key not in self._blobs: + self._blobs[key] = _FakeBlob(key, self) + return self._blobs[key] def exists(self): return True @@ -71,6 +76,7 @@ class _FakeGcsClient: def __init__(self): self._bucket = _FakeBucket("bucket-a") self.list_calls = [] + self._credentials = None def bucket(self, name): self._bucket.name = name @@ -106,6 +112,9 @@ def configured_gcs(monkeypatch): monkeypatch.setattr(gcs_module.settings, "GCS_PROJECT_ID", "proj-1", raising=False) monkeypatch.setattr(gcs_module.settings, "GCS_PREFIX", "", raising=False) monkeypatch.setattr(gcs_module.settings, "GCS_CREDENTIALS_PATH", None, raising=False) + monkeypatch.setattr( + gcs_module.settings, "GCS_SIGNING_SERVICE_ACCOUNT_EMAIL", None, raising=False + ) monkeypatch.setattr(gcs_module.settings, "ALLOWED_AUDIO_FORMATS", ["mp3", "wav"], raising=False) fake_client = _FakeGcsClient() @@ -205,3 +214,64 @@ def test_generate_presigned_url_by_key(configured_gcs, monkeypatch): url = service.generate_presigned_url_by_key("audio/test.mp3") assert "signed=1" in url + + +class _FakeAdcCredentials: + def __init__(self, service_account_email="workload-sa@project.iam.gserviceaccount.com"): + self.service_account_email = service_account_email + self.token = "adc-access-token" + self.valid = True + + def refresh(self, _request): + self.valid = True + + +def test_generate_presigned_url_by_key_uses_iam_signing_when_no_private_key( + configured_gcs, monkeypatch +): + service, fake_client = configured_gcs + fake_client._bucket.objects["audio/test.mp3"] = b"x" + fake_client._credentials = _FakeAdcCredentials() + monkeypatch.setattr(service, "_get_signing_credentials", lambda: None) + + url = service.generate_presigned_url_by_key("audio/test.mp3") + + blob = fake_client._bucket.blob("audio/test.mp3") + assert "signed=1" in url + assert blob.last_signed_url_kwargs["service_account_email"] == ( + "workload-sa@project.iam.gserviceaccount.com" + ) + assert blob.last_signed_url_kwargs["access_token"] == "adc-access-token" + assert "credentials" not in blob.last_signed_url_kwargs + + +def test_generate_presigned_url_by_key_uses_configured_signing_email( + configured_gcs, monkeypatch +): + service, fake_client = configured_gcs + fake_client._bucket.objects["audio/test.mp3"] = b"x" + fake_client._credentials = _FakeAdcCredentials(service_account_email=None) + monkeypatch.setattr( + gcs_module.settings, + "GCS_SIGNING_SERVICE_ACCOUNT_EMAIL", + "override-sa@project.iam.gserviceaccount.com", + raising=False, + ) + monkeypatch.setattr(service, "_get_signing_credentials", lambda: None) + + service.generate_presigned_url_by_key("audio/test.mp3") + + blob = fake_client._bucket.blob("audio/test.mp3") + assert blob.last_signed_url_kwargs["service_account_email"] == ( + "override-sa@project.iam.gserviceaccount.com" + ) + + +def test_generate_presigned_url_by_key_raises_when_signing_unavailable(configured_gcs, monkeypatch): + service, fake_client = configured_gcs + fake_client._bucket.objects["audio/test.mp3"] = b"x" + monkeypatch.setattr(service, "_get_signing_credentials", lambda: None) + monkeypatch.setattr(service, "_get_iam_signing_params", lambda: None) + + with pytest.raises(StorageError, match="roles/iam.serviceAccountTokenCreator"): + service.generate_presigned_url_by_key("audio/test.mp3") diff --git a/tests/test_services/test_telephony/test_exotel_client.py b/tests/test_services/test_telephony/test_exotel_client.py new file mode 100644 index 00000000..c7688f2a --- /dev/null +++ b/tests/test_services/test_telephony/test_exotel_client.py @@ -0,0 +1,82 @@ +"""Tests for Exotel client helpers.""" + +import pytest + +from app.services.telephony.exotel_client import ( + DEFAULT_API_BASE, + build_exotel_client_from_integration, + is_exotel_rest_api_host, + resolve_exotel_api_base, + validate_exotel_api_host_for_save, +) + + +def test_resolve_exotel_api_base_prefers_integration_api_host(monkeypatch): + monkeypatch.setattr( + "app.services.telephony.exotel_client.settings", + type("SettingsStub", (), {"EXOTEL_API_BASE": "https://api.exotel.com"})(), + ) + assert ( + resolve_exotel_api_base("api.in.exotel.com") + == "https://api.in.exotel.com" + ) + assert ( + resolve_exotel_api_base("https://api.in.exotel.com/") + == "https://api.in.exotel.com" + ) + + +def test_resolve_exotel_api_base_falls_back_to_default(monkeypatch): + monkeypatch.setattr( + "app.services.telephony.exotel_client.settings", + type("SettingsStub", (), {})(), + ) + assert resolve_exotel_api_base(None) == DEFAULT_API_BASE + assert resolve_exotel_api_base("") == DEFAULT_API_BASE + + +def test_resolve_exotel_api_base_uses_config_when_no_integration_host(monkeypatch): + monkeypatch.setattr( + "app.services.telephony.exotel_client.settings", + type( + "SettingsStub", + (), + {"EXOTEL_API_BASE": "https://api.in.exotel.com"}, + )(), + ) + assert resolve_exotel_api_base(None) == "https://api.in.exotel.com" + + +def test_build_exotel_client_from_integration_uses_api_host(): + client = build_exotel_client_from_integration( + auth_id="key", + auth_token="token", + account_sid="acct", + api_host="api.in.exotel.com", + ) + assert client._api_base == "https://api.in.exotel.com" + + +def test_is_exotel_rest_api_host_rejects_sip_domains(): + assert is_exotel_rest_api_host("api.exotel.com") is True + assert is_exotel_rest_api_host("https://api.in.exotel.com/") is True + assert is_exotel_rest_api_host("sip.exotel.com") is False + assert is_exotel_rest_api_host("pbx.example.com") is False + + +def test_resolve_exotel_api_base_ignores_sip_domain_host(monkeypatch): + monkeypatch.setattr( + "app.services.telephony.exotel_client.settings", + type( + "SettingsStub", + (), + {"EXOTEL_API_BASE": "https://api.in.exotel.com"}, + )(), + ) + assert resolve_exotel_api_base("sip.exotel.com") == "https://api.in.exotel.com" + assert resolve_exotel_api_base("pbx.customer.example") == "https://api.in.exotel.com" + + +def test_validate_exotel_api_host_for_save_rejects_sip_domain(): + with pytest.raises(ValueError, match="REST API base"): + validate_exotel_api_host_for_save("sip.exotel.com") diff --git a/tests/test_workers/test_process_call_import_row.py b/tests/test_workers/test_process_call_import_row.py index 5ba9bb8f..319a059a 100644 --- a/tests/test_workers/test_process_call_import_row.py +++ b/tests/test_workers/test_process_call_import_row.py @@ -259,12 +259,10 @@ def test_process_call_import_row_completes_and_rolls_up_to_completed(db_session, assert call_import.failed_rows == 0 assert call_import.status == CallImportStatus.COMPLETED - # Even though the row had a CSV-supplied recording_url, the worker - # MUST authenticate via the Calls API using conversation_id first - # and download from the freshly resolved URL. - expected_resolved_url = f"https://api.exotel.com/recordings/{row.conversation_id}.mp3" + # Tier 1 (Calls API lookup) is preferred even when the CSV supplies a URL. assert fake_client.resolved_calls == [row.conversation_id] - assert fake_client.calls == [expected_resolved_url] + expected_url = f"https://api.exotel.com/recordings/{row.conversation_id}.mp3" + assert fake_client.calls == [expected_url] # CSV-supplied URL is preserved on the row when present (only the # resolver-derived URL gets persisted when the CSV had none). assert row.recording_url == original_csv_url @@ -411,21 +409,63 @@ def test_process_call_import_row_resolves_url_when_csv_omits_it(db_session, monk assert call_import.status == CallImportStatus.COMPLETED -def test_process_call_import_row_falls_back_to_csv_url_when_lookup_fails( +def test_process_call_import_row_exotel_csv_url_uses_credentialed_download( db_session, monkeypatch ): - """When the credentialed call-id lookup fails non-retryably and the CSV - supplied a recording URL, the worker must fall back to that URL rather - than fail the row outright.""" + """When call-id lookup fails, Exotel batches fall back to the CSV URL + using credentialed download.""" _, call_import, rows = _seed(db_session, row_count=1) row = rows[0] csv_url = row.recording_url - assert csv_url # sanity: seeded with a CSV URL + assert csv_url + + from app.services.telephony.exotel_client import ExotelNotFoundError + + class _LookupFailsCsvSucceeds: + def __init__(self): + self.resolved_calls = [] + self.calls = [] + + def get_call_recording_url(self, call_sid): + self.resolved_calls.append(call_sid) + raise ExotelNotFoundError(f"call {call_sid} not found in API") + + def download_recording(self, recording_url): + self.calls.append(recording_url) + return b"csv-url-audio", "audio/mpeg" + + fake_client = _LookupFailsCsvSucceeds() + fake_s3 = _FakeS3(enabled=True) + task_module = _patch_dependencies(monkeypatch, db_session, fake_client, fake_s3) + public_calls = _patch_public_download( + monkeypatch, + return_value=(b"should-not-be-used", "audio/mpeg"), + ) + + result = task_module.process_call_import_row_task.run(str(row.id)) + + assert result["status"] == "completed" + db_session.refresh(row) + assert fake_client.resolved_calls == [row.conversation_id] + assert fake_client.calls == [csv_url] + assert public_calls == [] + assert row.status == CallImportRowStatus.COMPLETED + + +def test_process_call_import_row_fails_when_lookup_fails_without_csv_url( + db_session, monkeypatch +): + """When recording_url is absent and call-id lookup fails, the row fails.""" + + _, call_import, rows = _seed(db_session, row_count=1) + row = rows[0] + row.recording_url = None + db_session.commit() from app.services.telephony.exotel_client import ExotelNotFoundError - class _LookupFailsFallbackWorks: + class _LookupFailsNoUrl: def __init__(self): self.resolved_calls = [] self.calls = [] @@ -438,7 +478,7 @@ def download_recording(self, recording_url): self.calls.append(recording_url) return b"fallback-audio", "audio/mpeg" - fake_client = _LookupFailsFallbackWorks() + fake_client = _LookupFailsNoUrl() fake_s3 = _FakeS3(enabled=True) task_module = _patch_dependencies(monkeypatch, db_session, fake_client, fake_s3) public_calls = _patch_public_download( @@ -447,58 +487,45 @@ def download_recording(self, recording_url): result = task_module.process_call_import_row_task.run(str(row.id)) - assert result["status"] == "completed" + assert result["status"] == "failed" db_session.refresh(row) - db_session.refresh(call_import) - - # Lookup attempted exactly once with the call_sid, then download from - # the original CSV URL (no other URL was tried). assert fake_client.resolved_calls == [row.conversation_id] assert fake_client.calls == [] - assert public_calls == [csv_url] - - assert row.status == CallImportRowStatus.COMPLETED - assert row.recording_size_bytes == len(b"fallback-audio") - # CSV URL stays intact since the user supplied it. - assert row.recording_url == csv_url - assert call_import.status == CallImportStatus.COMPLETED + assert public_calls == [] + assert row.status == CallImportRowStatus.FAILED -def test_process_call_import_row_fails_when_both_lookup_and_csv_url_fail( +def test_process_call_import_row_fails_when_lookup_and_csv_url_both_fail( db_session, monkeypatch ): - """Both tiers exhausted with non-retryable errors -> row fails with a - composite error message that surfaces both failure reasons.""" + """When call-id lookup and CSV URL download both fail, the row is marked failed.""" _, call_import, rows = _seed(db_session, row_count=1) row = rows[0] assert row.recording_url - from app.services.telephony.exotel_client import ( - ExotelAuthError, - ExotelNotFoundError, - ) + from app.services.telephony.exotel_client import ExotelAuthError, ExotelNotFoundError - class _BothPathsFail: + class _BothTiersFail: def __init__(self): self.resolved_calls = [] self.calls = [] def get_call_recording_url(self, call_sid): self.resolved_calls.append(call_sid) - raise ExotelNotFoundError(f"call {call_sid} not found in API") + raise ExotelNotFoundError(f"call {call_sid} not found") def download_recording(self, recording_url): self.calls.append(recording_url) raise ExotelAuthError(f"auth rejected for {recording_url}") - fake_client = _BothPathsFail() + fake_client = _BothTiersFail() fake_s3 = _FakeS3(enabled=True) task_module = _patch_dependencies(monkeypatch, db_session, fake_client, fake_s3) _patch_public_download( monkeypatch, side_effect=lambda url: (_ for _ in ()).throw( - ExotelAuthError(f"auth rejected for {url}") + ExotelAuthError(f"public auth rejected for {url}") ), ) @@ -516,10 +543,65 @@ def download_recording(self, recording_url): db_session.refresh(row) db_session.refresh(call_import) assert row.status == CallImportRowStatus.FAILED - # Composite message surfaces both tiers' failure messages. + assert fake_client.resolved_calls == [row.conversation_id] + assert fake_client.calls == [row.recording_url] assert "call-id lookup" in (row.error_message or "") assert "recording URL" in (row.error_message or "") - assert "not found in API" in (row.error_message or "") + assert call_import.status == CallImportStatus.FAILED + assert fake_s3.uploads == [] + + +def test_process_call_import_row_fails_when_exotel_csv_url_download_fails( + db_session, monkeypatch +): + """Exotel credentialed CSV-URL download failure marks the row failed.""" + + _, call_import, rows = _seed(db_session, row_count=1) + row = rows[0] + assert row.recording_url + + from app.services.telephony.exotel_client import ExotelAuthError, ExotelNotFoundError + + class _CsvUrlAuthFails: + def __init__(self): + self.resolved_calls = [] + self.calls = [] + + def get_call_recording_url(self, call_sid): + self.resolved_calls.append(call_sid) + raise ExotelNotFoundError(f"call {call_sid} not found") + + def download_recording(self, recording_url): + self.calls.append(recording_url) + raise ExotelAuthError(f"auth rejected for {recording_url}") + + fake_client = _CsvUrlAuthFails() + fake_s3 = _FakeS3(enabled=True) + task_module = _patch_dependencies(monkeypatch, db_session, fake_client, fake_s3) + _patch_public_download( + monkeypatch, + side_effect=lambda url: (_ for _ in ()).throw( + ExotelAuthError(f"public auth rejected for {url}") + ), + ) + + monkeypatch.setattr( + task_module.process_call_import_row_task, + "retry", + lambda exc, countdown: (_ for _ in ()).throw(RetryCalled((exc, countdown))), + ) + + result = task_module.process_call_import_row_task.run(str(row.id)) + + assert result["status"] == "failed" + assert result["reason"] == "non_retryable_provider_error" + + db_session.refresh(row) + db_session.refresh(call_import) + assert row.status == CallImportRowStatus.FAILED + assert fake_client.resolved_calls == [row.conversation_id] + assert fake_client.calls == [row.recording_url] + assert "recording URL" in (row.error_message or "") assert "auth rejected" in (row.error_message or "") assert call_import.status == CallImportStatus.FAILED assert fake_s3.uploads == [] @@ -536,6 +618,9 @@ def test_process_call_import_row_uses_csv_url_when_provider_lacks_lookup( csv_url = row.recording_url assert csv_url + call_import.provider = TelephonyProvider.PLIVO.value + db_session.commit() + class _NoLookupClient: def __init__(self): self.calls = [] @@ -566,17 +651,17 @@ def download_recording(self, recording_url): def test_process_call_import_row_recovers_via_csv_url_after_transient_lookup( db_session, monkeypatch ): - """A transient blip in the call-id lookup should still let the CSV URL - serve as a fallback within the same attempt rather than always forcing - a 60s Celery retry.""" + """When call-id lookup hits a transient error and no CSV URL exists, the + worker schedules a retry.""" _, call_import, rows = _seed(db_session, row_count=1) row = rows[0] - csv_url = row.recording_url + row.recording_url = None + db_session.commit() from app.services.telephony.exotel_client import ExotelTransientError - class _LookupTransientCsvOk: + class _LookupTransientNoCsvUrl: def __init__(self): self.resolved_calls = [] self.calls = [] @@ -589,57 +674,62 @@ def download_recording(self, recording_url): self.calls.append(recording_url) return b"recovered-via-fallback", "audio/mpeg" - fake_client = _LookupTransientCsvOk() + fake_client = _LookupTransientNoCsvUrl() fake_s3 = _FakeS3(enabled=True) task_module = _patch_dependencies(monkeypatch, db_session, fake_client, fake_s3) public_calls = _patch_public_download( monkeypatch, return_value=(b"recovered-via-fallback", "audio/mpeg") ) - result = task_module.process_call_import_row_task.run(str(row.id)) + monkeypatch.setattr( + task_module.process_call_import_row_task, + "retry", + lambda exc, countdown: (_ for _ in ()).throw(RetryCalled((exc, countdown))), + ) + + with pytest.raises(RetryCalled): + task_module.process_call_import_row_task.run(str(row.id)) - assert result["status"] == "completed" db_session.refresh(row) - db_session.refresh(call_import) assert fake_client.resolved_calls == [row.conversation_id] assert fake_client.calls == [] - assert public_calls == [csv_url] - assert row.status == CallImportRowStatus.COMPLETED - assert row.recording_size_bytes == len(b"recovered-via-fallback") - assert call_import.status == CallImportStatus.COMPLETED + assert public_calls == [] + assert row.status == CallImportRowStatus.PENDING -def test_process_call_import_row_retries_when_both_tiers_transient( +def test_process_call_import_row_retries_when_exotel_csv_url_transient( db_session, monkeypatch ): - """When both the call-id lookup and the CSV-URL fallback hit transient - errors, the worker schedules a retry rather than failing the row.""" + """When call-id lookup fails and credentialed CSV-URL download is transient, retry.""" _, _call_import, rows = _seed(db_session, row_count=1) row = rows[0] - from app.services.telephony.exotel_client import ExotelTransientError + from app.services.telephony.exotel_client import ( + ExotelNotFoundError, + ExotelTransientError, + ) - class _BothTransient: + class _LookupFailsCsvTransient: def __init__(self): self.resolved_calls = [] self.calls = [] def get_call_recording_url(self, call_sid): self.resolved_calls.append(call_sid) - raise ExotelTransientError("502 bad gateway on lookup") + raise ExotelNotFoundError(f"call {call_sid} not found") def download_recording(self, recording_url): self.calls.append(recording_url) raise ExotelTransientError("503 fetching recording") - fake_client = _BothTransient() + fake_client = _LookupFailsCsvTransient() fake_s3 = _FakeS3(enabled=True) task_module = _patch_dependencies(monkeypatch, db_session, fake_client, fake_s3) public_calls = _patch_public_download( monkeypatch, side_effect=lambda url: (_ for _ in ()).throw( - ExotelTransientError("503 fetching recording") + ExotelTransientError("503 public fetch") ), ) @@ -656,16 +746,15 @@ def download_recording(self, recording_url): assert row.status == CallImportRowStatus.PENDING assert "Transient" in (row.error_message or "") assert fake_client.resolved_calls == [row.conversation_id] - # Fallback was attempted with the CSV URL before retry was scheduled. - assert fake_client.calls == [] - assert public_calls == [row.recording_url] + assert fake_client.calls == [row.recording_url] + assert public_calls == [] def test_process_call_import_row_uses_conversation_id_when_provider_set_without_pin( db_session, monkeypatch ): """Legacy/credentialed batches with provider but no pinned credential id - must still use the conversation-id lookup path.""" + still resolve recordings via call-id lookup when conversation_id is set.""" _, call_import, rows = _seed(db_session, row_count=1) row = rows[0] call_import.telephony_integration_id = None @@ -686,6 +775,8 @@ def test_process_call_import_row_uses_conversation_id_when_provider_set_without_ db_session.refresh(row) assert row.status == CallImportRowStatus.COMPLETED assert fake_client.resolved_calls == [row.conversation_id] + expected_url = f"https://api.exotel.com/recordings/{row.conversation_id}.mp3" + assert fake_client.calls == [expected_url] assert public_calls == []