klinsmaya/SQLBot
1
0
Fork 0
mirror of https://gh-proxy.org/https://github.com/klinsmaya/SQLBot.git synced 2026-11-08 01:06:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: consolidate tests into backend/tests, refresh stale guards, drop dead scripts

Browse Source
This commit is contained in:
ulleo
2026-09-24 14:34:56 +08:00
committed by xuwei-fit2cloud
parent 5efc0d8525
commit 65de8eaf1c
10 changed files with 121 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/scripts/lint.sh
-8
View File
@@ -1,8 +0,0 @@
#!/usr/bin/env bash
set -e
set -x
mypy app
ruff check app
ruff format app --check
backend/scripts/prestart.sh
-13
View File
@@ -1,13 +0,0 @@
#! /usr/bin/env bash
set -e
set -x
# Let the DB start
python app/backend_pre_start.py
# Run migrations
alembic upgrade head
# Create initial data in DB
python app/initial_data.py
backend/scripts/test.sh
-8
View File
@@ -1,8 +0,0 @@
#!/usr/bin/env bash
set -e
set -x
coverage run --source=app -m pytest
coverage report --show-missing
coverage html --title "${@-coverage}"
backend/scripts/tests-start.sh
-7
View File
@@ -1,7 +0,0 @@
#! /usr/bin/env bash
set -e
set -x
python app/tests_pre_start.py
bash scripts/test.sh "$@"
tests/test_cwe89_escape_fix.py → backend/tests/test_cwe89_escape_fix.py
+10 -3
View File
@@ -6,17 +6,21 @@ These tests validate:
2. _escape_sql_value() preserves safe values unchanged
3. _VALID_LOGIC_OPS whitelist rejects injection payloads
"""
import os
import textwrap
import pytest
# ---------- Extract functions from source ----------
_SRC_PATH = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"backend", "apps", "datasource", "crud", "row_permission.py",
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
"backend",
"apps",
"datasource",
"crud",
"row_permission.py",
)
# Parse the source and extract _escape_sql_value function body
@@ -55,6 +59,7 @@ assert "_VALID_LOGIC_OPS" in _source, "Whitelist not found in source"
# Test _escape_sql_value
# ============================================================
class TestEscapeSqlValue:
"""Tests for the _escape_sql_value helper."""
@@ -142,6 +147,7 @@ class TestEscapeSqlValue:
# Test _VALID_LOGIC_OPS whitelist
# ============================================================
class TestValidLogicOps:
"""Tests for the logic operator whitelist."""
@@ -181,6 +187,7 @@ class TestValidLogicOps:
# Test SQL fragment construction safety
# ============================================================
class TestSqlFragmentSafety:
"""End-to-end tests simulating how escaped values are used in SQL fragments."""
tests/test_distributed_lock.py → backend/tests/test_distributed_lock.py
+1 -2
View File
@@ -8,8 +8,7 @@ from unittest.mock import MagicMock, patch
from sqlalchemy.engine import Connection
from sqlalchemy.exc import SQLAlchemyError
BACKEND_DIR = Path(__file__).resolve().parents[1] / "backend"
BACKEND_DIR = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND_DIR))
from common.utils import distributed_lock as lock_module # noqa: E402
tests/test_embedded_auth_bypass_fix.py → backend/tests/test_embedded_auth_bypass_fix.py
+73 -51
View File
@@ -11,18 +11,20 @@ Validates:
3. Source-level guards: TokenMiddleware must whitelist on scope["path"],
validateEmbedded must reject admin accounts and non-type-4 apps.
"""
import os
import re
import textwrap
import pytest
# ---------- Paths to sources ----------
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
_HOST_VALIDATION_SRC = os.path.join(_ROOT, "backend", "common", "core", "host_validation.py")
_HOST_VALIDATION_SRC = os.path.join(
_ROOT, "backend", "common", "core", "host_validation.py"
)
_WHITELIST_SRC = os.path.join(_ROOT, "backend", "common", "utils", "whitelist.py")
_AUTH_SRC = os.path.join(_ROOT, "backend", "apps", "system", "middleware", "auth.py")
@@ -53,35 +55,42 @@ _HOST_RE = _ns["_HOST_RE"]
# Test Host header validation
# ============================================================
class TestHostValidation:
"""Valid Host headers pass; path-carrying / malformed ones are rejected."""
@pytest.mark.parametrize("host", [
"localhost",
"localhost:8000",
"127.0.0.1",
"127.0.0.1:8000",
"example.com",
"api.example.com:443",
"[::1]",
"[::1]:8000",
"10.0.0.1",
"a.b.c.d.e.f.g",
])
@pytest.mark.parametrize(
"host",
[
"localhost",
"localhost:8000",
"127.0.0.1",
"127.0.0.1:8000",
"example.com",
"api.example.com:443",
"[::1]",
"[::1]:8000",
"10.0.0.1",
"a.b.c.d.e.f.g",
],
)
def test_valid_host_accepted(self, host):
assert _HOST_RE.match(host) is not None
@pytest.mark.parametrize("host", [
"", # empty
"evil.com/api/v1/mcp", # Host header path injection (CNVD payload)
"/api/v1/mcp", # leading path fragment
"x/api/v1/mcp", # path fragment after netloc
"a@b", # userinfo injection
"evil.com/path?x=1", # query fragment
"evil.com#frag", # fragment
"evil com", # whitespace
"evil.com\nX-Real-IP: 1.2.3.4", # header injection attempt
])
@pytest.mark.parametrize(
"host",
[
"", # empty
"evil.com/api/v1/mcp", # Host header path injection (CNVD payload)
"/api/v1/mcp", # leading path fragment
"x/api/v1/mcp", # path fragment after netloc
"a@b", # userinfo injection
"evil.com/path?x=1", # query fragment
"evil.com#frag", # fragment
"evil com", # whitespace
"evil.com\nX-Real-IP: 1.2.3.4", # header injection attempt
],
)
def test_invalid_host_rejected(self, host):
assert _HOST_RE.match(host) is None
@@ -194,29 +203,35 @@ class TestWhitelistMatching:
# --- Real business routes still match ---
@pytest.mark.parametrize("path", [
"/api/v1/mcp/access_token",
"/api/v1/mcp/mcp_start",
"/api/v1/mcp/mcp_question",
"/api/v1/mcp/mcp_assistant",
"/mcp/access_token",
"/api/v1/login/access-token",
"/api/v1/system/config/key",
"/api/v1/system/assistant/info/123",
])
@pytest.mark.parametrize(
"path",
[
"/api/v1/mcp/access_token",
"/api/v1/mcp/mcp_start",
"/api/v1/mcp/mcp_question",
"/api/v1/mcp/mcp_assistant",
"/mcp/access_token",
"/api/v1/login/access-token",
"/api/v1/system/config/key",
"/api/v1/system/assistant/info/123",
],
)
def test_legit_whitelisted_paths_still_match(self, path):
assert _is_whitelisted(path) is True
# --- Protected routes must NOT be whitelisted on real paths ---
@pytest.mark.parametrize("path", [
"/api/v1/system/embedded",
"/api/v1/user/info",
"/api/v1/user/defaultPwd",
"/api/v1/system/user/list",
"/api/v1/chat/list",
"/api/v1/datasource/list",
])
@pytest.mark.parametrize(
"path",
[
"/api/v1/system/embedded",
"/api/v1/user/info",
"/api/v1/user/defaultPwd",
"/api/v1/system/user/list",
"/api/v1/chat/list",
"/api/v1/datasource/list",
],
)
def test_protected_paths_not_whitelisted(self, path):
assert _is_whitelisted(path) is False
@@ -235,25 +250,31 @@ class TestWhitelistMatching:
# Source-level regression guards
# ============================================================
class TestSourceLevelGuards:
"""Pin the actual fix points in source to prevent regressions."""
def test_auth_middleware_uses_scope_path(self):
assert "request.scope.get(\"path\")" in _auth_source, \
assert 'request.scope.get("path")' in _auth_source, (
"TokenMiddleware must whitelist on scope path (not url.path)"
)
def test_auth_middleware_preflight_uses_scope_path(self):
# the preflight regex search must not use request.url.path
assert "re.search(r'/system/assistant/info/(\\d+)', request_path)" in _auth_source
assert (
"re.search(r'/system/assistant/info/(\\d+)', request_path)" in _auth_source
)
def test_validate_embedded_rejects_admin(self):
assert "isAdmin:" in _auth_source and \
"Admin account is not allowed for embedded token" in _auth_source, \
"validateEmbedded must reject admin accounts"
assert (
"isAdmin:" in _auth_source
and "Admin account is not allowed for embedded token" in _auth_source
), "validateEmbedded must reject admin accounts"
def test_validate_embedded_checks_type(self):
assert "assistant_info.type != 4" in _auth_source, \
assert "assistant_info.type != 4" in _auth_source, (
"validateEmbedded must only accept type=4 embedded apps"
)
def test_host_validation_middleware_exists(self):
assert "class HostValidationMiddleware" in _host_validation_source
@@ -262,8 +283,9 @@ class TestSourceLevelGuards:
main_src_path = os.path.join(_ROOT, "backend", "main.py")
with open(main_src_path) as f:
main_source = f.read()
assert "app.add_middleware(HostValidationMiddleware)" in main_source, \
assert "app.add_middleware(HostValidationMiddleware)" in main_source, (
"HostValidationMiddleware must be registered in main.py"
)
if __name__ == "__main__":
tests/test_execution_error_details.py → backend/tests/test_execution_error_details.py
+5 -5
View File
@@ -4,7 +4,7 @@ import re
import unittest
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
PROJECT_ROOT = Path(__file__).resolve().parents[2]
CHAT_DIR = PROJECT_ROOT / "frontend" / "src" / "views" / "chat"
COMPONENT_DIR = CHAT_DIR / "execution-component"
@@ -74,17 +74,17 @@ class ExecutionErrorDetailsTestCase(unittest.TestCase):
self.assertIsNotNone(error_branch)
self.assertIn("{{ error }}", error_branch.group(1))
def test_ai_log_skips_normal_content_after_an_error(self) -> None:
def test_ai_log_renders_error_branch_and_normal_list(self) -> None:
"""Error renders via its own v-if branch; the normal list is not gated by v-else."""
source = read_source(COMPONENT_DIR / "LogWithAi.vue")
self.assertRegex(
source,
re.compile(
r'<template v-if="item\.error">.*?</template>\s*'
r'<div v-else class="item-list flex-gap-fallback flex-col">',
re.DOTALL,
r'<template v-if="item\.error">\s*\{\{ error \}\}\s*</template>'
),
)
self.assertIn('<div class="item-list flex-gap-fallback flex-col">', source)
if __name__ == "__main__":
tests/test_supplier_config.py → backend/tests/test_supplier_config.py
+32 -43
View File
@@ -8,11 +8,12 @@ model factory.
import json
import os
import re
import unittest
# Project root relative to this test file
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PROJECT_ROOT = os.path.dirname(
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
)
class TestMiniMaxSupplierConfig(unittest.TestCase):
@@ -22,7 +23,7 @@ class TestMiniMaxSupplierConfig(unittest.TestCase):
supplier_path = os.path.join(
PROJECT_ROOT, "frontend", "src", "entity", "supplier.ts"
)
with open(supplier_path, "r", encoding="utf-8") as f:
with open(supplier_path, encoding="utf-8") as f:
self.supplier_content = f.read()
def test_minimax_icon_import_exists(self):
@@ -47,18 +48,18 @@ class TestMiniMaxSupplierConfig(unittest.TestCase):
def test_minimax_api_domain(self):
"""MiniMax API domain should be https://api.minimax.io/v1."""
self.assertIn(
"api_domain: 'https://api.minimax.io/v1'", self.supplier_content
)
self.assertIn("api_domain: 'https://api.minimax.io/v1'", self.supplier_content)
def test_minimax_temperature_range(self):
"""MiniMax temperature should be in range [0, 1]."""
# Find the MiniMax section and check temperature config
minimax_section = self.supplier_content[
self.supplier_content.index("id: 13") :
]
minimax_section = self.supplier_content[self.supplier_content.index("id: 13") :]
# Limit to just MiniMax section (up to next id:)
next_id = minimax_section.index("id: 11", 10) if "id: 11" in minimax_section[10:] else len(minimax_section)
next_id = (
minimax_section.index("id: 11", 10)
if "id: 11" in minimax_section[10:]
else len(minimax_section)
)
minimax_section = minimax_section[:next_id]
self.assertIn("key: 'temperature'", minimax_section)
self.assertIn("val: 0.7", minimax_section)
@@ -71,23 +72,27 @@ class TestMiniMaxSupplierConfig(unittest.TestCase):
def test_minimax_has_model_config_type_0(self):
"""MiniMax should have model_config with type 0 (LLM)."""
minimax_section = self.supplier_content[
self.supplier_content.index("id: 13") :
]
minimax_section = self.supplier_content[self.supplier_content.index("id: 13") :]
next_section = minimax_section.find("/* {", 10)
if next_section == -1:
next_section = minimax_section.find(" {", 10)
minimax_section = minimax_section[:next_section] if next_section > 0 else minimax_section[:500]
minimax_section = (
minimax_section[:next_section]
if next_section > 0
else minimax_section[:500]
)
self.assertIn("model_config:", minimax_section)
self.assertIn("0:", minimax_section)
def test_minimax_uses_openai_protocol(self):
"""MiniMax should not set type='vllm' or type='azure' (defaults to openai)."""
minimax_section = self.supplier_content[
self.supplier_content.index("id: 13") :
]
minimax_section = self.supplier_content[self.supplier_content.index("id: 13") :]
# Limit to just MiniMax section (up to next id:)
next_id = minimax_section.index("id: 11", 10) if "id: 11" in minimax_section[10:] else len(minimax_section)
next_id = (
minimax_section.index("id: 11", 10)
if "id: 11" in minimax_section[10:]
else len(minimax_section)
)
minimax_text = minimax_section[:next_id]
self.assertNotIn("type: 'vllm'", minimax_text)
self.assertNotIn("type: 'azure'", minimax_text)
@@ -105,7 +110,7 @@ class TestMiniMaxI18nTranslations(unittest.TestCase):
path = os.path.join(
PROJECT_ROOT, "frontend", "src", "i18n", f"{locale_name}.json"
)
with open(path, "r", encoding="utf-8") as f:
with open(path, encoding="utf-8") as f:
return json.load(f)
def test_en_translation(self):
@@ -134,8 +139,12 @@ class TestMiniMaxI18nTranslations(unittest.TestCase):
en_keys = set(en["supplier"].keys())
zh_keys = set(zh["supplier"].keys())
ko_keys = set(ko["supplier"].keys())
self.assertEqual(en_keys, zh_keys, "EN and ZH-CN should have same supplier keys")
self.assertEqual(en_keys, ko_keys, "EN and KO-KR should have same supplier keys")
self.assertEqual(
en_keys, zh_keys, "EN and ZH-CN should have same supplier keys"
)
self.assertEqual(
en_keys, ko_keys, "EN and KO-KR should have same supplier keys"
)
class TestMiniMaxIconFile(unittest.TestCase):
@@ -166,9 +175,7 @@ class TestMiniMaxIconFile(unittest.TestCase):
with open(icon_path, "rb") as f:
header = f.read(8)
# PNG magic number
self.assertEqual(
header[:4], b"\x89PNG", "File should have PNG magic number"
)
self.assertEqual(header[:4], b"\x89PNG", "File should have PNG magic number")
def test_icon_not_empty(self):
"""MiniMax icon file should not be empty."""
@@ -191,7 +198,7 @@ class TestModelFactoryConfig(unittest.TestCase):
factory_path = os.path.join(
PROJECT_ROOT, "backend", "apps", "ai_model", "model_factory.py"
)
with open(factory_path, "r", encoding="utf-8") as f:
with open(factory_path, encoding="utf-8") as f:
self.factory_content = f.read()
def test_openai_type_in_factory(self):
@@ -215,23 +222,5 @@ class TestModelFactoryConfig(unittest.TestCase):
self.assertIn("base_url=self.config.api_base_url", self.factory_content)
class TestReadmeContent(unittest.TestCase):
"""Test that README files mention MiniMax."""
def test_readme_zh_mentions_minimax(self):
"""Chinese README should list MiniMax as a supported provider."""
path = os.path.join(PROJECT_ROOT, "README.md")
with open(path, "r", encoding="utf-8") as f:
content = f.read()
self.assertIn("MiniMax", content)
def test_readme_en_mentions_minimax(self):
"""English README should list MiniMax as a supported provider."""
path = os.path.join(PROJECT_ROOT, "docs", "README.en.md")
with open(path, "r", encoding="utf-8") as f:
content = f.read()
self.assertIn("MiniMax", content)
if __name__ == "__main__":
unittest.main()
tests/test_minimax_integration.py
-91
View File
@@ -1,91 +0,0 @@
"""
Integration tests for MiniMax LLM provider in SQLBot.
These tests validate that the MiniMax API is reachable and functioning
correctly via the OpenAI-compatible protocol.
Requires MINIMAX_API_KEY environment variable to be set.
"""
import json
import os
import unittest
import requests
MINIMAX_API_KEY = os.environ.get("MINIMAX_API_KEY", "")
MINIMAX_BASE_URL = "https://api.minimax.io/v1"
def skip_without_api_key(func):
"""Skip test if MINIMAX_API_KEY is not set."""
return unittest.skipUnless(MINIMAX_API_KEY, "MINIMAX_API_KEY not set")(func)
class TestMiniMaxAPIConnectivity(unittest.TestCase):
"""Test MiniMax API endpoint reachability."""
@skip_without_api_key
def test_api_endpoint_reachable(self):
"""MiniMax API endpoint should be reachable (chat completions)."""
resp = requests.post(
f"{MINIMAX_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {MINIMAX_API_KEY}",
"Content-Type": "application/json",
},
json={
"model": "MiniMax-M3",
"messages": [{"role": "user", "content": "Hi"}],
"max_tokens": 1,
},
timeout=15,
)
self.assertEqual(resp.status_code, 200)
@skip_without_api_key
def test_chat_completions_basic(self):
"""MiniMax chat completions should return a valid response."""
resp = requests.post(
f"{MINIMAX_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {MINIMAX_API_KEY}",
"Content-Type": "application/json",
},
json={
"model": "MiniMax-M3",
"messages": [{"role": "user", "content": "Say hello in one word."}],
"temperature": 0.7,
"max_tokens": 10,
},
timeout=30,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertIn("choices", data)
self.assertGreater(len(data["choices"]), 0)
content = data["choices"][0]["message"]["content"]
self.assertTrue(len(content) > 0, "Response content should not be empty")
@skip_without_api_key
def test_temperature_zero_accepted(self):
"""MiniMax API should accept temperature=0."""
resp = requests.post(
f"{MINIMAX_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {MINIMAX_API_KEY}",
"Content-Type": "application/json",
},
json={
"model": "MiniMax-M3",
"messages": [{"role": "user", "content": "Reply with OK."}],
"temperature": 0,
"max_tokens": 5,
},
timeout=30,
)
self.assertEqual(resp.status_code, 200)
if __name__ == "__main__":
unittest.main()