Compare commits

...
Author SHA1 Message Date
GareArc f1c32e1bc8 refactor(cli): split release scripts into pure rules and CLI shells
The release script tests could only observe behaviour through a process
exit code, because die() called process.exit from inside otherwise-pure
functions. That forced every test to spawn a subprocess, which forced
its input to come from cli/package.json, which is why editing that file
kept breaking tests — most recently #39658, and next the difyctl version
bump that cli-release.yml requires on every release.

Move the decisions into lib/release-rules.mjs and lib/edge-manifest.mjs,
which take every input as an argument and return values instead of
exiting. release-naming.mjs and release-r2-edge.mjs keep argv parsing,
manifest reading, stdout and exit codes, and nothing else.

Tests now split by what they ask. Logic is unit-tested against literal
inputs, the shells are tested for plumbing only, and release-config.test.ts
is the single place that reads the real manifest — asserting it is
internally consistent, never what it currently contains. Editing
cli/package.json no longer breaks a logic test; a malformed config still
fails. Verified by mutating each field in both directions.

Three intentional behaviour changes:

- compat-check compares the numeric A.B.C core only, ignoring prerelease
  and build suffixes, so Dify 1.16.0-rc1 now satisfies a 1.16.0 window.
  This matches the shipped runtime check in src/version/compat.ts, which
  already stripped suffixes; the release gate was the one disagreeing.
  Removes the hand-rolled prerelease ordering entirely.
- validate now rejects a missing or inverted compat window. It previously
  passed such a config while github-env emitted minDify=undefined.
- release-r2-edge reports a bad channel version under its own name rather
  than release-naming's, since it no longer borrows that script's die().

release-config.test.ts also pins the install scripts to the naming config.
They hardcode artifact names by necessity — they run standalone on a user
machine with no repo and no node — so changing tagPrefix or checksumsSuffix
now fails until all four installers are updated to match, instead of
silently publishing names no installer looks for.
2026-07-27 19:13:42 -07:00
Xiyuan ChenandGitHub a57b0b9b58 feat(plugin): allow disabling the tenant plugin model providers cache (#39632) 2026-07-27 23:53:59 +00:00
Xiyuan ChenandGitHub d8506efed6 fix(cli): decouple release script tests from the live compat window (#39658) 2026-07-27 23:50:05 +00:00
JingyiandGitHub b25b28cc76 fix(workflow): align block icon vector sizes (#39657) 2026-07-27 23:43:03 +00:00
yyhGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
f68cabe226 fix(web): align tour trigger DOM order (#39654)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-27 13:00:16 +00:00
Asuka MinatoGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang
f73f83c5d6 test: use SQLite sessions in core app (#39075)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Byron.wang <byron@dify.ai>
2026-07-27 10:45:16 +00:00
Asuka MinatoGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang
29801d6a65 test: use SQLite sessions in core tools (#39077)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Byron.wang <byron@dify.ai>
2026-07-27 10:45:11 +00:00
Asuka MinatoGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang
75016e8bfe test: use SQLite sessions in core ops (#39072)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Byron.wang <byron@dify.ai>
2026-07-27 10:41:17 +00:00
yyhandGitHub b6dea7b2ba fix(workflow): preserve latest collaboration session (#39646) 2026-07-27 10:37:45 +00:00
yyhandGitHub b2b1cd7e97 chore: update workspace dependencies (#39641) 2026-07-27 10:27:31 +00:00
Asuka MinatoGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang
ca63e01d45 test: use SQLite sessions in controllers service api (#39069)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Byron.wang <byron@dify.ai>
2026-07-27 10:26:23 +00:00
yyhandGitHub 271b6e1f5c fix(ui): unify combobox trigger focus rings (#39643) 2026-07-27 10:21:32 +00:00
Asuka MinatoGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang
609ddf9e01 test: use SQLite sessions in core memory (#39066)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Byron.wang <byron@dify.ai>
2026-07-27 10:15:08 +00:00
Asuka MinatoGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang
9fd1fea58c test: use SQLite sessions in services core (#39057)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Byron.wang <byron@dify.ai>
2026-07-27 09:36:44 +00:00
JoelandGitHub 8de3b4d033 fix: prevent the agent build draft UI from flashing during apply (#39642) 2026-07-27 09:32:52 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
d9c038daf2 chore(deps): bump the storage group across 1 directory with 3 updates (#39604)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-07-27 09:24:58 +00:00
Asuka MinatoGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang
c9c057fd04 test: use SQLite sessions in rag retrieval (#39049)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Byron.wang <byron@dify.ai>
2026-07-27 09:12:30 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
85189be53d chore: bump pymdown-extensions from 10.21.2 to 11.0 in /dify-agent (#39618)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-07-27 09:10:22 +00:00
Asuka MinatoGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
01e736aaf7 test: use sqlite3 session in test_extract_thread_messages (#38739)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-27 09:00:47 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
a28969f564 chore(deps): bump golang.org/x/net from 0.53.0 to 0.55.0 in /dify-agent-runtime (#39025)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-07-27 08:58:40 +00:00
Asuka MinatoandGitHub 42f9610ba5 test: use sqlite3 session in test_api (#38733) 2026-07-27 08:57:51 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
1a7ffbe5ee chore: bump gitpython from 3.1.52 to 3.1.54 in /api (#39619)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-07-27 08:57:44 +00:00
JoelandGitHub 481354ca70 refactor: make agent composer save and publish state consistent (#39637) 2026-07-27 08:43:45 +00:00
3c80857ea3 feat: add bearer auth to agent backend (#39622)
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
2026-07-27 08:37:30 +00:00
Xiyuan ChenandGitHub 94702efbc2 fix: gate service API, MCP and trigger surfaces on enterprise license (#39635) 2026-07-27 08:23:15 +00:00
Asuka MinatoandGitHub 755f7b0e8b test: use sqlite3 session in test_workflow_comment_service (#38700) 2026-07-27 08:22:45 +00:00
Asuka MinatoandGitHub a1c9564b30 test: use sqlite3 session in test_llm_generator_missing (#38742) 2026-07-27 08:22:16 +00:00
Asuka MinatoandGitHub 97acd9c70b test: use sqlite3 session in test_utils (#38729) 2026-07-27 08:21:39 +00:00
Asuka MinatoandGitHub 71601bf76c test: use sqlite3 session in test_draft_var_loader_simple (#38702) 2026-07-27 08:20:18 +00:00
Asuka MinatoandGitHub 460efbf285 test: use sqlite3 session in test_workflow_run_service (#38701) 2026-07-27 08:19:32 +00:00
a752f43b8e test: use pristine file-backed SQLite fixtures (#39624)
Co-authored-by: Asuka Minato <i@asukaminato.eu.org>
2026-07-27 08:07:05 +00:00
非法操作GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
b3774bfe1c perf: add lightweight recent apps endpoint (#39625)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-27 08:00:39 +00:00
yyhandGitHub 4313d23889 perf(web): lazy-load Home creation modals (#39634) 2026-07-27 07:48:35 +00:00
林玮 (Jade Lin)andGitHub 26a43db6a4 fix(api): wait for workflow worker cleanup before task completion (#39614) 2026-07-27 07:34:25 +00:00
yyhandGitHub e5d40336b3 fix(workflow): lazy-load Loro collaboration runtime (#39631) 2026-07-27 07:15:13 +00:00
yyhandGitHub 58e2bcbba1 fix(web): stop reporting workspace status (#39626) 2026-07-27 06:22:37 +00:00
Yunlu WenGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
e1b55f54e6 feat(agent): add a squid proxy for agent sandbox (#39544)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-27 06:21:20 +00:00
Yunlu WenGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
989039db6e fix: use jinja sandbox (#39609)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-27 06:03:01 +00:00
JoelGitHub林玮autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
52428df1bd feat: add amptitude to new agent (#39608)
Co-authored-by: 林玮 (Jade Lin) <linw1995@icloud.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-27 05:54:45 +00:00
yyhGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
28d174603f fix: align workspace card plan ownership (#39616)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-27 05:01:02 +00:00
Xiyuan ChenandGitHub b6099d09ff feat(inner_api): add endpoint to invalidate plugin model providers cache (#39468) 2026-07-27 04:33:36 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2962a7ea93 chore(deps): bump the github-actions-dependencies group with 6 updates (#39603)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-07-27 04:32:09 +00:00
zyssyz123andGitHub cc01189966 fix(agent): expose publish state from composer (#39615) 2026-07-27 04:05:38 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
d563d6e7df chore: bump pypdf from 6.10.2 to 6.14.2 in /dify-agent (#39502)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-07-27 03:56:57 +00:00
CYJ1226andGitHub e4e99d198b fix: preserve spaces during recursive text splitting (#39498) 2026-07-27 03:21:26 +00:00
yyhandGitHub 34e6e5a049 refactor(web): simplify Dify logo theming (#39580) 2026-07-27 03:01:24 +00:00
林玮 (Jade Lin)GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
7e6ba05464 feat(api): expose app mode in webapp site response (#39607)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-27 02:57:18 +00:00
Madan kumarandGitHub aeda37db68 fix(rag): stop the document cleaner from stripping valid characters ï, ¿, ¾ (#39215) 2026-07-27 02:53:33 +00:00
FFXNandGitHub 441f9f9ec0 fix: The expiration time for web app login to JWT is incorrectly set. (#39537) 2026-07-27 02:43:32 +00:00
f16c249b1e fix(api): honor array-element limit and byte budget for list[File] in VariableTruncator (#39218) (#39220)
Co-authored-by: Taranum Wasu <taranumwasu@Taranums-MacBook-Pro.local>
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-27 02:12:57 +00:00
FFXNGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
858b3b74dc feat: add CE telemetry report (#39452)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-27 02:04:28 +00:00
d5c0e927c6 fix(web): stop chunk length/overlap inputs collapsing to an unusable width in narrow containers (#39600)
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-27 02:03:17 +00:00
Asuka MinatoandGitHub b81737cfb3 test: use sqlite3 session in test_message_service (#38698) 2026-07-27 02:01:47 +00:00
247 changed files with 12087 additions and 9619 deletions
+6 -6
View File
@@ -29,13 +29,13 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
persist-credentials: false
- name: Setup UV and Python
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: true
python-version: ${{ matrix.python-version }}
@@ -88,13 +88,13 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
persist-credentials: false
- name: Setup UV and Python
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: true
python-version: ${{ matrix.python-version }}
@@ -139,13 +139,13 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
persist-credentials: false
- name: Setup UV and Python
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: true
python-version: "3.12"
+3 -3
View File
@@ -20,7 +20,7 @@ jobs:
run: echo "autofix.ci updates pull request branches, not merge group refs."
- if: github.event_name != 'merge_group'
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Check Docker Compose inputs
if: github.event_name != 'merge_group'
@@ -84,12 +84,12 @@ jobs:
dify-agent/pyproject.toml
dify-agent/uv.lock
- if: github.event_name != 'merge_group'
uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
python-version: "3.11"
- if: github.event_name != 'merge_group'
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
- name: Generate Docker Compose
if: github.event_name != 'merge_group' && steps.docker-compose-changes.outputs.any_changed == 'true'
+2 -2
View File
@@ -97,7 +97,7 @@ jobs:
echo "PLATFORM_PAIR=${platform//\//-}" >> $GITHUB_ENV
- name: Login to Docker Hub
uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0
uses: docker/login-action@abd2ef45e78c5afb21d64d4ca52ee8550d9572c7 # v4.5.1
with:
username: ${{ env.DOCKERHUB_USER }}
password: ${{ env.DOCKERHUB_TOKEN }}
@@ -199,7 +199,7 @@ jobs:
merge-multiple: true
- name: Login to Docker Hub
uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0
uses: docker/login-action@abd2ef45e78c5afb21d64d4ca52ee8550d9572c7 # v4.5.1
with:
username: ${{ env.DOCKERHUB_USER }}
password: ${{ env.DOCKERHUB_TOKEN }}
+6 -6
View File
@@ -79,7 +79,7 @@ jobs:
ws2_app_id: ${{ steps.out.outputs.DIFY_E2E_WS2_APP_ID }}
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
with:
ref: ${{ inputs.cli_ref || github.ref }}
persist-credentials: false
@@ -123,7 +123,7 @@ jobs:
shell: bash
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
with:
ref: ${{ inputs.cli_ref || github.ref }}
persist-credentials: false
@@ -170,7 +170,7 @@ jobs:
shell: bash
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
with:
ref: ${{ inputs.cli_ref || github.ref }}
persist-credentials: false
@@ -233,7 +233,7 @@ jobs:
shell: bash
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
with:
ref: ${{ inputs.cli_ref || github.ref }}
persist-credentials: false
@@ -295,7 +295,7 @@ jobs:
shell: bash
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
with:
ref: ${{ inputs.cli_ref || github.ref }}
persist-credentials: false
@@ -351,7 +351,7 @@ jobs:
shell: bash
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
with:
ref: ${{ inputs.cli_ref || github.ref }}
persist-credentials: false
+1 -1
View File
@@ -23,7 +23,7 @@ jobs:
working-directory: ./cli
steps:
- name: Checkout
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
fetch-depth: 0
+2 -2
View File
@@ -35,7 +35,7 @@ jobs:
dify_tag: ${{ steps.resolve.outputs.dify_tag }}
steps:
- name: Checkout
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
@@ -98,7 +98,7 @@ jobs:
DIFY_TAG: ${{ needs.validate.outputs.dify_tag }}
steps:
- name: Checkout
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
fetch-depth: 1
+1 -1
View File
@@ -24,7 +24,7 @@ jobs:
shell: bash
steps:
- name: Checkout cli ref
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
ref: ${{ inputs.cli_ref || github.ref }}
persist-credentials: false
+1 -1
View File
@@ -30,7 +30,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
+4 -4
View File
@@ -13,13 +13,13 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
persist-credentials: false
- name: Setup UV and Python
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: true
python-version: "3.12"
@@ -63,13 +63,13 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
persist-credentials: false
- name: Setup UV and Python
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: true
python-version: "3.12"
+1 -1
View File
@@ -24,7 +24,7 @@ jobs:
name: Require cherry-pick provenance
runs-on: depot-ubuntu-24.04
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
+1 -1
View File
@@ -9,6 +9,6 @@ jobs:
pull-requests: write
runs-on: depot-ubuntu-24.04
steps:
- uses: actions/labeler@b8dd2d9be0f68b860e7dae5dae7d772984eacd6d # v6.2.0
- uses: actions/labeler@bf12e9b00b37c5c0ca2b87b79b2daf7891dbda13 # v7.0.0
with:
sync-labels: true
+1 -1
View File
@@ -47,7 +47,7 @@ jobs:
migration-changed: ${{ steps.changes.outputs.migration }}
sandbox-runtime-changed: ${{ steps.changes.outputs.sandbox-runtime }}
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- uses: dorny/paths-filter@7b450fff21473bca461d4b92ce414b9d0420d706 # v4.0.2
id: changes
with:
+1 -1
View File
@@ -18,7 +18,7 @@ jobs:
outputs:
external-e2e-changed: ${{ steps.changes.outputs.external_e2e }}
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- uses: dorny/paths-filter@7b450fff21473bca461d4b92ce414b9d0420d706 # v4.0.2
id: changes
with:
+2 -2
View File
@@ -17,12 +17,12 @@ jobs:
pull-requests: write
steps:
- name: Checkout PR branch
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
- name: Setup Python & UV
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: true
@@ -21,10 +21,10 @@ jobs:
if: ${{ github.event.workflow_run.conclusion == 'success' && github.event.workflow_run.pull_requests[0].head.repo.full_name != github.repository }}
steps:
- name: Checkout default branch (trusted code)
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Setup Python & UV
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: true
+2 -2
View File
@@ -17,12 +17,12 @@ jobs:
pull-requests: write
steps:
- name: Checkout PR branch
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
- name: Setup Python & UV
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: true
+3 -3
View File
@@ -21,7 +21,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
persist-credentials: false
@@ -45,7 +45,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
persist-credentials: false
@@ -72,7 +72,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
persist-credentials: false
+5 -5
View File
@@ -23,7 +23,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
fetch-depth: 0
@@ -45,7 +45,7 @@ jobs:
- name: Setup UV and Python
if: steps.changed-files.outputs.any_changed == 'true'
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: false
python-version: "3.12"
@@ -93,7 +93,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
@@ -144,7 +144,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
@@ -186,7 +186,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
persist-credentials: false
+1 -1
View File
@@ -24,7 +24,7 @@ jobs:
working-directory: sdks/nodejs-client
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
+2 -2
View File
@@ -40,7 +40,7 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
token: ${{ secrets.GITHUB_TOKEN }}
@@ -158,7 +158,7 @@ jobs:
- name: Run Claude Code for Translation Sync
if: steps.context.outputs.CHANGED_FILES != ''
uses: anthropics/claude-code-action@af0559ee4f514d1ef21826982bed13f7edc3c35e # v1.0.178
uses: anthropics/claude-code-action@be7b93b1907a4abad570368f3c74b6fe3807510b # v1.0.183
with:
anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
github_token: ${{ secrets.GITHUB_TOKEN }}
+1 -1
View File
@@ -21,7 +21,7 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
fetch-depth: 0
+2 -2
View File
@@ -24,7 +24,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
@@ -36,7 +36,7 @@ jobs:
remove_tool_cache: true
- name: Setup UV and Python
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: true
python-version: ${{ matrix.python-version }}
+2 -2
View File
@@ -21,7 +21,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
@@ -33,7 +33,7 @@ jobs:
remove_tool_cache: true
- name: Setup UV and Python
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: true
python-version: ${{ matrix.python-version }}
+2 -2
View File
@@ -26,7 +26,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
@@ -34,7 +34,7 @@ jobs:
uses: ./.github/actions/setup-web
- name: Setup UV and Python
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
with:
enable-cache: true
python-version: "3.12"
+4 -4
View File
@@ -29,7 +29,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
@@ -62,7 +62,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
@@ -100,7 +100,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
@@ -132,7 +132,7 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
+3 -1
View File
@@ -666,6 +666,7 @@ PLUGIN_REMOTE_INSTALL_PORT=5003
PLUGIN_REMOTE_INSTALL_HOST=localhost
PLUGIN_MAX_PACKAGE_SIZE=15728640
PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
PLUGIN_MODEL_PROVIDERS_CACHE_ENABLED=true
PLUGIN_MODEL_PROVIDERS_CACHE_TTL=86400
# Comma-separated marketplace plugin IDs whose latest versions are installed for newly registered users.
# Example: langgenius/openai,langgenius/gemini
@@ -677,6 +678,8 @@ INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y
# Dify Agent backend
AGENT_BACKEND_BASE_URL=http://localhost:5050
# Bearer token sent to the Agent backend /runs API. Must match DIFY_AGENT_API_TOKEN on the server side.
AGENT_BACKEND_API_TOKEN=dify-agent-run-token-for-dev-only
AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS=30
AGENT_BACKEND_STREAM_MAX_RECONNECTS=3
AGENT_BACKEND_RUN_TIMEOUT_SECONDS=1200
@@ -729,7 +732,6 @@ OTEL_MAX_EXPORT_BATCH_SIZE=512
OTEL_METRIC_EXPORT_INTERVAL=60000
OTEL_BATCH_EXPORT_TIMEOUT=10000
OTEL_METRIC_EXPORT_TIMEOUT=30000
# Prevent Clickjacking
ALLOW_EMBED=false
+59 -30
View File
@@ -1,10 +1,13 @@
import logging
import time
from collections.abc import Callable
from typing import NamedTuple
import socketio
from flask import request
from opentelemetry.trace import get_current_span
from opentelemetry.trace.span import INVALID_SPAN_ID, INVALID_TRACE_ID
from werkzeug.exceptions import Forbidden, HTTPException, ServiceUnavailable
from configs import dify_config
from contexts.wrapper import RecyclableContextVar
@@ -42,6 +45,53 @@ _CONSOLE_EXEMPT_PREFIXES = (
"/console/api/activate/check",
)
_WEBAPP_EXEMPT_PREFIXES = ("/api/system-features",)
_INVALID_LICENSE_STATUSES = (LicenseStatus.INACTIVE, LicenseStatus.EXPIRED, LicenseStatus.LOST)
def _session_surface_error(license_status: LicenseStatus | None) -> HTTPException:
if license_status is None:
return UnauthorizedAndForceLogout("Unable to verify enterprise license. Please contact your administrator.")
return UnauthorizedAndForceLogout(f"Enterprise license is {license_status}. Please contact your administrator.")
def _bearer_surface_error(license_status: LicenseStatus | None) -> HTTPException:
"""Token-authed: forcing a logout is meaningless and license state must not leak."""
return Forbidden(description="license_required")
def _retryable_surface_error(license_status: LicenseStatus | None) -> HTTPException:
"""Webhook senders retry on 5xx but treat 4xx as permanent, disabling the subscription."""
return ServiceUnavailable(description="license_required")
class _LicenseGatedSurface(NamedTuple):
prefix: str
exempt_prefixes: tuple[str, ...]
build_error: Callable[[LicenseStatus | None], HTTPException]
# /files (plugin-daemon data plane), /inner/api (enterprise control plane) and /health
# stay ungated: blocking them breaks workflow execution or license recovery itself.
_LICENSE_GATED_SURFACES = (
_LicenseGatedSurface("/console/api/", _CONSOLE_EXEMPT_PREFIXES, _session_surface_error),
_LicenseGatedSurface("/api/", _WEBAPP_EXEMPT_PREFIXES, _session_surface_error),
_LicenseGatedSurface("/v1", (), _bearer_surface_error),
_LicenseGatedSurface("/mcp", (), _bearer_surface_error),
_LicenseGatedSurface("/triggers", (), _retryable_surface_error),
)
def _match_license_gated_surface(path: str) -> _LicenseGatedSurface | None:
for surface in _LICENSE_GATED_SURFACES:
if not path.startswith(surface.prefix):
continue
if any(path.startswith(exempt) for exempt in surface.exempt_prefixes):
return None
return surface
return None
# ----------------------------
# Application Factory Function
@@ -62,38 +112,17 @@ def create_flask_app_with_configs() -> DifyApp:
init_request_context()
RecyclableContextVar.increment_thread_recycles()
# Enterprise license validation for API endpoints (both console and webapp)
# When license expires, block all API access except bootstrap endpoints needed
# for the frontend to load the license expiration page without infinite reloads.
if dify_config.ENTERPRISE_ENABLED:
is_console_api = request.path.startswith("/console/api/")
is_webapp_api = request.path.startswith("/api/")
surface = _match_license_gated_surface(request.path)
if surface is not None:
try:
license_status = EnterpriseService.get_cached_license_status()
except Exception:
logger.exception("Failed to check enterprise license status")
license_status = None
if is_console_api or is_webapp_api:
if is_console_api:
is_exempt = any(request.path.startswith(p) for p in _CONSOLE_EXEMPT_PREFIXES)
else: # webapp API
is_exempt = request.path.startswith("/api/system-features")
if not is_exempt:
try:
# Check license status (cached — see EnterpriseService for TTL details)
license_status = EnterpriseService.get_cached_license_status()
if license_status in (LicenseStatus.INACTIVE, LicenseStatus.EXPIRED, LicenseStatus.LOST):
raise UnauthorizedAndForceLogout(
f"Enterprise license is {license_status}. Please contact your administrator."
)
if license_status is None:
raise UnauthorizedAndForceLogout(
"Unable to verify enterprise license. Please contact your administrator."
)
except UnauthorizedAndForceLogout:
raise
except Exception:
logger.exception("Failed to check enterprise license status")
raise UnauthorizedAndForceLogout(
"Unable to verify enterprise license. Please contact your administrator."
)
if license_status is None or license_status in _INVALID_LICENSE_STATUSES:
raise surface.build_error(license_status)
# add after request hook for injecting trace headers from OpenTelemetry span context
# Only adds headers when OTEL is enabled and has valid context
+5 -1
View File
@@ -11,6 +11,7 @@ from clients.agent_backend.fake_client import FakeAgentBackendRunClient, FakeAge
def create_agent_backend_run_client(
*,
base_url: str | None = None,
api_token: str | None = None,
use_fake: bool = False,
fake_scenario: str | FakeAgentBackendScenario = FakeAgentBackendScenario.SUCCESS,
stream_read_timeout_seconds: float = 30,
@@ -22,8 +23,11 @@ def create_agent_backend_run_client(
return FakeAgentBackendRunClient(scenario=FakeAgentBackendScenario(fake_scenario))
if base_url is None:
raise ValueError("base_url is required when creating a real Agent backend client")
headers: dict[str, str] = {}
if api_token:
headers["Authorization"] = f"Bearer {api_token}"
return DifyAgentBackendRunClient(
Client(base_url=base_url, stream_timeout=stream_read_timeout_seconds),
Client(base_url=base_url, stream_timeout=stream_read_timeout_seconds, headers=headers),
stream_max_reconnects=stream_max_reconnects,
stream_timeout_seconds=stream_run_timeout_seconds,
)
@@ -12,6 +12,11 @@ class AgentBackendConfig(BaseSettings):
default=None,
)
AGENT_BACKEND_API_TOKEN: str | None = Field(
description="Bearer token for authenticating with the Agent backend /runs API.",
default=None,
)
AGENT_BACKEND_USE_FAKE: bool = Field(
description="Use the deterministic in-process fake Agent backend client.",
default=False,
+42
View File
@@ -266,6 +266,12 @@ class PluginConfig(BaseSettings):
default=60 * 60,
)
PLUGIN_MODEL_PROVIDERS_CACHE_ENABLED: bool = Field(
description="Whether tenant plugin model providers are cached in Redis. Disable when plugins are installed "
"by a system other than this one, which cannot invalidate the cache when a tenant's plugins change.",
default=True,
)
PLUGIN_MODEL_PROVIDERS_CACHE_TTL: PositiveInt = Field(
description="TTL in seconds for caching tenant plugin model providers in Redis",
default=60 * 60 * 24,
@@ -816,6 +822,41 @@ class UpdateConfig(BaseSettings):
)
class CommunityTelemetryConfig(BaseSettings):
"""
Configuration for anonymous self-hosted community telemetry.
"""
DISABLE_TELEMETRY: bool = Field(
description="Disable anonymous community telemetry",
default=False,
)
DO_NOT_TRACK: bool = Field(
description="Respect the standard do-not-track opt-out signal for telemetry",
default=False,
)
TELEMETRY_ENDPOINT: str = Field(
description="Endpoint for anonymous community telemetry events",
default="https://otel.dify.ai/v1/events",
)
TELEMETRY_FALLBACK_ENDPOINT: str = Field(
description="Fallback endpoint for anonymous community telemetry events",
default="https://otel.dify.cn/v1/events",
)
TELEMETRY_TIMEOUT_SECONDS: PositiveInt = Field(
description="HTTP timeout in seconds for anonymous community telemetry requests",
default=3,
)
TELEMETRY_HEARTBEAT_INTERVAL_MINUTES: PositiveInt = Field(
description="Celery beat interval in minutes for checking whether heartbeat telemetry is due",
default=30,
)
CI: bool = Field(
description="Whether the process is running in CI; telemetry is skipped when true",
default=False,
)
class WorkflowVariableTruncationConfig(BaseSettings):
WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE: PositiveInt = Field(
# 1000 KiB
@@ -1599,6 +1640,7 @@ class FeatureConfig(
TenantIsolatedTaskQueueConfig,
ToolConfig,
UpdateConfig,
CommunityTelemetryConfig,
WorkflowConfig,
WorkflowNodeExecutionConfig,
WorkspaceConfig,
-5
View File
@@ -257,7 +257,6 @@ class AgentAppDetailWithSite(GenericAppDetailWithSite):
debug_conversation_has_messages: bool = False
debug_conversation_message_count: int = 0
role: str | None = None
active_config_is_published: bool = False
class AgentDebugConversationRefreshResponse(BaseModel):
@@ -410,10 +409,6 @@ def _serialize_agent_app_detail(
payload["debug_conversation_has_messages"] = message_count > 0
payload["debug_conversation_message_count"] = message_count
payload["role"] = agent.role or ""
payload["active_config_is_published"] = roster_service.active_config_is_published(
tenant_id=app_model.tenant_id,
agent=agent,
)
return payload
+77
View File
@@ -58,6 +58,7 @@ from services.app_service import (
AppResponseView,
AppService,
CreateAppParams,
RecentAppMode,
StarredAppListParams,
)
from services.enterprise import rbac_service as enterprise_rbac_service
@@ -139,6 +140,10 @@ class AppListBaseQuery(BaseModel):
raise ValueError("Invalid UUID format in creator_ids.") from exc
class RecentAppListQuery(BaseModel):
limit: int = Field(default=8, ge=1, le=8, description="Number of recently modified apps to return (1-8)")
class AppListQuery(AppListBaseQuery):
pass
@@ -411,6 +416,33 @@ class AppPartial(AppResponseModel):
return to_timestamp(value)
class RecentAppResponse(ResponseModel):
id: str
name: str
icon_type: IconType | None = None
icon: str | None = None
icon_background: str | None = None
mode: RecentAppMode
author_name: str | None = None
updated_at: int
permission_keys: list[str] = Field(default_factory=list)
maintainer: str | None = None
@computed_field(return_type=str | None) # type: ignore[prop-decorator]
@property
def icon_url(self) -> str | None:
return build_icon_url(self.icon_type, self.icon)
@field_validator("updated_at", mode="before")
@classmethod
def _normalize_timestamp(cls, value: datetime | int) -> int:
return to_timestamp(value)
class RecentAppListResponse(ResponseModel):
data: list[RecentAppResponse]
class AppDetail(AppResponseModel):
id: str
name: str
@@ -575,6 +607,8 @@ register_schema_models(
register_response_schema_models(
console_ns,
AppPartial,
RecentAppResponse,
RecentAppListResponse,
AppDetailWithSite,
AppPagination,
)
@@ -699,6 +733,49 @@ class AppListApi(Resource):
return app_detail.model_dump(mode="json"), 201
@console_ns.route("/apps/recent")
class RecentAppListApi(Resource):
@console_ns.doc("list_recent_apps")
@console_ns.doc(description="Get recently modified apps for the home Continue Work section")
@console_ns.doc(params=query_params_from_model(RecentAppListQuery))
@console_ns.response(200, "Success", console_ns.models[RecentAppListResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@enterprise_license_required
@with_session(write=False)
@with_current_user_id
@with_current_tenant_id
def get(self, current_tenant_id: str, current_user_id: str, session: Session):
"""Return the lightweight app cards needed by the Explore home page."""
args = query_params_from_request(RecentAppListQuery)
params = AppListParams(limit=args.limit)
permissions = enterprise_rbac_service.RBACService.MyPermissions.get(
current_tenant_id,
current_user_id,
session=session,
)
if dify_config.RBAC_ENABLED:
access_filter = resolve_app_access_filter(
current_tenant_id,
current_user_id,
session=session,
permissions=permissions,
)
access_filter.apply_to_params(params)
recent_apps = AppService().get_recent_apps(current_user_id, current_tenant_id, params, session)
permission_keys_map = permissions.app.permission_keys_by_resource_ids([app.id for app in recent_apps])
response_items = [
RecentAppResponse.model_validate(app, from_attributes=True).model_copy(
update={"permission_keys": permission_keys_map.get(app.id, [])}
)
for app in recent_apps
]
return dump_response(RecentAppListResponse, {"data": response_items}), 200
@console_ns.route("/apps/starred")
class StarredAppListApi(Resource):
@console_ns.doc("list_starred_apps")
+2 -1
View File
@@ -24,7 +24,7 @@ from extensions.ext_database import db
from fields.base import ResponseModel
from libs.helper import RateLimiter, dump_response, extract_remote_ip, to_timestamp
from models.account import TenantStatus
from models.model import App, Site
from models.model import App, AppMode, Site
from repositories.factory import DifyAPIRepositoryFactory
from services.feature_service import FeatureService
from services.human_input_file_upload_service import HumanInputFileUploadService
@@ -207,6 +207,7 @@ class HumanInputFormApi(Resource):
site=WebAppSiteResponse.from_app_site(
tenant=tenant,
app_model=app_model,
mode=AppMode.value_of(app_model.mode),
site=site,
end_user_id=None,
features=features,
+5 -1
View File
@@ -14,7 +14,7 @@ from extensions.storage.storage_type import StorageType
from fields.base import ResponseModel
from libs.helper import build_icon_url
from models.account import Tenant, TenantStatus
from models.model import App, EndUser, IconType, Site
from models.model import App, AppMode, EndUser, IconType, Site
from services.feature_service import FeatureModel, FeatureService
from services.file_service import FileService
@@ -67,6 +67,7 @@ class WebAppCustomConfigResponse(ResponseModel):
class WebAppSiteResponse(ResponseModel):
app_id: str
mode: AppMode
end_user_id: str | None = None
enable_site: bool
site: WebSiteResponse
@@ -83,6 +84,7 @@ class WebAppSiteResponse(ResponseModel):
*,
tenant: Tenant,
app_model: App,
mode: AppMode,
site: Site,
end_user_id: str | None,
features: FeatureModel,
@@ -109,6 +111,7 @@ class WebAppSiteResponse(ResponseModel):
return cls(
app_id=app_model.id,
mode=mode,
end_user_id=end_user_id,
enable_site=app_model.enable_site,
site=site_response,
@@ -167,6 +170,7 @@ class AppSiteApi(WebApiResource):
return WebAppSiteResponse.from_app_site(
tenant=tenant,
app_model=app_model,
mode=AppMode.value_of(app_model.mode_compatible_with_agent_with_session(session=db.session())),
site=site,
end_user_id=end_user.id,
features=features,
@@ -616,23 +616,34 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
message_snapshot = MessageSnapshot.from_message(message)
session.close()
# return response or stream generator
response = self._handle_advanced_chat_response(
application_generate_entity=application_generate_entity,
workflow=workflow_snapshot,
queue_manager=queue_manager,
conversation=conversation_snapshot,
message=message_snapshot,
user=user,
stream=stream,
draft_var_saver_factory=self._get_draft_var_saver_factory(
invoke_from,
account=user,
tenant_id=application_generate_entity.app_config.tenant_id,
),
)
try:
response = self._handle_advanced_chat_response(
application_generate_entity=application_generate_entity,
workflow=workflow_snapshot,
queue_manager=queue_manager,
conversation=conversation_snapshot,
message=message_snapshot,
user=user,
stream=stream,
draft_var_saver_factory=self._get_draft_var_saver_factory(
invoke_from,
account=user,
tenant_id=application_generate_entity.app_config.tenant_id,
),
)
converted_response = AdvancedChatAppGenerateResponseConverter.convert(
response=response,
invoke_from=invoke_from,
)
except BaseException:
self._join_worker_thread(worker_thread)
raise
return AdvancedChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
if isinstance(converted_response, Generator):
return self._wrap_stream_with_worker_thread_join(converted_response, worker_thread)
self._join_worker_thread(worker_thread)
return converted_response
def _generate_worker(
self,
@@ -538,6 +538,7 @@ class AgentAppGenerator(MessageBasedAppGenerator):
request_builder=AgentAppRuntimeRequestBuilder(credentials_provider=credentials_provider),
agent_backend_client=create_agent_backend_run_client(
base_url=dify_config.AGENT_BACKEND_BASE_URL,
api_token=dify_config.AGENT_BACKEND_API_TOKEN,
use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
+29
View File
@@ -1,3 +1,5 @@
import logging
import threading
from collections.abc import Generator, Mapping, Sequence
from contextlib import AbstractContextManager, nullcontext
from typing import TYPE_CHECKING, Any, Union, final
@@ -23,6 +25,10 @@ from services.workflow_draft_variable_service import DraftVariableSaver as Draft
if TYPE_CHECKING:
from graphon.variables.input_entities import VariableEntity
logger = logging.getLogger(__name__)
_WORKER_THREAD_JOIN_TIMEOUT_SECONDS = 300
@final
class _DebuggerDraftVariableSaver:
@@ -64,6 +70,29 @@ class _DebuggerDraftVariableSaver:
class BaseAppGenerator:
_file_access_controller: DatabaseFileAccessController = DatabaseFileAccessController()
@staticmethod
def _join_worker_thread(worker_thread: threading.Thread) -> None:
# Bound the wait so a leaked app worker cannot occupy an execution slot indefinitely.
worker_thread.join(timeout=_WORKER_THREAD_JOIN_TIMEOUT_SECONDS)
if worker_thread.is_alive():
logger.warning(
"Possible app worker thread leak: thread_name=%s timeout_seconds=%s; "
"continuing without waiting further to avoid occupying an execution slot indefinitely",
worker_thread.name,
_WORKER_THREAD_JOIN_TIMEOUT_SECONDS,
)
@staticmethod
def _wrap_stream_with_worker_thread_join[ResponseT](
response_stream: Generator[ResponseT, None, None],
worker_thread: threading.Thread,
) -> Generator[ResponseT, None, None]:
"""Keep the producer owned by the response stream until both finish."""
try:
yield from response_stream
finally:
BaseAppGenerator._join_worker_thread(worker_thread)
@staticmethod
def _bind_file_access_scope(
*,
@@ -351,17 +351,28 @@ class PipelineGenerator(BaseAppGenerator):
user,
tenant_id=pipeline.tenant_id,
)
# return response or stream generator
response = self._handle_response(
application_generate_entity=application_generate_entity,
workflow=workflow,
queue_manager=queue_manager,
user=user,
stream=streaming,
draft_var_saver_factory=draft_var_saver_factory,
)
try:
response = self._handle_response(
application_generate_entity=application_generate_entity,
workflow=workflow,
queue_manager=queue_manager,
user=user,
stream=streaming,
draft_var_saver_factory=draft_var_saver_factory,
)
converted_response = WorkflowAppGenerateResponseConverter.convert(
response=response,
invoke_from=invoke_from,
)
except BaseException:
self._join_worker_thread(worker_thread)
raise
return WorkflowAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
if isinstance(converted_response, Generator):
return self._wrap_stream_with_worker_thread_join(converted_response, worker_thread)
self._join_worker_thread(worker_thread)
return converted_response
def single_iteration_generate(
self,
+21 -10
View File
@@ -405,17 +405,28 @@ class WorkflowAppGenerator(BaseAppGenerator):
tenant_id=app_model.tenant_id,
)
# return response or stream generator
response = self._handle_response(
application_generate_entity=application_generate_entity,
workflow=workflow,
queue_manager=queue_manager,
user=user,
draft_var_saver_factory=draft_var_saver_factory,
stream=streaming,
)
try:
response = self._handle_response(
application_generate_entity=application_generate_entity,
workflow=workflow,
queue_manager=queue_manager,
user=user,
draft_var_saver_factory=draft_var_saver_factory,
stream=streaming,
)
converted_response = WorkflowAppGenerateResponseConverter.convert(
response=response,
invoke_from=invoke_from,
)
except BaseException:
self._join_worker_thread(worker_thread)
raise
return WorkflowAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
if isinstance(converted_response, Generator):
return self._wrap_stream_with_worker_thread_join(converted_response, worker_thread)
self._join_worker_thread(worker_thread)
return converted_response
def single_iteration_generate(
self,
@@ -39,15 +39,16 @@ class Jinja2TemplateTransformer(TemplateTransformer):
@override
def get_runner_script(cls) -> str:
runner_script = dedent(f"""
import jinja2
import json
from base64 import b64decode
from jinja2.sandbox import SandboxedEnvironment
# declare main function
def main(**inputs):
# Decode base64-encoded template to handle special characters safely
template_code = b64decode('{cls._template_b64_placeholder}').decode('utf-8')
template = jinja2.Template(template_code)
env = SandboxedEnvironment()
template = env.from_string(template_code)
return template.render(**inputs)
# decode and prepare input dict
@@ -67,12 +68,13 @@ class Jinja2TemplateTransformer(TemplateTransformer):
@override
def get_preload_script(cls) -> str:
preload_script = dedent("""
import jinja2
from jinja2.sandbox import SandboxedEnvironment
from base64 import b64decode
def _jinja2_preload_():
# prepare jinja2 environment, load template and render before to avoid sandbox issue
template = jinja2.Template('{{s}}')
# prepare jinja2 sandboxed environment, load template and render
env = SandboxedEnvironment()
template = env.from_string('{{s}}')
template.render(s='a')
if __name__ == '__main__':
+1 -1
View File
@@ -519,7 +519,7 @@ class IndexingRunner:
def filter_string(text):
text = re.sub(r"<\|", "<", text)
text = re.sub(r"\|>", ">", text)
text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F\xEF\xBF\xBE]", "", text)
text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]", "", text)
# Unicode U+FFFE
text = re.sub("\ufffe", "", text)
return text
+11 -4
View File
@@ -434,14 +434,18 @@ class PluginService:
exc_info=True,
)
@classmethod
def _fetch_plugin_model_providers_uncached(
cls, tenant_id: str, client: PluginModelClient | None
) -> tuple[ProviderEntity, ...]:
model_client = client or PluginModelClient()
return tuple(cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id))
@classmethod
def _fetch_and_cache_plugin_model_providers(
cls, tenant_id: str, client: PluginModelClient | None, *, refresh_generation: int | None
) -> tuple[ProviderEntity, ...]:
model_client = client or PluginModelClient()
providers = tuple(
cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id)
)
providers = cls._fetch_plugin_model_providers_uncached(tenant_id, client)
generation = cls._load_plugin_model_providers_generation(tenant_id)
if generation is not None and generation == refresh_generation:
cls._store_cached_plugin_model_providers(tenant_id, generation, providers)
@@ -471,6 +475,9 @@ class PluginService:
are intentionally owned by this service so tenant isolation and cache
expiry are handled in one place.
"""
if not dify_config.PLUGIN_MODEL_PROVIDERS_CACHE_ENABLED:
return cls._fetch_plugin_model_providers_uncached(tenant_id, client)
deadline = time.monotonic() + cls.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT
while True:
+1 -1
View File
@@ -9,7 +9,7 @@ class CleanProcessor:
# remove invalid symbol
text = re.sub(r"<\|", "<", text)
text = re.sub(r"\|>", ">", text)
text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F\xEF\xBF\xBE]", "", text)
text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]", "", text)
# Unicode U+FFFE
text = re.sub("\ufffe", "", text)
@@ -2023,6 +2023,8 @@ class DatasetRetrieval:
redis_client.zremrangebyscore(key, 0, current_time - 60000)
request_count = redis_client.zcard(key)
if request_count > knowledge_rate_limit.limit:
# The rate-limit exception is raised after this block, so commit the audit row
# explicitly instead of relying on the Session context, which only closes it.
with session_factory.create_session() as session:
rate_limit_log = RateLimitLog(
tenant_id=tenant_id,
@@ -2030,6 +2032,7 @@ class DatasetRetrieval:
operation="knowledge",
)
session.add(rate_limit_log)
session.commit()
raise exc.RateLimitExceededError(
"you have reached the knowledge base request rate limit of your subscription."
)
+2 -2
View File
@@ -91,8 +91,8 @@ class FixedRecursiveCharacterTextSplitter(EnhanceRecursiveCharacterTextSplitter)
splits = re.split(r" +", text)
else:
splits = text.split(separator)
if self._keep_separator:
splits = [s + separator for s in splits[:-1]] + splits[-1:]
if self._keep_separator:
splits = [s + separator for s in splits[:-1]] + splits[-1:]
else:
splits = list(text)
if separator == "\n":
+1
View File
@@ -497,6 +497,7 @@ class DifyNodeFactory(NodeFactory):
),
"agent_backend_client": create_agent_backend_run_client(
base_url=dify_config.AGENT_BACKEND_BASE_URL,
api_token=dify_config.AGENT_BACKEND_API_TOKEN,
use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
+27
View File
@@ -5,6 +5,7 @@ from typing import Any
import pytz # type: ignore[import-untyped]
from celery import Celery, Task
from celery.schedules import crontab
from celery.signals import beat_init
from typing_extensions import TypedDict
from configs import dify_config
@@ -36,6 +37,19 @@ class CeleryBeatScheduleEntry(TypedDict):
schedule: crontab | timedelta
def _enqueue_initial_community_telemetry_heartbeat(sender: Any, **_: Any) -> None:
task_name = "community_telemetry.send_heartbeat"
if "community_telemetry_heartbeat" not in sender.app.conf.beat_schedule:
return
task = sender.app.tasks.get(task_name)
if task is not None:
task.apply_async()
beat_init.connect(_enqueue_initial_community_telemetry_heartbeat, weak=False)
def get_celery_ssl_options() -> CelerySSLOptionsDict | None:
"""Get SSL configuration for Celery broker/backend connections."""
# Only apply SSL if we're using Redis as broker/backend
@@ -260,6 +274,19 @@ def init_app(app: DifyApp) -> Celery:
"schedule": timedelta(minutes=dify_config.API_TOKEN_LAST_USED_UPDATE_INTERVAL),
}
if (
dify_config.EDITION == "SELF_HOSTED"
and not dify_config.ENTERPRISE_ENABLED
and not dify_config.DISABLE_TELEMETRY
and not dify_config.DO_NOT_TRACK
and not dify_config.CI
):
imports.append("tasks.community_telemetry_task")
beat_schedule["community_telemetry_heartbeat"] = {
"task": "community_telemetry.send_heartbeat",
"schedule": timedelta(minutes=dify_config.TELEMETRY_HEARTBEAT_INTERVAL_MINUTES),
}
if dify_config.ENTERPRISE_ENABLED and dify_config.ENTERPRISE_TELEMETRY_ENABLED:
imports.append("tasks.enterprise_telemetry_task")
celery_app.conf.update(beat_schedule=beat_schedule, imports=imports)
+1
View File
@@ -383,6 +383,7 @@ class AgentAppComposerResponse(ResponseModel):
variant: Literal[ComposerVariant.AGENT_APP]
agent: AgentComposerAgentResponse
active_config_snapshot: AgentConfigSnapshotSummaryResponse | None = None
active_config_is_published: bool
draft: AgentConfigDraftSummaryResponse | None = None
agent_soul: AgentSoulConfig
save_options: list[ComposerSaveStrategy]
@@ -0,0 +1,30 @@
"""add telemetry fields to dify_setups
Revision ID: 6f5a9c2d8e1b
Revises: d2825e7b9c10
Create Date: 2026-07-23 12:00:00.000000
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = "6f5a9c2d8e1b"
down_revision = "d2825e7b9c10"
branch_labels = None
depends_on = None
def upgrade():
with op.batch_alter_table("dify_setups", schema=None) as batch_op:
batch_op.add_column(sa.Column("instance_id", sa.String(length=255), nullable=True))
batch_op.add_column(sa.Column("install_reported_at", sa.DateTime(), nullable=True))
batch_op.add_column(sa.Column("last_heartbeat_at", sa.DateTime(), nullable=True))
def downgrade():
with op.batch_alter_table("dify_setups", schema=None) as batch_op:
batch_op.drop_column("last_heartbeat_at")
batch_op.drop_column("install_reported_at")
batch_op.drop_column("instance_id")
+5 -2
View File
@@ -362,6 +362,9 @@ class DifySetup(TypeBase):
__table_args__ = (sa.PrimaryKeyConstraint("version", name="dify_setup_pkey"),)
version: Mapped[str] = mapped_column(String(255), nullable=False)
instance_id: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
install_reported_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True, default=None)
last_heartbeat_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True, default=None)
setup_at: Mapped[datetime] = mapped_column(
sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
)
@@ -1114,14 +1117,14 @@ class ExporleBanner(TypeBase):
status: Mapped[BannerStatus] = mapped_column(
EnumText(BannerStatus, length=255),
nullable=False,
server_default=sa.text("'enabled'::character varying"),
server_default=sa.text("'enabled'"),
default=BannerStatus.ENABLED,
)
created_at: Mapped[datetime] = mapped_column(
sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
)
language: Mapped[str] = mapped_column(
String(255), nullable=False, server_default=sa.text("'en-US'::character varying"), default="en-US"
String(255), nullable=False, server_default=sa.text("'en-US'"), default="en-US"
)
+40 -1
View File
@@ -1672,6 +1672,23 @@ Create a new application
| 200 | Import confirmed | **application/json**: [Import](#import)<br> |
| 400 | Import failed | **application/json**: [Import](#import)<br> |
### [GET] /apps/recent
**Return the lightweight app cards needed by the Explore home page**
Get recently modified apps for the home Continue Work section
#### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| limit | query | Number of recently modified apps to return (1-8) | No | integer, <br>**Default:** 8 |
#### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [RecentAppListResponse](#recentapplistresponse)<br> |
### [GET] /apps/starred
Get applications starred by the current account
@@ -13243,6 +13260,7 @@ Model class for AI model.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| active_config_is_published | boolean | | Yes |
| active_config_snapshot | [AgentConfigSnapshotSummaryResponse](#agentconfigsnapshotsummaryresponse) | | No |
| agent | [AgentComposerAgentResponse](#agentcomposeragentresponse) | | Yes |
| agent_soul | [AgentSoulConfig](#agentsoulconfig) | | Yes |
@@ -13282,7 +13300,6 @@ Model class for AI model.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| access_mode | string | | No |
| active_config_is_published | boolean | | No |
| api_base_url | string | | No |
| app_id | string | | No |
| backing_app_id | string | | No |
@@ -21018,6 +21035,28 @@ Whitelist scopes accepted by RBAC app and dataset access config APIs.
| result | string | | Yes |
| updated_at | integer | | Yes |
#### RecentAppListResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| data | [ [RecentAppResponse](#recentappresponse) ] | | Yes |
#### RecentAppResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| author_name | string | | No |
| icon | string | | No |
| icon_background | string | | No |
| icon_type | [IconType](#icontype) | | No |
| icon_url | string | | Yes |
| id | string | | Yes |
| maintainer | string | | No |
| mode | string, <br>**Available values:** "advanced-chat", "agent-chat", "chat", "completion", "workflow" | *Enum:* `"advanced-chat"`, `"agent-chat"`, `"chat"`, `"completion"`, `"workflow"` | Yes |
| name | string | | Yes |
| permission_keys | [ string ] | | No |
| updated_at | integer | | Yes |
#### RecommendedAppDetailNullableResponse
| Name | Type | Description | Required |
+7
View File
@@ -965,6 +965,12 @@ Returns Server-Sent Events stream.
| ---- | ---- | ----------- | -------- |
| tool_icons | object | Tool icon metadata keyed by tool name | No |
#### AppMode
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| AppMode | string | | |
#### AppPermissionQuery
| Name | Type | Description | Required |
@@ -1646,6 +1652,7 @@ in form definition, or a variable while the workflow is running.
| custom_config | [WebAppCustomConfigResponse](#webappcustomconfigresponse) | | No |
| enable_site | boolean | | Yes |
| end_user_id | string | | No |
| mode | [AppMode](#appmode) | | Yes |
| model_config | [WebModelConfigResponse](#webmodelconfigresponse) | | No |
| plan | string | | Yes |
| site | [WebSiteResponse](#websiteresponse) | | Yes |
+3 -3
View File
@@ -6,7 +6,7 @@ requires-python = "~=3.12.0"
dependencies = [
# Legacy: mature and widely deployed
"bleach>=6.4.0,<7.0.0",
"boto3>=1.43.46,<2.0.0",
"boto3>=1.43.56,<2.0.0",
"celery>=5.6.3,<6.0.0",
"croniter>=6.2.2,<7.0.0",
"dify-agent",
@@ -193,10 +193,10 @@ dev = [
############################################################
storage = [
"azure-storage-blob>=12.30.0,<13.0.0",
"bce-python-sdk==0.9.72",
"bce-python-sdk==0.9.76",
"cos-python-sdk-v5>=1.9.44,<2.0.0",
"esdk-obs-python>=3.26.6,<4.0.0",
"google-cloud-storage>=3.12.1,<4.0.0",
"google-cloud-storage>=3.13.0,<4.0.0",
"opendal==0.46.0",
"oss2>=2.19.1,<3.0.0",
"supabase>=2.31.0,<3.0.0",
+7 -1
View File
@@ -75,6 +75,7 @@ from services.errors.account import (
from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
from services.feature_service import FeatureService
from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService
from services.telemetry_service import CommunityTelemetryService
from tasks.delete_account_task import delete_account_task
from tasks.mail_account_deletion_task import send_account_deletion_verification_code
from tasks.mail_change_mail_task import (
@@ -1953,7 +1954,7 @@ class RegisterService:
TenantService.create_owner_tenant_if_not_exist(account=account, is_setup=True, session=session)
dify_setup = DifySetup(version=dify_config.project.version)
dify_setup = DifySetup(version=dify_config.project.version, instance_id=str(uuid.uuid4()))
session.add(dify_setup)
session.commit()
except Exception as e:
@@ -1966,6 +1967,11 @@ class RegisterService:
logger.exception("Setup account failed, email: %s, name: %s", email, name)
raise ValueError(f"Setup failed: {e}")
try:
CommunityTelemetryService.report_install(session=session)
except Exception:
logger.debug("Failed to report install telemetry", exc_info=True)
@classmethod
def register(
cls,
+1
View File
@@ -405,6 +405,7 @@ class AgentComposerService:
"variant": ComposerVariant.AGENT_APP.value,
"agent": cls._serialize_agent(agent),
"active_config_snapshot": cls._serialize_version(version),
"active_config_is_published": bool(agent.active_config_snapshot_id and agent.active_config_is_published),
"draft": cls._serialize_draft(draft),
"agent_soul": draft.config_snapshot_dict,
"save_options": [ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION.value],
+84
View File
@@ -1,6 +1,7 @@
import json
import logging
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Literal, NotRequired, TypedDict, cast, override
@@ -41,6 +42,20 @@ from tasks.remove_app_and_related_data_task import remove_app_and_related_data_t
logger = logging.getLogger(__name__)
AppListSortBy = Literal["last_modified", "recently_created", "earliest_created"]
RecentAppMode = Literal[
AppMode.COMPLETION,
AppMode.WORKFLOW,
AppMode.CHAT,
AppMode.ADVANCED_CHAT,
AppMode.AGENT_CHAT,
]
RECENT_APP_MODES: tuple[RecentAppMode, ...] = (
AppMode.COMPLETION,
AppMode.WORKFLOW,
AppMode.CHAT,
AppMode.ADVANCED_CHAT,
AppMode.AGENT_CHAT,
)
class AppListBaseParams(BaseModel):
@@ -65,6 +80,19 @@ class StarredAppListParams(AppListBaseParams):
pass
@dataclass(frozen=True)
class RecentAppListItem:
id: str
name: str
icon_type: IconType | None
icon: str | None
icon_background: str | None
mode: RecentAppMode
author_name: str | None
updated_at: datetime
maintainer: str | None
class CreateAppParams(BaseModel):
name: str = Field(min_length=1)
description: str | None = None
@@ -323,6 +351,62 @@ class AppService:
return app_models
def get_recent_apps(
self,
user_id: str,
tenant_id: str,
params: AppListParams,
session: Session,
) -> list[RecentAppListItem]:
"""Return recently modified apps as one lightweight, non-paginated projection."""
filters = self._build_app_list_filters(user_id, tenant_id, params, session)
if not filters:
return []
stmt = (
sa.select(
App.id,
App.name,
App.icon_type,
App.icon,
App.icon_background,
App.mode,
Account.name.label("author_name"),
App.updated_at,
App.maintainer,
)
.outerjoin(Account, Account.id == App.created_by)
.where(*filters, App.mode.in_(RECENT_APP_MODES))
.order_by(App.updated_at.desc())
.limit(params.limit)
)
rows = session.execute(stmt).all()
return [
RecentAppListItem(
id=str(app_id),
name=name,
icon_type=icon_type,
icon=icon,
icon_background=icon_background,
mode=cast(RecentAppMode, mode),
author_name=author_name,
updated_at=updated_at,
maintainer=maintainer,
)
for (
app_id,
name,
icon_type,
icon,
icon_background,
mode,
author_name,
updated_at,
maintainer,
) in rows
]
def get_paginate_starred_apps(
self,
user_id: str,
+165
View File
@@ -0,0 +1,165 @@
import logging
import platform
import uuid
from datetime import datetime
from typing import Literal
import httpx
from sqlalchemy import select
from sqlalchemy.orm import Session
from configs import dify_config
from libs.datetime_utils import naive_utc_now
from models.model import DifySetup
logger = logging.getLogger(__name__)
TelemetryEvent = Literal["install", "heartbeat"]
SCHEMA_VERSION = 1
class CommunityTelemetryService:
@classmethod
def report_install(cls, *, session: Session) -> bool:
setup = cls._get_setup(session)
if setup is None:
return False
if setup.instance_id is None:
setup.instance_id = str(uuid.uuid4())
session.add(setup)
session.commit()
payload = cls._build_payload(setup, "install")
if not cls._send_event(payload):
return False
setup.install_reported_at = naive_utc_now()
session.add(setup)
session.commit()
return True
@classmethod
def report_heartbeat(cls, *, session: Session, now: datetime | None = None) -> bool:
setup = cls._get_setup(session)
if setup is None:
return False
if setup.instance_id is None:
setup.instance_id = str(uuid.uuid4())
session.add(setup)
session.commit()
now = now or naive_utc_now()
if not cls._is_heartbeat_due(setup, now):
return False
if setup.install_reported_at is None:
cls.report_install(session=session)
payload = cls._build_payload(setup, "heartbeat")
if not cls._send_event(payload):
return False
setup.last_heartbeat_at = now
session.add(setup)
session.commit()
return True
@classmethod
def _get_setup(cls, session: Session) -> DifySetup | None:
return session.scalar(select(DifySetup).order_by(DifySetup.setup_at.asc()).limit(1))
@classmethod
def _is_enabled(cls) -> bool:
return (
dify_config.EDITION == "SELF_HOSTED"
and not dify_config.ENTERPRISE_ENABLED
and not dify_config.DISABLE_TELEMETRY
and not dify_config.DO_NOT_TRACK
and not dify_config.CI
and bool(dify_config.TELEMETRY_ENDPOINT)
)
@classmethod
def _build_payload(cls, setup: DifySetup, event: TelemetryEvent) -> dict[str, str | int]:
payload: dict[str, str | int] = {
"event": event,
"instance_id": setup.instance_id or "",
"version": setup.version if event == "install" else dify_config.project.version,
"edition": dify_config.EDITION,
"deployment_type": "unknown",
"schema_version": SCHEMA_VERSION,
"os": cls._normalize_os(platform.system()),
"arch": cls._normalize_arch(platform.machine()),
"sent_at": cls._format_datetime(naive_utc_now()),
}
if event == "install":
payload["installed_at"] = cls._format_datetime(setup.setup_at)
return payload
@classmethod
def _send_event(cls, payload: dict[str, str | int]) -> bool:
if not cls._is_enabled():
return False
endpoints = [dify_config.TELEMETRY_ENDPOINT]
if dify_config.TELEMETRY_FALLBACK_ENDPOINT not in endpoints:
endpoints.append(dify_config.TELEMETRY_FALLBACK_ENDPOINT)
for endpoint in endpoints:
if not endpoint:
continue
try:
response = httpx.post(
endpoint,
json=payload,
timeout=dify_config.TELEMETRY_TIMEOUT_SECONDS,
)
response.raise_for_status()
return True
except httpx.RequestError:
logger.debug("Failed to send community telemetry event to %s", endpoint, exc_info=True)
except httpx.HTTPStatusError:
logger.debug("Community telemetry endpoint returned an error: %s", endpoint, exc_info=True)
return False
return False
@classmethod
def _is_heartbeat_due(cls, setup: DifySetup, now: datetime) -> bool:
if setup.instance_id is None:
return False
if setup.last_heartbeat_at is not None and setup.last_heartbeat_at.date() >= now.date():
return False
return True
@staticmethod
def _format_datetime(value: datetime) -> str:
return value.replace(microsecond=0).isoformat() + "Z"
@staticmethod
def _normalize_os(value: str) -> str:
os_name = value.lower()
if os_name in {"linux", "darwin", "windows"}:
return os_name
return "unknown"
@staticmethod
def _normalize_arch(value: str) -> str:
arch = value.lower()
if arch in {"x86_64", "amd64"}:
return "amd64"
if arch in {"aarch64", "arm64"}:
return "arm64"
if arch.startswith("arm"):
return "arm"
if arch in {"i386", "i686", "x86"}:
return "386"
return "unknown"
+9 -9
View File
@@ -278,14 +278,14 @@ class VariableTruncator(BaseTruncator):
target_length = self._array_element_limit
for i, item in enumerate(value):
# Dirty fix:
# The output of `Start` node may contain list of `File` elements,
# causing `AssertionError` while invoking `_truncate_json_primitives`.
#
# This check ensures that `list[File]` are handled separately
if isinstance(item, File):
truncated_value.append(item)
continue
# ``File`` is routed through ``_truncate_json_primitives`` (whose
# dedicated ``File`` branch returns the file as-is with its real
# serialized size). That preserves the count cap
# (``array_element_limit``) and the byte budget (``target_size``)
# for ``list[File]`` — the original "Dirty fix" branch above this
# loop bypassed both guarantees and reported ``used_size=2`` even
# when the returned array serialized to well over the budget.
# See https://github.com/langgenius/dify/issues/39218.
if i >= target_length:
return _PartResult(truncated_value, used_size, True)
if i > 0:
@@ -295,7 +295,7 @@ class VariableTruncator(BaseTruncator):
break
remaining_budget = target_size - used_size
if item is None or isinstance(item, (str, list, dict, bool, int, float, UpdatedVariable)):
if item is None or isinstance(item, (str, list, dict, bool, int, float, File, UpdatedVariable)):
part_result = self._truncate_json_primitives(item, remaining_budget)
else:
raise UnknownTypeError(f"got unknown type {type(item)} in array truncation")
+1 -1
View File
@@ -116,7 +116,7 @@ class WebAppAuthService:
@classmethod
def _get_account_jwt_token(cls, account: Account) -> str:
exp_dt = datetime.now(UTC) + timedelta(minutes=dify_config.ACCESS_TOKEN_EXPIRE_MINUTES * 24)
exp_dt = datetime.now(UTC) + timedelta(minutes=dify_config.ACCESS_TOKEN_EXPIRE_MINUTES)
exp = int(exp_dt.timestamp())
payload = {
+1 -1
View File
@@ -27,7 +27,6 @@ class WorkspaceService:
tenant_info: dict[str, object] = {
"id": tenant.id,
"name": tenant.name,
"plan": tenant.plan,
"status": tenant.status,
"created_at": tenant.created_at,
"trial_end_reason": None,
@@ -44,6 +43,7 @@ class WorkspaceService:
tenant_info["role"] = tenant_account_join.role
feature = FeatureService.get_features(tenant.id, exclude_vector_space=True)
tenant_info["plan"] = feature.billing.subscription.plan if feature.billing.enabled else None
can_replace_logo = feature.can_replace_logo
if can_replace_logo and TenantService.has_roles(
@@ -22,6 +22,7 @@ def _create_agent_backend_client():
return None
return create_agent_backend_run_client(
base_url=dify_config.AGENT_BACKEND_BASE_URL,
api_token=dify_config.AGENT_BACKEND_API_TOKEN,
use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
@@ -457,7 +457,7 @@ def _publish_streaming_response(
@shared_task(queue=WORKFLOW_BASED_APP_EXECUTION_QUEUE)
def workflow_based_app_execution_task(
payload: str,
) -> Generator[Mapping[str, Any] | str, None, None] | Mapping[str, Any] | None:
) -> Mapping[str, Any] | None:
exec_params = AppExecutionParams.model_validate_json(payload)
logger.info("workflow_based_app_execution_task run with params: %s", exec_params)
+19
View File
@@ -0,0 +1,19 @@
import logging
from celery import shared_task
from sqlalchemy.orm import sessionmaker
from extensions.ext_database import db
from services.telemetry_service import CommunityTelemetryService
logger = logging.getLogger(__name__)
@shared_task(name="community_telemetry.send_heartbeat", queue="schedule_executor")
def send_community_telemetry_heartbeat() -> None:
session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
with session_factory() as session:
try:
CommunityTelemetryService.report_heartbeat(session=session)
except Exception:
logger.debug("Failed to process community telemetry heartbeat", exc_info=True)
@@ -17,10 +17,11 @@ class _SimpleJinja2Renderer:
"""Minimal Jinja2-based renderer for integration tests (no code executor)."""
def render_template(self, template: str, variables: dict[str, object]) -> str:
from jinja2 import Template
from jinja2.sandbox import SandboxedEnvironment
try:
return Template(template).render(**variables)
env = SandboxedEnvironment()
return env.from_string(template).render(**variables)
except Exception as exc:
raise TemplateRenderError(str(exc)) from exc
@@ -97,6 +97,7 @@ class TestAppSiteApi:
assert result["end_user_id"] == end_user.id
assert result["plan"] == "basic"
assert result["enable_site"] is True
assert result["mode"] == AppMode.CHAT
@patch("controllers.web.site.FileService.get_file_presigned_url")
@patch("controllers.web.site.FeatureService.get_features")
@@ -178,6 +179,7 @@ class TestWebAppSiteResponse:
response = WebAppSiteResponse.from_app_site(
tenant=tenant,
app_model=app_model,
mode=AppMode.CHAT,
site=_site_model(app_id=app_model.id),
end_user_id="eu-1",
features=FeatureModel(can_replace_logo=False, webapp_copyright_enabled=True),
@@ -185,6 +187,7 @@ class TestWebAppSiteResponse:
)
assert response.app_id == app_model.id
assert response.mode == AppMode.CHAT
assert response.end_user_id == "eu-1"
assert response.enable_site is True
assert response.plan == "basic"
@@ -209,6 +212,7 @@ class TestWebAppSiteResponse:
response = WebAppSiteResponse.from_app_site(
tenant=tenant,
app_model=app_model,
mode=AppMode.CHAT,
site=site,
end_user_id=None,
features=FeatureModel(can_replace_logo=False, webapp_copyright_enabled=True),
@@ -236,6 +240,7 @@ class TestWebAppSiteResponse:
response = WebAppSiteResponse.from_app_site(
tenant=tenant,
app_model=app_model,
mode=AppMode.CHAT,
site=_site_model(app_id=app_model.id),
end_user_id="eu-1",
features=FeatureModel(can_replace_logo=True, webapp_copyright_enabled=True),
@@ -24,7 +24,10 @@ class TestWorkspaceService:
patch("services.workspace_service.dify_config") as mock_dify_config,
):
# Setup default mock returns
mock_feature_service.get_features.return_value.can_replace_logo = True
feature = mock_feature_service.get_features.return_value
feature.can_replace_logo = True
feature.billing.enabled = True
feature.billing.subscription.plan = "professional"
mock_tenant_service.has_roles.return_value = True
mock_dify_config.FILES_URL = "https://example.com/files"
@@ -112,7 +115,7 @@ class TestWorkspaceService:
assert result is not None
assert result["id"] == tenant.id
assert result["name"] == tenant.name
assert result["plan"] == tenant.plan
assert result["plan"] == "professional"
assert result["status"] == tenant.status
assert result["role"] == TenantAccountRole.OWNER
assert result["created_at"] == tenant.created_at
@@ -159,7 +162,7 @@ class TestWorkspaceService:
assert result is not None
assert result["id"] == tenant.id
assert result["name"] == tenant.name
assert result["plan"] == tenant.plan
assert result["plan"] == "professional"
assert result["status"] == tenant.status
assert result["role"] == TenantAccountRole.OWNER
assert result["created_at"] == tenant.created_at
@@ -214,7 +217,7 @@ class TestWorkspaceService:
assert result is not None
assert result["id"] == tenant.id
assert result["name"] == tenant.name
assert result["plan"] == tenant.plan
assert result["plan"] == "professional"
assert result["status"] == tenant.status
assert result["role"] == TenantAccountRole.NORMAL
assert result["created_at"] == tenant.created_at
@@ -606,20 +609,23 @@ class TestWorkspaceService:
def test_get_tenant_info_should_not_include_cloud_fields_in_self_hosted(
self, db_session_with_containers: Session, mock_external_service_dependencies
):
"""next_credit_reset_date and trial_credits should NOT appear in SELF_HOSTED mode."""
"""Cloud-only billing data should not appear in SELF_HOSTED mode."""
fake = Faker()
account, tenant = self._create_test_account_and_tenant(
db_session_with_containers, mock_external_service_dependencies
)
mock_external_service_dependencies["dify_config"].DEPLOYMENT_EDITION = DeploymentEdition.COMMUNITY
mock_external_service_dependencies["feature_service"].get_features.return_value.can_replace_logo = False
feature = mock_external_service_dependencies["feature_service"].get_features.return_value
feature.can_replace_logo = False
feature.billing.enabled = False
mock_external_service_dependencies["tenant_service"].has_roles.return_value = False
with patch("services.workspace_service.current_user", account):
result = WorkspaceService.get_tenant_info(tenant, db_session_with_containers)
assert result is not None
assert result["plan"] is None
assert "next_credit_reset_date" not in result
assert "trial_credits" not in result
assert "trial_credits_used" not in result
@@ -4,6 +4,7 @@ from dotenv import dotenv_values
BASE_API_AND_DOCKER_CONFIG_SET_DIFF: frozenset[str] = frozenset(
(
"AGENT_BACKEND_API_TOKEN",
"APP_MAX_EXECUTION_TIME",
"BATCH_UPLOAD_LIMIT",
"CELERY_BEAT_SCHEDULER_TIME",
@@ -43,6 +44,7 @@ BASE_API_AND_DOCKER_CONFIG_SET_DIFF: frozenset[str] = frozenset(
BASE_API_AND_DOCKER_COMPOSE_CONFIG_SET_DIFF: frozenset[str] = frozenset(
(
"AGENT_BACKEND_API_TOKEN",
"BATCH_UPLOAD_LIMIT",
"CELERY_BEAT_SCHEDULER_TIME",
"HTTP_REQUEST_MAX_CONNECT_TIMEOUT",
+56 -26
View File
@@ -1,11 +1,13 @@
import os
import shutil
from collections.abc import Iterator
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from flask import Flask
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.engine import URL, Engine
from sqlalchemy.orm import Session, sessionmaker
# Getting the absolute path of the current file's directory
@@ -35,7 +37,7 @@ os.environ.setdefault("OPENDAL_SCHEME", "fs")
os.environ.setdefault("OPENDAL_FS_ROOT", "/tmp/dify-storage")
os.environ.setdefault("STORAGE_TYPE", "opendal")
from core.db.session_factory import configure_session_factory, session_factory
import core.db.session_factory as session_factory_module
from extensions import ext_redis
from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
from models.base import TypeBase
@@ -111,42 +113,70 @@ def reset_secret_key():
dify_config.SECRET_KEY = original
@pytest.fixture(scope="session")
def _unit_test_engine():
engine = create_engine("sqlite:///:memory:")
yield engine
engine.dispose()
@pytest.fixture
def sqlite_engine() -> Iterator[Engine]:
"""Create an isolated in-memory SQLite engine for tests that need a disposable database."""
def _sqlite_engine(_sqlite_database_template: Path, tmp_path: Path) -> Iterator[Engine]:
"""Create an engine over a pristine per-test copy of the SQLite schema."""
database_path = tmp_path / "unit-tests.sqlite3"
shutil.copyfile(_sqlite_database_template, database_path)
engine = create_engine(URL.create("sqlite", database=str(database_path)))
engine = create_engine("sqlite:///:memory:")
try:
yield engine
finally:
engine.dispose()
database_path.unlink(missing_ok=True)
@pytest.fixture
def sqlite_session(request: pytest.FixtureRequest, sqlite_engine: Engine) -> Iterator[Session]:
"""Yield a SQLite session after creating the model tables passed through ``request.param``."""
@pytest.fixture(scope="session")
def _sqlite_database_template(tmp_path_factory: pytest.TempPathFactory) -> Path:
"""Create one empty full-schema SQLite database per pytest worker."""
models: tuple[type[TypeBase], ...] = request.param
tables = [model.metadata.tables[model.__tablename__] for model in models]
TypeBase.metadata.create_all(sqlite_engine, tables=tables)
session_factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
with session_factory() as session:
yield session
database_path = tmp_path_factory.mktemp("sqlite-template") / "unit-tests.sqlite3"
engine = create_engine(URL.create("sqlite", database=str(database_path)))
try:
TypeBase.metadata.create_all(engine)
finally:
engine.dispose()
return database_path
@pytest.fixture(autouse=True)
def _configure_session_factory(_unit_test_engine):
try:
session_factory.get_session_maker()
except RuntimeError:
configure_session_factory(_unit_test_engine, expire_on_commit=False)
def _sqlite_session_factory(
_sqlite_engine: Engine,
monkeypatch: pytest.MonkeyPatch,
) -> sessionmaker[Session]:
"""Bind all unit-test Sessions to the pristine full-schema SQLite database."""
factory = sessionmaker(bind=_sqlite_engine, expire_on_commit=False)
monkeypatch.setattr(session_factory_module, "_session_maker", factory)
return factory
@pytest.fixture
def sqlite_engine(_sqlite_engine: Engine) -> Engine:
"""Expose the pristine full-schema SQLite engine to tests."""
return _sqlite_engine
@pytest.fixture
def sqlite_session_factory(_sqlite_session_factory: sessionmaker[Session]) -> sessionmaker[Session]:
"""Expose the shared SQLite session factory to tests."""
return _sqlite_session_factory
@pytest.fixture
def sqlite_session(_sqlite_session_factory: sessionmaker[Session]) -> Iterator[Session]:
"""Yield a session over the pristine full-schema SQLite database.
Legacy indirect model parameters remain accepted by pytest but are ignored.
Remove those decorators as their test files receive individual review.
"""
with _sqlite_session_factory() as session:
yield session
def persist_service_api_tenant_owner(session: Session, tenant: Tenant, owner: Account) -> TenantAccountJoin:
@@ -115,6 +115,7 @@ def _agent_app_composer_response() -> dict:
"active_config_snapshot_id": "version-1",
},
"active_config_snapshot": _version_response(),
"active_config_is_published": True,
"agent_soul": {},
"save_options": ["save_to_current_version"],
}
@@ -376,7 +377,7 @@ def test_agent_app_list_and_create_use_agent_route(
assert created["app_id"] == "app-created"
assert created["debug_conversation_id"] == "debug-conversation-created"
assert created["role"] == "Created role"
assert created["active_config_is_published"] is False
assert "active_config_is_published" not in created
assert "bound_agent_id" not in created
create_call = cast(dict[str, object], captured["create"])
create_params = cast(Any, create_call["params"])
@@ -487,7 +488,7 @@ def test_agent_app_detail_update_delete_resolve_app_from_agent_id(
assert detail["debug_conversation_has_messages"] is True
assert detail["debug_conversation_message_count"] == 2
assert detail["role"] == "Resolved role"
assert detail["active_config_is_published"] is False
assert "active_config_is_published" not in detail
assert "bound_agent_id" not in detail
assert captured["get_app"] == {"app": app_model, "session": session}
with app.test_request_context(
@@ -502,7 +503,7 @@ def test_agent_app_detail_update_delete_resolve_app_from_agent_id(
assert updated["debug_conversation_has_messages"] is True
assert updated["debug_conversation_message_count"] == 2
assert updated["role"] == "Resolved role"
assert updated["active_config_is_published"] is False
assert "active_config_is_published" not in updated
assert "bound_agent_id" not in updated
update_call = cast(dict[str, object], captured["update"])
assert update_call["app"] is app_model
@@ -845,9 +846,6 @@ def test_agent_app_update_allows_empty_role(app: Flask, monkeypatch: pytest.Monk
monkeypatch.setattr(
roster_controller.AgentRosterService, "count_agent_app_debug_conversation_messages", lambda _self, **kwargs: 0
)
monkeypatch.setattr(
roster_controller.AgentRosterService, "active_config_is_published", lambda _self, **kwargs: False
)
monkeypatch.setattr(
roster_controller.FeatureService,
"get_system_features",
@@ -1299,13 +1297,14 @@ def test_agent_composer_routes_resolve_app_from_agent_id(
composer_controller.AgentComposerService, "collect_validation_findings", collect_validation_findings
)
monkeypatch.setattr(composer_controller.AgentComposerService, "get_agent_app_candidates", get_agent_app_candidates)
assert unwrap(AgentComposerApi.get)(AgentComposerApi(), MagicMock(), "tenant-1", agent_id)["variant"] == "agent_app"
composer = unwrap(AgentComposerApi.get)(AgentComposerApi(), MagicMock(), "tenant-1", agent_id)
assert composer["variant"] == "agent_app"
assert composer["active_config_is_published"] is True
assert cast(dict[str, object], captured["load"])["agent_id"] == agent_id
with app.test_request_context(json=payload):
assert (
unwrap(AgentComposerApi.put)(AgentComposerApi(), MagicMock(), "tenant-1", account_id, agent_id)["variant"]
== "agent_app"
)
saved_composer = unwrap(AgentComposerApi.put)(AgentComposerApi(), MagicMock(), "tenant-1", account_id, agent_id)
assert saved_composer["variant"] == "agent_app"
assert saved_composer["active_config_is_published"] is True
assert cast(dict[str, object], captured["save"])["agent_id"] == agent_id
assert unwrap(AgentComposerValidateApi.post)(AgentComposerValidateApi(), MagicMock(), "tenant-1", agent_id) == {
"result": "success",
@@ -708,6 +708,121 @@ def test_app_list_api_attaches_permission_keys(app, app_module):
assert resp["data"][0]["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"]
def test_recent_app_list_api_returns_only_home_card_fields(app, app_module):
method = app_module.RecentAppListApi.get
while hasattr(method, "__wrapped__"):
method = method.__wrapped__
recent_app = SimpleNamespace(
id="app-1",
name="Recent App",
icon_type="emoji",
icon="🚀",
icon_background="#FFFFFF",
mode="chat",
author_name="Recent Author",
updated_at=_ts(15),
maintainer="acct-1",
)
get_recent_apps = MagicMock(return_value=[recent_app])
with app.test_request_context("/apps/recent?limit=8"):
with pytest.MonkeyPatch.context() as monkeypatch:
monkeypatch.setattr(dify_config, "RBAC_ENABLED", False)
monkeypatch.setattr(app_module.AppService, "get_recent_apps", get_recent_apps)
monkeypatch.setattr(
app_module.enterprise_rbac_service.RBACService.MyPermissions,
"get",
lambda tenant_id, account_id, session: app_module.enterprise_rbac_service.MyPermissionsResponse(
app=app_module.enterprise_rbac_service.ResourcePermissionSnapshot(
overrides=[
app_module.enterprise_rbac_service.ResourcePermissionKeys(
resource_id="app-1",
permission_keys=["app.acl.monitor"],
)
]
)
),
)
resp, status = method(app_module.RecentAppListApi(), "tenant-1", "acct-1", MagicMock())
assert status == 200
assert resp == {
"data": [
{
"id": "app-1",
"name": "Recent App",
"icon_type": "emoji",
"icon": "🚀",
"icon_background": "#FFFFFF",
"mode": "chat",
"author_name": "Recent Author",
"updated_at": int(_ts(15).timestamp()),
"permission_keys": ["app.acl.monitor"],
"maintainer": "acct-1",
"icon_url": None,
}
]
}
params = get_recent_apps.call_args.args[2]
assert params.limit == 8
assert "total" not in resp
assert "description" not in resp["data"][0]
assert "tags" not in resp["data"][0]
assert "workflow" not in resp["data"][0]
@pytest.mark.parametrize("mode", ["channel", "rag-pipeline", "agent"])
def test_recent_app_response_rejects_non_home_app_modes(app_module, mode: str) -> None:
with pytest.raises(ValidationError):
app_module.RecentAppResponse.model_validate(
{
"id": "app-1",
"name": "Recent App",
"mode": mode,
"updated_at": _ts(),
}
)
def test_recent_app_list_api_applies_rbac_visibility_filter(app, app_module):
method = app_module.RecentAppListApi.get
while hasattr(method, "__wrapped__"):
method = method.__wrapped__
get_recent_apps = MagicMock(return_value=[])
with app.test_request_context("/apps/recent"):
with pytest.MonkeyPatch.context() as monkeypatch:
monkeypatch.setattr(dify_config, "RBAC_ENABLED", True)
monkeypatch.setattr(app_module.AppService, "get_recent_apps", get_recent_apps)
monkeypatch.setattr(
app_module.enterprise_rbac_service.RBACService.MyPermissions,
"get",
lambda tenant_id, account_id, session: app_module.enterprise_rbac_service.MyPermissionsResponse(
workspace=app_module.enterprise_rbac_service.WorkspacePermissionSnapshot(
permission_keys=["app.create_and_management"]
)
),
)
monkeypatch.setattr(
app_module.enterprise_rbac_service.RBACService.AppAccess,
"whitelist_resources",
lambda tenant_id, account_id: SimpleNamespace(
unrestricted=False,
resource_ids=["app-shared"],
),
)
resp, status = method(app_module.RecentAppListApi(), "tenant-1", "acct-1", MagicMock())
assert status == 200
assert resp == {"data": []}
params = get_recent_apps.call_args.args[2]
assert params.accessible_app_ids == ["app-shared"]
assert params.include_own_apps is True
def test_app_list_api_limits_to_apps_created_by_current_user_without_view_permission(app, app_module):
method = app_module.AppListApi.get
while hasattr(method, "__wrapped__"):
@@ -1,17 +1,31 @@
"""
Unit tests for Service API File Preview endpoint
"""Unit tests for the Service API file-preview endpoint.
Ownership checks run against persisted message, file, app, and upload rows so the
tests exercise the same SQLAlchemy statements and tenant boundary as production.
Storage remains mocked because it is the external I/O boundary of the endpoint.
"""
import logging
import uuid
from collections.abc import Iterator
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from typing import Protocol, cast
from unittest.mock import Mock, patch
from uuid import uuid4
import pytest
from sqlalchemy import event
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session
from controllers.service_api.app.error import FileAccessDeniedError, FileNotFoundError
from controllers.service_api.app.file_preview import FilePreviewApi
from models.model import App, EndUser, Message, MessageFile, UploadFile
from extensions.storage.storage_type import StorageType
from graphon.file import FileTransferMethod, FileType
from models.base import TypeBase
from models.enums import ConversationFromSource, CreatorUserRole
from models.model import App, AppMode, Message, MessageFile, UploadFile
class _FilePreviewLogRecord(Protocol):
@@ -20,367 +34,252 @@ class _FilePreviewLogRecord(Protocol):
error: str
@dataclass(frozen=True)
class _Database:
"""Expose the real test session through the interface used by the controller."""
session: Session
@dataclass(frozen=True)
class _PreviewRecords:
app: App
message: Message
message_file: MessageFile
upload_file: UploadFile
@pytest.fixture
def database(sqlite_engine: Engine) -> Iterator[_Database]:
"""Create only the tables required by file ownership validation."""
models = (App, Message, MessageFile, UploadFile)
tables = [TypeBase.metadata.tables[model.__tablename__] for model in models]
TypeBase.metadata.create_all(sqlite_engine, tables=tables)
with Session(sqlite_engine, expire_on_commit=False) as session:
yield _Database(session)
@pytest.fixture
def file_preview_api() -> FilePreviewApi:
"""Create the resource instance under test."""
return FilePreviewApi()
def _upload_file(*, tenant_id: str, file_id: str | None = None) -> UploadFile:
upload_file = UploadFile(
tenant_id=tenant_id,
storage_type=StorageType.LOCAL,
key="storage/key/test_file.jpg",
name="test_file.jpg",
size=1024,
extension="jpg",
mime_type="image/jpeg",
created_by_role=CreatorUserRole.ACCOUNT,
created_by=str(uuid4()),
created_at=datetime(2026, 1, 1),
used=True,
)
if file_id is not None:
upload_file.id = file_id
return upload_file
def _persist_preview_records(
session: Session,
*,
app_id: str | None = None,
app_tenant_id: str | None = None,
upload_tenant_id: str | None = None,
) -> _PreviewRecords:
app_id = app_id or str(uuid4())
app_tenant_id = app_tenant_id or str(uuid4())
upload_file = _upload_file(tenant_id=upload_tenant_id or app_tenant_id)
app = App(
id=app_id,
tenant_id=app_tenant_id,
name="Preview app",
description="",
mode=AppMode.CHAT,
icon_type=None,
icon="",
icon_background=None,
enable_site=True,
enable_api=True,
)
message = Message(
id=str(uuid4()),
app_id=app_id,
conversation_id=str(uuid4()),
_inputs={},
query="preview",
message={},
message_unit_price=Decimal(0),
answer="answer",
answer_unit_price=Decimal(0),
currency="USD",
from_source=ConversationFromSource.API,
)
message_file = MessageFile(
message_id=message.id,
type=FileType.IMAGE,
transfer_method=FileTransferMethod.LOCAL_FILE,
created_by_role=CreatorUserRole.ACCOUNT,
created_by=str(uuid4()),
upload_file_id=upload_file.id,
)
session.add_all([app, message, message_file, upload_file])
session.commit()
return _PreviewRecords(app=app, message=message, message_file=message_file, upload_file=upload_file)
class TestFilePreviewApi:
"""Test suite for FilePreviewApi"""
"""Exercise ownership validation and response construction."""
@pytest.fixture
def file_preview_api(self):
"""Create FilePreviewApi instance for testing"""
return FilePreviewApi()
def test_validate_file_ownership_success(self, file_preview_api: FilePreviewApi, database: _Database):
records = _persist_preview_records(database.session)
@pytest.fixture
def mock_app(self):
"""Mock App model"""
app = Mock(spec=App)
app.id = str(uuid.uuid4())
app.tenant_id = str(uuid.uuid4())
return app
with patch("controllers.service_api.app.file_preview.db", database):
message_file, upload_file = file_preview_api._validate_file_ownership(
records.upload_file.id, records.app.id
)
@pytest.fixture
def mock_end_user(self):
"""Mock EndUser model"""
end_user = Mock(spec=EndUser)
end_user.id = str(uuid.uuid4())
return end_user
assert message_file.id == records.message_file.id
assert upload_file.id == records.upload_file.id
assert upload_file.tenant_id == records.app.tenant_id
@pytest.fixture
def mock_upload_file(self):
"""Mock UploadFile model"""
upload_file = Mock(spec=UploadFile)
upload_file.id = str(uuid.uuid4())
upload_file.name = "test_file.jpg"
upload_file.extension = "jpg"
upload_file.mime_type = "image/jpeg"
upload_file.size = 1024
upload_file.key = "storage/key/test_file.jpg"
upload_file.tenant_id = str(uuid.uuid4())
return upload_file
def test_validate_file_ownership_file_not_found(self, file_preview_api: FilePreviewApi, database: _Database):
with patch("controllers.service_api.app.file_preview.db", database):
with pytest.raises(FileNotFoundError, match="File not found in message context"):
file_preview_api._validate_file_ownership(str(uuid4()), str(uuid4()))
@pytest.fixture
def mock_message_file(self):
"""Mock MessageFile model"""
message_file = Mock(spec=MessageFile)
message_file.id = str(uuid.uuid4())
message_file.upload_file_id = str(uuid.uuid4())
message_file.message_id = str(uuid.uuid4())
return message_file
def test_validate_file_ownership_access_denied(self, file_preview_api: FilePreviewApi, database: _Database):
records = _persist_preview_records(database.session)
@pytest.fixture
def mock_message(self):
"""Mock Message model"""
message = Mock(spec=Message)
message.id = str(uuid.uuid4())
message.app_id = str(uuid.uuid4())
return message
with patch("controllers.service_api.app.file_preview.db", database):
with pytest.raises(FileAccessDeniedError, match="not owned by requesting app"):
file_preview_api._validate_file_ownership(records.upload_file.id, str(uuid4()))
def test_validate_file_ownership_success(
self, file_preview_api: FilePreviewApi, mock_app, mock_upload_file, mock_message_file, mock_message
):
"""Test successful file ownership validation"""
file_id = str(uuid.uuid4())
app_id = mock_app.id
def test_validate_file_ownership_upload_file_not_found(self, file_preview_api: FilePreviewApi, database: _Database):
records = _persist_preview_records(database.session)
database.session.delete(records.upload_file)
database.session.commit()
# Set up the mocks
mock_upload_file.tenant_id = mock_app.tenant_id
mock_message.app_id = app_id
mock_message_file.upload_file_id = file_id
mock_message_file.message_id = mock_message.id
with patch("controllers.service_api.app.file_preview.db", database):
with pytest.raises(FileNotFoundError, match="Upload file record not found"):
file_preview_api._validate_file_ownership(records.upload_file.id, records.app.id)
with patch("controllers.service_api.app.file_preview.db") as mock_db:
# Mock scalar() for MessageFile and Message queries
mock_db.session.scalar.side_effect = [
mock_message_file, # MessageFile query
mock_message, # Message query
]
# Mock get() for UploadFile and App PK lookups
mock_db.session.get.side_effect = [
mock_upload_file, # UploadFile query
mock_app, # App query for tenant validation
]
def test_validate_file_ownership_tenant_mismatch(self, file_preview_api: FilePreviewApi, database: _Database):
records = _persist_preview_records(database.session, upload_tenant_id=str(uuid4()))
# Execute the method
result_message_file, result_upload_file = file_preview_api._validate_file_ownership(file_id, app_id)
# Assertions
assert result_message_file == mock_message_file
assert result_upload_file == mock_upload_file
def test_validate_file_ownership_file_not_found(self, file_preview_api: FilePreviewApi):
"""Test file ownership validation when MessageFile not found"""
file_id = str(uuid.uuid4())
app_id = str(uuid.uuid4())
with patch("controllers.service_api.app.file_preview.db") as mock_db:
# Mock MessageFile not found via scalar()
mock_db.session.scalar.return_value = None
# Execute and assert exception
with pytest.raises(FileNotFoundError) as exc_info:
file_preview_api._validate_file_ownership(file_id, app_id)
assert "File not found in message context" in str(exc_info.value)
def test_validate_file_ownership_access_denied(self, file_preview_api: FilePreviewApi, mock_message_file):
"""Test file ownership validation when Message not owned by app"""
file_id = str(uuid.uuid4())
app_id = str(uuid.uuid4())
with patch("controllers.service_api.app.file_preview.db") as mock_db:
# Mock MessageFile found but Message not owned by app via scalar()
mock_db.session.scalar.side_effect = [
mock_message_file, # MessageFile query - found
None, # Message query - not found (access denied)
]
# Execute and assert exception
with pytest.raises(FileAccessDeniedError) as exc_info:
file_preview_api._validate_file_ownership(file_id, app_id)
assert "not owned by requesting app" in str(exc_info.value)
def test_validate_file_ownership_upload_file_not_found(
self, file_preview_api: FilePreviewApi, mock_message_file, mock_message
):
"""Test file ownership validation when UploadFile not found"""
file_id = str(uuid.uuid4())
app_id = str(uuid.uuid4())
with patch("controllers.service_api.app.file_preview.db") as mock_db:
# Mock scalar() for MessageFile and Message
mock_db.session.scalar.side_effect = [
mock_message_file, # MessageFile query - found
mock_message, # Message query - found
]
# Mock get() for UploadFile - not found
mock_db.session.get.return_value = None
# Execute and assert exception
with pytest.raises(FileNotFoundError) as exc_info:
file_preview_api._validate_file_ownership(file_id, app_id)
assert "Upload file record not found" in str(exc_info.value)
def test_validate_file_ownership_tenant_mismatch(
self, file_preview_api: FilePreviewApi, mock_app, mock_upload_file, mock_message_file, mock_message
):
"""Test file ownership validation with tenant mismatch"""
file_id = str(uuid.uuid4())
app_id = mock_app.id
# Set up tenant mismatch
mock_upload_file.tenant_id = "different_tenant_id"
mock_app.tenant_id = "app_tenant_id"
mock_message.app_id = app_id
mock_message_file.upload_file_id = file_id
mock_message_file.message_id = mock_message.id
with patch("controllers.service_api.app.file_preview.db") as mock_db:
# Mock scalar() for MessageFile and Message queries
mock_db.session.scalar.side_effect = [
mock_message_file, # MessageFile query
mock_message, # Message query
]
# Mock get() for UploadFile and App PK lookups
mock_db.session.get.side_effect = [
mock_upload_file, # UploadFile query
mock_app, # App query for tenant validation
]
# Execute and assert exception
with pytest.raises(FileAccessDeniedError) as exc_info:
file_preview_api._validate_file_ownership(file_id, app_id)
assert "tenant mismatch" in str(exc_info.value)
with patch("controllers.service_api.app.file_preview.db", database):
with pytest.raises(FileAccessDeniedError, match="tenant mismatch"):
file_preview_api._validate_file_ownership(records.upload_file.id, records.app.id)
def test_validate_file_ownership_invalid_input(self, file_preview_api: FilePreviewApi):
"""Test file ownership validation with invalid input"""
# Test with empty file_id
with pytest.raises(FileAccessDeniedError) as exc_info:
with pytest.raises(FileAccessDeniedError, match="Invalid file or app identifier"):
file_preview_api._validate_file_ownership("", "app_id")
assert "Invalid file or app identifier" in str(exc_info.value)
# Test with empty app_id
with pytest.raises(FileAccessDeniedError) as exc_info:
with pytest.raises(FileAccessDeniedError, match="Invalid file or app identifier"):
file_preview_api._validate_file_ownership("file_id", "")
assert "Invalid file or app identifier" in str(exc_info.value)
def test_build_file_response_basic(self, file_preview_api: FilePreviewApi, mock_upload_file):
"""Test basic file response building"""
mock_generator = Mock()
@pytest.mark.parametrize(
("as_attachment", "mime_type", "name", "extension", "size"),
[
(False, "image/jpeg", "test_file.jpg", "jpg", 1024),
(True, "image/jpeg", "test_file.jpg", "jpg", 1024),
(False, "text/html", "unsafe.html", "html", 1024),
(False, "video/mp4", "test_file.mp4", "mp4", 1024),
(False, "image/jpeg", "test_file.jpg", "jpg", 0),
],
)
def test_build_file_response(
self,
file_preview_api: FilePreviewApi,
as_attachment: bool,
mime_type: str,
name: str,
extension: str,
size: int,
):
upload_file = _upload_file(tenant_id=str(uuid4()))
upload_file.mime_type = mime_type
upload_file.name = name
upload_file.extension = extension
upload_file.size = size
response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False)
response = file_preview_api._build_file_response(Mock(), upload_file, as_attachment)
# Check response properties
assert response.mimetype == mock_upload_file.mime_type
assert response.direct_passthrough is True
assert response.headers["Content-Length"] == str(mock_upload_file.size)
assert "Cache-Control" in response.headers
def test_build_file_response_as_attachment(self, file_preview_api: FilePreviewApi, mock_upload_file):
"""Test file response building with attachment flag"""
mock_generator = Mock()
response = file_preview_api._build_file_response(mock_generator, mock_upload_file, True)
# Check attachment-specific headers
assert "attachment" in response.headers["Content-Disposition"]
assert mock_upload_file.name in response.headers["Content-Disposition"]
assert response.headers["Content-Type"] == "application/octet-stream"
def test_build_file_response_html_forces_attachment(self, file_preview_api: FilePreviewApi, mock_upload_file):
"""Test HTML files are forced to download"""
mock_generator = Mock()
mock_upload_file.mime_type = "text/html"
mock_upload_file.name = "unsafe.html"
mock_upload_file.extension = "html"
response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False)
assert "attachment" in response.headers["Content-Disposition"]
assert response.headers["Content-Type"] == "application/octet-stream"
assert response.headers["X-Content-Type-Options"] == "nosniff"
def test_build_file_response_audio_video(self, file_preview_api: FilePreviewApi, mock_upload_file):
"""Test file response building for audio/video files"""
mock_generator = Mock()
mock_upload_file.mime_type = "video/mp4"
response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False)
# Check Range support for media files
assert response.headers["Accept-Ranges"] == "bytes"
def test_build_file_response_no_size(self, file_preview_api: FilePreviewApi, mock_upload_file):
"""Test file response building when size is unknown"""
mock_generator = Mock()
mock_upload_file.size = 0 # Unknown size
response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False)
# Content-Length should not be set when size is unknown
assert "Content-Length" not in response.headers
assert ("Content-Length" in response.headers) is bool(size)
if as_attachment or mime_type == "text/html":
assert "attachment" in response.headers["Content-Disposition"]
assert response.headers["Content-Type"] == "application/octet-stream"
else:
assert response.mimetype == mime_type
if mime_type == "text/html":
assert response.headers["X-Content-Type-Options"] == "nosniff"
if mime_type.startswith("video/"):
assert response.headers["Accept-Ranges"] == "bytes"
@patch("controllers.service_api.app.file_preview.storage")
def test_get_method_integration(
self,
mock_storage,
file_preview_api: FilePreviewApi,
mock_app,
mock_end_user,
mock_upload_file,
mock_message_file,
mock_message,
def test_components_use_validated_file(
self, mock_storage: Mock, file_preview_api: FilePreviewApi, database: _Database
):
"""Test the full GET method integration (without decorator)"""
file_id = str(uuid.uuid4())
app_id = mock_app.id
records = _persist_preview_records(database.session)
generator = Mock()
# Set up mocks
mock_upload_file.tenant_id = mock_app.tenant_id
mock_message.app_id = app_id
mock_message_file.upload_file_id = file_id
mock_message_file.message_id = mock_message.id
with patch("controllers.service_api.app.file_preview.db", database):
message_file, upload_file = file_preview_api._validate_file_ownership(
records.upload_file.id, records.app.id
)
response = file_preview_api._build_file_response(generator, upload_file, False)
mock_generator = Mock()
mock_storage.load.return_value = mock_generator
with patch("controllers.service_api.app.file_preview.db") as mock_db:
# Mock scalar() for MessageFile and Message queries
mock_db.session.scalar.side_effect = [
mock_message_file, # MessageFile query
mock_message, # Message query
]
# Mock get() for UploadFile and App PK lookups
mock_db.session.get.side_effect = [
mock_upload_file, # UploadFile query
mock_app, # App query for tenant validation
]
# Test the core logic directly without Flask decorators
# Validate file ownership
result_message_file, result_upload_file = file_preview_api._validate_file_ownership(file_id, app_id)
assert result_message_file == mock_message_file
assert result_upload_file == mock_upload_file
# Test file response building
response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False)
assert response is not None
# Verify storage was called correctly
mock_storage.load.assert_not_called() # Since we're testing components separately
assert message_file.id == records.message_file.id
assert response.mimetype == "image/jpeg"
mock_storage.load.assert_not_called()
@patch("controllers.service_api.app.file_preview.storage")
def test_storage_error_handling(
self,
mock_storage,
file_preview_api: FilePreviewApi,
mock_app,
mock_upload_file,
mock_message_file,
mock_message,
def test_storage_error_remains_external(
self, mock_storage: Mock, file_preview_api: FilePreviewApi, database: _Database
):
"""Test storage error handling in the core logic"""
file_id = str(uuid.uuid4())
app_id = mock_app.id
records = _persist_preview_records(database.session)
mock_storage.load.side_effect = OSError("Storage error")
# Set up mocks
mock_upload_file.tenant_id = mock_app.tenant_id
mock_message.app_id = app_id
mock_message_file.upload_file_id = file_id
mock_message_file.message_id = mock_message.id
with patch("controllers.service_api.app.file_preview.db", database):
_, upload_file = file_preview_api._validate_file_ownership(records.upload_file.id, records.app.id)
# Mock storage error
mock_storage.load.side_effect = Exception("Storage error")
with patch("controllers.service_api.app.file_preview.db") as mock_db:
# Mock scalar() for MessageFile and Message queries
mock_db.session.scalar.side_effect = [
mock_message_file, # MessageFile query
mock_message, # Message query
]
# Mock get() for UploadFile and App PK lookups
mock_db.session.get.side_effect = [
mock_upload_file, # UploadFile query
mock_app, # App query for tenant validation
]
# First validate file ownership works
result_message_file, result_upload_file = file_preview_api._validate_file_ownership(file_id, app_id)
assert result_message_file == mock_message_file
assert result_upload_file == mock_upload_file
# Test storage error handling
with pytest.raises(Exception) as exc_info:
mock_storage.load(mock_upload_file.key, stream=True)
assert "Storage error" in str(exc_info.value)
with pytest.raises(OSError, match="Storage error"):
mock_storage.load(upload_file.key, stream=True)
def test_validate_file_ownership_unexpected_error_logging(
self, file_preview_api: FilePreviewApi, caplog: pytest.LogCaptureFixture
self,
file_preview_api: FilePreviewApi,
database: _Database,
sqlite_engine: Engine,
caplog: pytest.LogCaptureFixture,
):
"""Test that unexpected errors are logged properly"""
file_id = str(uuid.uuid4())
app_id = str(uuid.uuid4())
file_id = str(uuid4())
app_id = str(uuid4())
with patch("controllers.service_api.app.file_preview.db") as mock_db:
# Mock database scalar to raise unexpected exception
mock_db.session.scalar.side_effect = Exception("Unexpected database error")
def fail_statement(*_args: object) -> None:
raise RuntimeError("Unexpected database error")
# Execute and assert exception
with caplog.at_level(logging.ERROR, logger="controllers.service_api.app.file_preview"):
with pytest.raises(FileAccessDeniedError) as exc_info:
file_preview_api._validate_file_ownership(file_id, app_id)
event.listen(sqlite_engine, "before_cursor_execute", fail_statement)
try:
with patch("controllers.service_api.app.file_preview.db", database):
with caplog.at_level(logging.ERROR, logger="controllers.service_api.app.file_preview"):
with pytest.raises(FileAccessDeniedError, match="File access validation failed"):
file_preview_api._validate_file_ownership(file_id, app_id)
finally:
event.remove(sqlite_engine, "before_cursor_execute", fail_statement)
# Verify error message
assert "File access validation failed" in str(exc_info.value)
# Verify logging was called with the structured context fields. The ``extra`` keys
# are attached to the LogRecord as attributes, so they are not in ``caplog.text``.
assert len(caplog.records) == 1
log_record = caplog.records[0]
assert log_record.getMessage() == "Unexpected error during file ownership validation"
record = cast(_FilePreviewLogRecord, log_record)
assert record.file_id == file_id
assert record.app_id == app_id
assert record.error == "Unexpected database error"
assert len(caplog.records) == 1
log_record = caplog.records[0]
assert log_record.getMessage() == "Unexpected error during file ownership validation"
record = cast(_FilePreviewLogRecord, log_record)
assert record.file_id == file_id
assert record.app_id == app_id
assert record.error == "Unexpected database error"
@@ -163,6 +163,7 @@ def test_get_form_includes_site(monkeypatch: pytest.MonkeyPatch, app: Flask, dat
assert body["expiration_time"] == int(expiration_time.timestamp())
assert body["site"] == {
"app_id": app_model.id,
"mode": "chat",
"end_user_id": None,
"enable_site": True,
"site": {
@@ -383,6 +384,7 @@ def test_get_form_allows_backstage_token(monkeypatch: pytest.MonkeyPatch, app: F
assert body["expiration_time"] == int(expiration_time.timestamp())
assert body["site"] == {
"app_id": app_model.id,
"mode": "chat",
"end_user_id": None,
"enable_site": True,
"site": {
@@ -3,7 +3,42 @@ from unittest.mock import MagicMock, patch
from configs import dify_config
from controllers.web import site as site_module
from extensions.storage.storage_type import StorageType
from models.model import IconType, Site
from models.model import AppMode, IconType, Site
from services.feature_service import FeatureModel
def test_app_site_api_returns_legacy_agent_compatible_mode() -> None:
app_model = MagicMock()
app_model.id = "app-id"
app_model.tenant_id = "tenant-id"
app_model.tenant = MagicMock(id="tenant-id", status="normal")
app_model.mode_compatible_with_agent_with_session.return_value = AppMode.AGENT_CHAT
end_user = MagicMock(id="end-user-id")
site = MagicMock(spec=Site)
response = MagicMock()
response.model_dump.return_value = {"mode": AppMode.AGENT_CHAT}
with (
patch.object(site_module, "db") as mock_db,
patch.object(site_module.FeatureService, "get_features", return_value=FeatureModel(can_replace_logo=False)),
patch.object(site_module, "_build_site_icon_url", return_value=None),
patch.object(site_module.WebAppSiteResponse, "from_app_site", return_value=response) as mock_from_app_site,
):
mock_db.session.scalar.return_value = site
result = site_module.AppSiteApi().get(app_model, end_user)
assert result["mode"] == AppMode.AGENT_CHAT
app_model.mode_compatible_with_agent_with_session.assert_called_once_with(session=mock_db.session())
mock_from_app_site.assert_called_once_with(
tenant=app_model.tenant,
app_model=app_model,
mode=AppMode.AGENT_CHAT,
site=site,
end_user_id=end_user.id,
features=FeatureModel(can_replace_logo=False),
can_replace_logo=False,
icon_url=None,
)
def test_build_site_icon_url_uses_s3_presigned_url() -> None:
@@ -442,6 +442,13 @@ class TestAdvancedChatAppGeneratorInternals:
def start(self):
thread_data["started"] = True
def join(self, timeout):
thread_data["joined"] = True
thread_data["join_timeout"] = timeout
def is_alive(self):
return False
monkeypatch.setattr("core.app.apps.advanced_chat.app_generator.threading.Thread", _Thread)
monkeypatch.setattr(
"core.app.apps.advanced_chat.app_generator.db", SimpleNamespace(engine=object(), session=db_session)
@@ -475,6 +482,8 @@ class TestAdvancedChatAppGeneratorInternals:
assert response["response"] == {"raw": True}
assert thread_data["started"] is True
assert thread_data["joined"] is True
assert thread_data["join_timeout"] == 300
assert "pause-layer" in thread_data["kwargs"]["graph_engine_layers"]
assert generator._dialogue_count == 3
assert init_records.call_args.kwargs["session"] is db_session
@@ -542,6 +551,13 @@ class TestAdvancedChatAppGeneratorInternals:
def start(self):
thread_data["started"] = True
def join(self, timeout):
thread_data["joined"] = True
thread_data["join_timeout"] = timeout
def is_alive(self):
return False
monkeypatch.setattr("core.app.apps.advanced_chat.app_generator.threading.Thread", _Thread)
monkeypatch.setattr(
"core.app.apps.advanced_chat.app_generator.db", SimpleNamespace(engine=object(), session=db_session)
@@ -574,6 +590,8 @@ class TestAdvancedChatAppGeneratorInternals:
init_records.assert_not_called()
get_thread_messages_length.assert_called_once_with(conversation.id, session=db_session)
assert thread_data["started"] is True
assert thread_data["joined"] is True
assert thread_data["join_timeout"] == 300
db_session.commit.assert_not_called()
db_session.refresh.assert_not_called()
db_session.close.assert_called_once()
@@ -435,6 +435,7 @@ def test_generate_success_returns_converted(generator, mocker: MockerFixture):
mocker.patch.object(module, "PipelineQueueManager", return_value=queue_manager)
worker_thread = MagicMock()
worker_thread.is_alive.return_value = False
mocker.patch.object(module.threading, "Thread", return_value=worker_thread)
mocker.patch.object(generator, "_get_draft_var_saver_factory", return_value=MagicMock())
@@ -461,6 +462,7 @@ def test_generate_success_returns_converted(generator, mocker: MockerFixture):
)
assert result == "converted"
worker_thread.join.assert_called_once_with(timeout=300)
def test_single_iteration_generate_validates_inputs(generator, mocker: MockerFixture):
@@ -1,3 +1,6 @@
import logging
from unittest.mock import Mock
import pytest
from core.app.apps.base_app_generator import BaseAppGenerator
@@ -369,6 +372,58 @@ def test_validate_inputs_optional_file_with_empty_string_ignores_default():
class TestBaseAppGeneratorExtras:
def test_wrap_stream_joins_worker_after_stream_exhaustion(self):
base_app_generator = BaseAppGenerator()
worker_thread = Mock()
worker_thread.is_alive.return_value = False
def response_stream():
yield {"event": "workflow_finished"}
managed_stream = base_app_generator._wrap_stream_with_worker_thread_join(
response_stream(),
worker_thread,
)
assert next(managed_stream) == {"event": "workflow_finished"}
worker_thread.join.assert_not_called()
with pytest.raises(StopIteration):
next(managed_stream)
worker_thread.join.assert_called_once_with(timeout=300)
def test_wrap_stream_joins_worker_when_stream_closes(self):
base_app_generator = BaseAppGenerator()
worker_thread = Mock()
worker_thread.is_alive.return_value = False
def response_stream():
yield {"event": "workflow_started"}
yield {"event": "workflow_finished"}
managed_stream = base_app_generator._wrap_stream_with_worker_thread_join(
response_stream(),
worker_thread,
)
assert next(managed_stream) == {"event": "workflow_started"}
managed_stream.close()
worker_thread.join.assert_called_once_with(timeout=300)
def test_join_worker_thread_warns_when_thread_remains_alive(self, caplog: pytest.LogCaptureFixture):
worker_thread = Mock()
worker_thread.name = "leaked-app-worker"
worker_thread.is_alive.return_value = True
with caplog.at_level(logging.WARNING, logger="core.app.apps.base_app_generator"):
BaseAppGenerator._join_worker_thread(worker_thread)
worker_thread.join.assert_called_once_with(timeout=300)
assert "Possible app worker thread leak" in caplog.text
assert "leaked-app-worker" in caplog.text
def test_prepare_user_inputs_converts_files_and_lists(self, monkeypatch: pytest.MonkeyPatch):
base_app_generator = BaseAppGenerator()
@@ -211,6 +211,13 @@ def test_generate_appends_pause_layer_and_forwards_state(mocker: MockerFixture):
def start(self):
return None
def join(self, timeout):
worker_kwargs["joined"] = True
worker_kwargs["join_timeout"] = timeout
def is_alive(self):
return False
mocker.patch("core.app.apps.workflow.app_generator.threading.Thread", DummyThread)
app_model = SimpleNamespace(mode="workflow", tenant_id="tenant")
@@ -244,6 +251,8 @@ def test_generate_appends_pause_layer_and_forwards_state(mocker: MockerFixture):
assert result == "converted"
assert worker_kwargs["kwargs"]["graph_engine_layers"] == ("base-layer", pause_layer)
assert worker_kwargs["kwargs"]["graph_runtime_state"] is graph_runtime_state
assert worker_kwargs["joined"] is True
assert worker_kwargs["join_timeout"] == 300
assert draft_saver_factory.call_args.kwargs["tenant_id"] == app_model.tenant_id
@@ -286,6 +295,8 @@ def test_resume_path_runs_worker_with_runtime_state(mocker: MockerFixture):
mocker.patch("core.app.apps.workflow.app_generator.WorkflowAppRunner", side_effect=runner_ctor)
worker_lifecycle: dict[str, bool] = {}
class ImmediateThread:
def __init__(self, target, kwargs):
target(**kwargs)
@@ -293,6 +304,13 @@ def test_resume_path_runs_worker_with_runtime_state(mocker: MockerFixture):
def start(self):
return None
def join(self, timeout):
worker_lifecycle["joined"] = True
worker_lifecycle["join_timeout"] = timeout
def is_alive(self):
return False
mocker.patch("core.app.apps.workflow.app_generator.threading.Thread", ImmediateThread)
mocker.patch(
@@ -331,5 +349,7 @@ def test_resume_path_runs_worker_with_runtime_state(mocker: MockerFixture):
)
assert result == "raw-response"
assert worker_lifecycle["joined"] is True
assert worker_lifecycle["join_timeout"] == 300
runner_instance.run.assert_called_once()
queue_manager.graph_runtime_state = runtime_state
@@ -1,5 +1,9 @@
import threading
from collections.abc import Generator
import pytest
from core.app.apps.base_app_generator import BaseAppGenerator
from core.app.apps.workflow.active_workflow_tasks import (
active_workflow_task,
get_active_workflow_task_count,
@@ -28,3 +32,51 @@ def test_active_workflow_task_rejects_duplicate_task_id() -> None:
with pytest.raises(ValueError, match="already active"):
with active_workflow_task("task-a"):
pass
def test_managed_stream_waits_for_active_worker_cleanup() -> None:
worker_started = threading.Event()
release_worker = threading.Event()
stream_exhausted = threading.Event()
consumer_finished = threading.Event()
consumer_errors: list[BaseException] = []
def run_worker() -> None:
with active_workflow_task("task-a"):
worker_started.set()
release_worker.wait()
def response_stream() -> Generator[dict[str, str], None, None]:
yield {"event": "workflow_finished"}
stream_exhausted.set()
worker_thread = threading.Thread(target=run_worker)
worker_thread.start()
assert worker_started.wait(timeout=2)
managed_stream = BaseAppGenerator._wrap_stream_with_worker_thread_join(response_stream(), worker_thread)
assert next(managed_stream) == {"event": "workflow_finished"}
def finish_stream() -> None:
try:
list(managed_stream)
except BaseException as exc:
consumer_errors.append(exc)
finally:
consumer_finished.set()
consumer_thread = threading.Thread(target=finish_stream)
consumer_thread.start()
try:
assert stream_exhausted.wait(timeout=2)
assert not consumer_finished.is_set()
assert get_active_workflow_task_count() == 1
finally:
release_worker.set()
consumer_thread.join(timeout=2)
worker_thread.join(timeout=2)
assert not consumer_thread.is_alive()
assert not worker_thread.is_alive()
assert consumer_errors == []
assert get_active_workflow_task_count() == 0
@@ -15,6 +15,70 @@ from models.model import AppMode
class TestWorkflowAppGeneratorValidation:
def test_generate_stream_joins_worker_after_response_exhaustion(self, monkeypatch: pytest.MonkeyPatch):
generator = WorkflowAppGenerator()
worker_thread = Mock()
worker_thread.is_alive.return_value = False
app_config = WorkflowUIBasedAppConfig(
tenant_id="tenant",
app_id="app",
app_mode=AppMode.WORKFLOW,
additional_features=AppAdditionalFeatures(),
variables=[],
workflow_id="workflow-id",
)
application_generate_entity = WorkflowAppGenerateEntity.model_construct(
task_id="task",
app_config=app_config,
inputs={},
files=[],
user_id="user",
stream=True,
invoke_from=InvokeFrom.WEB_APP,
extras={},
)
def response_stream():
yield {"event": "workflow_finished"}
monkeypatch.setattr(generator, "_bind_file_access_scope", lambda **kwargs: contextlib.nullcontext())
monkeypatch.setattr(
"core.app.apps.workflow.app_generator.WorkflowAppQueueManager",
lambda **kwargs: SimpleNamespace(**kwargs),
)
monkeypatch.setattr(
"core.app.apps.workflow.app_generator.current_app",
SimpleNamespace(_get_current_object=lambda: SimpleNamespace(name="flask")),
)
monkeypatch.setattr("core.app.apps.workflow.app_generator.contextvars.copy_context", lambda: "ctx")
monkeypatch.setattr("core.app.apps.workflow.app_generator.threading.Thread", lambda **kwargs: worker_thread)
monkeypatch.setattr(
"core.app.apps.workflow.app_generator.db",
SimpleNamespace(session=SimpleNamespace(close=Mock())),
)
monkeypatch.setattr(generator, "_get_draft_var_saver_factory", lambda *args, **kwargs: "draft-factory")
monkeypatch.setattr(generator, "_handle_response", lambda **kwargs: response_stream())
monkeypatch.setattr(
"core.app.apps.workflow.app_generator.WorkflowAppGenerateResponseConverter.convert",
lambda response, invoke_from: response,
)
managed_stream = generator._generate(
app_model=SimpleNamespace(mode=AppMode.WORKFLOW, tenant_id="tenant"),
workflow=SimpleNamespace(id="workflow-id"),
user=SimpleNamespace(id="user"),
application_generate_entity=application_generate_entity,
invoke_from=InvokeFrom.WEB_APP,
workflow_execution_repository=SimpleNamespace(),
workflow_node_execution_repository=SimpleNamespace(),
streaming=True,
)
worker_thread.start.assert_called_once_with()
worker_thread.join.assert_not_called()
assert list(managed_stream) == [{"event": "workflow_finished"}]
worker_thread.join.assert_called_once_with(timeout=300)
def test_ensure_snippet_start_node_returns_original_for_non_snippet_workflow(self):
workflow = SimpleNamespace(kind_or_standard="workflow")
session = SimpleNamespace(scalar=Mock())
@@ -1,24 +1,95 @@
"""Unit tests for the message cycle manager optimization."""
import logging
from collections.abc import Iterator
from dataclasses import dataclass
from types import SimpleNamespace
from unittest.mock import Mock, patch
import pytest
from flask import Flask, current_app
from sqlalchemy import Engine, event, select
from sqlalchemy.orm import Session, sessionmaker
from core.app.entities.queue_entities import QueueAnnotationReplyEvent, QueueRetrieverResourcesEvent
from core.app.entities.task_entities import MessageStreamResponse, StreamEvent, TaskStateMetadata
from core.app.task_pipeline import message_cycle_manager as message_cycle_manager_module
from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
from core.rag.entities import RetrievalSourceMetadata
from models.model import App, AppMode
from graphon.file import FileTransferMethod, FileType
from models import model as model_module
from models.base import TypeBase
from models.enums import ConversationFromSource, CreatorUserRole, MessageFileBelongsTo
from models.model import App, AppMode, Conversation, MessageFile
def _patch_create_session(mock_session):
session_cm = Mock()
session_cm.__enter__ = Mock(return_value=mock_session)
session_cm.__exit__ = Mock(return_value=False)
return patch("core.app.task_pipeline.message_cycle_manager.session_factory.create_session", return_value=session_cm)
@dataclass(frozen=True)
class _SQLiteDb:
engine: Engine
session: Session
@pytest.fixture
def cycle_db(sqlite_engine: Engine, monkeypatch: pytest.MonkeyPatch) -> Iterator[Session]:
"""Bind request-owned and cycle-manager-owned sessions to isolated SQLite."""
TypeBase.metadata.create_all(
sqlite_engine,
tables=[App.__table__, Conversation.__table__, MessageFile.__table__],
)
owned_session_factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
with owned_session_factory() as request_session:
sqlite_db = _SQLiteDb(engine=sqlite_engine, session=request_session)
monkeypatch.setattr(message_cycle_manager_module, "db", sqlite_db)
monkeypatch.setattr(model_module, "db", sqlite_db)
monkeypatch.setattr(message_cycle_manager_module.session_factory, "create_session", owned_session_factory)
yield request_session
def _app(*, app_id: str = "app-id", tenant_id: str = "tenant-1") -> App:
return App(
id=app_id,
tenant_id=tenant_id,
name="Test App",
description="",
mode=AppMode.CHAT,
enable_site=True,
enable_api=True,
max_active_requests=0,
)
def _conversation(*, conversation_id: str = "conv-1", app_id: str = "app-id") -> Conversation:
conversation = Conversation(
app_id=app_id,
mode=AppMode.CHAT,
name="",
status="normal",
from_source=ConversationFromSource.API,
inputs={},
)
conversation.id = conversation_id
return conversation
def _message_file(
*,
file_id: str = "file-1",
message_id: str = "test-message-id",
belongs_to: MessageFileBelongsTo | None = MessageFileBelongsTo.ASSISTANT,
url: str | None = "http://example.com/image.png",
file_type: FileType = FileType.IMAGE,
) -> MessageFile:
message_file = MessageFile(
message_id=message_id,
type=file_type,
transfer_method=FileTransferMethod.TOOL_FILE,
created_by_role=CreatorUserRole.ACCOUNT,
created_by="account-id",
belongs_to=belongs_to,
url=url,
)
message_file.id = file_id
return message_file
class TestMessageCycleManagerOptimization:
@@ -37,30 +108,22 @@ class TestMessageCycleManagerOptimization:
task_state = Mock()
return MessageCycleManager(application_generate_entity=mock_application_generate_entity, task_state=task_state)
def test_get_message_event_type_with_assistant_file(self, message_cycle_manager):
def test_get_message_event_type_with_assistant_file(self, message_cycle_manager, cycle_db: Session):
"""Test get_message_event_type returns MESSAGE_FILE when message has assistant-generated files.
This ensures that AI-generated images (belongs_to='assistant') trigger the MESSAGE_FILE event,
allowing the frontend to properly display generated image files with url field.
"""
with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
# Setup mock session and message file
mock_session = Mock()
mock_session_factory.create_session.return_value.__enter__.return_value = mock_session
cycle_db.add(_message_file())
cycle_db.commit()
mock_message_file = Mock()
mock_message_file.belongs_to = "assistant"
mock_session.scalar.return_value = mock_message_file
with current_app.app_context():
result = message_cycle_manager.get_message_event_type("test-message-id")
# Execute
with current_app.app_context():
result = message_cycle_manager.get_message_event_type("test-message-id")
assert result == StreamEvent.MESSAGE_FILE
assert "test-message-id" in message_cycle_manager._message_has_file
# Assert
assert result == StreamEvent.MESSAGE_FILE
mock_session.scalar.assert_called_once()
def test_get_message_event_type_with_user_file(self, message_cycle_manager):
def test_get_message_event_type_with_user_file(self, message_cycle_manager, cycle_db: Session):
"""Test get_message_event_type returns MESSAGE when message only has user-uploaded files.
This is a regression test for the issue where user-uploaded images (belongs_to='user')
@@ -68,90 +131,81 @@ class TestMessageCycleManagerOptimization:
resulting in broken images in the chat UI. The query filters for belongs_to='assistant',
so when only user files exist, the database query returns None, resulting in MESSAGE event type.
"""
with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
# Setup mock session and message file
mock_session = Mock()
mock_session_factory.create_session.return_value.__enter__.return_value = mock_session
cycle_db.add(_message_file(belongs_to=MessageFileBelongsTo.USER))
cycle_db.commit()
# When querying for assistant files with only user files present, return None
# (simulates database query with belongs_to='assistant' filter returning no results)
mock_session.scalar.return_value = None
with current_app.app_context():
result = message_cycle_manager.get_message_event_type("test-message-id")
# Execute
with current_app.app_context():
result = message_cycle_manager.get_message_event_type("test-message-id")
assert result == StreamEvent.MESSAGE
assert "test-message-id" not in message_cycle_manager._message_has_file
# Assert
assert result == StreamEvent.MESSAGE
mock_session.scalar.assert_called_once()
def test_get_message_event_type_without_message_file(self, message_cycle_manager):
def test_get_message_event_type_without_message_file(self, message_cycle_manager, cycle_db: Session):
"""Test get_message_event_type returns MESSAGE when message has no files."""
with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
# Setup mock session and no message file
mock_session = Mock()
mock_session_factory.create_session.return_value.__enter__.return_value = mock_session
# Current implementation uses session.scalar(select(...))
mock_session.scalar.return_value = None
assert list(cycle_db.scalars(select(MessageFile)).all()) == []
# Execute
with current_app.app_context():
result = message_cycle_manager.get_message_event_type("test-message-id")
with current_app.app_context():
result = message_cycle_manager.get_message_event_type("test-message-id")
# Assert
assert result == StreamEvent.MESSAGE
mock_session.scalar.assert_called_once()
assert result == StreamEvent.MESSAGE
def test_get_message_event_type_uses_cache_without_query(self, message_cycle_manager):
def test_get_message_event_type_uses_cache_without_query(
self, message_cycle_manager, cycle_db: Session, sqlite_engine: Engine
):
"""Return MESSAGE_FILE directly from in-memory cache without opening a DB session."""
message_cycle_manager._message_has_file.add("cached-message")
statements: list[str] = []
with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
def record_statement(_conn, _cursor, statement, _parameters, _context, _executemany) -> None:
statements.append(statement)
event.listen(sqlite_engine, "before_cursor_execute", record_statement)
try:
result = message_cycle_manager.get_message_event_type("cached-message")
finally:
event.remove(sqlite_engine, "before_cursor_execute", record_statement)
assert result == StreamEvent.MESSAGE_FILE
mock_session_factory.create_session.assert_not_called()
assert statements == []
def test_message_to_stream_response_with_precomputed_event_type(self, message_cycle_manager):
def test_message_to_stream_response_with_precomputed_event_type(self, message_cycle_manager, cycle_db: Session):
"""MessageCycleManager.message_to_stream_response expects a valid event_type; callers should precompute it."""
with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
# Setup mock session and message file
mock_session = Mock()
mock_session_factory.create_session.return_value.__enter__.return_value = mock_session
cycle_db.add(_message_file())
cycle_db.commit()
mock_message_file = Mock()
mock_message_file.belongs_to = "assistant"
mock_session.scalar.return_value = mock_message_file
with current_app.app_context():
event_type = message_cycle_manager.get_message_event_type("test-message-id")
result = message_cycle_manager.message_to_stream_response(
answer="Hello world", message_id="test-message-id", event_type=event_type
)
# Execute: compute event type once, then pass to message_to_stream_response
with current_app.app_context():
event_type = message_cycle_manager.get_message_event_type("test-message-id")
result = message_cycle_manager.message_to_stream_response(
answer="Hello world", message_id="test-message-id", event_type=event_type
)
assert isinstance(result, MessageStreamResponse)
assert result.answer == "Hello world"
assert result.id == "test-message-id"
assert result.event == StreamEvent.MESSAGE_FILE
# Assert
assert isinstance(result, MessageStreamResponse)
assert result.answer == "Hello world"
assert result.id == "test-message-id"
assert result.event == StreamEvent.MESSAGE_FILE
mock_session.scalar.assert_called_once()
def test_message_to_stream_response_with_event_type_skips_query(self, message_cycle_manager):
def test_message_to_stream_response_with_event_type_skips_query(
self, message_cycle_manager, cycle_db: Session, sqlite_engine: Engine
):
"""Test that message_to_stream_response skips database query when event_type is provided."""
with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
# Execute with event_type provided
statements: list[str] = []
def record_statement(_conn, _cursor, statement, _parameters, _context, _executemany) -> None:
statements.append(statement)
event.listen(sqlite_engine, "before_cursor_execute", record_statement)
try:
result = message_cycle_manager.message_to_stream_response(
answer="Hello world", message_id="test-message-id", event_type=StreamEvent.MESSAGE
)
finally:
event.remove(sqlite_engine, "before_cursor_execute", record_statement)
# Assert
assert isinstance(result, MessageStreamResponse)
assert result.answer == "Hello world"
assert result.id == "test-message-id"
assert result.event == StreamEvent.MESSAGE
# Should not open a session when event_type is provided
mock_session_factory.create_session.assert_not_called()
assert isinstance(result, MessageStreamResponse)
assert result.answer == "Hello world"
assert result.id == "test-message-id"
assert result.event == StreamEvent.MESSAGE
assert statements == []
def test_message_to_stream_response_with_from_variable_selector(self, message_cycle_manager):
"""Test message_to_stream_response with from_variable_selector parameter."""
@@ -168,40 +222,32 @@ class TestMessageCycleManagerOptimization:
assert result.from_variable_selector == ["var1", "var2"]
assert result.event == StreamEvent.MESSAGE
def test_optimization_usage_example(self, message_cycle_manager):
def test_optimization_usage_example(self, message_cycle_manager, cycle_db: Session, sqlite_engine: Engine):
"""Test the optimization pattern that should be used by callers."""
# Step 1: Get event type once (this queries database)
with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
mock_session = Mock()
mock_session_factory.create_session.return_value.__enter__.return_value = mock_session
# Current implementation uses session.scalar(select(...))
mock_session.scalar.return_value = None # No files
statements: list[str] = []
def record_statement(_conn, _cursor, statement, _parameters, _context, _executemany) -> None:
statements.append(statement)
event.listen(sqlite_engine, "before_cursor_execute", record_statement)
try:
with current_app.app_context():
event_type = message_cycle_manager.get_message_event_type("test-message-id")
# Should open session once
mock_session_factory.create_session.assert_called_once()
assert event_type == StreamEvent.MESSAGE
# Step 2: Use event_type for multiple calls (no additional queries)
with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
mock_session_factory.create_session.return_value.__enter__.return_value = Mock()
chunk1_response = message_cycle_manager.message_to_stream_response(
answer="Chunk 1", message_id="test-message-id", event_type=event_type
)
chunk2_response = message_cycle_manager.message_to_stream_response(
answer="Chunk 2", message_id="test-message-id", event_type=event_type
)
finally:
event.remove(sqlite_engine, "before_cursor_execute", record_statement)
# Should not open session again when event_type provided
mock_session_factory.create_session.assert_not_called()
assert chunk1_response.event == StreamEvent.MESSAGE
assert chunk2_response.event == StreamEvent.MESSAGE
assert chunk1_response.answer == "Chunk 1"
assert chunk2_response.answer == "Chunk 2"
assert event_type == StreamEvent.MESSAGE
assert len([statement for statement in statements if statement.lstrip().upper().startswith("SELECT")]) == 1
assert chunk1_response.event == StreamEvent.MESSAGE
assert chunk2_response.event == StreamEvent.MESSAGE
assert chunk1_response.answer == "Chunk 1"
assert chunk2_response.answer == "Chunk 2"
def test_generate_conversation_name_returns_none_for_completion(self, message_cycle_manager):
"""Return None when completion entities are used for conversation naming.
@@ -269,51 +315,38 @@ class TestMessageCycleManagerOptimization:
assert message_cycle_manager._application_generate_entity.is_new_conversation is False
mock_timer.assert_not_called()
def test_generate_conversation_name_worker_returns_when_conversation_missing(self, message_cycle_manager):
def test_generate_conversation_name_worker_returns_when_conversation_missing(
self, message_cycle_manager, cycle_db: Session
):
"""Return early when the conversation cannot be found."""
flask_app = Flask(__name__)
db_session = Mock()
db_session.scalar.return_value = None
assert list(cycle_db.scalars(select(Conversation)).all()) == []
with _patch_create_session(db_session):
message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-missing", "hello")
message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-missing", "hello")
db_session.commit.assert_not_called()
assert list(cycle_db.scalars(select(Conversation)).all()) == []
def test_generate_conversation_name_worker_returns_when_app_missing(self, message_cycle_manager):
def test_generate_conversation_name_worker_returns_when_app_missing(self, message_cycle_manager, cycle_db: Session):
"""Return early when non-completion conversation has no app relation."""
flask_app = Flask(__name__)
conversation = SimpleNamespace(mode=AppMode.CHAT, app=None, app_id="app-id")
db_session = Mock()
db_session.scalar.return_value = conversation
db_session.get.return_value = None
conversation = _conversation()
cycle_db.add(conversation)
cycle_db.commit()
with _patch_create_session(db_session):
message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-1", "hello")
message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-1", "hello")
db_session.commit.assert_not_called()
assert cycle_db.get(Conversation, "conv-1").name == ""
assert cycle_db.get(App, "app-id") is None
def test_generate_conversation_name_worker_uses_cached_name(self, message_cycle_manager):
def test_generate_conversation_name_worker_uses_cached_name(
self, message_cycle_manager, cycle_db: Session, sqlite_engine: Engine
):
"""Use cached conversation name when present and avoid LLM call."""
flask_app = Flask(__name__)
class ConversationWithPoisonedApp:
mode = AppMode.CHAT
app_id = "app-id"
name = ""
@property
def app(self):
raise AssertionError("conversation.app must not open an implicit session")
conversation = ConversationWithPoisonedApp()
app_model = SimpleNamespace(tenant_id="tenant-1")
db_session = Mock()
db_session.scalar.return_value = conversation
db_session.get.return_value = app_model
cycle_db.add_all([_app(), _conversation()])
cycle_db.commit()
with (
_patch_create_session(db_session) as create_session,
patch("core.app.task_pipeline.message_cycle_manager.redis_client") as mock_redis,
patch("core.app.task_pipeline.message_cycle_manager.LLMGenerator") as mock_llm_generator,
):
@@ -321,27 +354,23 @@ class TestMessageCycleManagerOptimization:
message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-1", "hello")
assert cycle_db.in_transaction() is False
with Session(sqlite_engine) as verification_session:
conversation = verification_session.get(Conversation, "conv-1")
assert conversation is not None
assert conversation.name == "cached-title"
create_session.assert_called_once_with()
db_session.get.assert_called_once_with(App, "app-id")
db_session.commit.assert_called_once()
mock_llm_generator.generate_conversation_name.assert_not_called()
mock_redis.setex.assert_not_called()
def test_generate_conversation_name_worker_generates_and_caches_name(self, message_cycle_manager):
def test_generate_conversation_name_worker_generates_and_caches_name(
self, message_cycle_manager, cycle_db: Session, sqlite_engine: Engine
):
"""Generate conversation name and write it to redis cache on cache miss."""
flask_app = Flask(__name__)
conversation = SimpleNamespace(
mode=AppMode.CHAT,
app=SimpleNamespace(tenant_id="tenant-1"),
app_id="app-id",
name="",
)
db_session = Mock()
db_session.scalar.return_value = conversation
cycle_db.add_all([_app(), _conversation()])
cycle_db.commit()
with (
_patch_create_session(db_session),
patch("core.app.task_pipeline.message_cycle_manager.redis_client") as mock_redis,
patch("core.app.task_pipeline.message_cycle_manager.LLMGenerator") as mock_llm_generator,
):
@@ -350,27 +379,27 @@ class TestMessageCycleManagerOptimization:
message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-1", "hello")
assert cycle_db.in_transaction() is False
with Session(sqlite_engine) as verification_session:
conversation = verification_session.get(Conversation, "conv-1")
assert conversation is not None
assert conversation.name == "generated-title"
db_session.commit.assert_called_once()
mock_redis.setex.assert_called_once()
def test_generate_conversation_name_worker_falls_back_when_generation_fails(
self, message_cycle_manager, caplog: pytest.LogCaptureFixture
self,
message_cycle_manager,
cycle_db: Session,
sqlite_engine: Engine,
caplog: pytest.LogCaptureFixture,
):
"""Fallback to truncated query when LLM generation fails."""
flask_app = Flask(__name__)
conversation = SimpleNamespace(
mode=AppMode.CHAT,
app=SimpleNamespace(tenant_id="tenant-1"),
app_id="app-id",
name="",
)
db_session = Mock()
db_session.scalar.return_value = conversation
cycle_db.add_all([_app(), _conversation()])
cycle_db.commit()
long_query = "q" * 60
with (
_patch_create_session(db_session),
patch("core.app.task_pipeline.message_cycle_manager.redis_client") as mock_redis,
patch("core.app.task_pipeline.message_cycle_manager.LLMGenerator") as mock_llm_generator,
patch("core.app.task_pipeline.message_cycle_manager.dify_config") as mock_dify_config,
@@ -382,8 +411,11 @@ class TestMessageCycleManagerOptimization:
with caplog.at_level(logging.ERROR, logger="core.app.task_pipeline.message_cycle_manager"):
message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-1", long_query)
assert cycle_db.in_transaction() is False
with Session(sqlite_engine) as verification_session:
conversation = verification_session.get(Conversation, "conv-1")
assert conversation is not None
assert conversation.name == (long_query[:47] + "...")
db_session.commit.assert_called_once()
assert any(record.levelno == logging.ERROR for record in caplog.records)
def test_handle_annotation_reply_sets_metadata(self, message_cycle_manager):
@@ -454,33 +486,25 @@ class TestMessageCycleManagerOptimization:
assert message_cycle_manager._task_state.metadata.retriever_resources[0].position == 1
assert message_cycle_manager._task_state.metadata.retriever_resources[1].position == 2
def test_message_file_to_stream_response_builds_signed_url(self, message_cycle_manager):
def test_message_file_to_stream_response_builds_signed_url(self, message_cycle_manager, cycle_db: Session):
"""Build a stream response with a signed tool file URL.
Args: message_cycle_manager with mocked Session/db and sign_tool_file.
Args: message_cycle_manager with a persisted MessageFile and mocked sign_tool_file.
Returns: MessageStreamResponse with signed url and belongs_to normalized to user.
Side effects: Calls sign_tool_file for tool file ids.
"""
message_cycle_manager._application_generate_entity.task_id = "task-1"
message_file = SimpleNamespace(
id="file-1",
type="image",
belongs_to=None,
url="tool://file.verylongextension",
message_id="msg-1",
cycle_db.add(
_message_file(
file_id="file-1",
message_id="msg-1",
belongs_to=None,
url="tool://file.verylongextension",
)
)
cycle_db.commit()
session = Mock()
session.scalar.return_value = message_file
with (
patch("core.app.task_pipeline.message_cycle_manager.Session") as mock_session_cls,
patch("core.app.task_pipeline.message_cycle_manager.sign_tool_file") as mock_sign,
patch("core.app.task_pipeline.message_cycle_manager.db") as mock_db,
):
mock_db.engine = Mock()
mock_session_cls.return_value.__enter__.return_value = session
with patch("core.app.task_pipeline.message_cycle_manager.sign_tool_file") as mock_sign:
mock_sign.return_value = "signed-url"
response = message_cycle_manager.message_file_to_stream_response(SimpleNamespace(message_file_id="file-1"))
@@ -514,56 +538,42 @@ class TestMessageCycleManagerOptimization:
assert len(message_cycle_manager._task_state.metadata.retriever_resources) == 1
assert message_cycle_manager._task_state.metadata.retriever_resources[0].position == 1
def test_message_file_to_stream_response_uses_http_url_directly(self, message_cycle_manager):
def test_message_file_to_stream_response_uses_http_url_directly(self, message_cycle_manager, cycle_db: Session):
"""Use original URL when message file URL is already HTTP."""
message_cycle_manager._application_generate_entity.task_id = "task-http"
message_file = SimpleNamespace(
id="file-http",
type="image",
belongs_to="assistant",
url="http://example.com/pic.png",
message_id="msg-http",
)
session = Mock()
session.scalar.return_value = message_file
with (
patch("core.app.task_pipeline.message_cycle_manager.Session") as mock_session_cls,
patch("core.app.task_pipeline.message_cycle_manager.db") as mock_db,
):
mock_db.engine = Mock()
mock_session_cls.return_value.__enter__.return_value = session
response = message_cycle_manager.message_file_to_stream_response(
SimpleNamespace(message_file_id="file-http")
cycle_db.add(
_message_file(
file_id="file-http",
message_id="msg-http",
belongs_to=MessageFileBelongsTo.ASSISTANT,
url="http://example.com/pic.png",
)
)
cycle_db.commit()
response = message_cycle_manager.message_file_to_stream_response(SimpleNamespace(message_file_id="file-http"))
assert response is not None
assert response.url == "http://example.com/pic.png"
assert "msg-http" in message_cycle_manager._message_has_file
def test_message_file_to_stream_response_defaults_extension_to_bin_without_dot(self, message_cycle_manager):
def test_message_file_to_stream_response_defaults_extension_to_bin_without_dot(
self, message_cycle_manager, cycle_db: Session
):
"""Default tool file extension to .bin when URL has no extension part."""
message_cycle_manager._application_generate_entity.task_id = "task-bin"
message_file = SimpleNamespace(
id="file-bin",
type="file",
belongs_to="assistant",
url="tool-file-id",
message_id="msg-bin",
cycle_db.add(
_message_file(
file_id="file-bin",
message_id="msg-bin",
belongs_to=MessageFileBelongsTo.ASSISTANT,
url="tool-file-id",
file_type=FileType.CUSTOM,
)
)
cycle_db.commit()
session = Mock()
session.scalar.return_value = message_file
with (
patch("core.app.task_pipeline.message_cycle_manager.Session") as mock_session_cls,
patch("core.app.task_pipeline.message_cycle_manager.sign_tool_file") as mock_sign,
patch("core.app.task_pipeline.message_cycle_manager.db") as mock_db,
):
mock_db.engine = Mock()
mock_session_cls.return_value.__enter__.return_value = session
with patch("core.app.task_pipeline.message_cycle_manager.sign_tool_file") as mock_sign:
mock_sign.return_value = "signed-bin-url"
response = message_cycle_manager.message_file_to_stream_response(
@@ -574,19 +584,13 @@ class TestMessageCycleManagerOptimization:
assert response.url == "signed-bin-url"
mock_sign.assert_called_once_with(tool_file_id="tool-file-id", extension=".bin")
def test_message_file_to_stream_response_returns_none_when_file_missing(self, message_cycle_manager):
def test_message_file_to_stream_response_returns_none_when_file_missing(
self, message_cycle_manager, cycle_db: Session
):
"""Return None when message file lookup does not find a record."""
session = Mock()
session.scalar.return_value = None
assert list(cycle_db.scalars(select(MessageFile)).all()) == []
with (
patch("core.app.task_pipeline.message_cycle_manager.Session") as mock_session_cls,
patch("core.app.task_pipeline.message_cycle_manager.db") as mock_db,
):
mock_db.engine = Mock()
mock_session_cls.return_value.__enter__.return_value = session
response = message_cycle_manager.message_file_to_stream_response(SimpleNamespace(message_file_id="missing"))
response = message_cycle_manager.message_file_to_stream_response(SimpleNamespace(message_file_id="missing"))
assert response is None
@@ -0,0 +1,67 @@
"""Verify that Jinja2 transform_caller scripts block unsafe template attacks via SandboxedEnvironment."""
import io
import sys
import pytest
from core.helper.code_executor.jinja2.jinja2_transformer import Jinja2TemplateTransformer
MALICIOUS_TEMPLATES = [
pytest.param(
"{{ lipsum.__globals__.__builtins__.__import__('os').popen('id').read() }}",
id="lipsum_globals_builtins",
),
pytest.param(
"{{ ''.__class__.__mro__[1].__subclasses__() }}",
id="string_class_mro",
),
pytest.param(
"{{ cycler.__init__.__globals__.os.popen('whoami').read() }}",
id="cycler_init_globals",
),
pytest.param(
"{{ namespace.__init__.__globals__['__builtins__']['__import__']('os').system('id') }}",
id="namespace_init_globals",
),
]
def _exec_scripts(runner: str, preload: str) -> str:
"""Execute preload then runner in a shared namespace, return captured stdout."""
ns: dict = {}
exec(compile(preload, "<preload>", "exec"), ns) # noqa: S102
captured = io.StringIO()
old_stdout = sys.stdout
sys.stdout = captured
try:
exec(compile(runner, "<runner>", "exec"), ns) # noqa: S102
finally:
sys.stdout = old_stdout
return captured.getvalue()
class TestJinja2TransformCallerSandbox:
"""Test transform_caller output (runner + preload) blocks attacks and allows safe templates."""
@pytest.mark.parametrize("malicious_template", MALICIOUS_TEMPLATES)
def test_blocks_unsafe_template(self, malicious_template: str) -> None:
runner, preload = Jinja2TemplateTransformer.transform_caller(malicious_template, {})
ns: dict = {}
exec(compile(preload, "<preload>", "exec"), ns) # noqa: S102
with pytest.raises(Exception) as exc_info:
exec(compile(runner, "<runner>", "exec"), ns) # noqa: S102
assert "unsafe" in str(exc_info.value).lower() or "security" in str(exc_info.value).lower()
def test_renders_safe_template(self) -> None:
runner, preload = Jinja2TemplateTransformer.transform_caller(
"Hello {{ name }}, you are {{ age }} years old!",
{"name": "Alice", "age": 30},
)
output = _exec_scripts(runner, preload)
assert "Hello Alice, you are 30 years old!" in output
def test_scripts_use_sandboxed_environment(self) -> None:
runner, preload = Jinja2TemplateTransformer.transform_caller("{{ x }}", {"x": 1})
assert "SandboxedEnvironment" in runner
assert "SandboxedEnvironment" in preload
@@ -1,7 +1,57 @@
import sys
from datetime import datetime, timedelta
from unittest.mock import MagicMock, patch
import pytest
from sqlalchemy import event
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session, sessionmaker
import core.llm_generator.llm_generator as generator_module
from core.llm_generator.llm_generator import LLMGenerator, _parse_string_list
from core.model_manager import ModelInstance, ModelManager
from core.workflow.generator import tool_catalogue as tool_catalogue_module
from core.workflow.generator.tool_catalogue import ToolCatalogueEntry
from graphon.model_runtime.entities.llm_entities import LLMResult, LLMUsage
from graphon.model_runtime.entities.message_entities import AssistantPromptMessage
from models.dataset import Dataset
from services.workflow_service import WorkflowService
@pytest.fixture
def dataset_session(sqlite_session: Session, monkeypatch: pytest.MonkeyPatch) -> Session:
"""Bind the real SQLite session to the production database extension."""
monkeypatch.setattr(generator_module.db, "session", sqlite_session)
return sqlite_session
def _llm_result(content: str) -> LLMResult:
"""Build a real non-streaming LLM response around deterministic test content."""
return LLMResult(
model="test-model",
message=AssistantPromptMessage(content=content),
usage=LLMUsage.empty_usage(),
)
def _model_manager() -> tuple[MagicMock, MagicMock]:
"""Build spec-constrained mocks for the model-manager boundary and its default model."""
model_manager = MagicMock(spec=ModelManager)
model_instance = MagicMock(spec=ModelInstance)
model_manager.get_default_model_instance.return_value = model_instance
return model_manager, model_instance
def _dataset(*, dataset_id: str, tenant_id: str, name: str, created_at: datetime) -> Dataset:
return Dataset(
id=dataset_id,
tenant_id=tenant_id,
name=name,
created_by="account-id",
created_at=created_at,
)
class TestParseStringList:
@@ -34,95 +84,115 @@ class TestParseStringList:
class TestGenerateWorkflowInstructionSuggestions:
@patch("core.llm_generator.llm_generator.ModelManager.for_tenant")
def test_no_default_model(self, mock_for_tenant):
mock_for_tenant.return_value.get_default_model_instance.side_effect = Exception("No model")
model_manager, _ = _model_manager()
model_manager.get_default_model_instance.side_effect = RuntimeError("no default model")
mock_for_tenant.return_value = model_manager
assert LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow") == []
@patch("core.llm_generator.llm_generator.ModelManager.for_tenant")
@patch("core.llm_generator.llm_generator.LLMGenerator._build_suggestion_context")
def test_llm_success(self, mock_build_context, mock_for_tenant):
mock_build_context.return_value = "context"
mock_model = MagicMock()
mock_model.invoke_llm.return_value = MagicMock()
mock_model.invoke_llm.return_value.message.get_text_content.return_value = '["idea 1", "idea 2"]'
mock_for_tenant.return_value.get_default_model_instance.return_value = mock_model
model_manager, model_instance = _model_manager()
model_instance.invoke_llm.return_value = _llm_result('["idea 1", "idea 2"]')
mock_for_tenant.return_value = model_manager
result = LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow")
assert result == ["idea 1", "idea 2"]
model_instance.invoke_llm.assert_called_once()
@patch("core.llm_generator.llm_generator.ModelManager.for_tenant")
@patch("core.llm_generator.llm_generator.LLMGenerator._build_suggestion_context")
def test_llm_error(self, mock_build_context, mock_for_tenant):
mock_build_context.return_value = "context"
model_manager, model_instance = _model_manager()
model_instance.invoke_llm.side_effect = RuntimeError("API error")
mock_for_tenant.return_value = model_manager
mock_model = MagicMock()
mock_model.invoke_llm.side_effect = Exception("API error")
mock_for_tenant.return_value.get_default_model_instance.return_value = mock_model
assert LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow") == []
result = LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow")
assert result == []
model_instance.invoke_llm.assert_called_once()
@patch("core.llm_generator.llm_generator.ModelManager.for_tenant")
@patch("core.llm_generator.llm_generator.LLMGenerator._build_suggestion_context")
def test_llm_bad_output(self, mock_build_context, mock_for_tenant):
mock_build_context.return_value = "context"
model_manager, model_instance = _model_manager()
model_instance.invoke_llm.return_value = _llm_result("Not a list")
mock_for_tenant.return_value = model_manager
mock_model = MagicMock()
mock_model.invoke_llm.return_value = MagicMock()
mock_model.invoke_llm.return_value.message.get_text_content.return_value = "Not a list"
mock_for_tenant.return_value.get_default_model_instance.return_value = mock_model
assert LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow") == []
result = LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow")
assert result == []
model_instance.invoke_llm.assert_called_once()
@pytest.mark.parametrize("sqlite_session", [(Dataset,)], indirect=True)
class TestBuildSuggestionContext:
@patch("core.llm_generator.llm_generator.db.session.scalars")
def test_both_success(self, mock_scalars, monkeypatch):
mock_scalars.return_value.all.return_value = ["kb1", "kb2"]
def test_both_success(self, dataset_session: Session, monkeypatch: pytest.MonkeyPatch):
now = datetime.now()
dataset_session.add_all(
(
_dataset(dataset_id="kb-1", tenant_id="tenant", name="kb1", created_at=now),
_dataset(
dataset_id="kb-2",
tenant_id="tenant",
name="kb2",
created_at=now - timedelta(seconds=1),
),
_dataset(dataset_id="other-kb", tenant_id="other", name="private", created_at=now),
)
)
dataset_session.commit()
# ``_build_suggestion_context`` imports the tool catalogue lazily, so we
# stub the module in ``sys.modules``. Use ``monkeypatch.setitem`` so the
# ORIGINAL module is RESTORED on teardown — a bare ``del`` would evict it
# from sys.modules entirely, after which a sibling test that imported
# ``build_tool_catalogue`` at collection time (e.g. test_tool_catalogue)
# diverges from a freshly re-imported module and its @patch targets stop
# applying, silently breaking it under xdist.
mock_tool_catalogue = MagicMock()
mock_tool_catalogue.build_tool_catalogue.return_value = "catalog"
mock_tool_catalogue.format_tool_catalogue.return_value = "tool1\ntool2"
monkeypatch.setitem(sys.modules, "core.workflow.generator.tool_catalogue", mock_tool_catalogue)
def build_tool_catalogue(_tenant_id: str) -> list[ToolCatalogueEntry]:
return [
ToolCatalogueEntry(
provider_name="provider",
provider_type="builtin",
plugin_id="",
tool_name="tool1",
tool_label="tool1",
description="First tool",
),
ToolCatalogueEntry(
provider_name="provider",
provider_type="builtin",
plugin_id="",
tool_name="tool2",
tool_label="tool2",
description="Second tool",
),
]
# Keep the real module and formatter; only isolate provider/plugin discovery.
monkeypatch.setattr(tool_catalogue_module, "build_tool_catalogue", build_tool_catalogue)
result = LLMGenerator._build_suggestion_context("tenant")
assert "Knowledge bases:\n- kb1\n- kb2" in result
assert "Installed tools:\ntool1\ntool2" in result
assert "Installed tools:\n- provider/tool1 — First tool\n- provider/tool2 — Second tool" in result
@patch("core.llm_generator.llm_generator.db.session.scalars")
def test_both_fail(self, mock_scalars, monkeypatch):
mock_scalars.side_effect = Exception("DB error")
def test_both_fail(self, dataset_session: Session, monkeypatch: pytest.MonkeyPatch):
def fail_query(_orm_execute_state: object) -> None:
raise SQLAlchemyError("DB error")
# See ``test_both_success``: restore the original module via monkeypatch
# rather than ``del``-ing it, so we don't evict it for sibling tests.
mock_tool_catalogue = MagicMock()
mock_tool_catalogue.build_tool_catalogue.side_effect = Exception("Tool error")
monkeypatch.setitem(sys.modules, "core.workflow.generator.tool_catalogue", mock_tool_catalogue)
def fail_tool_catalogue(_tenant_id: str) -> list[ToolCatalogueEntry]:
raise RuntimeError("Tool error")
assert LLMGenerator._build_suggestion_context("tenant") == ""
event.listen(dataset_session, "do_orm_execute", fail_query)
monkeypatch.setattr(tool_catalogue_module, "build_tool_catalogue", fail_tool_catalogue)
try:
assert LLMGenerator._build_suggestion_context("tenant") == ""
finally:
event.remove(dataset_session, "do_orm_execute", fail_query)
class TestWorkflowServiceInterface:
def test_protocol_methods(self):
# Just to cover the 'pass' statements in the Protocol definition
def test_real_workflow_service_exposes_protocol_methods(self):
from core.llm_generator.llm_generator import WorkflowServiceInterface
class MockService(WorkflowServiceInterface):
def get_draft_workflow(self, app_model, workflow_id=None, *, session):
return super().get_draft_workflow(app_model, workflow_id, session=session)
service: WorkflowServiceInterface = WorkflowService(sessionmaker())
def get_node_last_run(self, app_model, workflow, node_id):
return super().get_node_last_run(app_model, workflow, node_id)
service = MockService()
service.get_draft_workflow(None, session=None)
service.get_node_last_run(None, None, "node")
assert callable(service.get_draft_workflow)
assert callable(service.get_node_last_run)
@@ -1,11 +1,19 @@
"""Comprehensive unit tests for core/memory/token_buffer_memory.py"""
"""Comprehensive SQLite-backed tests for token-buffer memory."""
from collections.abc import Iterator
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from unittest.mock import MagicMock, patch
from uuid import uuid4
import pytest
from sqlalchemy import Engine, event
from sqlalchemy.orm import Session
from core.memory import token_buffer_memory as memory_module
from core.memory.token_buffer_memory import TokenBufferMemory
from graphon.file import FileTransferMethod, FileType
from graphon.model_runtime.entities import (
AssistantPromptMessage,
ImagePromptMessageContent,
@@ -13,13 +21,44 @@ from graphon.model_runtime.entities import (
TextPromptMessageContent,
UserPromptMessage,
)
from models.model import AppMode
from models.base import TypeBase
from models.enums import ConversationFromSource, CreatorUserRole, MessageFileBelongsTo
from models.model import AppMode, Message, MessageFile
from models.workflow import Workflow, WorkflowType
# ---------------------------------------------------------------------------
# Helpers / shared fixtures
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Database:
"""Typed SQLite binding plus executed SQL for query-count assertions."""
engine: Engine
session: Session
statements: list[tuple[str, object]]
@pytest.fixture
def database(sqlite_engine: Engine, monkeypatch: pytest.MonkeyPatch) -> Iterator[Database]:
TypeBase.metadata.create_all(
sqlite_engine,
tables=[Message.__table__, MessageFile.__table__, Workflow.__table__],
)
statements: list[tuple[str, object]] = []
def record_statement(_connection, _cursor, statement, parameters, _context, _executemany) -> None:
statements.append((statement, parameters))
event.listen(sqlite_engine, "before_cursor_execute", record_statement)
with Session(sqlite_engine, expire_on_commit=False) as session:
database = Database(engine=sqlite_engine, session=session, statements=statements)
monkeypatch.setattr(memory_module, "db", database)
yield database
event.remove(sqlite_engine, "before_cursor_execute", record_statement)
def _make_conversation(mode: AppMode = AppMode.CHAT) -> MagicMock:
"""Return a minimal Conversation mock."""
conv = MagicMock()
@@ -47,6 +86,73 @@ def _make_message(answer: str = "hello", answer_tokens: int = 5) -> MagicMock:
return msg
def _persist_message(
database: Database,
conversation_id: str,
*,
query: str = "user query",
answer: str = "hello",
answer_tokens: int = 5,
created_at: datetime | None = None,
workflow_run_id: str | None = None,
) -> Message:
message = Message(
id=str(uuid4()),
app_id="app-1",
conversation_id=conversation_id,
_inputs={},
query=query,
message={},
message_unit_price=Decimal(0),
answer=answer,
answer_tokens=answer_tokens,
answer_unit_price=Decimal(0),
currency="USD",
from_source=ConversationFromSource.API,
workflow_run_id=workflow_run_id,
created_at=created_at or datetime.now(UTC).replace(tzinfo=None),
)
database.session.add(message)
database.session.commit()
return message
def _persist_message_file(
database: Database,
message: Message,
*,
belongs_to: MessageFileBelongsTo | None,
) -> MessageFile:
message_file = MessageFile(
message_id=message.id,
type=FileType.IMAGE,
transfer_method=FileTransferMethod.REMOTE_URL,
created_by_role=CreatorUserRole.ACCOUNT,
created_by="account-1",
belongs_to=belongs_to,
url="https://example.com/image.png",
)
database.session.add(message_file)
database.session.commit()
return message_file
def _persist_workflow(database: Database, *, workflow_id: str) -> Workflow:
workflow = Workflow(
id=workflow_id,
tenant_id="tenant-1",
app_id="app-1",
type=WorkflowType.CHAT,
version="1",
graph="{}",
features="{}",
created_by="account-1",
)
database.session.add(workflow)
database.session.commit()
return workflow
# ===========================================================================
# Tests for __init__ and workflow_run_repo property
# ===========================================================================
@@ -61,25 +167,25 @@ class TestInit:
assert mem.model_instance is mi
assert mem._workflow_run_repo is None
def test_workflow_run_repo_is_created_lazily(self):
def test_workflow_run_repo_is_created_lazily(self, database: Database):
conv = _make_conversation()
mi = _make_model_instance()
mem = TokenBufferMemory(conversation=conv, model_instance=mi)
mock_repo = MagicMock()
with (
patch("core.memory.token_buffer_memory.sessionmaker") as mock_sm,
patch("core.memory.token_buffer_memory.db") as mock_db,
patch(
"core.memory.token_buffer_memory.DifyAPIRepositoryFactory.create_api_workflow_run_repository",
return_value=mock_repo,
),
):
mock_db.engine = MagicMock()
with patch(
"core.memory.token_buffer_memory.DifyAPIRepositoryFactory.create_api_workflow_run_repository",
return_value=mock_repo,
) as repository_factory:
repo = mem.workflow_run_repo
assert repo is mock_repo
assert mem._workflow_run_repo is mock_repo
session_factory = repository_factory.call_args.args[0]
with session_factory() as session:
assert isinstance(session, Session)
assert session.get_bind() is database.engine
def test_workflow_run_repo_cached_after_first_access(self):
conv = _make_conversation()
mi = _make_model_instance()
@@ -410,7 +516,7 @@ class TestBuildPromptMessageWithFiles:
)
@pytest.mark.parametrize("mode", [AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
def test_workflow_mode_workflow_not_found_raises(self, mode):
def test_workflow_mode_workflow_not_found_raises(self, mode, database: Database):
"""Raises ValueError when Workflow lookup returns None."""
conv = _make_conversation(mode)
conv.app = MagicMock()
@@ -422,22 +528,17 @@ class TestBuildPromptMessageWithFiles:
mem._workflow_run_repo = MagicMock()
mem._workflow_run_repo.get_workflow_run_by_id.return_value = mock_workflow_run
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
):
mock_db.session.scalar.return_value = None # workflow not found
with pytest.raises(ValueError, match="Workflow not found"):
mem._build_prompt_message_with_files(
message_files=[],
text_content="text",
message=_make_message(),
app_record=MagicMock(),
is_user_message=True,
)
with pytest.raises(ValueError, match="Workflow not found"):
mem._build_prompt_message_with_files(
message_files=[],
text_content="text",
message=_make_message(),
app_record=MagicMock(),
is_user_message=True,
)
@pytest.mark.parametrize("mode", [AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
def test_workflow_mode_success_no_files_user(self, mode):
def test_workflow_mode_success_no_files_user(self, mode, database: Database):
"""Happy path: workflow mode, no message files → plain UserPromptMessage."""
conv = _make_conversation(mode)
conv.app = MagicMock()
@@ -445,22 +546,16 @@ class TestBuildPromptMessageWithFiles:
mock_workflow_run = MagicMock()
mock_workflow_run.workflow_id = str(uuid4())
mock_workflow = MagicMock()
mock_workflow.features_dict = {}
workflow = _persist_workflow(database, workflow_id=mock_workflow_run.workflow_id)
mem = TokenBufferMemory(conversation=conv, model_instance=_make_model_instance())
mem._workflow_run_repo = MagicMock()
mem._workflow_run_repo.get_workflow_run_by_id.return_value = mock_workflow_run
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch(
"core.memory.token_buffer_memory.FileUploadConfigManager.convert",
return_value=None,
),
with patch(
"core.memory.token_buffer_memory.FileUploadConfigManager.convert",
return_value=None,
):
mock_db.session.scalar.return_value = mock_workflow
result = mem._build_prompt_message_with_files(
message_files=[],
text_content="wf text",
@@ -471,6 +566,7 @@ class TestBuildPromptMessageWithFiles:
assert isinstance(result, UserPromptMessage)
assert result.content == "wf text"
assert database.session.get(Workflow, workflow.id) is workflow
# ------------------------------------------------------------------
# Invalid mode
@@ -498,417 +594,140 @@ class TestBuildPromptMessageWithFiles:
class TestGetHistoryPromptMessages:
"""Tests for get_history_prompt_messages."""
"""Tests for persisted history retrieval, file batching, and pruning."""
def _make_memory(self, mode: AppMode = AppMode.CHAT) -> TokenBufferMemory:
conv = _make_conversation(mode)
conv.app = MagicMock()
return TokenBufferMemory(conversation=conv, model_instance=_make_model_instance())
def test_returns_empty_when_no_messages(self):
def test_returns_empty_when_no_messages(self, database: Database) -> None:
assert self._make_memory().get_history_prompt_messages() == []
def test_skips_newest_message_without_answer(self, database: Database) -> None:
mem = self._make_memory()
with patch("core.memory.token_buffer_memory.db") as mock_db:
mock_db.session.scalars.return_value.all.return_value = []
result = mem.get_history_prompt_messages()
assert result == []
message = _persist_message(database, mem.conversation.id, answer="", answer_tokens=0)
def test_skips_first_message_without_answer(self):
"""The newest message (index 0 after extraction) without answer and tokens==0 is skipped."""
assert mem.get_history_prompt_messages() == []
assert database.session.get(Message, message.id) is message
def test_message_with_answer_returns_user_and_assistant_prompts(self, database: Database) -> None:
mem = self._make_memory()
_persist_message(database, mem.conversation.id, query="My query", answer="My answer", answer_tokens=10)
msg_no_answer = _make_message(answer="", answer_tokens=0)
msg_no_answer.parent_message_id = None # ensures extract_thread_messages returns it
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch(
"core.memory.token_buffer_memory.extract_thread_messages",
return_value=[msg_no_answer],
),
):
mock_db.session.scalars.return_value.all.side_effect = [
[msg_no_answer], # first call: messages query
[], # second call: user files query (never hit, but safe)
]
result = mem.get_history_prompt_messages()
assert result == []
def test_message_with_answer_not_skipped(self):
"""A message with a non-empty answer is NOT popped."""
mem = self._make_memory()
msg = _make_message(answer="some answer", answer_tokens=10)
msg.parent_message_id = None
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch(
"core.memory.token_buffer_memory.extract_thread_messages",
return_value=[msg],
),
patch(
"core.memory.token_buffer_memory.FileUploadConfigManager.convert",
return_value=None,
),
):
# user files query → empty; assistant files query → empty
mock_db.session.scalars.return_value.all.return_value = []
result = mem.get_history_prompt_messages()
assert len(result) == 2 # one user + one assistant
def test_message_limit_default_is_500(self):
"""When message_limit is None the stmt is limited to 500."""
mem = self._make_memory()
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch("core.memory.token_buffer_memory.select") as mock_select,
patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=[]),
):
mock_stmt = MagicMock()
mock_select.return_value.where.return_value.order_by.return_value = mock_stmt
mock_stmt.limit.return_value = mock_stmt
mock_db.session.scalars.return_value.all.return_value = []
mem.get_history_prompt_messages(message_limit=None)
mock_stmt.limit.assert_called_with(500)
def test_message_limit_clipped_to_500(self):
"""A message_limit > 500 is clamped to 500."""
mem = self._make_memory()
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch("core.memory.token_buffer_memory.select") as mock_select,
patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=[]),
):
mock_stmt = MagicMock()
mock_select.return_value.where.return_value.order_by.return_value = mock_stmt
mock_stmt.limit.return_value = mock_stmt
mock_db.session.scalars.return_value.all.return_value = []
mem.get_history_prompt_messages(message_limit=9999)
mock_stmt.limit.assert_called_with(500)
def test_message_limit_positive_used(self):
"""A positive message_limit < 500 is used as-is."""
mem = self._make_memory()
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch("core.memory.token_buffer_memory.select") as mock_select,
patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=[]),
):
mock_stmt = MagicMock()
mock_select.return_value.where.return_value.order_by.return_value = mock_stmt
mock_stmt.limit.return_value = mock_stmt
mock_db.session.scalars.return_value.all.return_value = []
mem.get_history_prompt_messages(message_limit=10)
mock_stmt.limit.assert_called_with(10)
def test_message_limit_zero_uses_default(self):
"""message_limit=0 triggers the else branch → default 500."""
mem = self._make_memory()
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch("core.memory.token_buffer_memory.select") as mock_select,
patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=[]),
):
mock_stmt = MagicMock()
mock_select.return_value.where.return_value.order_by.return_value = mock_stmt
mock_stmt.limit.return_value = mock_stmt
mock_db.session.scalars.return_value.all.return_value = []
mem.get_history_prompt_messages(message_limit=0)
mock_stmt.limit.assert_called_with(500)
def test_user_files_cause_build_with_files_call(self):
"""When user_files is non-empty _build_prompt_message_with_files is invoked."""
mem = self._make_memory()
msg = _make_message()
msg.parent_message_id = None
mock_user_file = MagicMock()
mock_user_file.message_id = msg.id # must match so batched grouping keys it to this message
mock_user_prompt = UserPromptMessage(content="from build")
mock_assistant_prompt = AssistantPromptMessage(content="answer")
call_count = {"n": 0}
def scalars_side_effect(stmt):
r = MagicMock()
if call_count["n"] == 0:
# messages query
r.all.return_value = [msg]
elif call_count["n"] == 1:
# user files
r.all.return_value = [mock_user_file]
else:
# assistant files
r.all.return_value = []
call_count["n"] += 1
return r
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch(
"core.memory.token_buffer_memory.extract_thread_messages",
return_value=[msg],
),
patch.object(
mem,
"_build_prompt_message_with_files",
side_effect=[mock_user_prompt, mock_assistant_prompt],
) as mock_build,
patch(
"core.memory.token_buffer_memory.FileUploadConfigManager.convert",
return_value=None,
),
):
mock_db.session.scalars.side_effect = scalars_side_effect
result = mem.get_history_prompt_messages()
assert mock_build.call_count >= 1
# First call should be user message
first_call_kwargs = mock_build.call_args_list[0][1]
assert first_call_kwargs["is_user_message"] is True
def test_assistant_files_cause_build_with_files_call(self):
"""When assistant_files is non-empty, build is called with is_user_message=False."""
mem = self._make_memory()
msg = _make_message()
msg.parent_message_id = None
mock_assistant_file = MagicMock()
mock_assistant_file.message_id = msg.id # must match so batched grouping keys it to this message
mock_user_prompt = UserPromptMessage(content="query")
mock_assistant_prompt = AssistantPromptMessage(content="built")
call_count = {"n": 0}
def scalars_side_effect(stmt):
r = MagicMock()
if call_count["n"] == 0:
r.all.return_value = [msg]
elif call_count["n"] == 1:
r.all.return_value = [] # no user files
else:
r.all.return_value = [mock_assistant_file]
call_count["n"] += 1
return r
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch(
"core.memory.token_buffer_memory.extract_thread_messages",
return_value=[msg],
),
patch.object(
mem,
"_build_prompt_message_with_files",
return_value=mock_assistant_prompt,
) as mock_build,
):
mock_db.session.scalars.side_effect = scalars_side_effect
result = mem.get_history_prompt_messages()
mock_build.assert_called_once()
call_kwargs = mock_build.call_args[1]
assert call_kwargs["is_user_message"] is False
def test_message_files_loaded_with_constant_query_count(self):
"""Regression guard against N+1: message files must be batch-loaded.
Regardless of the number of messages in the thread, file loading must use a
constant number of queries (1 messages query + 2 batched file queries),
never 2 queries per message.
"""
mem = self._make_memory()
messages = [_make_message() for _ in range(5)]
for m in messages:
m.parent_message_id = None
scalars_calls = {"n": 0}
def scalars_side_effect(stmt):
r = MagicMock()
# First call returns the thread messages; the batched file queries return none.
r.all.return_value = messages if scalars_calls["n"] == 0 else []
scalars_calls["n"] += 1
return r
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=messages),
patch("core.memory.token_buffer_memory.FileUploadConfigManager.convert", return_value=None),
):
mock_db.session.scalars.side_effect = scalars_side_effect
mem.get_history_prompt_messages()
# 1 (messages) + 2 (batched user/assistant files) = 3, independent of message count.
# Before this fix it would have been 1 + 2 * 5 = 11 (an N+1 pattern).
assert scalars_calls["n"] == 3
def test_token_pruning_removes_oldest_messages(self):
"""If tokens exceed limit, oldest messages are removed until within limit."""
conv = _make_conversation()
conv.app = MagicMock()
# Model returns tokens that decrease only after removing pairs
token_values = [3000, 1500] # first call over limit, second within
mi = MagicMock()
mi.get_llm_num_tokens.side_effect = token_values
mem = TokenBufferMemory(conversation=conv, model_instance=mi)
msg = _make_message()
msg.parent_message_id = None
call_count = {"n": 0}
def scalars_side_effect(stmt):
r = MagicMock()
if call_count["n"] == 0:
r.all.return_value = [msg]
else:
r.all.return_value = []
call_count["n"] += 1
return r
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch(
"core.memory.token_buffer_memory.extract_thread_messages",
return_value=[msg],
),
patch(
"core.memory.token_buffer_memory.FileUploadConfigManager.convert",
return_value=None,
),
):
mock_db.session.scalars.side_effect = scalars_side_effect
result = mem.get_history_prompt_messages(max_token_limit=2000)
# After pruning, we should have fewer than the 2 initial messages
assert len(result) <= 1
def test_token_pruning_stops_at_single_message(self):
"""Pruning stops when only 1 message remains (to prevent empty list)."""
conv = _make_conversation()
conv.app = MagicMock()
# Always over limit
mi = MagicMock()
mi.get_llm_num_tokens.return_value = 99999
mem = TokenBufferMemory(conversation=conv, model_instance=mi)
msg = _make_message()
msg.parent_message_id = None
call_count = {"n": 0}
def scalars_side_effect(stmt):
r = MagicMock()
if call_count["n"] == 0:
r.all.return_value = [msg]
else:
r.all.return_value = []
call_count["n"] += 1
return r
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch(
"core.memory.token_buffer_memory.extract_thread_messages",
return_value=[msg],
),
patch(
"core.memory.token_buffer_memory.FileUploadConfigManager.convert",
return_value=None,
),
):
mock_db.session.scalars.side_effect = scalars_side_effect
result = mem.get_history_prompt_messages(max_token_limit=1)
# At least 1 message should remain
assert len(result) >= 1
def test_no_pruning_when_within_limit(self):
"""When tokens ≤ limit, no pruning occurs."""
mem = self._make_memory()
mem.model_instance.get_llm_num_tokens.return_value = 50 # well under default 2000
msg = _make_message()
msg.parent_message_id = None
call_count = {"n": 0}
def scalars_side_effect(stmt):
r = MagicMock()
if call_count["n"] == 0:
r.all.return_value = [msg]
else:
r.all.return_value = []
call_count["n"] += 1
return r
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch(
"core.memory.token_buffer_memory.extract_thread_messages",
return_value=[msg],
),
patch(
"core.memory.token_buffer_memory.FileUploadConfigManager.convert",
return_value=None,
),
):
mock_db.session.scalars.side_effect = scalars_side_effect
result = mem.get_history_prompt_messages(max_token_limit=2000)
assert len(result) == 2 # user + assistant
def test_plain_user_and_assistant_messages_returned(self):
"""Without files, plain UserPromptMessage and AssistantPromptMessage appear."""
mem = self._make_memory()
msg = _make_message(answer="My answer")
msg.query = "My query"
msg.parent_message_id = None
call_count = {"n": 0}
def scalars_side_effect(stmt):
r = MagicMock()
if call_count["n"] == 0:
r.all.return_value = [msg]
else:
r.all.return_value = []
call_count["n"] += 1
return r
with (
patch("core.memory.token_buffer_memory.db") as mock_db,
patch(
"core.memory.token_buffer_memory.extract_thread_messages",
return_value=[msg],
),
patch(
"core.memory.token_buffer_memory.FileUploadConfigManager.convert",
return_value=None,
),
):
mock_db.session.scalars.side_effect = scalars_side_effect
result = mem.get_history_prompt_messages()
result = mem.get_history_prompt_messages()
assert len(result) == 2
user_msg, ai_msg = result
assert isinstance(user_msg, UserPromptMessage)
assert user_msg.content == "My query"
assert isinstance(ai_msg, AssistantPromptMessage)
assert ai_msg.content == "My answer"
assert isinstance(result[0], UserPromptMessage)
assert result[0].content == "My query"
assert isinstance(result[1], AssistantPromptMessage)
assert result[1].content == "My answer"
def test_history_is_conversation_scoped(self, database: Database) -> None:
mem = self._make_memory()
_persist_message(database, mem.conversation.id, answer="visible")
_persist_message(database, "other-conversation", answer="hidden")
result = mem.get_history_prompt_messages()
assert [prompt.content for prompt in result] == ["user query", "visible"]
@pytest.mark.parametrize(
("message_limit", "expected_limit"),
[(None, 500), (9999, 500), (10, 10), (0, 500)],
)
def test_message_limit_is_applied_to_executable_query(
self,
database: Database,
message_limit: int | None,
expected_limit: int,
) -> None:
mem = self._make_memory()
before = len(database.statements)
mem.get_history_prompt_messages(message_limit=message_limit)
statements = database.statements[before:]
assert len(statements) == 1
sql, parameters = statements[0]
assert "LIMIT" in sql
assert expected_limit in parameters
@pytest.mark.parametrize(
("belongs_to", "is_user_message"),
[
(MessageFileBelongsTo.USER, True),
(None, True),
(MessageFileBelongsTo.ASSISTANT, False),
],
)
def test_message_files_use_persisted_ownership(
self,
database: Database,
belongs_to: MessageFileBelongsTo | None,
is_user_message: bool,
) -> None:
mem = self._make_memory()
message = _persist_message(database, mem.conversation.id)
message_file = _persist_message_file(database, message, belongs_to=belongs_to)
built_prompt = (
UserPromptMessage(content="built user")
if is_user_message
else AssistantPromptMessage(content="built assistant")
)
with patch.object(mem, "_build_prompt_message_with_files", return_value=built_prompt) as build_prompt:
result = mem.get_history_prompt_messages()
build_prompt.assert_called_once()
assert build_prompt.call_args.kwargs["message_files"] == [message_file]
assert build_prompt.call_args.kwargs["is_user_message"] is is_user_message
assert built_prompt in result
def test_message_files_are_batch_loaded_with_constant_query_count(self, database: Database) -> None:
mem = self._make_memory()
base_time = datetime.now(UTC).replace(tzinfo=None)
messages = [
_persist_message(
database,
mem.conversation.id,
query=f"query-{index}",
answer=f"answer-{index}",
created_at=base_time + timedelta(seconds=index),
)
for index in range(5)
]
before = len(database.statements)
with patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=messages):
result = mem.get_history_prompt_messages()
selects = [sql for sql, _ in database.statements[before:] if sql.lstrip().upper().startswith("SELECT")]
assert len(selects) == 3
assert len(result) == 10
@pytest.mark.parametrize(
("token_values", "max_token_limit", "expected_length"),
[
([3000, 1500], 2000, 1),
([99999, 99999], 1, 1),
([50], 2000, 2),
],
)
def test_token_pruning_uses_persisted_history(
self,
database: Database,
token_values: list[int],
max_token_limit: int,
expected_length: int,
) -> None:
mem = self._make_memory()
mem.model_instance.get_llm_num_tokens.side_effect = token_values
_persist_message(database, mem.conversation.id)
result = mem.get_history_prompt_messages(max_token_limit=max_token_limit)
assert len(result) == expected_length
# ===========================================================================
@@ -2,13 +2,24 @@ from unittest.mock import MagicMock, patch
import pytest
from pydantic import ValidationError
from sqlalchemy.orm import Session
import core.moderation.api.api as moderation_module
from core.extension.api_based_extension_requestor import APIBasedExtensionPoint
from core.moderation.api.api import ApiModeration, ModerationInputParams, ModerationOutputParams
from core.moderation.base import ModerationAction, ModerationInputsResult, ModerationOutputsResult
from models.api_based_extension import APIBasedExtension
class _DatabaseBinding:
"""Expose the real SQLite session used by extension lookup."""
session: Session
def __init__(self, session: Session) -> None:
self.session = session
class TestApiModeration:
@pytest.fixture
def api_config(self):
@@ -165,17 +176,27 @@ class TestApiModeration:
with pytest.raises(ValueError, match="API-based Extension not found"):
api_moderation._get_config_by_requestor(APIBasedExtensionPoint.APP_MODERATION_INPUT, {})
@patch("core.moderation.api.api.db.session.scalar")
def test_get_api_based_extension(self, mock_scalar):
mock_ext = MagicMock(spec=APIBasedExtension)
mock_scalar.return_value = mock_ext
@pytest.mark.parametrize("sqlite_session", [(APIBasedExtension,)], indirect=True)
def test_get_api_based_extension(self, sqlite_session: Session, monkeypatch: pytest.MonkeyPatch) -> None:
target = APIBasedExtension(
tenant_id="tenant-1",
name="Target extension",
api_endpoint="https://example.com/moderate",
api_key="encrypted-key",
)
target.id = "ext-1"
other_tenant = APIBasedExtension(
tenant_id="tenant-2",
name="Other extension",
api_endpoint="https://example.com/other",
api_key="other-key",
)
other_tenant.id = "ext-2"
sqlite_session.add_all((target, other_tenant))
sqlite_session.commit()
monkeypatch.setattr(moderation_module, "db", _DatabaseBinding(sqlite_session))
result = ApiModeration._get_api_based_extension("tenant-1", "ext-1")
assert result == mock_ext
mock_scalar.assert_called_once()
# Verify the call has the correct filters
args, kwargs = mock_scalar.call_args
stmt = args[0]
# We can't easily inspect the statement without complex sqlalchemy tricks,
# but calling it is usually enough for unit tests if we mock the result.
assert result is target
assert ApiModeration._get_api_based_extension("tenant-1", "ext-2") is None
@@ -7,36 +7,212 @@ Covers:
- TraceTask._get_user_id_from_metadata
"""
from unittest.mock import MagicMock, patch
import uuid
from collections.abc import Iterator
from contextlib import contextmanager
from unittest.mock import PropertyMock, patch
import pytest
from sqlalchemy import Engine, event
from sqlalchemy.orm import Session
from core.tools.entities.tool_entities import ApiProviderSchemaType
from extensions.ext_database import db
from graphon.model_runtime.entities.model_entities import ModelType
from models.account import Tenant
from models.base import TypeBase
from models.model import App, AppMode, IconType
from models.provider import Provider, ProviderCredential, ProviderModel, ProviderModelCredential, ProviderType
from models.tools import ApiToolProvider, BuiltinToolProvider, MCPToolProvider, WorkflowToolProvider
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_db_and_session_patches(scalar_side_effect=None, scalar_return_value=None):
"""Return (mock_db, cm, session) ready to patch 'core.ops.ops_trace_manager.db'
and 'core.ops.ops_trace_manager.Session'.
@pytest.fixture
def orm_session(sqlite_engine: Engine) -> Iterator[Session]:
models = (
App,
Tenant,
Provider,
ProviderCredential,
ProviderModel,
ProviderModelCredential,
BuiltinToolProvider,
ApiToolProvider,
WorkflowToolProvider,
MCPToolProvider,
)
tables = [model.metadata.tables[model.__tablename__] for model in models]
TypeBase.metadata.create_all(sqlite_engine, tables=tables)
Provide either scalar_side_effect (list, for multiple calls) or
scalar_return_value (single value).
"""
mock_db = MagicMock()
mock_db.engine = MagicMock()
with patch.object(type(db), "engine", new_callable=PropertyMock, return_value=sqlite_engine):
with Session(sqlite_engine, expire_on_commit=False) as session:
yield session
session = MagicMock()
if scalar_side_effect is not None:
session.scalar.side_effect = scalar_side_effect
def _persist_app(session: Session, *, tenant_id: str, name: str = "MyApp") -> App:
app = App(
id=str(uuid.uuid4()),
tenant_id=tenant_id,
name=name,
mode=AppMode.WORKFLOW,
icon_type=IconType.EMOJI,
icon="workflow",
icon_background="#FFFFFF",
enable_site=True,
enable_api=False,
)
session.add(app)
session.commit()
return app
def _persist_tenant(session: Session, *, name: str = "MyWorkspace") -> Tenant:
tenant = Tenant(name=name)
session.add(tenant)
session.commit()
return tenant
def _persist_tool_provider(
session: Session, provider_type: str
) -> BuiltinToolProvider | ApiToolProvider | WorkflowToolProvider | MCPToolProvider:
tenant_id = str(uuid.uuid4())
user_id = str(uuid.uuid4())
if provider_type in {"builtin", "plugin"}:
provider = BuiltinToolProvider(
name="CredentialA",
tenant_id=tenant_id,
user_id=user_id,
provider="test/provider",
)
elif provider_type == "api":
provider = ApiToolProvider(
name="CredentialA",
icon="icon.svg",
schema="{}",
schema_type_str=ApiProviderSchemaType.OPENAPI,
user_id=user_id,
tenant_id=tenant_id,
description="API provider",
tools_str="[]",
credentials_str="{}",
)
elif provider_type == "workflow":
provider = WorkflowToolProvider(
name="CredentialA",
label="CredentialA",
icon="icon.svg",
app_id=str(uuid.uuid4()),
version="1",
user_id=user_id,
tenant_id=tenant_id,
description="Workflow provider",
)
elif provider_type == "mcp":
provider = MCPToolProvider(
name="CredentialA",
server_identifier="credential-a",
server_url="https://example.com/mcp",
server_url_hash="credential-a-hash",
icon="icon.svg",
tenant_id=tenant_id,
user_id=user_id,
)
else:
session.scalar.return_value = scalar_return_value
raise ValueError(f"unsupported provider type: {provider_type}")
cm = MagicMock()
cm.__enter__ = MagicMock(return_value=session)
cm.__exit__ = MagicMock(return_value=False)
session.add(provider)
session.commit()
return provider
return mock_db, cm, session
def _persist_provider_credential(
session: Session,
*,
tenant_id: str,
credential_name: str = "ProvCredName",
) -> ProviderCredential:
credential = ProviderCredential(
tenant_id=tenant_id,
provider_name="openai",
credential_name=credential_name,
encrypted_config="{}",
)
session.add(credential)
session.commit()
return credential
def _persist_model_credential(
session: Session,
*,
tenant_id: str,
credential_name: str = "ModelCredName",
) -> ProviderModelCredential:
credential = ProviderModelCredential(
tenant_id=tenant_id,
provider_name="openai",
model_name="gpt-4",
model_type=ModelType.LLM,
credential_name=credential_name,
encrypted_config="{}",
)
session.add(credential)
session.commit()
return credential
def _persist_provider(
session: Session,
*,
tenant_id: str,
credential_id: str | None,
) -> Provider:
provider = Provider(
tenant_id=tenant_id,
provider_name="openai",
provider_type=ProviderType.CUSTOM,
credential_id=credential_id,
)
session.add(provider)
session.commit()
return provider
def _persist_provider_model(
session: Session,
*,
tenant_id: str,
credential_id: str | None,
) -> ProviderModel:
model = ProviderModel(
tenant_id=tenant_id,
provider_name="openai",
model_name="gpt-4",
model_type=ModelType.LLM,
credential_id=credential_id,
)
session.add(model)
session.commit()
return model
@contextmanager
def _raise_on_table(engine: Engine, table_name: str) -> Iterator[None]:
"""Raise only when SQL targets the named table, leaving other real lookups intact."""
def fail_target_query(_conn, _cursor, statement, _parameters, _context, _executemany):
if f"FROM {table_name}" in statement:
raise RuntimeError(f"forced failure for {table_name}")
event.listen(engine, "before_cursor_execute", fail_target_query)
try:
yield
finally:
event.remove(engine, "before_cursor_execute", fail_target_query)
# ---------------------------------------------------------------------------
@@ -47,62 +223,42 @@ def _make_db_and_session_patches(scalar_side_effect=None, scalar_return_value=No
class TestLookupAppAndWorkspaceNames:
"""Tests for _lookup_app_and_workspace_names(app_id, tenant_id)."""
def test_both_found(self):
def test_both_found(self, orm_session: Session):
"""Returns (app_name, workspace_name) when both records exist."""
from core.ops.ops_trace_manager import _lookup_app_and_workspace_names
mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=["MyApp", "MyWorkspace"])
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
app_name, workspace_name = _lookup_app_and_workspace_names("app-123", "tenant-456")
tenant = _persist_tenant(orm_session)
app = _persist_app(orm_session, tenant_id=tenant.id)
app_name, workspace_name = _lookup_app_and_workspace_names(app.id, tenant.id)
assert app_name == "MyApp"
assert workspace_name == "MyWorkspace"
def test_app_only_found(self):
def test_app_only_found(self, orm_session: Session):
"""Returns (app_name, '') when tenant record is absent."""
from core.ops.ops_trace_manager import _lookup_app_and_workspace_names
mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=["MyApp", None])
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
app_name, workspace_name = _lookup_app_and_workspace_names("app-123", "tenant-456")
app = _persist_app(orm_session, tenant_id=str(uuid.uuid4()))
app_name, workspace_name = _lookup_app_and_workspace_names(app.id, str(uuid.uuid4()))
assert app_name == "MyApp"
assert workspace_name == ""
def test_tenant_only_found(self):
def test_tenant_only_found(self, orm_session: Session):
"""Returns ('', workspace_name) when app record is absent."""
from core.ops.ops_trace_manager import _lookup_app_and_workspace_names
mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=[None, "MyWorkspace"])
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
app_name, workspace_name = _lookup_app_and_workspace_names("app-123", "tenant-456")
tenant = _persist_tenant(orm_session)
app_name, workspace_name = _lookup_app_and_workspace_names(str(uuid.uuid4()), tenant.id)
assert app_name == ""
assert workspace_name == "MyWorkspace"
def test_neither_found(self):
def test_neither_found(self, orm_session: Session):
"""Returns ('', '') when both DB lookups return None."""
from core.ops.ops_trace_manager import _lookup_app_and_workspace_names
mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=[None, None])
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
app_name, workspace_name = _lookup_app_and_workspace_names("app-123", "tenant-456")
app_name, workspace_name = _lookup_app_and_workspace_names(str(uuid.uuid4()), str(uuid.uuid4()))
assert app_name == ""
assert workspace_name == ""
@@ -111,50 +267,30 @@ class TestLookupAppAndWorkspaceNames:
"""Returns ('', '') immediately when both IDs are None — no DB access."""
from core.ops.ops_trace_manager import _lookup_app_and_workspace_names
mock_db = MagicMock()
mock_session_cls = MagicMock()
app_name, workspace_name = _lookup_app_and_workspace_names(None, None)
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", mock_session_cls),
):
app_name, workspace_name = _lookup_app_and_workspace_names(None, None)
mock_session_cls.assert_not_called()
assert app_name == ""
assert workspace_name == ""
def test_app_id_none_only_queries_tenant(self):
def test_app_id_none_only_queries_tenant(self, orm_session: Session):
"""When app_id is None, only the tenant query is issued."""
from core.ops.ops_trace_manager import _lookup_app_and_workspace_names
mock_db, cm, session = _make_db_and_session_patches(scalar_return_value="OnlyWorkspace")
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
app_name, workspace_name = _lookup_app_and_workspace_names(None, "tenant-456")
tenant = _persist_tenant(orm_session, name="OnlyWorkspace")
app_name, workspace_name = _lookup_app_and_workspace_names(None, tenant.id)
assert app_name == ""
assert workspace_name == "OnlyWorkspace"
assert session.scalar.call_count == 1
def test_tenant_id_none_only_queries_app(self):
def test_tenant_id_none_only_queries_app(self, orm_session: Session):
"""When tenant_id is None, only the app query is issued."""
from core.ops.ops_trace_manager import _lookup_app_and_workspace_names
mock_db, cm, session = _make_db_and_session_patches(scalar_return_value="OnlyApp")
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
app_name, workspace_name = _lookup_app_and_workspace_names("app-123", None)
app = _persist_app(orm_session, tenant_id=str(uuid.uuid4()), name="OnlyApp")
app_name, workspace_name = _lookup_app_and_workspace_names(app.id, None)
assert app_name == "OnlyApp"
assert workspace_name == ""
assert session.scalar.call_count == 1
# ---------------------------------------------------------------------------
@@ -166,32 +302,20 @@ class TestLookupCredentialName:
"""Tests for _lookup_credential_name(credential_id, provider_type)."""
@pytest.mark.parametrize("provider_type", ["builtin", "plugin", "api", "workflow", "mcp"])
def test_known_provider_types_return_name(self, provider_type):
def test_known_provider_types_return_name(self, provider_type: str, orm_session: Session):
"""Each valid provider_type results in a DB query and returns the credential name."""
from core.ops.ops_trace_manager import _lookup_credential_name
mock_db, cm, session = _make_db_and_session_patches(scalar_return_value="CredentialA")
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
result = _lookup_credential_name("cred-123", provider_type)
provider = _persist_tool_provider(orm_session, provider_type)
result = _lookup_credential_name(provider.id, provider_type)
assert result == "CredentialA"
session.scalar.assert_called_once()
def test_credential_not_found_returns_empty_string(self):
def test_credential_not_found_returns_empty_string(self, orm_session: Session):
"""Returns '' when DB yields None for the given credential_id."""
from core.ops.ops_trace_manager import _lookup_credential_name
mock_db, cm, _session = _make_db_and_session_patches(scalar_return_value=None)
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
result = _lookup_credential_name("cred-999", "api")
result = _lookup_credential_name(str(uuid.uuid4()), "api")
assert result == ""
@@ -199,48 +323,24 @@ class TestLookupCredentialName:
"""Returns '' immediately for an unrecognised provider_type — no DB access."""
from core.ops.ops_trace_manager import _lookup_credential_name
mock_db = MagicMock()
mock_session_cls = MagicMock()
result = _lookup_credential_name(str(uuid.uuid4()), "unknown_type")
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", mock_session_cls),
):
result = _lookup_credential_name("cred-123", "unknown_type")
mock_session_cls.assert_not_called()
assert result == ""
def test_none_credential_id_returns_empty_string_without_db(self):
"""Returns '' immediately when credential_id is None — no DB access."""
from core.ops.ops_trace_manager import _lookup_credential_name
mock_db = MagicMock()
mock_session_cls = MagicMock()
result = _lookup_credential_name(None, "api")
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", mock_session_cls),
):
result = _lookup_credential_name(None, "api")
mock_session_cls.assert_not_called()
assert result == ""
def test_none_provider_type_returns_empty_string_without_db(self):
"""Returns '' immediately when provider_type is None — no DB access."""
from core.ops.ops_trace_manager import _lookup_credential_name
mock_db = MagicMock()
mock_session_cls = MagicMock()
result = _lookup_credential_name(str(uuid.uuid4()), None)
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", mock_session_cls),
):
result = _lookup_credential_name("cred-123", None)
mock_session_cls.assert_not_called()
assert result == ""
def test_builtin_and_plugin_map_to_same_model(self):
@@ -281,106 +381,78 @@ class TestLookupCredentialName:
class TestLookupLlmCredentialInfo:
"""Tests for _lookup_llm_credential_info(tenant_id, provider, model, model_type)."""
def _provider_record(self, credential_id: str | None = None) -> MagicMock:
record = MagicMock()
record.credential_id = credential_id
return record
def _model_record(self, credential_id: str | None = None) -> MagicMock:
record = MagicMock()
record.credential_id = credential_id
return record
def test_model_level_credential_found(self):
def test_model_level_credential_found(self, orm_session: Session):
"""Returns model-level credential_id and name when ProviderModel has a credential."""
from core.ops.ops_trace_manager import _lookup_llm_credential_info
provider_record = self._provider_record(credential_id=None)
model_record = self._model_record(credential_id="model-cred-id")
tenant_id = str(uuid.uuid4())
model_credential = _persist_model_credential(orm_session, tenant_id=tenant_id)
_persist_provider(orm_session, tenant_id=tenant_id, credential_id=None)
_persist_provider_model(orm_session, tenant_id=tenant_id, credential_id=model_credential.id)
# scalar calls: (1) Provider, (2) ProviderModel, (3) ProviderModelCredential.credential_name
mock_db, cm, _session = _make_db_and_session_patches(
scalar_side_effect=[provider_record, model_record, "ModelCredName"]
decoy_tenant_id = str(uuid.uuid4())
decoy_credential = _persist_model_credential(
orm_session,
tenant_id=decoy_tenant_id,
credential_name="WrongTenantCredential",
)
_persist_provider(orm_session, tenant_id=decoy_tenant_id, credential_id=None)
_persist_provider_model(orm_session, tenant_id=decoy_tenant_id, credential_id=decoy_credential.id)
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")
assert cred_id == "model-cred-id"
assert cred_id == model_credential.id
assert cred_name == "ModelCredName"
def test_provider_level_fallback_when_no_model_credential(self):
def test_provider_level_fallback_when_no_model_credential(self, orm_session: Session):
"""Falls back to provider-level credential when ProviderModel has no credential_id."""
from core.ops.ops_trace_manager import _lookup_llm_credential_info
provider_record = self._provider_record(credential_id="prov-cred-id")
model_record = self._model_record(credential_id=None)
tenant_id = str(uuid.uuid4())
provider_credential = _persist_provider_credential(orm_session, tenant_id=tenant_id)
_persist_provider(orm_session, tenant_id=tenant_id, credential_id=provider_credential.id)
_persist_provider_model(orm_session, tenant_id=tenant_id, credential_id=None)
# scalar calls: (1) Provider, (2) ProviderModel (no cred), (3) ProviderCredential.credential_name
mock_db, cm, _session = _make_db_and_session_patches(
scalar_side_effect=[provider_record, model_record, "ProvCredName"]
)
cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
assert cred_id == "prov-cred-id"
assert cred_id == provider_credential.id
assert cred_name == "ProvCredName"
def test_provider_level_fallback_when_no_model_record(self):
def test_provider_level_fallback_when_no_model_record(self, orm_session: Session):
"""Falls back to provider-level credential when no ProviderModel row exists."""
from core.ops.ops_trace_manager import _lookup_llm_credential_info
provider_record = self._provider_record(credential_id="prov-cred-id")
tenant_id = str(uuid.uuid4())
provider_credential = _persist_provider_credential(orm_session, tenant_id=tenant_id)
_persist_provider(orm_session, tenant_id=tenant_id, credential_id=provider_credential.id)
# scalar calls: (1) Provider, (2) ProviderModel → None, (3) ProviderCredential.credential_name
mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=[provider_record, None, "ProvCredName"])
cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
assert cred_id == "prov-cred-id"
assert cred_id == provider_credential.id
assert cred_name == "ProvCredName"
def test_no_model_arg_uses_provider_level_only(self):
def test_no_model_arg_uses_provider_level_only(self, orm_session: Session):
"""When model is None, skips ProviderModel query and uses provider credential."""
from core.ops.ops_trace_manager import _lookup_llm_credential_info
provider_record = self._provider_record(credential_id="prov-cred-id")
tenant_id = str(uuid.uuid4())
provider_credential = _persist_provider_credential(orm_session, tenant_id=tenant_id)
_persist_provider(orm_session, tenant_id=tenant_id, credential_id=provider_credential.id)
# scalar calls: (1) Provider, (2) ProviderCredential.credential_name — no ProviderModel
mock_db, cm, session = _make_db_and_session_patches(scalar_side_effect=[provider_record, "ProvCredName"])
cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", None)
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", None)
assert cred_id == "prov-cred-id"
assert cred_id == provider_credential.id
assert cred_name == "ProvCredName"
assert session.scalar.call_count == 2
def test_provider_not_found_returns_none_and_empty(self):
def test_provider_not_found_returns_none_and_empty(self, orm_session: Session):
"""Returns (None, '') when Provider record does not exist."""
from core.ops.ops_trace_manager import _lookup_llm_credential_info
mock_db, cm, _session = _make_db_and_session_patches(scalar_return_value=None)
other_tenant_id = str(uuid.uuid4())
_persist_provider(orm_session, tenant_id=other_tenant_id, credential_id=None)
tenant_id = str(uuid.uuid4())
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")
assert cred_id is None
assert cred_name == ""
@@ -389,16 +461,8 @@ class TestLookupLlmCredentialInfo:
"""Returns (None, '') immediately when tenant_id is None — no DB access."""
from core.ops.ops_trace_manager import _lookup_llm_credential_info
mock_db = MagicMock()
mock_session_cls = MagicMock()
cred_id, cred_name = _lookup_llm_credential_info(None, "openai", "gpt-4")
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", mock_session_cls),
):
cred_id, cred_name = _lookup_llm_credential_info(None, "openai", "gpt-4")
mock_session_cls.assert_not_called()
assert cred_id is None
assert cred_name == ""
@@ -406,69 +470,46 @@ class TestLookupLlmCredentialInfo:
"""Returns (None, '') immediately when provider is None — no DB access."""
from core.ops.ops_trace_manager import _lookup_llm_credential_info
mock_db = MagicMock()
mock_session_cls = MagicMock()
cred_id, cred_name = _lookup_llm_credential_info(str(uuid.uuid4()), None, "gpt-4")
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", mock_session_cls),
):
cred_id, cred_name = _lookup_llm_credential_info("tenant-1", None, "gpt-4")
mock_session_cls.assert_not_called()
assert cred_id is None
assert cred_name == ""
def test_db_error_on_outer_query_returns_none_and_empty(self):
def test_db_error_on_outer_query_returns_none_and_empty(self, orm_session: Session, sqlite_engine: Engine):
"""Returns (None, '') and logs a warning when the outer DB query raises."""
from core.ops.ops_trace_manager import _lookup_llm_credential_info
mock_db, cm, session = _make_db_and_session_patches()
session.scalar.side_effect = Exception("DB connection failed")
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
with _raise_on_table(sqlite_engine, "providers"):
cred_id, cred_name = _lookup_llm_credential_info(str(uuid.uuid4()), "openai", "gpt-4")
assert cred_id is None
assert cred_name == ""
def test_credential_name_lookup_failure_returns_id_with_empty_name(self):
def test_credential_name_lookup_failure_returns_id_with_empty_name(
self, orm_session: Session, sqlite_engine: Engine
):
"""When credential name sub-query fails, returns cred_id but '' for name."""
from core.ops.ops_trace_manager import _lookup_llm_credential_info
provider_record = self._provider_record(credential_id="prov-cred-id")
tenant_id = str(uuid.uuid4())
provider_credential = _persist_provider_credential(orm_session, tenant_id=tenant_id)
_persist_provider(orm_session, tenant_id=tenant_id, credential_id=provider_credential.id)
# Provider found, no model record, then name lookup raises
mock_db, cm, _session = _make_db_and_session_patches(
scalar_side_effect=[provider_record, None, Exception("deleted")]
)
with _raise_on_table(sqlite_engine, "provider_credentials"):
cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
assert cred_id == "prov-cred-id"
assert cred_id == provider_credential.id
assert cred_name == ""
def test_no_credential_on_provider_or_model_returns_none_id(self):
def test_no_credential_on_provider_or_model_returns_none_id(self, orm_session: Session):
"""Returns (None, '') when neither provider nor model has a credential_id."""
from core.ops.ops_trace_manager import _lookup_llm_credential_info
provider_record = self._provider_record(credential_id=None)
model_record = self._model_record(credential_id=None)
tenant_id = str(uuid.uuid4())
_persist_provider(orm_session, tenant_id=tenant_id, credential_id=None)
_persist_provider_model(orm_session, tenant_id=tenant_id, credential_id=None)
mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=[provider_record, model_record])
with (
patch("core.ops.ops_trace_manager.db", mock_db),
patch("core.ops.ops_trace_manager.Session", return_value=cm),
):
cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")
assert cred_id is None
assert cred_name == ""
+47 -11
View File
@@ -1,9 +1,11 @@
import re
from datetime import datetime
from unittest.mock import MagicMock, patch
from decimal import Decimal
import pytest
from sqlalchemy.orm import Session
import core.ops.utils as utils_module
from core.ops.utils import (
filter_none_values,
generate_dotted_order,
@@ -15,6 +17,42 @@ from core.ops.utils import (
validate_url,
validate_url_with_path,
)
from models.enums import ConversationFromSource
from models.model import Message
class _DatabaseBinding:
"""Expose the real SQLite session used by the message lookup helper."""
session: Session
def __init__(self, session: Session) -> None:
self.session = session
@pytest.fixture
def message_session(sqlite_session: Session, monkeypatch: pytest.MonkeyPatch) -> Session:
"""Bind the message lookup helper to the shared SQLite test session."""
monkeypatch.setattr(utils_module, "db", _DatabaseBinding(sqlite_session))
return sqlite_session
def _message(message_id: str) -> Message:
message = Message(
id=message_id,
app_id="app-id",
conversation_id="conversation-id",
query="question",
message={"role": "user", "content": "question"},
answer="answer",
message_unit_price=Decimal("0.0001"),
answer_unit_price=Decimal("0.0001"),
currency="USD",
from_source=ConversationFromSource.API,
)
message._inputs = {}
return message
class TestValidateUrl:
@@ -220,22 +258,20 @@ class TestFilterNoneValues:
assert filter_none_values({}) == {}
@pytest.mark.parametrize("sqlite_session", [(Message,)], indirect=True)
class TestGetMessageData:
"""Test cases for get_message_data function"""
@patch("core.ops.utils.db")
@patch("core.ops.utils.Message")
@patch("core.ops.utils.select")
def test_get_message_data(self, mock_select, mock_message, mock_db):
mock_scalar = mock_db.session.scalar
mock_msg_instance = MagicMock()
mock_scalar.return_value = mock_msg_instance
def test_get_message_data(self, message_session: Session):
target = _message("message-id")
unrelated = _message("other-message-id")
message_session.add_all((target, unrelated))
message_session.commit()
result = get_message_data("message-id")
assert result == mock_msg_instance
mock_select.assert_called_once()
mock_scalar.assert_called_once()
assert result is target
assert result.id == "message-id"
class TestMeasureTime:
@@ -1,9 +1,42 @@
from unittest.mock import MagicMock
from datetime import datetime, timedelta
from decimal import Decimal
from uuid import uuid4
import pytest
from sqlalchemy.orm import Session
from constants import UUID_NIL
from core.prompt.utils.extract_thread_messages import extract_thread_messages
from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
from models.enums import ConversationFromSource
from models.model import Message
def _persisted_message(
*,
message_id: str,
conversation_id: str,
parent_message_id: str,
answer: str,
created_at: datetime,
) -> Message:
message = Message(
id=message_id,
app_id="app-id",
conversation_id=conversation_id,
query="question",
message={"role": "user", "content": "question"},
answer=answer,
message_unit_price=Decimal("0.0001"),
answer_unit_price=Decimal("0.0001"),
currency="USD",
from_source=ConversationFromSource.API,
parent_message_id=parent_message_id,
created_at=created_at,
updated_at=created_at,
)
message._inputs = {}
return message
class MockMessage:
@@ -104,33 +137,64 @@ def test_extract_thread_messages_breaks_when_parent_is_none():
assert result[0].id == id2
def test_get_thread_messages_length_excludes_newly_created_empty_answer():
@pytest.mark.parametrize("sqlite_session", [(Message,)], indirect=True)
def test_get_thread_messages_length_excludes_newly_created_empty_answer(sqlite_session: Session):
id1, id2 = str(uuid4()), str(uuid4())
now = datetime.now()
messages = [
MockMessage(id2, id1, answer=""), # newest generated message should be excluded
MockMessage(id1, UUID_NIL, answer="ok"),
_persisted_message(
message_id=id2,
conversation_id="conversation-1",
parent_message_id=id1,
answer="",
created_at=now,
),
_persisted_message(
message_id=id1,
conversation_id="conversation-1",
parent_message_id=UUID_NIL,
answer="ok",
created_at=now - timedelta(seconds=1),
),
_persisted_message(
message_id=str(uuid4()),
conversation_id="other-conversation",
parent_message_id=UUID_NIL,
answer="unrelated",
created_at=now + timedelta(seconds=1),
),
]
sqlite_session.add_all(messages)
sqlite_session.commit()
session = MagicMock()
session.scalars.return_value.all.return_value = messages
length = get_thread_messages_length("conversation-1", session=session)
length = get_thread_messages_length("conversation-1", session=sqlite_session)
assert length == 1
session.scalars.assert_called_once()
def test_get_thread_messages_length_keeps_non_empty_latest_answer():
@pytest.mark.parametrize("sqlite_session", [(Message,)], indirect=True)
def test_get_thread_messages_length_keeps_non_empty_latest_answer(sqlite_session: Session):
id1, id2 = str(uuid4()), str(uuid4())
now = datetime.now()
messages = [
MockMessage(id2, id1, answer="latest-answer"),
MockMessage(id1, UUID_NIL, answer="older-answer"),
_persisted_message(
message_id=id2,
conversation_id="conversation-2",
parent_message_id=id1,
answer="latest-answer",
created_at=now,
),
_persisted_message(
message_id=id1,
conversation_id="conversation-2",
parent_message_id=UUID_NIL,
answer="older-answer",
created_at=now - timedelta(seconds=1),
),
]
sqlite_session.add_all(messages)
sqlite_session.commit()
session = MagicMock()
session.scalars.return_value.all.return_value = messages
length = get_thread_messages_length("conversation-2", session=session)
length = get_thread_messages_length("conversation-2", session=sqlite_session)
assert length == 2
session.scalars.assert_called_once()
@@ -22,6 +22,23 @@ class TestCleanProcessor:
expected = "normalpadding"
assert CleanProcessor.clean(text_with_ufffe, None) == expected
def test_clean_preserves_valid_extended_characters(self):
"""Default cleaning must not strip valid printable characters.
The invalid-symbol filter used to include the UTF-8 bytes of U+FFFE
(0xEF 0xBF 0xBE) inside a character class. On a decoded string those
bytes are the code points U+00EF, U+00BF and U+00BE, i.e. the valid
characters 'ï', '¿' and '¾', so words like "naïve" and Spanish
questions like "¿Cómo?" were being silently corrupted on ingest.
"""
assert CleanProcessor.clean("naïve", None) == "naïve"
assert CleanProcessor.clean("¿Cómo estás?", None) == "¿Cómo estás?"
assert CleanProcessor.clean("¾ cup sugar", None) == "¾ cup sugar"
assert CleanProcessor.clean("￾", None) == "￾"
# The U+FFFE noncharacter is still stripped by its dedicated substitution.
assert CleanProcessor.clean("keep\ufffedrop", None) == "keepdrop"
def test_clean_with_none_process_rule(self):
"""Test cleaning with None process_rule - only default cleaning applied."""
text = "Hello<|World\x00"
@@ -1372,6 +1372,19 @@ class TestIndexingRunnerDocumentCleaning:
assert "\ufffe" not in result
assert "Text with" in result
def test_filter_string_preserves_valid_extended_characters(self):
"""filter_string must keep valid printable characters like 'ï', '¿', '¾'."""
# Arrange
text = "naïve ¿Cómo? ¾ done"
# Act
result = IndexingRunner.filter_string(text)
# Assert
assert result == text
# The U+FFFE noncharacter is still stripped.
assert IndexingRunner.filter_string("keep\ufffedrop") == "keepdrop"
class TestIndexingRunnerSplitter:
"""Unit tests for text splitter configuration.
File diff suppressed because it is too large Load Diff
@@ -952,6 +952,21 @@ class TestFixedRecursiveCharacterTextSplitter:
assert "word1" in combined
assert "word2" in combined
def test_preserves_spaces_when_recursively_splitting_long_paragraph(self):
"""Ensure recursive space splitting preserves word boundaries."""
text = "여름철에는 항상 기상상황에 주목하며 주변 사람들과 함께 정보를 공유합니다."
splitter = FixedRecursiveCharacterTextSplitter(
fixed_separator="\n\n",
chunk_size=20,
chunk_overlap=0,
keep_separator=True,
)
result = splitter.split_text(text)
assert len(result) > 1
assert " ".join(result) == text
def test_character_level_splitting(self):
"""Test character-level splitting when no separator works."""
text = "verylongwordwithoutspaces"
@@ -1,16 +1,16 @@
"""Unit tests for workflow-as-tool behavior.
StubSession/StubScalars emulate SQLAlchemy session/scalars with minimal methods
(`scalar`, `scalars`, `expunge`, `commit`, `refresh`, context manager) to keep
database access mocked and predictable in tests.
"""
"""Unit tests for workflow-as-tool behavior with real SQLite ORM boundaries."""
import json
import uuid
from collections.abc import Iterator
from dataclasses import dataclass
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock, Mock, patch
import pytest
from sqlalchemy import Engine, inspect
from sqlalchemy.orm import Session, sessionmaker
from core.app.entities.app_invoke_entities import InvokeFrom
from core.tools.__base.tool_runtime import ToolRuntime
@@ -23,74 +23,142 @@ from core.tools.entities.tool_entities import (
ToolProviderType,
)
from core.tools.errors import ToolInvokeError
from core.tools.workflow_as_tool import tool as workflow_tool_module
from core.tools.workflow_as_tool.tool import WorkflowTool
from graphon.file import FILE_MODEL_IDENTITY, FileTransferMethod, FileType
from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
from models.base import TypeBase
from models.enums import EndUserType
from models.model import App, AppMode, EndUser
from models.workflow import Workflow, WorkflowType
TENANT_ID = "00000000-0000-0000-0000-000000000001"
OTHER_TENANT_ID = "00000000-0000-0000-0000-000000000002"
APP_ID = "00000000-0000-0000-0000-000000000003"
ACCOUNT_ID = "00000000-0000-0000-0000-000000000004"
END_USER_ID = "00000000-0000-0000-0000-000000000005"
CREATOR_ID = "00000000-0000-0000-0000-000000000006"
class StubScalars:
"""Minimal stub for SQLAlchemy scalar results."""
_value: Any
def __init__(self, value: Any) -> None:
self._value = value
def first(self) -> Any:
return self._value
@dataclass(frozen=True)
class SqliteToolDb:
engine: Engine
session_maker: sessionmaker[Session]
caller_session: Session
class StubSession:
"""Minimal stub for session_factory-created sessions."""
@pytest.fixture
def sqlite_tool_db(
monkeypatch: pytest.MonkeyPatch,
sqlite_engine: Engine,
) -> Iterator[SqliteToolDb]:
"""Bind service-owned sessions and Account tenant reloads to SQLite."""
models = (App, Workflow, EndUser, Account, Tenant, TenantAccountJoin)
TypeBase.metadata.create_all(sqlite_engine, tables=[model.__table__ for model in models])
session_maker = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
monkeypatch.setattr(workflow_tool_module.session_factory, "create_session", session_maker)
scalar_results: list[Any]
scalars_results: list[Any]
expunge_calls: list[object]
from models import account as account_module
def __init__(self, *, scalar_results: list[Any] | None = None, scalars_results: list[Any] | None = None) -> None:
self.scalar_results = list(scalar_results or [])
self.scalars_results = list(scalars_results or [])
self.expunge_calls: list[object] = []
def scalar(self, _stmt: Any) -> Any:
return self.scalar_results.pop(0)
def scalars(self, _stmt: Any) -> StubScalars:
return StubScalars(self.scalars_results.pop(0))
def expunge(self, value: Any) -> None:
self.expunge_calls.append(value)
def begin(self) -> "StubSession":
return self
def commit(self) -> None:
pass
def refresh(self, _value: Any) -> None:
pass
def close(self) -> None:
pass
def __enter__(self) -> "StubSession":
return self
def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> bool:
return False
monkeypatch.setattr(account_module, "db", SimpleNamespace(engine=sqlite_engine))
with session_maker() as caller_session:
yield SqliteToolDb(engine=sqlite_engine, session_maker=session_maker, caller_session=caller_session)
def _build_tool() -> WorkflowTool:
def _persist_tenant(db: SqliteToolDb, *, tenant_id: str = TENANT_ID) -> Tenant:
tenant = Tenant(name="Tenant")
tenant.id = tenant_id
db.caller_session.add(tenant)
db.caller_session.commit()
return tenant
def _persist_account(db: SqliteToolDb, *, tenant_id: str = TENANT_ID) -> Account:
account = Account(name="Account", email="account@example.com")
account.id = ACCOUNT_ID
join = TenantAccountJoin(
tenant_id=tenant_id,
account_id=account.id,
current=True,
role=TenantAccountRole.NORMAL,
)
db.caller_session.add_all([account, join])
db.caller_session.commit()
return account
def _persist_end_user(
db: SqliteToolDb,
*,
end_user_id: str = END_USER_ID,
tenant_id: str = TENANT_ID,
) -> EndUser:
end_user = EndUser(
id=end_user_id,
tenant_id=tenant_id,
app_id=APP_ID,
type=EndUserType.SERVICE_API,
name="End user",
session_id="end-user-session",
)
db.caller_session.add(end_user)
db.caller_session.commit()
return end_user
def _persist_app(db: SqliteToolDb) -> App:
app = App(
id=APP_ID,
tenant_id=TENANT_ID,
name="Workflow app",
description="",
mode=AppMode.WORKFLOW,
icon_type=None,
icon="",
icon_background=None,
app_model_config_id=None,
workflow_id=None,
enable_site=False,
enable_api=True,
max_active_requests=None,
created_by=CREATOR_ID,
)
db.caller_session.add(app)
db.caller_session.commit()
return app
def _persist_workflow(db: SqliteToolDb, *, version: str, workflow_id: str | None = None) -> Workflow:
workflow = Workflow.new(
tenant_id=TENANT_ID,
app_id=APP_ID,
type=WorkflowType.WORKFLOW.value,
version=version,
graph=json.dumps({"nodes": [], "edges": []}),
features="{}",
created_by=CREATOR_ID,
environment_variables=[],
conversation_variables=[],
rag_pipeline_variables=[],
)
workflow.id = workflow_id or str(uuid.uuid4())
db.caller_session.add(workflow)
db.caller_session.commit()
return workflow
def _build_tool(*, tenant_id: str = "test_tool", workflow_app_id: str = "app-1", version: str = "1") -> WorkflowTool:
entity = ToolEntity(
identity=ToolIdentity(author="test", name="test tool", label=I18nObject(en_US="test tool"), provider="test"),
parameters=[],
description=None,
has_runtime_parameters=False,
)
runtime = ToolRuntime(tenant_id="test_tool", invoke_from=InvokeFrom.EXPLORE)
runtime = ToolRuntime(tenant_id=tenant_id, invoke_from=InvokeFrom.EXPLORE)
return WorkflowTool(
workflow_app_id="app-1",
workflow_app_id=workflow_app_id,
workflow_as_tool_id="wf-tool-1",
version="1",
version=version,
workflow_entities={},
workflow_call_depth=1,
entity=entity,
@@ -98,7 +166,10 @@ def _build_tool() -> WorkflowTool:
)
def test_workflow_tool_should_raise_tool_invoke_error_when_result_has_error_field(monkeypatch: pytest.MonkeyPatch):
def test_workflow_tool_should_raise_tool_invoke_error_when_result_has_error_field(
monkeypatch: pytest.MonkeyPatch,
sqlite_tool_db: SqliteToolDb,
):
"""Ensure that WorkflowTool will throw a `ToolInvokeError` exception when
`WorkflowAppGenerator.generate` returns a result with `error` key inside
the `data` element.
@@ -122,11 +193,14 @@ def test_workflow_tool_should_raise_tool_invoke_error_when_result_has_error_fiel
with pytest.raises(ToolInvokeError) as exc_info:
# WorkflowTool always returns a generator, so we need to iterate to
# actually `run` the tool.
list(tool.invoke(MagicMock(), "test_user", {}))
list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))
assert exc_info.value.args == ("oops",)
def test_workflow_tool_does_not_use_pause_state_config(monkeypatch: pytest.MonkeyPatch):
def test_workflow_tool_does_not_use_pause_state_config(
monkeypatch: pytest.MonkeyPatch,
sqlite_tool_db: SqliteToolDb,
):
"""Ensure pause_state_config is passed as None."""
tool = _build_tool()
@@ -140,14 +214,17 @@ def test_workflow_tool_does_not_use_pause_state_config(monkeypatch: pytest.Monke
monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)
list(tool.invoke(MagicMock(), "test_user", {}))
list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))
call_kwargs = generate_mock.call_args.kwargs
assert "pause_state_config" in call_kwargs
assert call_kwargs["pause_state_config"] is None
def test_workflow_tool_passes_parent_trace_context_from_runtime(monkeypatch: pytest.MonkeyPatch):
def test_workflow_tool_passes_parent_trace_context_from_runtime(
monkeypatch: pytest.MonkeyPatch,
sqlite_tool_db: SqliteToolDb,
):
"""Ensure nested workflow runtime metadata is forwarded as parent trace context."""
tool = _build_tool()
tool.set_parent_trace_context(
@@ -165,7 +242,7 @@ def test_workflow_tool_passes_parent_trace_context_from_runtime(monkeypatch: pyt
monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)
list(tool.invoke(MagicMock(), "test_user", {}))
list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))
call_kwargs = generate_mock.call_args.kwargs
assert call_kwargs["args"]["parent_trace_context"].model_dump() == {
@@ -174,7 +251,10 @@ def test_workflow_tool_passes_parent_trace_context_from_runtime(monkeypatch: pyt
}
def test_workflow_tool_passes_parent_trace_session_id(monkeypatch: pytest.MonkeyPatch):
def test_workflow_tool_passes_parent_trace_session_id(
monkeypatch: pytest.MonkeyPatch,
sqlite_tool_db: SqliteToolDb,
):
"""Ensure nested workflows inherit the parent observability session ID."""
tool = _build_tool()
tool.entity.parameters = [
@@ -197,14 +277,17 @@ def test_workflow_tool_passes_parent_trace_session_id(monkeypatch: pytest.Monkey
monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)
list(tool.invoke(MagicMock(), "test_user", {"trace_session_id": "user-input-session"}))
list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {"trace_session_id": "user-input-session"}))
call_kwargs = generate_mock.call_args.kwargs
assert call_kwargs["args"]["inputs"]["trace_session_id"] == "user-input-session"
assert call_kwargs["args"]["trace_session_id"] == "session-1"
def test_workflow_tool_keeps_user_inputs_named_like_trace_runtime_keys(monkeypatch: pytest.MonkeyPatch):
def test_workflow_tool_keeps_user_inputs_named_like_trace_runtime_keys(
monkeypatch: pytest.MonkeyPatch,
sqlite_tool_db: SqliteToolDb,
):
"""Ensure private trace context does not overwrite same-named workflow inputs."""
tool = _build_tool()
tool.entity.parameters = [
@@ -238,7 +321,7 @@ def test_workflow_tool_keeps_user_inputs_named_like_trace_runtime_keys(monkeypat
list(
tool.invoke(
MagicMock(),
sqlite_tool_db.caller_session,
"test_user",
{
"outer_workflow_run_id": "user-workflow-input",
@@ -256,7 +339,10 @@ def test_workflow_tool_keeps_user_inputs_named_like_trace_runtime_keys(monkeypat
}
def test_workflow_tool_can_clear_parent_trace_context(monkeypatch: pytest.MonkeyPatch):
def test_workflow_tool_can_clear_parent_trace_context(
monkeypatch: pytest.MonkeyPatch,
sqlite_tool_db: SqliteToolDb,
):
"""Ensure reused WorkflowTool instances do not keep stale parent trace context."""
tool = _build_tool()
tool.set_parent_trace_context(
@@ -275,13 +361,16 @@ def test_workflow_tool_can_clear_parent_trace_context(monkeypatch: pytest.Monkey
monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)
list(tool.invoke(MagicMock(), "test_user", {}))
list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))
call_kwargs = generate_mock.call_args.kwargs
assert "parent_trace_context" not in call_kwargs["args"]
def test_workflow_tool_can_clear_trace_session_id(monkeypatch: pytest.MonkeyPatch):
def test_workflow_tool_can_clear_trace_session_id(
monkeypatch: pytest.MonkeyPatch,
sqlite_tool_db: SqliteToolDb,
):
"""Ensure reused WorkflowTool instances do not keep stale trace session IDs."""
tool = _build_tool()
tool.set_trace_session_id("session-1")
@@ -297,7 +386,7 @@ def test_workflow_tool_can_clear_trace_session_id(monkeypatch: pytest.MonkeyPatc
monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)
list(tool.invoke(MagicMock(), "test_user", {}))
list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))
call_kwargs = generate_mock.call_args.kwargs
assert "trace_session_id" not in call_kwargs["args"]
@@ -315,6 +404,7 @@ def test_workflow_tool_can_clear_trace_session_id(monkeypatch: pytest.MonkeyPatc
def test_workflow_tool_omits_parent_trace_context_when_runtime_is_incomplete(
monkeypatch: pytest.MonkeyPatch,
runtime_parameters: dict[str, Any],
sqlite_tool_db: SqliteToolDb,
):
"""Ensure incomplete runtime metadata does not leak parent trace context into generator args."""
tool = _build_tool()
@@ -330,13 +420,16 @@ def test_workflow_tool_omits_parent_trace_context_when_runtime_is_incomplete(
monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)
list(tool.invoke(MagicMock(), "test_user", {}))
list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))
call_kwargs = generate_mock.call_args.kwargs
assert "parent_trace_context" not in call_kwargs["args"]
def test_workflow_tool_should_generate_variable_messages_for_outputs(monkeypatch: pytest.MonkeyPatch):
def test_workflow_tool_should_generate_variable_messages_for_outputs(
monkeypatch: pytest.MonkeyPatch,
sqlite_tool_db: SqliteToolDb,
):
"""Test that WorkflowTool should generate variable messages when there are outputs"""
tool = _build_tool()
@@ -359,7 +452,7 @@ def test_workflow_tool_should_generate_variable_messages_for_outputs(monkeypatch
monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)
# Execute tool invocation
messages = list(tool.invoke(MagicMock(), "test_user", {}))
messages = list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))
# Verify variable messages
variable_messages = [msg for msg in messages if msg.type == ToolInvokeMessage.MessageType.VARIABLE]
@@ -382,7 +475,10 @@ def test_workflow_tool_should_generate_variable_messages_for_outputs(monkeypatch
assert json_messages[0].message.json_object == mock_outputs
def test_workflow_tool_should_handle_empty_outputs(monkeypatch: pytest.MonkeyPatch):
def test_workflow_tool_should_handle_empty_outputs(
monkeypatch: pytest.MonkeyPatch,
sqlite_tool_db: SqliteToolDb,
):
"""Test that WorkflowTool should handle empty outputs correctly"""
tool = _build_tool()
@@ -402,7 +498,7 @@ def test_workflow_tool_should_handle_empty_outputs(monkeypatch: pytest.MonkeyPat
monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)
# Execute tool invocation
messages = list(tool.invoke(MagicMock(), "test_user", {}))
messages = list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))
# Verify generated messages
# Should contain: 0 variable messages + 1 text message + 1 JSON message = 2 messages
@@ -458,41 +554,32 @@ def test_create_file_message_should_include_file_marker():
assert message.meta == {"file": file_obj}
def test_resolve_user_from_database_falls_back_to_end_user(monkeypatch: pytest.MonkeyPatch):
def test_resolve_user_from_database_falls_back_to_end_user(sqlite_tool_db: SqliteToolDb):
"""Ensure worker context can resolve EndUser when Account is missing."""
tenant = SimpleNamespace(id="tenant_id")
end_user = SimpleNamespace(id="end_user_id", tenant_id="tenant_id")
# Monkeypatch session factory to return our stub session
stub_session = StubSession(scalar_results=[tenant, None, end_user])
monkeypatch.setattr(
"core.tools.workflow_as_tool.tool.session_factory.create_session",
lambda: stub_session,
_persist_tenant(sqlite_tool_db)
end_user = _persist_end_user(sqlite_tool_db)
other_tenant_end_user = _persist_end_user(
sqlite_tool_db,
end_user_id="00000000-0000-0000-0000-000000000007",
tenant_id=OTHER_TENANT_ID,
)
tool = _build_tool()
tool = _build_tool(tenant_id=TENANT_ID)
tool.runtime.invoke_from = InvokeFrom.SERVICE_API
tool.runtime.tenant_id = "tenant_id"
resolved_user = tool._resolve_user_from_database(user_id=end_user.id)
assert resolved_user is end_user
assert stub_session.expunge_calls == [end_user]
assert isinstance(resolved_user, EndUser)
assert resolved_user.id == end_user.id
assert resolved_user.tenant_id == TENANT_ID
assert inspect(resolved_user).detached is True
assert tool._resolve_user_from_database(user_id=other_tenant_end_user.id) is None
def test_resolve_user_from_database_returns_none_when_no_tenant(monkeypatch: pytest.MonkeyPatch):
def test_resolve_user_from_database_returns_none_when_no_tenant(sqlite_tool_db: SqliteToolDb):
"""Return None if tenant cannot be found in worker context."""
# Monkeypatch session factory to return our stub session with no tenant
monkeypatch.setattr(
"core.tools.workflow_as_tool.tool.session_factory.create_session",
lambda: StubSession(scalar_results=[None]),
)
tool = _build_tool()
tool = _build_tool(tenant_id=OTHER_TENANT_ID)
tool.runtime.invoke_from = InvokeFrom.SERVICE_API
tool.runtime.tenant_id = "missing_tenant"
resolved_user = tool._resolve_user_from_database(user_id="any")
@@ -544,7 +631,10 @@ def test_extract_usage_from_nested():
assert nested == {"total_tokens": 3}
def test_invoke_raises_when_user_not_found(monkeypatch: pytest.MonkeyPatch):
def test_invoke_raises_when_user_not_found(
monkeypatch: pytest.MonkeyPatch,
sqlite_tool_db: SqliteToolDb,
):
"""Raise ToolInvokeError when user resolution fails."""
tool = _build_tool()
monkeypatch.setattr(tool, "_get_app", lambda *args, **kwargs: None)
@@ -552,58 +642,45 @@ def test_invoke_raises_when_user_not_found(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(tool, "_resolve_user", lambda *args, **kwargs: None)
with pytest.raises(ToolInvokeError, match="User not found"):
list(tool.invoke(MagicMock(), "missing", {}))
list(tool.invoke(sqlite_tool_db.caller_session, "missing", {}))
def test_resolve_user_from_database_returns_account(monkeypatch: pytest.MonkeyPatch):
def test_resolve_user_from_database_returns_account(sqlite_tool_db: SqliteToolDb):
"""Resolve Account and set tenant in worker context."""
tenant = SimpleNamespace(id="tenant_id")
account = SimpleNamespace(id="account_id", current_tenant=None)
set_current_tenant = Mock(side_effect=lambda tenant, *, session: setattr(account, "current_tenant", tenant))
account.set_current_tenant_with_session = set_current_tenant
session = StubSession(scalar_results=[tenant, account])
tenant = _persist_tenant(sqlite_tool_db)
account = _persist_account(sqlite_tool_db)
tool = _build_tool(tenant_id=TENANT_ID)
monkeypatch.setattr("core.tools.workflow_as_tool.tool.session_factory.create_session", lambda: session)
tool = _build_tool()
tool.runtime.tenant_id = "tenant_id"
resolved = tool._resolve_user_from_database(user_id="account_id")
assert resolved is account
assert account.current_tenant is tenant
set_current_tenant.assert_called_once_with(tenant, session=session)
assert session.expunge_calls == [account]
resolved = tool._resolve_user_from_database(user_id=account.id)
assert isinstance(resolved, Account)
assert resolved.id == account.id
assert resolved.current_tenant_id == tenant.id
assert inspect(resolved).detached is True
def test_get_workflow_and_get_app_db_branches(monkeypatch: pytest.MonkeyPatch):
def test_get_workflow_and_get_app_db_branches(sqlite_tool_db: SqliteToolDb):
"""Cover workflow/app retrieval branches and error cases."""
tool = _build_tool()
latest_workflow = SimpleNamespace(id="wf-latest")
specific_workflow = SimpleNamespace(id="wf-v1")
app = SimpleNamespace(id="app-1")
sessions = iter(
[
StubSession(scalar_results=[], scalars_results=[latest_workflow]),
StubSession(scalar_results=[specific_workflow], scalars_results=[]),
StubSession(scalar_results=[app], scalars_results=[]),
]
)
monkeypatch.setattr(
"core.tools.workflow_as_tool.tool.session_factory.create_session",
lambda: next(sessions),
)
app = _persist_app(sqlite_tool_db)
specific_workflow = _persist_workflow(sqlite_tool_db, version="1")
latest_workflow = _persist_workflow(sqlite_tool_db, version="2")
_persist_workflow(sqlite_tool_db, version=Workflow.VERSION_DRAFT)
tool = _build_tool(tenant_id=TENANT_ID, workflow_app_id=APP_ID)
assert tool._get_workflow("app-1", "") is latest_workflow
assert tool._get_workflow("app-1", "1") is specific_workflow
assert tool._get_app("app-1") is app
latest = tool._get_workflow(APP_ID, "")
specific = tool._get_workflow(APP_ID, "1")
resolved_app = tool._get_app(APP_ID)
assert latest.id == latest_workflow.id
assert specific.id == specific_workflow.id
assert resolved_app.id == app.id
assert inspect(latest).detached is True
assert inspect(specific).detached is True
assert inspect(resolved_app).detached is True
monkeypatch.setattr(
"core.tools.workflow_as_tool.tool.session_factory.create_session",
lambda: StubSession(scalar_results=[None, None], scalars_results=[None]),
)
with pytest.raises(ValueError, match="workflow not found"):
tool._get_workflow("app-1", "1")
tool._get_workflow(APP_ID, "missing")
with pytest.raises(ValueError, match="app not found"):
tool._get_app("app-1")
tool._get_app("00000000-0000-0000-0000-000000000099")
def _setup_transform_args_tool(monkeypatch: pytest.MonkeyPatch) -> WorkflowTool:
@@ -722,7 +799,10 @@ def test_transform_args_normalizes_optional_files_parameter(
assert files == []
def test_workflow_tool_invocation_normalizes_optional_files_parameter(monkeypatch: pytest.MonkeyPatch):
def test_workflow_tool_invocation_normalizes_optional_files_parameter(
monkeypatch: pytest.MonkeyPatch,
sqlite_tool_db: SqliteToolDb,
):
"""Ensure casted empty FILES values do not reach workflow input validation as [None]."""
tool = _build_tool()
images_param = ToolParameter.get_simple_instance(
@@ -741,7 +821,7 @@ def test_workflow_tool_invocation_normalizes_optional_files_parameter(monkeypatc
generate_mock = MagicMock(return_value={"data": {}})
monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
list(tool.invoke(MagicMock(), "test_user", {"images": None}))
list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {"images": None}))
call_kwargs = generate_mock.call_args.kwargs
assert call_kwargs["args"]["inputs"]["images"] == []
@@ -18,7 +18,6 @@ from models.provider import ProviderType
@pytest.fixture
def credit_pool_session_factory(sqlite_engine: Engine) -> Iterator[sessionmaker[Session]]:
"""Bind message-created accounting to fixture-owned SQLite sessions."""
TenantCreditPool.__table__.create(sqlite_engine)
session_factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
with patch("events.event_handlers.update_provider_when_message_created.db.session", session_factory):
yield session_factory
@@ -0,0 +1,43 @@
from types import SimpleNamespace
from unittest.mock import Mock
from extensions.ext_celery import _enqueue_initial_community_telemetry_heartbeat
def test_beat_start_enqueues_community_telemetry_heartbeat() -> None:
task = Mock()
sender = SimpleNamespace(
app=SimpleNamespace(
conf=SimpleNamespace(beat_schedule={"community_telemetry_heartbeat": {}}),
tasks={"community_telemetry.send_heartbeat": task},
)
)
_enqueue_initial_community_telemetry_heartbeat(sender)
task.apply_async.assert_called_once_with()
def test_beat_start_skips_community_telemetry_when_not_scheduled() -> None:
task = Mock()
sender = SimpleNamespace(
app=SimpleNamespace(
conf=SimpleNamespace(beat_schedule={}),
tasks={"community_telemetry.send_heartbeat": task},
)
)
_enqueue_initial_community_telemetry_heartbeat(sender)
task.apply_async.assert_not_called()
def test_beat_start_skips_community_telemetry_when_task_is_unavailable() -> None:
sender = SimpleNamespace(
app=SimpleNamespace(
conf=SimpleNamespace(beat_schedule={"community_telemetry_heartbeat": {}}),
tasks={},
)
)
_enqueue_initial_community_telemetry_heartbeat(sender)
@@ -548,6 +548,7 @@ def test_load_agent_app_composer_exposes_draft_save_only(monkeypatch: pytest.Mon
agent = SimpleNamespace(
id="agent-1",
active_config_snapshot_id="version-1",
active_config_is_published=True,
updated_by="account-1",
created_by="account-1",
app_id="app-1",
@@ -567,6 +568,7 @@ def test_load_agent_app_composer_exposes_draft_save_only(monkeypatch: pytest.Mon
result = AgentComposerService.load_agent_app_composer(session=session, tenant_id="tenant-1", app_id="app-1")
assert result["save_options"] == [ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION.value]
assert result["active_config_is_published"] is True
def test_save_agent_app_composer_rejects_version_save_strategy():
@@ -610,7 +612,11 @@ def test_save_agent_app_composer_updates_normal_draft(monkeypatch: pytest.Monkey
lambda **kwargs: saved.update(kwargs) or SimpleNamespace(id="draft-1"),
)
monkeypatch.setattr(AgentComposerService, "_get_version_if_present", lambda **_kwargs: active_version)
monkeypatch.setattr(AgentComposerService, "load_agent_composer", lambda **kwargs: {"loaded": True})
monkeypatch.setattr(
AgentComposerService,
"load_agent_composer",
lambda **kwargs: {"loaded": True, "active_config_is_published": agent.active_config_is_published},
)
payload = ComposerSavePayload.model_validate(
{
"variant": ComposerVariant.AGENT_APP.value,
@@ -628,7 +634,7 @@ def test_save_agent_app_composer_updates_normal_draft(monkeypatch: pytest.Monkey
)
assert result.pop("validation") == {"warnings": [], "knowledge_retrieval_placeholder": []}
assert result == {"loaded": True}
assert result == {"loaded": True, "active_config_is_published": False}
assert saved["draft_type"] == AgentConfigDraftType.DRAFT
assert saved["agent_soul"].model_dump(mode="json") == _agent_soul_with_model().model_dump(mode="json")
assert agent.active_config_is_published is False
@@ -657,7 +663,11 @@ def test_save_agent_app_composer_keeps_published_when_draft_matches_active_snaps
lambda **_kwargs: SimpleNamespace(id="draft-1"),
)
monkeypatch.setattr(AgentComposerService, "_get_version_if_present", lambda **_kwargs: active_version)
monkeypatch.setattr(AgentComposerService, "load_agent_composer", lambda **_kwargs: {"loaded": True})
monkeypatch.setattr(
AgentComposerService,
"load_agent_composer",
lambda **_kwargs: {"loaded": True, "active_config_is_published": agent.active_config_is_published},
)
payload = ComposerSavePayload.model_validate(
{
"variant": ComposerVariant.AGENT_APP.value,
@@ -666,7 +676,7 @@ def test_save_agent_app_composer_keeps_published_when_draft_matches_active_snaps
}
)
AgentComposerService.save_agent_app_composer(
result = AgentComposerService.save_agent_app_composer(
session=session,
tenant_id="tenant-1",
app_id="app-1",
@@ -675,6 +685,7 @@ def test_save_agent_app_composer_keeps_published_when_draft_matches_active_snaps
)
assert agent.active_config_is_published is True
assert result["active_config_is_published"] is True
assert fake_session.flushes >= 1
@@ -246,6 +246,26 @@ class TestPluginModelProviderCache:
call([cache_key]),
]
def test_fetch_plugin_model_providers_bypasses_redis_when_cache_disabled(self) -> None:
"""With the cache disabled the daemon is the only source, and Redis is never touched."""
with patch(f"{MODULE}.redis_client") as redis_client, patch(f"{MODULE}.dify_config") as config:
config.PLUGIN_MODEL_PROVIDERS_CACHE_ENABLED = False
client = Mock()
client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
from core.plugin.plugin_service import PluginService
first = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
second = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
assert [provider.provider for provider in first] == ["langgenius/openai/openai"]
assert [provider.provider for provider in second] == ["langgenius/openai/openai"]
assert client.fetch_model_providers.call_count == 2
redis_client.get.assert_not_called()
redis_client.mget.assert_not_called()
redis_client.setex.assert_not_called()
redis_client.lock.assert_not_called()
def test_fetch_plugin_model_providers_refetches_when_cache_read_fails(self) -> None:
"""Redis read failures do not block provider discovery for the tenant."""
with patch(f"{MODULE}.redis_client") as redis_client:
@@ -1,7 +1,7 @@
import json
from collections.abc import Iterator
from datetime import datetime, timedelta
from unittest.mock import MagicMock, patch
from uuid import UUID
import pytest
from sqlalchemy import event, select
@@ -10,7 +10,6 @@ from sqlalchemy.orm import Session
from configs import dify_config
from models.account import (
Account,
AccountIntegrate,
AccountStatus,
Tenant,
TenantAccountJoin,
@@ -113,22 +112,6 @@ class TestAccountService:
- Error conditions and edge cases
"""
@pytest.fixture
def sqlite_session(self, sqlite_engine) -> Iterator[Session]:
"""SQLite session with the account/workspace tables these service tests touch."""
tables = [
model.metadata.tables[model.__tablename__]
for model in (
Account,
Tenant,
TenantAccountJoin,
TenantPluginAutoUpgradeStrategy,
)
]
Account.metadata.create_all(sqlite_engine, tables=tables)
with Session(sqlite_engine, expire_on_commit=False) as session:
yield session
@pytest.fixture
def mock_password_dependencies(self):
"""Mock setup for password-related functions."""
@@ -1263,24 +1246,6 @@ class TestRegisterService:
- Error conditions and edge cases
"""
@pytest.fixture
def sqlite_session(self, sqlite_engine) -> Iterator[Session]:
"""SQLite session with the account/workspace tables registration flows touch."""
tables = [
model.metadata.tables[model.__tablename__]
for model in (
Account,
AccountIntegrate,
Tenant,
TenantAccountJoin,
TenantPluginAutoUpgradeStrategy,
DifySetup,
)
]
Account.metadata.create_all(sqlite_engine, tables=tables)
with Session(sqlite_engine, expire_on_commit=False) as session:
yield session
@pytest.fixture
def mock_redis_dependencies(self):
"""Mock setup for Redis-related functions."""
@@ -1325,7 +1290,10 @@ class TestRegisterService:
with patch("services.account_service.AccountService.create_account") as mock_create_account:
mock_create_account.return_value = mock_account
with patch("services.account_service.TenantService.create_owner_tenant_if_not_exist") as mock_create_tenant:
with (
patch("services.account_service.TenantService.create_owner_tenant_if_not_exist") as mock_create_tenant,
patch("services.account_service.CommunityTelemetryService.report_install") as mock_report_install,
):
RegisterService.setup(
"admin@example.com",
"Admin User",
@@ -1344,7 +1312,39 @@ class TestRegisterService:
session=sqlite_session,
)
mock_create_tenant.assert_called_once_with(account=mock_account, is_setup=True, session=sqlite_session)
assert sqlite_session.scalar(select(DifySetup)) is not None
dify_setup = sqlite_session.scalar(select(DifySetup))
assert dify_setup is not None
assert dify_setup.instance_id is not None
assert str(UUID(dify_setup.instance_id)) == dify_setup.instance_id
assert dify_setup.install_reported_at is None
assert dify_setup.last_heartbeat_at is None
mock_report_install.assert_called_once_with(session=sqlite_session)
def test_setup_succeeds_when_telemetry_install_report_fails(
self, sqlite_session: Session, mock_external_service_dependencies
):
mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
mock_account = TestAccountAssociatedDataFactory.create_account_mock()
with (
patch("services.account_service.AccountService.create_account", return_value=mock_account),
patch("services.account_service.TenantService.create_owner_tenant_if_not_exist"),
patch(
"services.account_service.CommunityTelemetryService.report_install",
side_effect=RuntimeError("telemetry unavailable"),
),
):
RegisterService.setup(
"admin@example.com",
"Admin User",
"password123",
"192.168.1.1",
"en-US",
session=sqlite_session,
)
assert sqlite_session.scalar(select(DifySetup)) is not None
def test_setup_failure_rollback(self, sqlite_session: Session, mock_external_service_dependencies):
"""Test setup failure with proper rollback."""
@@ -1,19 +1,23 @@
from __future__ import annotations
from collections.abc import Callable
from datetime import datetime
from types import SimpleNamespace
from typing import cast
from unittest.mock import MagicMock, patch
from uuid import uuid4
import pytest
from sqlalchemy import event
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from graphon.model_runtime.entities.model_entities import ModelType
from models import Account
from models.model import App, AppMode, AppModelConfig
from models.model import App, AppMode, AppModelConfig, IconType
from models.workflow import Workflow
from services.agent.errors import AgentNameConflictError
from services.app_service import AppService, CreateAppParams
from services.app_service import AppListParams, AppService, CreateAppParams
class TestCreateAppTransactionBoundary:
@@ -236,6 +240,92 @@ class TestOpenapiVisibilityHelpers:
mock_session.execute.assert_called_once()
@pytest.mark.parametrize("sqlite_session", [(Account, App, AppModelConfig)], indirect=True)
def test_get_recent_apps_uses_one_tenant_scoped_projection_query(sqlite_session: Session) -> None:
tenant_id = str(uuid4())
other_tenant_id = str(uuid4())
account = Account(name="Recent Apps Author", email="recent-apps@example.com")
sqlite_session.add(account)
sqlite_session.flush()
def create_app(*, name: str, tenant_id: str, updated_at: datetime, mode: AppMode = AppMode.CHAT) -> App:
app = App()
app.id = str(uuid4())
app.tenant_id = tenant_id
app.name = name
app.description = ""
app.mode = mode
app.icon_type = IconType.EMOJI
app.icon = "🚀"
app.icon_background = "#FFFFFF"
app.enable_site = False
app.enable_api = False
app.created_by = account.id
app.maintainer = account.id
app.created_at = updated_at
app.updated_at = updated_at
app.use_icon_as_answer_icon = False
return app
newest = create_app(name="Newest", tenant_id=tenant_id, updated_at=datetime(2026, 7, 3))
legacy_agent = AppModelConfig(app_id=newest.id)
legacy_agent.agent_mode = '{"enabled": true, "strategy": "react"}'
newest.app_model_config_id = legacy_agent.id
second = create_app(
name="Second",
tenant_id=tenant_id,
updated_at=datetime(2026, 7, 2),
mode=AppMode.WORKFLOW,
)
second.icon_type = None
second.icon = None
second.icon_background = None
second.created_by = None
second.maintainer = None
channel = create_app(
name="Channel",
tenant_id=tenant_id,
updated_at=datetime(2026, 7, 5),
mode=AppMode.CHANNEL,
)
rag_pipeline = create_app(
name="RAG Pipeline",
tenant_id=tenant_id,
updated_at=datetime(2026, 7, 4),
mode=AppMode.RAG_PIPELINE,
)
oldest = create_app(name="Oldest", tenant_id=tenant_id, updated_at=datetime(2026, 7, 1))
foreign = create_app(name="Foreign", tenant_id=other_tenant_id, updated_at=datetime(2026, 7, 4))
sqlite_session.add_all([newest, legacy_agent, second, channel, rag_pipeline, oldest, foreign])
sqlite_session.commit()
statements: list[str] = []
bind = sqlite_session.get_bind()
def record_sql(_conn, _cursor, statement, _parameters, _context, _executemany) -> None:
statements.append(statement)
event.listen(bind, "before_cursor_execute", record_sql)
try:
recent_apps = AppService().get_recent_apps(
account.id,
tenant_id,
AppListParams(limit=2),
sqlite_session,
)
finally:
event.remove(bind, "before_cursor_execute", record_sql)
assert [(app.name, app.mode, app.icon_type, app.author_name, app.maintainer) for app in recent_apps] == [
("Newest", AppMode.CHAT, IconType.EMOJI, "Recent Apps Author", account.id),
("Second", AppMode.WORKFLOW, None, None, None),
]
select_statements = [statement for statement in statements if statement.lstrip().upper().startswith("SELECT")]
assert len(select_statements) == 1
assert "count(" not in select_statements[0].lower()
assert "app_model_configs" not in select_statements[0].lower()
class TestAppMeta:
def test_loads_workflow_with_caller_session(self):
session = MagicMock()
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,333 @@
import uuid
from datetime import datetime
from unittest.mock import Mock
import httpx
import pytest
from sqlalchemy import select
from sqlalchemy.orm import Session
from models.model import DifySetup
from services import telemetry_service
from services.telemetry_service import CommunityTelemetryService
@pytest.fixture
def telemetry_enabled(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(telemetry_service.dify_config, "EDITION", "SELF_HOSTED")
monkeypatch.setattr(telemetry_service.dify_config, "ENTERPRISE_ENABLED", False)
monkeypatch.setattr(telemetry_service.dify_config, "DISABLE_TELEMETRY", False)
monkeypatch.setattr(telemetry_service.dify_config, "DO_NOT_TRACK", False)
monkeypatch.setattr(telemetry_service.dify_config, "CI", False)
monkeypatch.setattr(telemetry_service.dify_config, "TELEMETRY_ENDPOINT", "https://telemetry.example.test/v1/events")
monkeypatch.setattr(
telemetry_service.dify_config,
"TELEMETRY_FALLBACK_ENDPOINT",
"https://telemetry-cn.example.test/v1/events",
)
monkeypatch.setattr(telemetry_service.dify_config, "TELEMETRY_TIMEOUT_SECONDS", 2)
def test_telemetry_is_disabled_for_enterprise(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(telemetry_service.dify_config, "EDITION", "SELF_HOSTED")
monkeypatch.setattr(telemetry_service.dify_config, "ENTERPRISE_ENABLED", True)
assert CommunityTelemetryService._is_enabled() is False
@pytest.mark.parametrize(
("setting", "value"),
[
("EDITION", "CLOUD"),
("DISABLE_TELEMETRY", True),
("DO_NOT_TRACK", True),
("CI", True),
("TELEMETRY_ENDPOINT", ""),
],
)
def test_telemetry_is_disabled_when_a_required_condition_is_not_met(
telemetry_enabled, monkeypatch: pytest.MonkeyPatch, setting: str, value: str | bool
):
monkeypatch.setattr(telemetry_service.dify_config, setting, value)
assert CommunityTelemetryService._is_enabled() is False
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
def test_reporting_without_setup_is_skipped(sqlite_session: Session, telemetry_enabled):
assert CommunityTelemetryService.report_install(session=sqlite_session) is False
assert CommunityTelemetryService.report_heartbeat(session=sqlite_session) is False
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
def test_report_install_marks_reported_at(sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch):
setup = DifySetup(version="installed-version", instance_id="d246c3a1-350b-406c-92c7-6043df680758")
sqlite_session.add(setup)
sqlite_session.commit()
monkeypatch.setattr(telemetry_service.dify_config.project, "version", "running-version")
sent_payloads: list[dict[str, str | int]] = []
def fake_post(url: str, json: dict[str, str | int], timeout: int):
sent_payloads.append(json)
return httpx.Response(204, request=httpx.Request("POST", url))
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
assert CommunityTelemetryService.report_install(session=sqlite_session) is True
saved_setup = sqlite_session.scalar(select(DifySetup))
assert saved_setup is not None
assert saved_setup.install_reported_at is not None
assert sent_payloads[0]["event"] == "install"
assert sent_payloads[0]["instance_id"] == setup.instance_id
assert sent_payloads[0]["version"] == "installed-version"
assert "installed_at" in sent_payloads[0]
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
def test_report_install_generates_missing_instance_id(
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
):
setup = DifySetup(version="installed-version")
sqlite_session.add(setup)
sqlite_session.commit()
monkeypatch.setattr(
telemetry_service.httpx,
"post",
lambda url, json, timeout: httpx.Response(204, request=httpx.Request("POST", url)),
)
assert CommunityTelemetryService.report_install(session=sqlite_session) is True
assert setup.instance_id is not None
assert str(uuid.UUID(setup.instance_id)) == setup.instance_id
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
def test_report_heartbeat_generates_missing_instance_id(
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
):
setup = DifySetup(version="1.0.0", install_reported_at=datetime(2026, 7, 12, 8, 0, 0))
sqlite_session.add(setup)
sqlite_session.commit()
monkeypatch.setattr(
telemetry_service.httpx,
"post",
lambda url, json, timeout: httpx.Response(204, request=httpx.Request("POST", url)),
)
assert (
CommunityTelemetryService.report_heartbeat(session=sqlite_session, now=datetime(2026, 7, 13, 12, 0, 0)) is True
)
assert setup.instance_id is not None
assert str(uuid.UUID(setup.instance_id)) == setup.instance_id
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
def test_report_install_failure_keeps_install_pending(
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
):
setup = DifySetup(version="1.0.0", instance_id="d246c3a1-350b-406c-92c7-6043df680758")
sqlite_session.add(setup)
sqlite_session.commit()
def fake_post(url: str, json: dict[str, str | int], timeout: int):
raise httpx.ConnectError("offline", request=httpx.Request("POST", url))
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
assert CommunityTelemetryService.report_install(session=sqlite_session) is False
saved_setup = sqlite_session.scalar(select(DifySetup))
assert saved_setup is not None
assert saved_setup.install_reported_at is None
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
def test_report_install_uses_fallback_endpoint_after_network_failure(
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
):
setup = DifySetup(version="1.0.0", instance_id="d246c3a1-350b-406c-92c7-6043df680758")
sqlite_session.add(setup)
sqlite_session.commit()
urls: list[str] = []
def fake_post(url: str, json: dict[str, str | int], timeout: int):
urls.append(url)
if url == telemetry_service.dify_config.TELEMETRY_ENDPOINT:
raise httpx.ConnectError("offline", request=httpx.Request("POST", url))
return httpx.Response(204, request=httpx.Request("POST", url))
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
assert CommunityTelemetryService.report_install(session=sqlite_session) is True
assert urls == [
telemetry_service.dify_config.TELEMETRY_ENDPOINT,
telemetry_service.dify_config.TELEMETRY_FALLBACK_ENDPOINT,
]
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
def test_report_install_does_not_use_fallback_endpoint_after_http_error(
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
):
setup = DifySetup(version="1.0.0", instance_id="d246c3a1-350b-406c-92c7-6043df680758")
sqlite_session.add(setup)
sqlite_session.commit()
post_mock = Mock(
return_value=httpx.Response(
500,
request=httpx.Request("POST", telemetry_service.dify_config.TELEMETRY_ENDPOINT),
)
)
monkeypatch.setattr(telemetry_service.httpx, "post", post_mock)
assert CommunityTelemetryService.report_install(session=sqlite_session) is False
post_mock.assert_called_once()
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
def test_report_heartbeat_retries_pending_install_before_heartbeat(
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
):
setup = DifySetup(version="installed-version", instance_id="d246c3a1-350b-406c-92c7-6043df680758")
sqlite_session.add(setup)
sqlite_session.commit()
monkeypatch.setattr(telemetry_service.dify_config.project, "version", "running-version")
sent_payloads: list[dict[str, str | int]] = []
def fake_post(url: str, json: dict[str, str | int], timeout: int):
sent_payloads.append(json)
return httpx.Response(204, request=httpx.Request("POST", url))
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
now = datetime(2026, 7, 13, 0, 0, 0)
assert CommunityTelemetryService.report_heartbeat(session=sqlite_session, now=now) is True
saved_setup = sqlite_session.scalar(select(DifySetup))
assert saved_setup is not None
assert saved_setup.install_reported_at is not None
assert saved_setup.last_heartbeat_at == now
assert [(payload["event"], payload["version"]) for payload in sent_payloads] == [
("install", "installed-version"),
("heartbeat", "running-version"),
]
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
def test_report_heartbeat_skips_when_already_sent_today(
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
):
setup = DifySetup(
version="1.0.0",
instance_id="d246c3a1-350b-406c-92c7-6043df680758",
install_reported_at=datetime(2026, 7, 13, 8, 0, 0),
last_heartbeat_at=datetime(2026, 7, 13, 9, 0, 0),
)
sqlite_session.add(setup)
sqlite_session.commit()
post_mock = Mock()
monkeypatch.setattr(telemetry_service.httpx, "post", post_mock)
assert (
CommunityTelemetryService.report_heartbeat(session=sqlite_session, now=datetime(2026, 7, 13, 12, 0, 0)) is False
)
post_mock.assert_not_called()
@pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True)
def test_report_heartbeat_failure_does_not_mark_the_day_reported(
sqlite_session: Session, telemetry_enabled, monkeypatch: pytest.MonkeyPatch
):
setup = DifySetup(
version="1.0.0",
instance_id="d246c3a1-350b-406c-92c7-6043df680758",
install_reported_at=datetime(2026, 7, 13, 8, 0, 0),
)
sqlite_session.add(setup)
sqlite_session.commit()
def fake_post(url: str, json: dict[str, str | int], timeout: int):
raise httpx.ConnectError("offline", request=httpx.Request("POST", url))
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
assert (
CommunityTelemetryService.report_heartbeat(session=sqlite_session, now=datetime(2026, 7, 13, 12, 0, 0)) is False
)
assert setup.last_heartbeat_at is None
def test_send_event_skips_when_telemetry_is_disabled(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(telemetry_service.dify_config, "DISABLE_TELEMETRY", True)
post_mock = Mock()
monkeypatch.setattr(telemetry_service.httpx, "post", post_mock)
assert CommunityTelemetryService._send_event({"event": "heartbeat"}) is False
post_mock.assert_not_called()
def test_send_event_skips_an_empty_fallback_endpoint(telemetry_enabled, monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(telemetry_service.dify_config, "TELEMETRY_FALLBACK_ENDPOINT", "")
def fake_post(url: str, json: dict[str, str], timeout: int):
raise httpx.ConnectError("offline", request=httpx.Request("POST", url))
monkeypatch.setattr(telemetry_service.httpx, "post", fake_post)
assert CommunityTelemetryService._send_event({"event": "heartbeat"}) is False
def test_send_event_does_not_retry_the_same_endpoint(telemetry_enabled, monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(
telemetry_service.dify_config,
"TELEMETRY_FALLBACK_ENDPOINT",
telemetry_service.dify_config.TELEMETRY_ENDPOINT,
)
post_mock = Mock(
return_value=httpx.Response(
204,
request=httpx.Request("POST", telemetry_service.dify_config.TELEMETRY_ENDPOINT),
)
)
monkeypatch.setattr(telemetry_service.httpx, "post", post_mock)
assert CommunityTelemetryService._send_event({"event": "heartbeat"}) is True
post_mock.assert_called_once()
def test_heartbeat_is_not_due_without_instance_id():
setup = DifySetup(version="1.0.0")
assert CommunityTelemetryService._is_heartbeat_due(setup, datetime(2026, 7, 13, 12, 0, 0)) is False
@pytest.mark.parametrize(
("value", "expected"),
[
("Linux", "linux"),
("Plan9", "unknown"),
],
)
def test_normalize_os(value: str, expected: str):
assert CommunityTelemetryService._normalize_os(value) == expected
@pytest.mark.parametrize(
("value", "expected"),
[
("x86_64", "amd64"),
("aarch64", "arm64"),
("armv7l", "arm"),
("i686", "386"),
("riscv64", "unknown"),
],
)
def test_normalize_arch(value: str, expected: str):
assert CommunityTelemetryService._normalize_arch(value) == expected

Some files were not shown because too many files have changed in this diff Show More