mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-13 14:15:52 +00:00
Alex
574f96341e
The backend import package is now docsgpt, the name it will carry on PyPI; application was far too generic to install into anyone's site-packages. git mv plus a mechanical rewrite of every import, dotted string and path reference: 734 Python files, the compose files, Dockerfile, workflows, docs, setup scripts, devcontainer, k8s manifests, vscode config, pytest and coverage config, .gitignore. Behaviour is unchanged. Kept for one release: - A top-level application package whose meta-path finder resolves application.x.y to the already-imported docsgpt.x.y object, so old imports and entry points (celery -A application.app.celery, uvicorn application.asgi:asgi_app) keep working with a FutureWarning. - Celery registers every application.* task name as an alias of its docsgpt.* task on start-up, so messages queued by the previous release still run. The redbeat key prefix moves to redbeat:docsgpt:v2: so schedule entries the previous release wrote are left unread instead of firing twice. The backend image builds from the repository root (docker build -f docsgpt/Dockerfile .) so it can ship the alias package; a root .dockerignore allow-lists docsgpt/ and application/ and keeps caches, local data, .env files, the sample index files and the Dockerfile out. Compose and the image workflows point at the new context.
269 lines
8.5 KiB
Python
269 lines
8.5 KiB
Python
"""Tests for helper functions in docsgpt/api/user/agents/routes.py."""
|
|
|
|
import json
|
|
from contextlib import contextmanager
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from flask import Flask
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
return Flask(__name__)
|
|
|
|
|
|
@contextmanager
|
|
def _patch_db(conn):
|
|
@contextmanager
|
|
def _yield():
|
|
yield conn
|
|
|
|
with patch(
|
|
"docsgpt.api.user.agents.routes.db_session", _yield
|
|
), patch(
|
|
"docsgpt.api.user.agents.routes.db_readonly", _yield
|
|
):
|
|
yield
|
|
|
|
|
|
class TestNormalizeWorkflowReference:
|
|
def test_none_returns_none(self):
|
|
from docsgpt.api.user.agents.routes import (
|
|
normalize_workflow_reference,
|
|
)
|
|
assert normalize_workflow_reference(None) is None
|
|
|
|
def test_dict_returns_id_field(self):
|
|
from docsgpt.api.user.agents.routes import (
|
|
normalize_workflow_reference,
|
|
)
|
|
assert normalize_workflow_reference({"id": "w1"}) == "w1"
|
|
|
|
def test_dict_returns_workflow_id_field(self):
|
|
from docsgpt.api.user.agents.routes import (
|
|
normalize_workflow_reference,
|
|
)
|
|
assert (
|
|
normalize_workflow_reference({"workflow_id": "wf99"}) == "wf99"
|
|
)
|
|
|
|
def test_dict_with_underscore_id(self):
|
|
from docsgpt.api.user.agents.routes import (
|
|
normalize_workflow_reference,
|
|
)
|
|
assert normalize_workflow_reference({"_id": "wf-2"}) == "wf-2"
|
|
|
|
def test_empty_string_returns_empty(self):
|
|
from docsgpt.api.user.agents.routes import (
|
|
normalize_workflow_reference,
|
|
)
|
|
assert normalize_workflow_reference("") == ""
|
|
|
|
def test_plain_string_returned(self):
|
|
from docsgpt.api.user.agents.routes import (
|
|
normalize_workflow_reference,
|
|
)
|
|
assert normalize_workflow_reference("wf-123") == "wf-123"
|
|
|
|
def test_json_string_dict(self):
|
|
from docsgpt.api.user.agents.routes import (
|
|
normalize_workflow_reference,
|
|
)
|
|
assert (
|
|
normalize_workflow_reference(json.dumps({"id": "x"})) == "x"
|
|
)
|
|
|
|
def test_json_string_as_string(self):
|
|
from docsgpt.api.user.agents.routes import (
|
|
normalize_workflow_reference,
|
|
)
|
|
assert normalize_workflow_reference('"wf-str"') == "wf-str"
|
|
|
|
def test_invalid_json_returns_original(self):
|
|
from docsgpt.api.user.agents.routes import (
|
|
normalize_workflow_reference,
|
|
)
|
|
assert normalize_workflow_reference("not-json-wf") == "not-json-wf"
|
|
|
|
def test_non_string_non_dict_coerced(self):
|
|
from docsgpt.api.user.agents.routes import (
|
|
normalize_workflow_reference,
|
|
)
|
|
assert normalize_workflow_reference(42) == "42"
|
|
|
|
|
|
class TestResolveWorkflowForUser:
|
|
def test_none_workflow_returns_none(self, pg_conn):
|
|
from docsgpt.api.user.agents.routes import (
|
|
_resolve_workflow_for_user,
|
|
)
|
|
pg_id, err = _resolve_workflow_for_user(pg_conn, None, "u")
|
|
assert pg_id is None and err is None
|
|
|
|
def test_not_found_returns_error(self, pg_conn, app):
|
|
from docsgpt.api.user.agents.routes import (
|
|
_resolve_workflow_for_user,
|
|
)
|
|
with app.app_context():
|
|
pg_id, err = _resolve_workflow_for_user(
|
|
pg_conn,
|
|
"00000000-0000-0000-0000-000000000000",
|
|
"u",
|
|
)
|
|
assert pg_id is None
|
|
assert err is not None
|
|
assert err.status_code == 404
|
|
|
|
def test_resolves_owned_workflow(self, pg_conn, app):
|
|
from docsgpt.api.user.agents.routes import (
|
|
_resolve_workflow_for_user,
|
|
)
|
|
from docsgpt.storage.db.repositories.workflows import (
|
|
WorkflowsRepository,
|
|
)
|
|
|
|
user = "owner"
|
|
wf = WorkflowsRepository(pg_conn).create(user, "wf")
|
|
with app.app_context():
|
|
pg_id, err = _resolve_workflow_for_user(
|
|
pg_conn, str(wf["id"]), user,
|
|
)
|
|
assert pg_id == str(wf["id"])
|
|
assert err is None
|
|
|
|
|
|
class TestResolveFolderId:
|
|
def test_none_returns_none(self, pg_conn):
|
|
from docsgpt.api.user.agents.routes import _resolve_folder_id
|
|
pg_id, err = _resolve_folder_id(pg_conn, None, "u")
|
|
assert pg_id is None
|
|
assert err is None
|
|
|
|
def test_not_found_returns_error(self, pg_conn, app):
|
|
from docsgpt.api.user.agents.routes import _resolve_folder_id
|
|
|
|
with app.app_context():
|
|
pg_id, err = _resolve_folder_id(
|
|
pg_conn, "00000000-0000-0000-0000-000000000000", "u",
|
|
)
|
|
assert pg_id is None
|
|
assert err.status_code == 404
|
|
|
|
def test_resolves_owned_folder(self, pg_conn, app):
|
|
from docsgpt.api.user.agents.routes import _resolve_folder_id
|
|
from docsgpt.storage.db.repositories.agent_folders import (
|
|
AgentFoldersRepository,
|
|
)
|
|
|
|
user = "u"
|
|
folder = AgentFoldersRepository(pg_conn).create(user, "F")
|
|
with app.app_context():
|
|
pg_id, err = _resolve_folder_id(pg_conn, str(folder["id"]), user)
|
|
assert pg_id == str(folder["id"])
|
|
assert err is None
|
|
|
|
|
|
class TestFormatAgentOutput:
|
|
def test_basic_shape(self):
|
|
from docsgpt.api.user.agents.routes import _format_agent_output
|
|
|
|
agent = {
|
|
"id": "agent-1",
|
|
"name": "My",
|
|
"description": "D",
|
|
"source_id": "src-1",
|
|
"extra_source_ids": ["s1", "s2"],
|
|
"chunks": 4,
|
|
"retriever": "classic",
|
|
"prompt_id": "p-1",
|
|
"tools": [],
|
|
"agent_type": "classic",
|
|
"status": "draft",
|
|
"key": "secret-api-key-long",
|
|
}
|
|
out = _format_agent_output(agent)
|
|
assert out["id"] == "agent-1"
|
|
assert out["name"] == "My"
|
|
assert out["source"] == "src-1"
|
|
assert out["sources"] == ["s1", "s2"]
|
|
# masked key: first4...last4
|
|
assert out["key"].startswith("secr") and out["key"].endswith("long")
|
|
|
|
def test_no_key_masking(self):
|
|
from docsgpt.api.user.agents.routes import _format_agent_output
|
|
|
|
agent = {"id": "a", "name": "n", "chunks": None}
|
|
out = _format_agent_output(agent, include_key_masked=False)
|
|
assert "key" not in out
|
|
|
|
def test_empty_key_returns_empty_string(self):
|
|
from docsgpt.api.user.agents.routes import _format_agent_output
|
|
|
|
agent = {"id": "a", "name": "n", "key": ""}
|
|
out = _format_agent_output(agent)
|
|
assert out["key"] == ""
|
|
|
|
def test_with_folder_and_workflow(self):
|
|
from docsgpt.api.user.agents.routes import _format_agent_output
|
|
|
|
agent = {
|
|
"id": "a", "name": "n",
|
|
"folder_id": "f-1", "workflow_id": "w-1",
|
|
}
|
|
out = _format_agent_output(agent)
|
|
assert out["folder_id"] == "f-1"
|
|
assert out["workflow"] == "w-1"
|
|
|
|
|
|
class TestBuildCreateKwargs:
|
|
def test_classic_kwargs(self):
|
|
from docsgpt.api.user.agents.routes import _build_create_kwargs
|
|
|
|
data = {
|
|
"description": "d",
|
|
"agent_type": "classic",
|
|
"chunks": "3",
|
|
"retriever": "classic",
|
|
}
|
|
out = _build_create_kwargs(
|
|
data, image_url="", agent_type="classic",
|
|
)
|
|
assert out["description"] == "d"
|
|
assert out["chunks"] == 3
|
|
|
|
def test_invalid_chunks_skipped(self, app):
|
|
from docsgpt.api.user.agents.routes import _build_create_kwargs
|
|
|
|
data = {"chunks": "abc"}
|
|
with app.app_context():
|
|
out = _build_create_kwargs(
|
|
data, image_url="", agent_type="classic",
|
|
)
|
|
assert "chunks" not in out
|
|
|
|
def test_prompt_id_default_not_set(self):
|
|
from docsgpt.api.user.agents.routes import _build_create_kwargs
|
|
|
|
data = {"prompt_id": "default"}
|
|
out = _build_create_kwargs(
|
|
data, image_url="", agent_type="classic",
|
|
)
|
|
assert "prompt_id" not in out
|
|
|
|
def test_image_url_used_when_provided(self):
|
|
from docsgpt.api.user.agents.routes import _build_create_kwargs
|
|
|
|
out = _build_create_kwargs(
|
|
{}, image_url="/upload/img.png", agent_type="classic",
|
|
)
|
|
assert out.get("image") == "/upload/img.png"
|
|
|
|
def test_client_image_path_is_never_persisted(self):
|
|
from docsgpt.api.user.agents.routes import _build_create_kwargs
|
|
|
|
out = _build_create_kwargs(
|
|
{"image": ".env"}, image_url="", agent_type="classic",
|
|
)
|
|
assert "image" not in out
|