mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-16 04:13:35 +00:00
stt init (#2306)
* stt init * fix: limits * fix: errors * fix: error messages
This commit is contained in:
38 files changed
+3096
-188
No files matched your search
@@ -0,0 +1,437 @@
|
||||
import io
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from flask import Flask, request
|
||||
|
||||
|
||||
def _get_response_status(response):
|
||||
if isinstance(response, tuple):
|
||||
return response[1]
|
||||
return response.status_code
|
||||
|
||||
|
||||
def _get_response_json(response):
|
||||
if isinstance(response, tuple):
|
||||
return response[0].json
|
||||
return response.json
|
||||
|
||||
|
||||
class FakeRedis:
|
||||
def __init__(self):
|
||||
self.values = {}
|
||||
|
||||
def get(self, key):
|
||||
return self.values.get(key)
|
||||
|
||||
def setex(self, key, ttl, value):
|
||||
_ = ttl
|
||||
self.values[key] = value
|
||||
|
||||
def delete(self, key):
|
||||
self.values.pop(key, None)
|
||||
|
||||
|
||||
class TestStoreAttachmentEndpoint:
|
||||
@patch("application.api.user.tasks.store_attachment.delay")
|
||||
def test_store_attachment_preserves_upload_indexes_for_partial_failures(
|
||||
self, mock_store_attachment, flask_app, mock_mongo_db
|
||||
):
|
||||
from application.api.user.attachments.routes import StoreAttachment
|
||||
|
||||
app = Flask(__name__)
|
||||
mock_storage = MagicMock()
|
||||
mock_store_attachment.side_effect = [
|
||||
SimpleNamespace(id="task-alpha"),
|
||||
SimpleNamespace(id="task-gamma"),
|
||||
]
|
||||
|
||||
def save_file(file, path):
|
||||
_ = path
|
||||
if file.filename == "beta.txt":
|
||||
raise ValueError("Failed to save file")
|
||||
return {"storage_type": "local"}
|
||||
|
||||
mock_storage.save_file.side_effect = save_file
|
||||
|
||||
with patch("application.api.user.base.storage", mock_storage):
|
||||
with app.test_request_context(
|
||||
"/api/store_attachment",
|
||||
method="POST",
|
||||
data={
|
||||
"file": [
|
||||
(io.BytesIO(b"alpha"), "alpha.txt"),
|
||||
(io.BytesIO(b"beta"), "beta.txt"),
|
||||
(io.BytesIO(b"gamma"), "gamma.txt"),
|
||||
]
|
||||
},
|
||||
content_type="multipart/form-data",
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
|
||||
resource = StoreAttachment()
|
||||
response = resource.post()
|
||||
payload = _get_response_json(response)
|
||||
|
||||
assert _get_response_status(response) == 200
|
||||
assert [task["upload_index"] for task in payload["tasks"]] == [0, 2]
|
||||
assert payload["errors"][0]["upload_index"] == 1
|
||||
assert payload["errors"][0]["error"] == "Failed to process file"
|
||||
|
||||
@patch("application.api.user.tasks.store_attachment.delay")
|
||||
@patch("application.stt.upload_limits.settings")
|
||||
def test_store_attachment_rejects_oversized_audio_files(
|
||||
self, mock_limit_settings, mock_store_attachment, flask_app, mock_mongo_db
|
||||
):
|
||||
from application.api.user.attachments.routes import StoreAttachment
|
||||
|
||||
app = Flask(__name__)
|
||||
mock_limit_settings.STT_MAX_FILE_SIZE_MB = 1
|
||||
|
||||
with app.test_request_context(
|
||||
"/api/store_attachment",
|
||||
method="POST",
|
||||
data={
|
||||
"file": (
|
||||
io.BytesIO(b"x" * (2 * 1024 * 1024)),
|
||||
"meeting.wav",
|
||||
)
|
||||
},
|
||||
content_type="multipart/form-data",
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
|
||||
resource = StoreAttachment()
|
||||
response = resource.post()
|
||||
|
||||
assert _get_response_status(response) == 413
|
||||
assert "exceeds" in _get_response_json(response)["message"]
|
||||
mock_store_attachment.assert_not_called()
|
||||
|
||||
|
||||
class TestSpeechToTextEndpoint:
|
||||
def test_stt_returns_400_when_file_is_missing(self, flask_app, mock_mongo_db):
|
||||
from application.api.user.attachments.routes import SpeechToText
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context("/api/stt", method="POST", data={}):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
|
||||
resource = SpeechToText()
|
||||
response = resource.post()
|
||||
|
||||
assert _get_response_status(response) == 400
|
||||
assert _get_response_json(response)["message"] == "Missing file"
|
||||
|
||||
def test_stt_returns_401_when_authentication_is_missing(
|
||||
self, flask_app, mock_mongo_db
|
||||
):
|
||||
from application.api.user.attachments.routes import SpeechToText
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(
|
||||
"/api/stt",
|
||||
method="POST",
|
||||
data={"file": (io.BytesIO(b"audio-bytes"), "clip.wav")},
|
||||
content_type="multipart/form-data",
|
||||
):
|
||||
request.decoded_token = None
|
||||
|
||||
resource = SpeechToText()
|
||||
response = resource.post()
|
||||
|
||||
assert _get_response_status(response) == 401
|
||||
assert _get_response_json(response)["message"] == "Authentication required"
|
||||
|
||||
@patch("application.api.user.attachments.routes.STTCreator.create_stt")
|
||||
def test_stt_transcribes_audio_for_authenticated_user(
|
||||
self, mock_create_stt, flask_app, mock_mongo_db
|
||||
):
|
||||
from application.api.user.attachments.routes import SpeechToText
|
||||
|
||||
app = Flask(__name__)
|
||||
mock_stt = MagicMock()
|
||||
mock_stt.transcribe.return_value = {
|
||||
"text": "hello from audio",
|
||||
"language": "en",
|
||||
"duration_s": 1.2,
|
||||
"segments": [],
|
||||
"provider": "openai",
|
||||
}
|
||||
mock_create_stt.return_value = mock_stt
|
||||
|
||||
with app.test_request_context(
|
||||
"/api/stt",
|
||||
method="POST",
|
||||
data={
|
||||
"file": (io.BytesIO(b"audio-bytes"), "clip.wav"),
|
||||
"language": "en",
|
||||
},
|
||||
content_type="multipart/form-data",
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
|
||||
resource = SpeechToText()
|
||||
response = resource.post()
|
||||
|
||||
assert _get_response_status(response) == 200
|
||||
assert _get_response_json(response) == {
|
||||
"success": True,
|
||||
"text": "hello from audio",
|
||||
"language": "en",
|
||||
"duration_s": 1.2,
|
||||
"segments": [],
|
||||
"provider": "openai",
|
||||
}
|
||||
mock_create_stt.assert_called_once()
|
||||
mock_stt.transcribe.assert_called_once()
|
||||
|
||||
def test_stt_rejects_unsupported_extension(self, flask_app, mock_mongo_db):
|
||||
from application.api.user.attachments.routes import SpeechToText
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
with app.test_request_context(
|
||||
"/api/stt",
|
||||
method="POST",
|
||||
data={"file": (io.BytesIO(b"audio-bytes"), "clip.exe")},
|
||||
content_type="multipart/form-data",
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
|
||||
resource = SpeechToText()
|
||||
response = resource.post()
|
||||
|
||||
assert _get_response_status(response) == 400
|
||||
assert "Unsupported audio format" in _get_response_json(response)["message"]
|
||||
|
||||
|
||||
class TestLiveSpeechToTextEndpoint:
|
||||
@patch("application.api.user.attachments.routes.get_redis_instance")
|
||||
def test_live_stt_start_creates_session(
|
||||
self, mock_get_redis, flask_app, mock_mongo_db
|
||||
):
|
||||
from application.api.user.attachments.routes import LiveSpeechToTextStart
|
||||
|
||||
app = Flask(__name__)
|
||||
mock_get_redis.return_value = FakeRedis()
|
||||
|
||||
with app.test_request_context(
|
||||
"/api/stt/live/start",
|
||||
method="POST",
|
||||
json={"language": "ru"},
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
|
||||
resource = LiveSpeechToTextStart()
|
||||
response = resource.post()
|
||||
payload = _get_response_json(response)
|
||||
|
||||
assert _get_response_status(response) == 200
|
||||
assert payload["success"] is True
|
||||
assert payload["language"] == "ru"
|
||||
assert payload["session_id"]
|
||||
assert payload["transcript_text"] == ""
|
||||
|
||||
@patch("application.api.user.attachments.routes.STTCreator.create_stt")
|
||||
@patch("application.api.user.attachments.routes.get_redis_instance")
|
||||
def test_live_stt_chunk_reconciles_transcript_progressively(
|
||||
self, mock_get_redis, mock_create_stt, flask_app, mock_mongo_db
|
||||
):
|
||||
from application.api.user.attachments.routes import (
|
||||
LiveSpeechToTextChunk,
|
||||
LiveSpeechToTextFinish,
|
||||
LiveSpeechToTextStart,
|
||||
)
|
||||
|
||||
app = Flask(__name__)
|
||||
fake_redis = FakeRedis()
|
||||
mock_get_redis.return_value = fake_redis
|
||||
|
||||
start_resource = LiveSpeechToTextStart()
|
||||
with app.test_request_context(
|
||||
"/api/stt/live/start",
|
||||
method="POST",
|
||||
json={"language": "ru"},
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
start_response = start_resource.post()
|
||||
session_id = _get_response_json(start_response)["session_id"]
|
||||
|
||||
mock_stt = MagicMock()
|
||||
mock_stt.transcribe.side_effect = [
|
||||
{
|
||||
"text": "hello this is a longer test phrase for transcript stabilization today now",
|
||||
"language": "ru",
|
||||
"duration_s": 1.0,
|
||||
"segments": [],
|
||||
"provider": "openai",
|
||||
},
|
||||
{
|
||||
"text": "hello this is a longer test phrase for transcript stabilization today now again later",
|
||||
"language": "ru",
|
||||
"duration_s": 1.0,
|
||||
"segments": [],
|
||||
"provider": "openai",
|
||||
},
|
||||
]
|
||||
mock_create_stt.return_value = mock_stt
|
||||
|
||||
chunk_resource = LiveSpeechToTextChunk()
|
||||
with app.test_request_context(
|
||||
"/api/stt/live/chunk",
|
||||
method="POST",
|
||||
data={
|
||||
"session_id": session_id,
|
||||
"chunk_index": "0",
|
||||
"file": (io.BytesIO(b"chunk-0"), "chunk-0.wav"),
|
||||
},
|
||||
content_type="multipart/form-data",
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
chunk_response = chunk_resource.post()
|
||||
chunk_payload = _get_response_json(chunk_response)
|
||||
|
||||
assert _get_response_status(chunk_response) == 200
|
||||
assert (
|
||||
chunk_payload["transcript_text"]
|
||||
== "hello this is a longer test phrase for transcript stabilization today now"
|
||||
)
|
||||
assert chunk_payload["committed_text"] == ""
|
||||
assert (
|
||||
chunk_payload["mutable_text"]
|
||||
== "hello this is a longer test phrase for transcript stabilization today now"
|
||||
)
|
||||
assert chunk_payload["finalized_text"] == ""
|
||||
assert (
|
||||
chunk_payload["pending_text"]
|
||||
== "hello this is a longer test phrase for transcript stabilization today now"
|
||||
)
|
||||
|
||||
with app.test_request_context(
|
||||
"/api/stt/live/chunk",
|
||||
method="POST",
|
||||
data={
|
||||
"session_id": session_id,
|
||||
"chunk_index": "1",
|
||||
"is_silence": "true",
|
||||
"file": (io.BytesIO(b"chunk-1"), "chunk-1.wav"),
|
||||
},
|
||||
content_type="multipart/form-data",
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
chunk_response = chunk_resource.post()
|
||||
chunk_payload = _get_response_json(chunk_response)
|
||||
|
||||
assert _get_response_status(chunk_response) == 200
|
||||
assert (
|
||||
chunk_payload["transcript_text"]
|
||||
== "hello this is a longer test phrase for transcript stabilization today now again later"
|
||||
)
|
||||
assert (
|
||||
chunk_payload["committed_text"]
|
||||
== "hello this is a longer test phrase for transcript stabilization today now"
|
||||
)
|
||||
assert chunk_payload["mutable_text"] == "again later"
|
||||
assert chunk_payload["finalized_text"] == chunk_payload["committed_text"]
|
||||
assert chunk_payload["pending_text"] == "again later"
|
||||
assert chunk_payload["is_silence"] is True
|
||||
|
||||
finish_resource = LiveSpeechToTextFinish()
|
||||
with app.test_request_context(
|
||||
"/api/stt/live/finish",
|
||||
method="POST",
|
||||
json={"session_id": session_id},
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
finish_response = finish_resource.post()
|
||||
finish_payload = _get_response_json(finish_response)
|
||||
|
||||
assert _get_response_status(finish_response) == 200
|
||||
assert (
|
||||
finish_payload["text"]
|
||||
== "hello this is a longer test phrase for transcript stabilization today now again later"
|
||||
)
|
||||
|
||||
@patch("application.api.user.attachments.routes.get_redis_instance")
|
||||
def test_live_stt_chunk_rejects_missing_session(
|
||||
self, mock_get_redis, flask_app, mock_mongo_db
|
||||
):
|
||||
from application.api.user.attachments.routes import LiveSpeechToTextChunk
|
||||
|
||||
app = Flask(__name__)
|
||||
mock_get_redis.return_value = FakeRedis()
|
||||
|
||||
with app.test_request_context(
|
||||
"/api/stt/live/chunk",
|
||||
method="POST",
|
||||
data={
|
||||
"session_id": "missing-session",
|
||||
"chunk_index": "0",
|
||||
"file": (io.BytesIO(b"chunk-0"), "chunk-0.wav"),
|
||||
},
|
||||
content_type="multipart/form-data",
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
|
||||
resource = LiveSpeechToTextChunk()
|
||||
response = resource.post()
|
||||
|
||||
assert _get_response_status(response) == 404
|
||||
assert _get_response_json(response)["message"] == "Live transcription session not found"
|
||||
|
||||
@patch("application.api.user.attachments.routes.STTCreator.create_stt")
|
||||
@patch("application.api.user.attachments.routes.get_redis_instance")
|
||||
def test_live_stt_chunk_hides_internal_value_errors(
|
||||
self, mock_get_redis, mock_create_stt, flask_app, mock_mongo_db
|
||||
):
|
||||
from application.api.user.attachments.routes import (
|
||||
LiveSpeechToTextChunk,
|
||||
LiveSpeechToTextStart,
|
||||
)
|
||||
|
||||
app = Flask(__name__)
|
||||
fake_redis = FakeRedis()
|
||||
mock_get_redis.return_value = fake_redis
|
||||
|
||||
start_resource = LiveSpeechToTextStart()
|
||||
with app.test_request_context(
|
||||
"/api/stt/live/start",
|
||||
method="POST",
|
||||
json={"language": "ru"},
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
start_response = start_resource.post()
|
||||
session_id = _get_response_json(start_response)["session_id"]
|
||||
|
||||
mock_stt = MagicMock()
|
||||
mock_stt.transcribe.return_value = {
|
||||
"text": "hello there",
|
||||
"language": "ru",
|
||||
"duration_s": 1.0,
|
||||
"segments": [],
|
||||
"provider": "openai",
|
||||
}
|
||||
mock_create_stt.return_value = mock_stt
|
||||
|
||||
chunk_resource = LiveSpeechToTextChunk()
|
||||
with app.test_request_context(
|
||||
"/api/stt/live/chunk",
|
||||
method="POST",
|
||||
data={
|
||||
"session_id": session_id,
|
||||
"chunk_index": "-1",
|
||||
"file": (io.BytesIO(b"chunk-neg"), "chunk-neg.wav"),
|
||||
},
|
||||
content_type="multipart/form-data",
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
response = chunk_resource.post()
|
||||
|
||||
assert _get_response_status(response) == 409
|
||||
assert (
|
||||
_get_response_json(response)["message"]
|
||||
== "Invalid live transcription chunk"
|
||||
)
|
||||
@@ -0,0 +1,71 @@
|
||||
import io
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from flask import Flask, request
|
||||
from application.parser.file.constants import SUPPORTED_SOURCE_EXTENSIONS
|
||||
|
||||
|
||||
def test_upload_route_passes_audio_extensions_to_ingest(flask_app, mock_mongo_db):
|
||||
from application.api.user.sources.upload import UploadFile
|
||||
|
||||
app = Flask(__name__)
|
||||
mock_storage = MagicMock()
|
||||
mock_task = SimpleNamespace(id="task-123")
|
||||
|
||||
with app.test_request_context(
|
||||
"/api/upload",
|
||||
method="POST",
|
||||
data={
|
||||
"user": "test_user",
|
||||
"name": "Meeting Notes",
|
||||
"file": (io.BytesIO(b"audio-bytes"), "meeting.wav"),
|
||||
},
|
||||
content_type="multipart/form-data",
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
|
||||
with patch(
|
||||
"application.api.user.sources.upload.StorageCreator.get_storage",
|
||||
return_value=mock_storage,
|
||||
), patch(
|
||||
"application.api.user.sources.upload.ingest.delay",
|
||||
return_value=mock_task,
|
||||
) as mock_delay:
|
||||
resource = UploadFile()
|
||||
response = resource.post()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json["success"] is True
|
||||
|
||||
formats = mock_delay.call_args.args[1]
|
||||
assert formats == list(SUPPORTED_SOURCE_EXTENSIONS)
|
||||
|
||||
|
||||
@patch("application.stt.upload_limits.settings")
|
||||
def test_upload_route_rejects_oversized_audio(
|
||||
mock_limit_settings, flask_app, mock_mongo_db
|
||||
):
|
||||
from application.api.user.sources.upload import UploadFile
|
||||
|
||||
app = Flask(__name__)
|
||||
mock_limit_settings.STT_MAX_FILE_SIZE_MB = 1
|
||||
|
||||
with app.test_request_context(
|
||||
"/api/upload",
|
||||
method="POST",
|
||||
data={
|
||||
"user": "test_user",
|
||||
"name": "Meeting Notes",
|
||||
"file": (io.BytesIO(b"x" * (2 * 1024 * 1024)), "meeting.wav"),
|
||||
},
|
||||
content_type="multipart/form-data",
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
|
||||
with patch("application.api.user.sources.upload.StorageCreator.get_storage"):
|
||||
resource = UploadFile()
|
||||
response = resource.post()
|
||||
|
||||
assert response.status_code == 413
|
||||
assert "exceeds" in response.json["message"]
|
||||
@@ -0,0 +1,122 @@
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from bson import ObjectId
|
||||
from flask import Flask, jsonify, make_response, request
|
||||
|
||||
|
||||
def test_safe_db_operation_hides_exception_details():
|
||||
from application.api.user.utils import safe_db_operation
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
def failing_operation():
|
||||
raise RuntimeError("database credentials leaked")
|
||||
|
||||
with app.app_context():
|
||||
_, error = safe_db_operation(
|
||||
failing_operation,
|
||||
"Failed to create workflow",
|
||||
)
|
||||
|
||||
assert error.status_code == 400
|
||||
assert error.json["message"] == "Failed to create workflow"
|
||||
assert "credentials" not in error.json["message"]
|
||||
|
||||
|
||||
def test_agent_folders_hides_exception_details():
|
||||
from application.api.user.agents.folders import AgentFolders
|
||||
|
||||
app = Flask(__name__)
|
||||
failing_collection = Mock()
|
||||
failing_collection.find.side_effect = RuntimeError("folder backend secret")
|
||||
|
||||
with patch(
|
||||
"application.api.user.agents.folders.agent_folders_collection",
|
||||
failing_collection,
|
||||
):
|
||||
with app.test_request_context("/api/agents/folders/", method="GET"):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
response = AgentFolders().get()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json["message"] == "Failed to fetch folders"
|
||||
assert "secret" not in response.json["message"]
|
||||
|
||||
|
||||
def test_workflow_create_hides_structure_exception_details():
|
||||
from application.api.user.workflows.routes import WorkflowList
|
||||
|
||||
app = Flask(__name__)
|
||||
insert_result = SimpleNamespace(inserted_id=ObjectId())
|
||||
|
||||
with patch(
|
||||
"application.api.user.workflows.routes.safe_db_operation",
|
||||
return_value=(insert_result, None),
|
||||
), patch(
|
||||
"application.api.user.workflows.routes.create_workflow_nodes",
|
||||
side_effect=RuntimeError("storage bucket credentials leaked"),
|
||||
), patch(
|
||||
"application.api.user.workflows.routes.workflow_nodes_collection"
|
||||
) as mock_nodes, patch(
|
||||
"application.api.user.workflows.routes.workflow_edges_collection"
|
||||
) as mock_edges, patch(
|
||||
"application.api.user.workflows.routes.workflows_collection"
|
||||
) as mock_workflows:
|
||||
with app.test_request_context(
|
||||
"/api/workflows",
|
||||
method="POST",
|
||||
json={
|
||||
"name": "Workflow",
|
||||
"nodes": [
|
||||
{"id": "start", "type": "start"},
|
||||
{"id": "end", "type": "end"},
|
||||
],
|
||||
"edges": [{"id": "edge-1", "source": "start", "target": "end"}],
|
||||
},
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
response = WorkflowList().post()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json["message"] == "Failed to create workflow structure"
|
||||
assert "credentials" not in response.json["message"]
|
||||
mock_nodes.delete_many.assert_called_once_with(
|
||||
{"workflow_id": str(insert_result.inserted_id)}
|
||||
)
|
||||
mock_edges.delete_many.assert_called_once_with(
|
||||
{"workflow_id": str(insert_result.inserted_id)}
|
||||
)
|
||||
mock_workflows.delete_one.assert_called_once_with(
|
||||
{"_id": insert_result.inserted_id}
|
||||
)
|
||||
|
||||
|
||||
def test_update_agent_reuses_sanitized_image_upload_error():
|
||||
from application.api.user.agents.routes import UpdateAgent
|
||||
|
||||
app = Flask(__name__)
|
||||
agent_id = str(ObjectId())
|
||||
|
||||
with app.test_request_context(
|
||||
f"/api/agents/update_agent/{agent_id}",
|
||||
method="PUT",
|
||||
json={},
|
||||
):
|
||||
request.decoded_token = {"sub": "test_user"}
|
||||
sanitized_error = make_response(
|
||||
jsonify({"success": False, "message": "Image upload failed"}),
|
||||
400,
|
||||
)
|
||||
|
||||
with patch(
|
||||
"application.api.user.agents.routes.agents_collection.find_one",
|
||||
return_value={"_id": ObjectId(agent_id), "user": "test_user"},
|
||||
), patch(
|
||||
"application.api.user.agents.routes.handle_image_upload",
|
||||
return_value=(None, sanitized_error),
|
||||
):
|
||||
response = UpdateAgent().put(agent_id)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json["message"] == "Image upload failed"
|
||||
Reference in new issue
Block a user