wehub-resource-sync
fed8b2eed7
Backend release / release (push) Waiting to run
Bandit Security Scan / bandit_scan (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / manifest (push) Blocked by required conditions
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / manifest (push) Blocked by required conditions
Python linting / ruff (push) Waiting to run
Run python tests with pytest / Run tests and count coverage (3.12) (push) Waiting to run
React Widget Build / build (push) Waiting to run
3646 lines
133 KiB
Python
3646 lines
133 KiB
Python
"""Tests for application.api.user.agents.routes module."""
|
|
|
|
import uuid
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
from flask import Flask
|
|
|
|
pytestmark = pytest.mark.skip(
|
|
reason="Asserts Mongo-era *_collection call shapes + references removed helpers "
|
|
"(validate_workflow_access, build_agent_document); needs PG repository-based rewrite. "
|
|
"Tracked as migration debt."
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
app = Flask(__name__)
|
|
return app
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper functions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestNormalizeWorkflowReference:
|
|
|
|
def test_returns_none_for_none(self):
|
|
from application.api.user.agents.routes import normalize_workflow_reference
|
|
|
|
assert normalize_workflow_reference(None) is None
|
|
|
|
def test_extracts_id_from_dict(self):
|
|
from application.api.user.agents.routes import normalize_workflow_reference
|
|
|
|
result = normalize_workflow_reference({"id": "abc123"})
|
|
assert result == "abc123"
|
|
|
|
def test_extracts_underscore_id_from_dict(self):
|
|
from application.api.user.agents.routes import normalize_workflow_reference
|
|
|
|
result = normalize_workflow_reference({"_id": "abc123"})
|
|
assert result == "abc123"
|
|
|
|
def test_extracts_workflow_id_from_dict(self):
|
|
from application.api.user.agents.routes import normalize_workflow_reference
|
|
|
|
result = normalize_workflow_reference({"workflow_id": "abc123"})
|
|
assert result == "abc123"
|
|
|
|
def test_returns_empty_string_for_blank_string(self):
|
|
from application.api.user.agents.routes import normalize_workflow_reference
|
|
|
|
assert normalize_workflow_reference(" ") == ""
|
|
|
|
def test_returns_plain_string_value(self):
|
|
from application.api.user.agents.routes import normalize_workflow_reference
|
|
|
|
oid = str(uuid.uuid4().hex)
|
|
assert normalize_workflow_reference(oid) == oid
|
|
|
|
def test_parses_json_string_value(self):
|
|
from application.api.user.agents.routes import normalize_workflow_reference
|
|
|
|
result = normalize_workflow_reference('"some_id"')
|
|
assert result == "some_id"
|
|
|
|
def test_parses_json_dict_value(self):
|
|
from application.api.user.agents.routes import normalize_workflow_reference
|
|
|
|
result = normalize_workflow_reference('{"id": "xyz"}')
|
|
assert result == "xyz"
|
|
|
|
def test_returns_string_for_non_json_string(self):
|
|
from application.api.user.agents.routes import normalize_workflow_reference
|
|
|
|
assert normalize_workflow_reference("plain_value") == "plain_value"
|
|
|
|
def test_converts_non_string_non_dict_to_str(self):
|
|
from application.api.user.agents.routes import normalize_workflow_reference
|
|
|
|
assert normalize_workflow_reference(42) == "42"
|
|
|
|
def test_dict_priority_id_over_workflow_id(self):
|
|
from application.api.user.agents.routes import normalize_workflow_reference
|
|
|
|
result = normalize_workflow_reference(
|
|
{"id": "first", "_id": "second", "workflow_id": "third"}
|
|
)
|
|
assert result == "first"
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestValidateWorkflowAccess:
|
|
|
|
def test_returns_none_when_not_required_and_empty(self, app):
|
|
from application.api.user.agents.routes import validate_workflow_access
|
|
|
|
with app.app_context():
|
|
wf_id, err = validate_workflow_access(None, "user1", required=False)
|
|
assert wf_id is None
|
|
assert err is None
|
|
|
|
def test_returns_error_when_required_and_empty(self, app):
|
|
from application.api.user.agents.routes import validate_workflow_access
|
|
|
|
with app.app_context():
|
|
wf_id, err = validate_workflow_access(None, "user1", required=True)
|
|
assert wf_id is None
|
|
assert err is not None
|
|
assert err.status_code == 400
|
|
|
|
def test_returns_error_for_invalid_oid_format(self, app):
|
|
from application.api.user.agents.routes import validate_workflow_access
|
|
|
|
with app.app_context():
|
|
wf_id, err = validate_workflow_access("not-a-valid-oid", "user1")
|
|
assert wf_id is None
|
|
assert err.status_code == 400
|
|
|
|
def test_returns_404_when_workflow_not_found(self, app):
|
|
from application.api.user.agents.routes import validate_workflow_access
|
|
|
|
mock_wf_col = Mock()
|
|
mock_wf_col.find_one.return_value = None
|
|
oid = str(uuid.uuid4().hex)
|
|
|
|
with app.app_context():
|
|
with patch(
|
|
"application.api.user.agents.routes.workflows_collection", mock_wf_col
|
|
):
|
|
wf_id, err = validate_workflow_access(oid, "user1")
|
|
assert err.status_code == 404
|
|
|
|
def test_returns_workflow_id_on_success(self, app):
|
|
from application.api.user.agents.routes import validate_workflow_access
|
|
|
|
mock_wf_col = Mock()
|
|
mock_wf_col.find_one.return_value = {"_id": uuid.uuid4().hex, "user": "user1"}
|
|
oid = str(uuid.uuid4().hex)
|
|
|
|
with app.app_context():
|
|
with patch(
|
|
"application.api.user.agents.routes.workflows_collection", mock_wf_col
|
|
):
|
|
wf_id, err = validate_workflow_access(oid, "user1")
|
|
assert wf_id == oid
|
|
assert err is None
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestBuildAgentDocument:
|
|
|
|
def test_builds_classic_agent(self):
|
|
from application.api.user.agents.routes import build_agent_document
|
|
|
|
data = {
|
|
"name": "Test Agent",
|
|
"description": "desc",
|
|
"status": "draft",
|
|
"chunks": "5",
|
|
"retriever": "classic",
|
|
"prompt_id": "default",
|
|
}
|
|
doc = build_agent_document(
|
|
data, "user1", "key123", "classic", image_url="img.png"
|
|
)
|
|
assert doc["user"] == "user1"
|
|
assert doc["name"] == "Test Agent"
|
|
assert doc["image"] == "img.png"
|
|
assert doc["agent_type"] == "classic"
|
|
assert "createdAt" in doc
|
|
assert "updatedAt" in doc
|
|
|
|
def test_builds_workflow_agent(self):
|
|
from application.api.user.agents.routes import build_agent_document
|
|
|
|
data = {
|
|
"name": "WF Agent",
|
|
"status": "published",
|
|
"workflow": "wf123",
|
|
"folder_id": "folder1",
|
|
}
|
|
doc = build_agent_document(data, "user1", "key123", "workflow")
|
|
assert doc["workflow"] == "wf123"
|
|
assert doc["folder_id"] == "folder1"
|
|
# Workflow agents should not have classic-specific fields
|
|
assert "image" not in doc
|
|
assert "source" not in doc
|
|
|
|
def test_defaults_to_classic_for_unknown_type(self):
|
|
from application.api.user.agents.routes import build_agent_document
|
|
|
|
data = {"name": "Agent", "status": "draft"}
|
|
doc = build_agent_document(data, "user1", "k", "unknown_type")
|
|
assert doc["agent_type"] == "classic"
|
|
|
|
def test_defaults_to_classic_for_empty_type(self):
|
|
from application.api.user.agents.routes import build_agent_document
|
|
|
|
data = {"name": "Agent", "status": "draft"}
|
|
doc = build_agent_document(data, "user1", "k", "")
|
|
assert doc["agent_type"] == "classic"
|
|
|
|
def test_limited_token_mode_string_true(self):
|
|
from application.api.user.agents.routes import build_agent_document
|
|
|
|
data = {"name": "A", "status": "draft", "limited_token_mode": "True"}
|
|
doc = build_agent_document(data, "user1", "k", "classic")
|
|
assert doc["limited_token_mode"] is True
|
|
|
|
def test_limited_token_mode_string_false(self):
|
|
from application.api.user.agents.routes import build_agent_document
|
|
|
|
data = {"name": "A", "status": "draft", "limited_token_mode": "False"}
|
|
doc = build_agent_document(data, "user1", "k", "classic")
|
|
assert doc["limited_token_mode"] is False
|
|
|
|
def test_limited_request_mode_bool(self):
|
|
from application.api.user.agents.routes import build_agent_document
|
|
|
|
data = {"name": "A", "status": "draft", "limited_request_mode": True}
|
|
doc = build_agent_document(data, "user1", "k", "classic")
|
|
assert doc["limited_request_mode"] is True
|
|
|
|
def test_filters_to_allowed_fields(self):
|
|
from application.api.user.agents.routes import build_agent_document
|
|
|
|
data = {"name": "A", "status": "draft"}
|
|
doc = build_agent_document(data, "user1", "k", "workflow")
|
|
# Workflow doc should not have classic-only fields
|
|
for classic_field in ["image", "source", "sources", "chunks", "retriever"]:
|
|
assert classic_field not in doc
|
|
|
|
def test_source_and_sources_passed_through(self):
|
|
from application.api.user.agents.routes import build_agent_document
|
|
|
|
source_ref = uuid.uuid4().hex # TODO: was DBRef("sources", uuid.uuid4().hex)
|
|
doc = build_agent_document(
|
|
{"name": "A", "status": "draft"},
|
|
"user1",
|
|
"k",
|
|
"classic",
|
|
source_field=source_ref,
|
|
sources_list=[source_ref],
|
|
)
|
|
assert doc["source"] == source_ref
|
|
assert doc["sources"] == [source_ref]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Route classes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetAgent:
|
|
|
|
def test_returns_401_unauthenticated(self, app):
|
|
from application.api.user.agents.routes import GetAgent
|
|
|
|
with app.test_request_context("/api/get_agent?id=abc"):
|
|
from flask import request
|
|
|
|
request.decoded_token = None
|
|
result = GetAgent().get()
|
|
# Returns tuple (dict, status_code)
|
|
assert result[1] == 401
|
|
|
|
def test_returns_400_missing_id(self, app):
|
|
from application.api.user.agents.routes import GetAgent
|
|
|
|
with app.test_request_context("/api/get_agent"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
result = GetAgent().get()
|
|
assert result[1] == 400
|
|
|
|
def test_returns_404_agent_not_found(self, app):
|
|
from application.api.user.agents.routes import GetAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = None
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context(f"/api/get_agent?id={agent_id}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
result = GetAgent().get()
|
|
assert result[1] == 404
|
|
|
|
def test_returns_agent_data_on_success(self, app):
|
|
from application.api.user.agents.routes import GetAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
key = str(uuid.uuid4())
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "Test Agent",
|
|
"description": "desc",
|
|
"chunks": "5",
|
|
"retriever": "classic",
|
|
"prompt_id": "default",
|
|
"tools": [],
|
|
"agent_type": "classic",
|
|
"status": "published",
|
|
"key": key,
|
|
}
|
|
|
|
mock_resolve = Mock(return_value=[])
|
|
mock_db = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.resolve_tool_details", mock_resolve
|
|
), patch(
|
|
"application.api.user.agents.routes.db", mock_db
|
|
):
|
|
with app.test_request_context(f"/api/get_agent?id={agent_id}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = GetAgent().get()
|
|
assert response.status_code == 200
|
|
data = response.json
|
|
assert data["id"] == str(agent_id)
|
|
assert data["name"] == "Test Agent"
|
|
assert data["key"].startswith(key[:4])
|
|
|
|
def test_returns_400_on_exception(self, app):
|
|
from application.api.user.agents.routes import GetAgent
|
|
|
|
mock_col = Mock()
|
|
mock_col.find_one.side_effect = Exception("DB error")
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context(f"/api/get_agent?id={uuid.uuid4().hex}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
result = GetAgent().get()
|
|
assert result[1] == 400
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetAgents:
|
|
|
|
def test_returns_401_unauthenticated(self, app):
|
|
from application.api.user.agents.routes import GetAgents
|
|
|
|
with app.test_request_context("/api/get_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = None
|
|
result = GetAgents().get()
|
|
assert result[1] == 401
|
|
|
|
def test_returns_agents_list(self, app):
|
|
from application.api.user.agents.routes import GetAgents
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
key = str(uuid.uuid4())
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.find.return_value = [
|
|
{
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "Agent1",
|
|
"source": "default",
|
|
"retriever": "classic",
|
|
"key": key,
|
|
}
|
|
]
|
|
|
|
mock_ensure = Mock(
|
|
return_value={
|
|
"user_id": "user1",
|
|
"agent_preferences": {"pinned": [str(agent_id)]},
|
|
}
|
|
)
|
|
mock_resolve = Mock(return_value=[])
|
|
mock_db = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure
|
|
), patch(
|
|
"application.api.user.agents.routes.resolve_tool_details", mock_resolve
|
|
), patch(
|
|
"application.api.user.agents.routes.db", mock_db
|
|
):
|
|
with app.test_request_context("/api/get_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = GetAgents().get()
|
|
assert response.status_code == 200
|
|
data = response.json
|
|
assert len(data) == 1
|
|
assert data[0]["name"] == "Agent1"
|
|
assert data[0]["pinned"] is True
|
|
|
|
def test_filters_agents_without_source_or_retriever(self, app):
|
|
from application.api.user.agents.routes import GetAgents
|
|
|
|
mock_agents_col = Mock()
|
|
# Agent without source/retriever and not workflow type -> filtered out
|
|
mock_agents_col.find.return_value = [
|
|
{
|
|
"_id": uuid.uuid4().hex,
|
|
"user": "user1",
|
|
"name": "BadAgent",
|
|
"agent_type": "classic",
|
|
}
|
|
]
|
|
|
|
mock_ensure = Mock(
|
|
return_value={
|
|
"user_id": "user1",
|
|
"agent_preferences": {"pinned": []},
|
|
}
|
|
)
|
|
mock_resolve = Mock(return_value=[])
|
|
mock_db = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure
|
|
), patch(
|
|
"application.api.user.agents.routes.resolve_tool_details", mock_resolve
|
|
), patch(
|
|
"application.api.user.agents.routes.db", mock_db
|
|
):
|
|
with app.test_request_context("/api/get_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = GetAgents().get()
|
|
assert response.status_code == 200
|
|
assert len(response.json) == 0
|
|
|
|
def test_includes_workflow_agent_without_source(self, app):
|
|
from application.api.user.agents.routes import GetAgents
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.find.return_value = [
|
|
{
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "WFAgent",
|
|
"agent_type": "workflow",
|
|
"key": "abcd1234efgh",
|
|
}
|
|
]
|
|
|
|
mock_ensure = Mock(
|
|
return_value={
|
|
"user_id": "user1",
|
|
"agent_preferences": {"pinned": []},
|
|
}
|
|
)
|
|
mock_resolve = Mock(return_value=[])
|
|
mock_db = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure
|
|
), patch(
|
|
"application.api.user.agents.routes.resolve_tool_details", mock_resolve
|
|
), patch(
|
|
"application.api.user.agents.routes.db", mock_db
|
|
):
|
|
with app.test_request_context("/api/get_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = GetAgents().get()
|
|
assert response.status_code == 200
|
|
assert len(response.json) == 1
|
|
assert response.json[0]["name"] == "WFAgent"
|
|
|
|
def test_returns_400_on_exception(self, app):
|
|
from application.api.user.agents.routes import GetAgents
|
|
|
|
mock_ensure = Mock(side_effect=Exception("DB error"))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure
|
|
):
|
|
with app.test_request_context("/api/get_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = GetAgents().get()
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCreateAgent:
|
|
|
|
def test_returns_401_unauthenticated(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={"name": "A"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = None
|
|
result = CreateAgent().post()
|
|
assert result[1] == 401
|
|
|
|
def test_returns_400_invalid_status(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={"name": "A", "status": "invalid"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 400
|
|
|
|
def test_creates_draft_agent_success(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
inserted_id = uuid.uuid4().hex
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.insert_one.return_value = Mock(inserted_id=inserted_id)
|
|
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "Draft Agent",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 201
|
|
data = response.json
|
|
assert data["id"] == str(inserted_id)
|
|
# Draft agents get empty key
|
|
assert data["key"] == ""
|
|
|
|
def test_creates_published_classic_agent(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
inserted_id = uuid.uuid4().hex
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.insert_one.return_value = Mock(inserted_id=inserted_id)
|
|
mock_handle_img = Mock(return_value=("img.png", None))
|
|
source_id = str(uuid.uuid4().hex)
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "Published Agent",
|
|
"description": "A test agent",
|
|
"status": "published",
|
|
"agent_type": "classic",
|
|
"source": source_id,
|
|
"chunks": "5",
|
|
"retriever": "classic",
|
|
"prompt_id": "default",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 201
|
|
data = response.json
|
|
assert data["id"] == str(inserted_id)
|
|
# Published agents get a uuid key
|
|
assert data["key"] != ""
|
|
|
|
def test_published_classic_requires_source(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "No Source",
|
|
"description": "desc",
|
|
"status": "published",
|
|
"agent_type": "classic",
|
|
"chunks": "5",
|
|
"retriever": "classic",
|
|
"prompt_id": "default",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 400
|
|
|
|
def test_creates_workflow_agent(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
inserted_id = uuid.uuid4().hex
|
|
wf_id = str(uuid.uuid4().hex)
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.insert_one.return_value = Mock(inserted_id=inserted_id)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
mock_wf_col = Mock()
|
|
mock_wf_col.find_one.return_value = {"_id": wf_id, "user": "user1"}
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
), patch(
|
|
"application.api.user.agents.routes.workflows_collection", mock_wf_col
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "WF Agent",
|
|
"status": "published",
|
|
"agent_type": "workflow",
|
|
"workflow": wf_id,
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 201
|
|
|
|
def test_image_upload_failure(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
mock_handle_img = Mock(return_value=(None, Mock()))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "Agent",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 400
|
|
|
|
def test_invalid_json_schema(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
from application.core.json_schema_utils import JsonSchemaValidationError
|
|
|
|
def raise_exc(val):
|
|
raise JsonSchemaValidationError("is invalid")
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.normalize_json_schema_payload",
|
|
side_effect=raise_exc,
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "Agent",
|
|
"status": "draft",
|
|
"json_schema": {"bad": True},
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 400
|
|
|
|
def test_folder_id_not_found(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
mock_folders = Mock()
|
|
mock_folders.find_one.return_value = None
|
|
folder_id = str(uuid.uuid4().hex)
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
), patch(
|
|
"application.api.user.agents.routes.agent_folders_collection", mock_folders
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "Agent",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
"folder_id": folder_id,
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 404
|
|
|
|
def test_invalid_folder_id_format(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "Agent",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
"folder_id": "not-valid",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 400
|
|
|
|
def test_form_data_with_json_fields(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
inserted_id = uuid.uuid4().hex
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.insert_one.return_value = Mock(inserted_id=inserted_id)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
content_type="multipart/form-data",
|
|
data={
|
|
"name": "FormAgent",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
"tools": '["tool1", "tool2"]',
|
|
"sources": '["src1"]',
|
|
"models": '["model1"]',
|
|
"json_schema": "null",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 201
|
|
|
|
def test_form_data_invalid_json_tools(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
inserted_id = uuid.uuid4().hex
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.insert_one.return_value = Mock(inserted_id=inserted_id)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
content_type="multipart/form-data",
|
|
data={
|
|
"name": "FormAgent",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
"tools": "not-valid-json",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
# invalid json for tools falls back to []
|
|
assert response.status_code == 201
|
|
|
|
def test_create_with_sources_list(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
inserted_id = uuid.uuid4().hex
|
|
src_id = str(uuid.uuid4().hex)
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.insert_one.return_value = Mock(inserted_id=inserted_id)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "MultiSource",
|
|
"description": "desc",
|
|
"status": "published",
|
|
"agent_type": "classic",
|
|
"sources": [src_id, "default"],
|
|
"chunks": "5",
|
|
"retriever": "classic",
|
|
"prompt_id": "default",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 201
|
|
|
|
def test_db_insert_failure(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.insert_one.side_effect = Exception("insert error")
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "Agent",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestUpdateAgent:
|
|
|
|
def _make_existing_agent(self, agent_id=None):
|
|
return {
|
|
"_id": agent_id or uuid.uuid4().hex,
|
|
"user": "user1",
|
|
"name": "Existing Agent",
|
|
"description": "existing desc",
|
|
"source": "default",
|
|
"chunks": "5",
|
|
"retriever": "classic",
|
|
"prompt_id": "default",
|
|
"status": "published",
|
|
"agent_type": "classic",
|
|
"key": str(uuid.uuid4()),
|
|
}
|
|
|
|
def test_returns_401_unauthenticated(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
with app.test_request_context(
|
|
"/api/update_agent/abc",
|
|
method="PUT",
|
|
json={"name": "A"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = None
|
|
response = UpdateAgent().put("abc")
|
|
assert response.status_code == 401
|
|
|
|
def test_returns_400_invalid_agent_id(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
with app.test_request_context(
|
|
"/api/update_agent/not-valid",
|
|
method="PUT",
|
|
json={"name": "A"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put("not-valid")
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_404_agent_not_found(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = str(uuid.uuid4().hex)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = None
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"name": "Updated"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(agent_id)
|
|
assert response.status_code == 404
|
|
|
|
def test_updates_agent_name_success(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_repo = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
def _run_dual_write(_repo_cls, fn):
|
|
fn(mock_repo)
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
), patch(
|
|
"application.api.user.agents.routes.dual_write",
|
|
side_effect=_run_dual_write,
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"name": "Updated Name"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
assert response.json["success"] is True
|
|
mock_repo.update_by_legacy_id.assert_called_once_with(
|
|
str(agent_id),
|
|
"user1",
|
|
{"name": "Updated Name"},
|
|
)
|
|
|
|
def test_returns_400_invalid_status(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"status": "invalid_status"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_400_negative_chunks(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"chunks": -1},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_400_tools_not_list(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"tools": "not-a-list"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_400_no_update_data(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"unknown_field": "value"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_limited_token_mode_without_limit(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"limited_token_mode": True},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_limited_request_mode_without_limit(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"limited_request_mode": True},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_token_limit_without_mode(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"token_limit": 1000},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_request_limit_without_mode(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"request_limit": 100},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_source_with_invalid_oid(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"source": "not-valid-oid"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_sources_list_with_invalid_oid(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"sources": ["bad-id"]},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_update_source_to_default(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"source": "default"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_update_empty_source_on_draft(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
existing["status"] = "draft"
|
|
existing["key"] = ""
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"source": ""},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_update_empty_source_on_published_fails(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
# Published agent with only "default" source
|
|
existing["source"] = "default"
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"source": ""},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_update_chunks_empty_defaults_to_2(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"chunks": ""},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
call_args = mock_col.update_one.call_args[0][1]["$set"]
|
|
assert call_args["chunks"] == "2"
|
|
|
|
def test_update_invalid_chunks_value(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"chunks": "abc"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_update_matched_but_not_modified(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=0)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"name": "Same Name"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
assert "No changes detected" in response.json["message"]
|
|
|
|
def test_update_matched_zero_returns_404(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=0, modified_count=0)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"name": "New Name"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 404
|
|
|
|
def test_publish_draft_generates_key(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
existing["status"] = "draft"
|
|
existing["key"] = ""
|
|
existing["source"] = "default"
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"status": "published"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
assert "key" in response.json
|
|
|
|
def test_publish_missing_required_fields(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "",
|
|
"description": "",
|
|
"source": "",
|
|
"chunks": "",
|
|
"retriever": "",
|
|
"prompt_id": "",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
"key": "",
|
|
}
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"status": "published"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_db_update_exception(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.side_effect = Exception("DB error")
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"name": "Updated"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 500
|
|
|
|
def test_update_json_schema_valid(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
schema_val = {"type": "object", "properties": {"x": {"type": "string"}}}
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
), patch(
|
|
"application.api.user.agents.routes.normalize_json_schema_payload",
|
|
return_value=schema_val,
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"json_schema": schema_val},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_update_json_schema_invalid(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
from application.core.json_schema_utils import JsonSchemaValidationError
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
def raise_exc(val):
|
|
raise JsonSchemaValidationError("is invalid")
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
), patch(
|
|
"application.api.user.agents.routes.normalize_json_schema_payload",
|
|
side_effect=raise_exc,
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"json_schema": {"bad": True}},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_update_json_schema_none(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"json_schema": None},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_update_folder_id_valid(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
folder_id = str(uuid.uuid4().hex)
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
mock_folders = Mock()
|
|
mock_folders.find_one.return_value = {"_id": folder_id, "user": "user1"}
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
), patch(
|
|
"application.api.user.agents.routes.agent_folders_collection", mock_folders
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"folder_id": folder_id},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_update_folder_id_invalid_format(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"folder_id": "invalid"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_update_folder_not_found(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
folder_id = str(uuid.uuid4().hex)
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
mock_folders = Mock()
|
|
mock_folders.find_one.return_value = None
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
), patch(
|
|
"application.api.user.agents.routes.agent_folders_collection", mock_folders
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"folder_id": folder_id},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 404
|
|
|
|
def test_update_folder_id_empty_sets_none(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"folder_id": ""},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
call_args = mock_col.update_one.call_args[0][1]["$set"]
|
|
assert call_args["folder_id"] is None
|
|
|
|
def test_empty_name_field_rejected(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"name": ""},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_update_workflow_field(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
wf_id = str(uuid.uuid4().hex)
|
|
existing = self._make_existing_agent(agent_id)
|
|
existing["agent_type"] = "workflow"
|
|
existing["status"] = "draft"
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
mock_wf_col = Mock()
|
|
mock_wf_col.find_one.return_value = {"_id": wf_id, "user": "user1"}
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
), patch(
|
|
"application.api.user.agents.routes.workflows_collection", mock_wf_col
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"workflow": wf_id},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_publish_workflow_without_workflow_field(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "WF Agent",
|
|
"status": "draft",
|
|
"agent_type": "workflow",
|
|
"key": "",
|
|
"workflow": None,
|
|
}
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"status": "published"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_form_data_json_parse_error(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = str(uuid.uuid4().hex)
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
content_type="multipart/form-data",
|
|
data={"tools": "not-valid-json"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(agent_id)
|
|
assert response.status_code == 400
|
|
|
|
def test_limited_token_mode_string_true(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={
|
|
"limited_token_mode": "True",
|
|
"token_limit": 5000,
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_update_sources_with_default(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
src_id = str(uuid.uuid4().hex)
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"sources": ["default", src_id]},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestDeleteAgent:
|
|
|
|
def test_returns_401_unauthenticated(self, app):
|
|
from application.api.user.agents.routes import DeleteAgent
|
|
|
|
with app.test_request_context("/api/delete_agent"):
|
|
from flask import request
|
|
|
|
request.decoded_token = None
|
|
response = DeleteAgent().delete()
|
|
assert response.status_code == 401
|
|
|
|
def test_returns_400_missing_id(self, app):
|
|
from application.api.user.agents.routes import DeleteAgent
|
|
|
|
with app.test_request_context("/api/delete_agent"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = DeleteAgent().delete()
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_404_agent_not_found(self, app):
|
|
from application.api.user.agents.routes import DeleteAgent
|
|
|
|
mock_col = Mock()
|
|
mock_col.find_one_and_delete.return_value = None
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context(f"/api/delete_agent?id={uuid.uuid4().hex}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = DeleteAgent().delete()
|
|
assert response.status_code == 404
|
|
|
|
def test_deletes_classic_agent_success(self, app):
|
|
from application.api.user.agents.routes import DeleteAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
mock_col = Mock()
|
|
mock_repo = Mock()
|
|
mock_col.find_one_and_delete.return_value = {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"agent_type": "classic",
|
|
}
|
|
|
|
def _run_dual_write(_repo_cls, fn):
|
|
fn(mock_repo)
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.dual_write",
|
|
side_effect=_run_dual_write,
|
|
):
|
|
with app.test_request_context(f"/api/delete_agent?id={agent_id}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = DeleteAgent().delete()
|
|
assert response.status_code == 200
|
|
assert response.json["id"] == str(agent_id)
|
|
mock_repo.delete_by_legacy_id.assert_called_once_with(str(agent_id), "user1")
|
|
|
|
def test_deletes_workflow_agent_cleans_up(self, app):
|
|
from application.api.user.agents.routes import DeleteAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
wf_id = str(uuid.uuid4().hex)
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.find_one_and_delete.return_value = {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"agent_type": "workflow",
|
|
"workflow": wf_id,
|
|
}
|
|
mock_wf_col = Mock()
|
|
mock_wf_col.find_one.return_value = {"_id": wf_id, "user": "user1"}
|
|
mock_nodes_col = Mock()
|
|
mock_edges_col = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.workflows_collection", mock_wf_col
|
|
), patch(
|
|
"application.api.user.agents.routes.workflow_nodes_collection", mock_nodes_col
|
|
), patch(
|
|
"application.api.user.agents.routes.workflow_edges_collection", mock_edges_col
|
|
):
|
|
with app.test_request_context(f"/api/delete_agent?id={agent_id}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = DeleteAgent().delete()
|
|
assert response.status_code == 200
|
|
mock_nodes_col.delete_many.assert_called_once()
|
|
mock_edges_col.delete_many.assert_called_once()
|
|
mock_wf_col.delete_one.assert_called_once()
|
|
|
|
def test_deletes_workflow_agent_non_owned_skips_cleanup(self, app):
|
|
from application.api.user.agents.routes import DeleteAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
wf_id = str(uuid.uuid4().hex)
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.find_one_and_delete.return_value = {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"agent_type": "workflow",
|
|
"workflow": wf_id,
|
|
}
|
|
mock_wf_col = Mock()
|
|
mock_wf_col.find_one.return_value = None # Not owned
|
|
mock_nodes_col = Mock()
|
|
mock_edges_col = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.workflows_collection", mock_wf_col
|
|
), patch(
|
|
"application.api.user.agents.routes.workflow_nodes_collection", mock_nodes_col
|
|
), patch(
|
|
"application.api.user.agents.routes.workflow_edges_collection", mock_edges_col
|
|
):
|
|
with app.test_request_context(f"/api/delete_agent?id={agent_id}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = DeleteAgent().delete()
|
|
assert response.status_code == 200
|
|
mock_nodes_col.delete_many.assert_not_called()
|
|
|
|
def test_returns_400_on_exception(self, app):
|
|
from application.api.user.agents.routes import DeleteAgent
|
|
|
|
mock_col = Mock()
|
|
mock_col.find_one_and_delete.side_effect = Exception("DB error")
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context(f"/api/delete_agent?id={uuid.uuid4().hex}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = DeleteAgent().delete()
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPinnedAgents:
|
|
|
|
def test_returns_401_unauthenticated(self, app):
|
|
from application.api.user.agents.routes import PinnedAgents
|
|
|
|
with app.test_request_context("/api/pinned_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = None
|
|
response = PinnedAgents().get()
|
|
assert response.status_code == 401
|
|
|
|
def test_returns_empty_when_no_pinned(self, app):
|
|
from application.api.user.agents.routes import PinnedAgents
|
|
|
|
mock_ensure = Mock(
|
|
return_value={
|
|
"user_id": "user1",
|
|
"agent_preferences": {"pinned": []},
|
|
}
|
|
)
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure
|
|
):
|
|
with app.test_request_context("/api/pinned_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = PinnedAgents().get()
|
|
assert response.status_code == 200
|
|
assert response.json == []
|
|
|
|
def test_returns_pinned_agents(self, app):
|
|
from application.api.user.agents.routes import PinnedAgents
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
key = str(uuid.uuid4())
|
|
mock_ensure = Mock(
|
|
return_value={
|
|
"user_id": "user1",
|
|
"agent_preferences": {"pinned": [str(agent_id)]},
|
|
}
|
|
)
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.find.return_value = [
|
|
{
|
|
"_id": agent_id,
|
|
"name": "Pinned Agent",
|
|
"source": "default",
|
|
"retriever": "classic",
|
|
"key": key,
|
|
}
|
|
]
|
|
mock_resolve = Mock(return_value=[])
|
|
mock_users_col = Mock()
|
|
mock_db = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure
|
|
), patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.resolve_tool_details", mock_resolve
|
|
), patch(
|
|
"application.api.user.agents.routes.users_collection", mock_users_col
|
|
), patch(
|
|
"application.api.user.agents.routes.db", mock_db
|
|
):
|
|
with app.test_request_context("/api/pinned_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = PinnedAgents().get()
|
|
assert response.status_code == 200
|
|
data = response.json
|
|
assert len(data) == 1
|
|
assert data[0]["pinned"] is True
|
|
|
|
def test_cleans_up_stale_pinned_ids(self, app):
|
|
from application.api.user.agents.routes import PinnedAgents
|
|
|
|
stale_id = str(uuid.uuid4().hex)
|
|
mock_ensure = Mock(
|
|
return_value={
|
|
"user_id": "user1",
|
|
"agent_preferences": {"pinned": [stale_id]},
|
|
}
|
|
)
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.find.return_value = []
|
|
mock_users_col = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure
|
|
), patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.users_collection", mock_users_col
|
|
):
|
|
with app.test_request_context("/api/pinned_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = PinnedAgents().get()
|
|
assert response.status_code == 200
|
|
mock_users_col.update_one.assert_called_once()
|
|
|
|
def test_returns_400_on_exception(self, app):
|
|
from application.api.user.agents.routes import PinnedAgents
|
|
|
|
mock_ensure = Mock(side_effect=Exception("DB error"))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure
|
|
):
|
|
with app.test_request_context("/api/pinned_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = PinnedAgents().get()
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetTemplateAgents:
|
|
|
|
def test_returns_template_agents(self, app):
|
|
from application.api.user.agents.routes import GetTemplateAgents
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
mock_col = Mock()
|
|
mock_col.find.return_value = [
|
|
{
|
|
"_id": agent_id,
|
|
"name": "Template1",
|
|
"description": "A template",
|
|
"image": "img.png",
|
|
}
|
|
]
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context("/api/template_agents"):
|
|
response = GetTemplateAgents().get()
|
|
assert response.status_code == 200
|
|
data = response.json
|
|
assert len(data) == 1
|
|
assert data[0]["name"] == "Template1"
|
|
|
|
def test_returns_400_on_exception(self, app):
|
|
from application.api.user.agents.routes import GetTemplateAgents
|
|
|
|
mock_col = Mock()
|
|
mock_col.find.side_effect = Exception("DB error")
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context("/api/template_agents"):
|
|
response = GetTemplateAgents().get()
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestAdoptAgent:
|
|
|
|
def test_returns_401_unauthenticated(self, app):
|
|
from application.api.user.agents.routes import AdoptAgent
|
|
|
|
with app.test_request_context("/api/adopt_agent?id=abc"):
|
|
from flask import request
|
|
|
|
request.decoded_token = None
|
|
response = AdoptAgent().post()
|
|
assert response.status_code == 401
|
|
|
|
def test_returns_400_missing_id(self, app):
|
|
from application.api.user.agents.routes import AdoptAgent
|
|
|
|
with app.test_request_context("/api/adopt_agent"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = AdoptAgent().post()
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_404_template_not_found(self, app):
|
|
from application.api.user.agents.routes import AdoptAgent
|
|
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = None
|
|
agent_id = str(uuid.uuid4().hex)
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context(f"/api/adopt_agent?id={agent_id}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = AdoptAgent().post()
|
|
assert response.status_code == 404
|
|
|
|
def test_adopts_agent_success(self, app):
|
|
from application.api.user.agents.routes import AdoptAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
new_id = uuid.uuid4().hex
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = {
|
|
"_id": agent_id,
|
|
"user": "system",
|
|
"name": "Template Agent",
|
|
"description": "A template",
|
|
"source": "default",
|
|
"tools": [],
|
|
}
|
|
mock_col.insert_one.return_value = Mock(inserted_id=new_id)
|
|
mock_resolve = Mock(return_value=[])
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.resolve_tool_details", mock_resolve
|
|
):
|
|
with app.test_request_context(f"/api/adopt_agent?id={agent_id}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = AdoptAgent().post()
|
|
assert response.status_code == 200
|
|
data = response.json
|
|
assert data["success"] is True
|
|
assert data["agent"]["id"] == str(new_id)
|
|
|
|
def test_returns_400_on_exception(self, app):
|
|
from application.api.user.agents.routes import AdoptAgent
|
|
|
|
mock_col = Mock()
|
|
mock_col.find_one.side_effect = Exception("DB error")
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context(f"/api/adopt_agent?id={uuid.uuid4().hex}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = AdoptAgent().post()
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPinAgent:
|
|
|
|
def test_returns_401_unauthenticated(self, app):
|
|
from application.api.user.agents.routes import PinAgent
|
|
|
|
with app.test_request_context("/api/pin_agent"):
|
|
from flask import request
|
|
|
|
request.decoded_token = None
|
|
response = PinAgent().post()
|
|
assert response.status_code == 401
|
|
|
|
def test_returns_400_missing_id(self, app):
|
|
from application.api.user.agents.routes import PinAgent
|
|
|
|
with app.test_request_context("/api/pin_agent"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = PinAgent().post()
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_404_agent_not_found(self, app):
|
|
from application.api.user.agents.routes import PinAgent
|
|
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = None
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context(f"/api/pin_agent?id={uuid.uuid4().hex}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = PinAgent().post()
|
|
assert response.status_code == 404
|
|
|
|
def test_pins_agent(self, app):
|
|
from application.api.user.agents.routes import PinAgent
|
|
|
|
agent_id = str(uuid.uuid4().hex)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = {"_id": agent_id}
|
|
|
|
mock_ensure = Mock(
|
|
return_value={
|
|
"user_id": "user1",
|
|
"agent_preferences": {"pinned": []},
|
|
}
|
|
)
|
|
mock_users_col = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure
|
|
), patch(
|
|
"application.api.user.agents.routes.users_collection", mock_users_col
|
|
):
|
|
with app.test_request_context(f"/api/pin_agent?id={agent_id}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = PinAgent().post()
|
|
assert response.status_code == 200
|
|
assert response.json["action"] == "pinned"
|
|
|
|
def test_unpins_agent(self, app):
|
|
from application.api.user.agents.routes import PinAgent
|
|
|
|
agent_id = str(uuid.uuid4().hex)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = {"_id": agent_id}
|
|
|
|
mock_ensure = Mock(
|
|
return_value={
|
|
"user_id": "user1",
|
|
"agent_preferences": {"pinned": [agent_id]},
|
|
}
|
|
)
|
|
mock_users_col = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure
|
|
), patch(
|
|
"application.api.user.agents.routes.users_collection", mock_users_col
|
|
):
|
|
with app.test_request_context(f"/api/pin_agent?id={agent_id}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = PinAgent().post()
|
|
assert response.status_code == 200
|
|
assert response.json["action"] == "unpinned"
|
|
|
|
def test_returns_500_on_exception(self, app):
|
|
from application.api.user.agents.routes import PinAgent
|
|
|
|
mock_col = Mock()
|
|
mock_col.find_one.side_effect = Exception("DB error")
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context(f"/api/pin_agent?id={uuid.uuid4().hex}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = PinAgent().post()
|
|
assert response.status_code == 500
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRemoveSharedAgent:
|
|
|
|
def test_returns_401_unauthenticated(self, app):
|
|
from application.api.user.agents.routes import RemoveSharedAgent
|
|
|
|
with app.test_request_context("/api/remove_shared_agent"):
|
|
from flask import request
|
|
|
|
request.decoded_token = None
|
|
response = RemoveSharedAgent().delete()
|
|
assert response.status_code == 401
|
|
|
|
def test_returns_400_missing_id(self, app):
|
|
from application.api.user.agents.routes import RemoveSharedAgent
|
|
|
|
with app.test_request_context("/api/remove_shared_agent"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = RemoveSharedAgent().delete()
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_404_shared_agent_not_found(self, app):
|
|
from application.api.user.agents.routes import RemoveSharedAgent
|
|
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = None
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/remove_shared_agent?id={uuid.uuid4().hex}"
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = RemoveSharedAgent().delete()
|
|
assert response.status_code == 404
|
|
|
|
def test_removes_shared_agent_success(self, app):
|
|
from application.api.user.agents.routes import RemoveSharedAgent
|
|
|
|
agent_id = str(uuid.uuid4().hex)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = {
|
|
"_id": agent_id,
|
|
"shared_publicly": True,
|
|
}
|
|
|
|
mock_ensure = Mock(
|
|
return_value={
|
|
"user_id": "user1",
|
|
"agent_preferences": {"pinned": [], "shared_with_me": [agent_id]},
|
|
}
|
|
)
|
|
mock_users_col = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure
|
|
), patch(
|
|
"application.api.user.agents.routes.users_collection", mock_users_col
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/remove_shared_agent?id={agent_id}"
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = RemoveSharedAgent().delete()
|
|
assert response.status_code == 200
|
|
assert response.json["action"] == "removed"
|
|
mock_users_col.update_one.assert_called_once()
|
|
|
|
def test_returns_500_on_exception(self, app):
|
|
from application.api.user.agents.routes import RemoveSharedAgent
|
|
|
|
mock_col = Mock()
|
|
mock_col.find_one.side_effect = Exception("DB error")
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/remove_shared_agent?id={uuid.uuid4().hex}"
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = RemoveSharedAgent().delete()
|
|
assert response.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Additional coverage tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCreateAgentFormDataEdgeCases:
|
|
"""Cover form-data JSON-parse fallback branches in CreateAgent."""
|
|
|
|
def test_form_data_invalid_sources_json_falls_back(self, app):
|
|
"""Lines 474-475: invalid JSON for 'sources' falls back to []."""
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
inserted_id = uuid.uuid4().hex
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.insert_one.return_value = Mock(inserted_id=inserted_id)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
content_type="multipart/form-data",
|
|
data={
|
|
"name": "Agent",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
"sources": "not-valid-json",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 201
|
|
|
|
def test_form_data_invalid_json_schema_falls_back(self, app):
|
|
"""Lines 479-480: invalid JSON for 'json_schema' falls back to None."""
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
inserted_id = uuid.uuid4().hex
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.insert_one.return_value = Mock(inserted_id=inserted_id)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
content_type="multipart/form-data",
|
|
data={
|
|
"name": "Agent",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
"json_schema": "{bad json",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 201
|
|
|
|
def test_form_data_invalid_models_json_falls_back(self, app):
|
|
"""Lines 484-485: invalid JSON for 'models' falls back to []."""
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
inserted_id = uuid.uuid4().hex
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.insert_one.return_value = Mock(inserted_id=inserted_id)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
content_type="multipart/form-data",
|
|
data={
|
|
"name": "Agent",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
"models": "not-json",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 201
|
|
|
|
def test_create_with_unknown_agent_type_defaults_classic(self, app):
|
|
"""Line 511-517: unknown agent_type not in AGENT_TYPE_SCHEMAS."""
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
inserted_id = uuid.uuid4().hex
|
|
mock_agents_col = Mock()
|
|
mock_agents_col.insert_one.return_value = Mock(inserted_id=inserted_id)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_agents_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "Agent",
|
|
"status": "draft",
|
|
"agent_type": "totally_unknown",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 201
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestUpdateAgentFormDataEdgeCases:
|
|
"""Cover form-data parsing and edge cases in UpdateAgent."""
|
|
|
|
def _make_existing_agent(self, agent_id=None):
|
|
return {
|
|
"_id": agent_id or uuid.uuid4().hex,
|
|
"user": "user1",
|
|
"name": "Existing Agent",
|
|
"description": "existing desc",
|
|
"source": "default",
|
|
"chunks": "5",
|
|
"retriever": "classic",
|
|
"prompt_id": "default",
|
|
"status": "published",
|
|
"agent_type": "classic",
|
|
"key": str(uuid.uuid4()),
|
|
}
|
|
|
|
def test_form_data_parses_json_fields(self, app):
|
|
"""Line 699: form.to_dict() branch in UpdateAgent."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
content_type="multipart/form-data",
|
|
data={
|
|
"name": "Updated",
|
|
"tools": '["tool1"]',
|
|
"sources": '["default"]',
|
|
"models": '["m1"]',
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_form_data_invalid_json_sources_returns_400(self, app):
|
|
"""Lines 712-713: invalid JSON for sources in form data."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
content_type="multipart/form-data",
|
|
data={"sources": "{bad json"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
assert "Invalid JSON format" in response.json["message"]
|
|
|
|
def test_form_data_invalid_json_models_returns_400(self, app):
|
|
"""Lines 712-713: invalid JSON for models in form data."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
content_type="multipart/form-data",
|
|
data={"models": "not-json{"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
assert "Invalid JSON format" in response.json["message"]
|
|
|
|
def test_form_data_empty_json_schema_sets_none(self, app):
|
|
"""Line 715-716: empty json_schema string sets None."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
content_type="multipart/form-data",
|
|
data={"json_schema": ""},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_find_agent_db_error(self, app):
|
|
"""Line 730: exception when finding agent in DB."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = str(uuid.uuid4().hex)
|
|
mock_col = Mock()
|
|
mock_col.find_one.side_effect = Exception("DB connection lost")
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"name": "Updated"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(agent_id)
|
|
assert response.status_code == 500
|
|
assert "Database error" in response.json["message"]
|
|
|
|
def test_handle_image_upload_error_forwarded(self, app):
|
|
"""Line 746: image upload error is returned directly."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
error_response = Mock(status_code=400)
|
|
mock_handle_img = Mock(return_value=(None, error_response))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"name": "Updated"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
result = UpdateAgent().put(str(agent_id))
|
|
assert result == error_response
|
|
|
|
def test_limited_request_mode_string_true(self, app):
|
|
"""Lines 904-907: limited_request_mode as string 'True'."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={
|
|
"limited_request_mode": "True",
|
|
"request_limit": 100,
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_publish_workflow_missing_workflow_id(self, app):
|
|
"""Line 1031: workflow_id is missing when publishing workflow."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "WF Agent",
|
|
"status": "draft",
|
|
"agent_type": "workflow",
|
|
"key": "",
|
|
"workflow": "",
|
|
}
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"status": "published"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
assert "Workflow" in response.json["message"]
|
|
|
|
def test_publish_workflow_invalid_workflow_id_format(self, app):
|
|
"""Line 1032-1033: workflow_id is present but not valid ObjectId."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "WF Agent",
|
|
"status": "draft",
|
|
"agent_type": "workflow",
|
|
"key": "",
|
|
"workflow": "not-an-objectid",
|
|
}
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"status": "published"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_publish_workflow_workflow_not_found(self, app):
|
|
"""Lines 1035-1039: workflow exists but not found in DB."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
wf_id = str(uuid.uuid4().hex)
|
|
existing = {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "WF Agent",
|
|
"status": "draft",
|
|
"agent_type": "workflow",
|
|
"key": "",
|
|
"workflow": wf_id,
|
|
}
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
mock_wf_col = Mock()
|
|
mock_wf_col.find_one.return_value = None
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
), patch(
|
|
"application.api.user.agents.routes.workflows_collection", mock_wf_col
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"status": "published"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
assert "Workflow access" in response.json["message"]
|
|
|
|
def test_update_with_image_url_set(self, app):
|
|
"""Line 1000-1001: image_url is truthy, added to update_fields."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("new_img.png", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
content_type="multipart/form-data",
|
|
data={"name": "Updated"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_publish_generates_key_in_response(self, app):
|
|
"""Line 1134-1136: newly_generated_key is included in response."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
existing["status"] = "draft"
|
|
existing["key"] = ""
|
|
existing["source"] = "default"
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"status": "published"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
assert "key" in response.json
|
|
assert len(response.json["key"]) > 0
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestDeleteAgentWorkflowEdgeCases:
|
|
"""Cover workflow cleanup edge cases in DeleteAgent."""
|
|
|
|
def test_delete_workflow_invalid_workflow_id(self, app):
|
|
"""Lines 1179-1181: invalid workflow id skips cleanup."""
|
|
from application.api.user.agents.routes import DeleteAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
mock_col = Mock()
|
|
mock_col.find_one_and_delete.return_value = {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"agent_type": "workflow",
|
|
"workflow": "not-a-valid-oid",
|
|
}
|
|
mock_wf_col = Mock()
|
|
mock_nodes_col = Mock()
|
|
mock_edges_col = Mock()
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.workflows_collection", mock_wf_col
|
|
), patch(
|
|
"application.api.user.agents.routes.workflow_nodes_collection",
|
|
mock_nodes_col,
|
|
), patch(
|
|
"application.api.user.agents.routes.workflow_edges_collection",
|
|
mock_edges_col,
|
|
):
|
|
with app.test_request_context(f"/api/delete_agent?id={agent_id}"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = DeleteAgent().delete()
|
|
assert response.status_code == 200
|
|
mock_nodes_col.delete_many.assert_not_called()
|
|
mock_wf_col.delete_one.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Coverage — additional uncovered lines
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCreateAgentEmptyAgentType:
|
|
"""Cover line 517: agent_type defaults to 'classic' when empty."""
|
|
|
|
def test_empty_agent_type_defaults_to_classic(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
mock_col = Mock()
|
|
mock_col.insert_one.return_value = Mock(inserted_id=uuid.uuid4().hex)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "Test",
|
|
"status": "draft",
|
|
"agent_type": "",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 201
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestUpdateAgentFormDataPath:
|
|
"""Cover lines 666, 699, 815-816, 859, 887, 948-951, 1073, 1108, 1136."""
|
|
|
|
def _make_existing_agent(self, agent_id):
|
|
return {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "Existing",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
"key": "abcdefghijklmnop",
|
|
"source": "default",
|
|
}
|
|
|
|
def test_form_data_parsing(self, app):
|
|
"""Cover line 699: form data parsing path."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
content_type="application/x-www-form-urlencoded",
|
|
data={"name": "Updated"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_invalid_source_id_in_sources_list(self, app):
|
|
"""Cover lines 815-816: invalid source ID in sources list."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"sources": ["not-a-valid-oid"]},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
assert "Invalid source ID" in response.json["message"]
|
|
|
|
def test_tools_not_a_list(self, app):
|
|
"""Cover line 859: tools is not a list."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"tools": "not-a-list"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
assert "Tools must be a list" in response.json["message"]
|
|
|
|
def test_limited_token_mode_string_true(self, app):
|
|
"""Cover line 887: limited_token_mode string 'True' converted."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"limited_token_mode": "True", "token_limit": 5000},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_request_limit_without_limited_mode(self, app):
|
|
"""Cover lines 948-951: request_limit without limited_request_mode."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={
|
|
"request_limit": 100,
|
|
"limited_request_mode": False,
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
assert "Request limit cannot be set" in response.json["message"]
|
|
|
|
def test_folder_id_update(self, app):
|
|
"""Cover lines 950-951: folder_id field update."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
folder_oid = str(uuid.uuid4().hex)
|
|
mock_folders_col = Mock()
|
|
mock_folders_col.find_one.return_value = {"_id": folder_oid, "user": "user1"}
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
), patch(
|
|
"application.api.user.agents.routes.agent_folders_collection", mock_folders_col
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"folder_id": folder_oid},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
def test_publish_missing_source_returns_400(self, app):
|
|
"""Cover line 1073: published with source_val check."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "Agent",
|
|
"description": "desc",
|
|
"prompt_id": "default",
|
|
"status": "draft",
|
|
"agent_type": "classic",
|
|
"key": "abcdefghijklmnop",
|
|
"retriever": "classic",
|
|
"chunks": 5,
|
|
}
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"status": "published"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
|
|
def test_update_not_found_returns_404(self, app):
|
|
"""Cover line 1108: matched_count==0 returns 404."""
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=0, modified_count=0)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"name": "Updated"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetPinnedAgents:
|
|
"""Cover lines 1253, 1262: pinned agents key masking and return."""
|
|
|
|
def test_returns_pinned_agents_with_masked_key(self, app):
|
|
from application.api.user.agents.routes import PinnedAgents
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
mock_col = Mock()
|
|
mock_col.find.return_value = [
|
|
{
|
|
"_id": agent_id,
|
|
"name": "Pinned",
|
|
"description": "desc",
|
|
"agent_type": "classic",
|
|
"source": "default",
|
|
"retriever": "classic",
|
|
"status": "published",
|
|
"key": "abcdefghijklmnop",
|
|
"createdAt": "",
|
|
"updatedAt": "",
|
|
"lastUsedAt": "",
|
|
}
|
|
]
|
|
mock_ensure_user = Mock(
|
|
return_value={"agent_preferences": {"pinned": [str(agent_id)]}}
|
|
)
|
|
mock_resolve_tools = Mock(return_value=[])
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure_user
|
|
), patch(
|
|
"application.api.user.agents.routes.resolve_tool_details", mock_resolve_tools
|
|
):
|
|
with app.test_request_context("/api/pinned_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = PinnedAgents().get()
|
|
assert response.status_code == 200
|
|
data = response.json
|
|
assert len(data) == 1
|
|
assert data[0]["key"] == "abcd...mnop"
|
|
|
|
def test_returns_empty_for_no_pinned(self, app):
|
|
from application.api.user.agents.routes import PinnedAgents
|
|
|
|
mock_ensure_user = Mock(
|
|
return_value={"agent_preferences": {"pinned": []}}
|
|
)
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.ensure_user_doc", mock_ensure_user
|
|
):
|
|
with app.test_request_context("/api/pinned_agents"):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = PinnedAgents().get()
|
|
assert response.status_code == 200
|
|
assert response.json == []
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetTemplateAgentsErrorPath:
|
|
"""Cover line 1283: template agents error path."""
|
|
|
|
def test_template_agents_error_returns_400(self, app):
|
|
from application.api.user.agents.routes import GetTemplateAgents
|
|
|
|
mock_col = Mock()
|
|
mock_col.find.side_effect = Exception("DB error")
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
):
|
|
with app.test_request_context("/api/template_agents"):
|
|
response = GetTemplateAgents().get()
|
|
assert response.status_code == 400
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Additional coverage for agents/routes.py
|
|
# Lines: 526 (workflow validation error), 666 (models field),
|
|
# 815-816 (invalid source id in list), 887 (limited_token_mode bool),
|
|
# 948-951 (request_limit without mode), 1073 (has_valid_source),
|
|
# 1253 (key masking), 1262 (return pinned agents)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCreateAgentWorkflowError:
|
|
"""Cover lines 525-526: workflow validation returns error."""
|
|
|
|
def test_workflow_validation_error_on_create(self, app):
|
|
from application.api.user.agents.routes import CreateAgent
|
|
|
|
mock_col = Mock()
|
|
mock_col.insert_one.return_value = Mock(inserted_id=uuid.uuid4().hex)
|
|
|
|
error_response = Mock()
|
|
error_response.status_code = 400
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.validate_workflow_access",
|
|
return_value=(None, error_response),
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload",
|
|
return_value=("", None),
|
|
):
|
|
with app.test_request_context(
|
|
"/api/create_agent",
|
|
method="POST",
|
|
json={
|
|
"name": "Test",
|
|
"agent_type": "workflow",
|
|
"status": "draft",
|
|
},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = CreateAgent().post()
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestUpdateAgentInvalidSourceInList:
|
|
"""Cover lines 815-816: invalid source ID in sources list."""
|
|
|
|
def _make_existing_agent(self, agent_id):
|
|
return {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "Test",
|
|
"agent_type": "classic",
|
|
"status": "draft",
|
|
}
|
|
|
|
def test_invalid_source_in_list(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"sources": ["not_valid_source_id!!!"]},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 400
|
|
assert "Invalid source ID" in response.json["message"]
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestUpdateAgentModelsField:
|
|
"""Cover line 666: models list field in update."""
|
|
|
|
def _make_existing_agent(self, agent_id):
|
|
return {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "Test",
|
|
"agent_type": "classic",
|
|
"status": "draft",
|
|
}
|
|
|
|
def test_models_field_update(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"models": ["model-1", "model-2"]},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestUpdateAgentHasValidSourceDefault:
|
|
"""Cover line 1073: has_valid_source with source == 'default'."""
|
|
|
|
def _make_existing_agent(self, agent_id):
|
|
return {
|
|
"_id": agent_id,
|
|
"user": "user1",
|
|
"name": "Test",
|
|
"agent_type": "classic",
|
|
"status": "published",
|
|
"description": "desc",
|
|
"prompt_id": "default",
|
|
"source": "default",
|
|
"chunks": 2,
|
|
}
|
|
|
|
def test_has_valid_source_with_default(self, app):
|
|
from application.api.user.agents.routes import UpdateAgent
|
|
|
|
agent_id = uuid.uuid4().hex
|
|
existing = self._make_existing_agent(agent_id)
|
|
mock_col = Mock()
|
|
mock_col.find_one.return_value = existing
|
|
mock_col.update_one.return_value = Mock(matched_count=1, modified_count=1)
|
|
mock_handle_img = Mock(return_value=("", None))
|
|
|
|
with patch(
|
|
"application.api.user.agents.routes.agents_collection", mock_col
|
|
), patch(
|
|
"application.api.user.agents.routes.handle_image_upload", mock_handle_img
|
|
):
|
|
with app.test_request_context(
|
|
f"/api/update_agent/{agent_id}",
|
|
method="PUT",
|
|
json={"name": "UpdatedName"},
|
|
):
|
|
from flask import request
|
|
|
|
request.decoded_token = {"sub": "user1"}
|
|
response = UpdateAgent().put(str(agent_id))
|
|
assert response.status_code == 200
|