logan
4c9427e546
- P0: 实例访问控制全线接入权限组 (829测试) - P1: permissions字段激活 + 操作级权限检查 (843测试) - 前端: 删除数据权限页面, 权限组合并为唯一入口 - 新增: CodeMirror编辑器 + 查询模板 + 慢查询分析 - 新增: MongoDB管理页面 + Dashboard增强 - 修复: Pydantic V2迁移 + 前端TS错误修复
607 lines
21 KiB
Python
607 lines
21 KiB
Python
"""TicketExecutor 多数据库工单执行引擎测试。
|
|
|
|
覆盖:
|
|
1. ExecutionReport 序列化/反序列化
|
|
2. MySQL 语法预校验
|
|
3. PostgreSQL 语法预校验
|
|
4. Redis 命令解析和校验
|
|
5. MongoDB 操作解析和校验
|
|
6. SQL 工具方法: 表名提取、WHERE 子句提取、语句分割
|
|
"""
|
|
|
|
import json
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
|
|
from app.services.ticket_executor import (
|
|
TicketExecutor,
|
|
ExecutionReport,
|
|
SyntaxValidationError,
|
|
REDIS_WRITE_COMMANDS,
|
|
MONGO_SUPPORTED_OPERATIONS,
|
|
)
|
|
from app.models.instance import Instance
|
|
from app.models.ticket import Ticket
|
|
|
|
|
|
# ===== ExecutionReport 测试 =====
|
|
|
|
|
|
class TestExecutionReport:
|
|
"""ExecutionReport 数据类测试。"""
|
|
|
|
def test_default_values(self):
|
|
"""默认值应为 success。"""
|
|
report = ExecutionReport()
|
|
assert report.status == "success"
|
|
assert report.affected_rows == 0
|
|
assert report.duration_ms == 0
|
|
assert report.validation_passed is True
|
|
assert report.warnings == []
|
|
assert report.errors == []
|
|
|
|
def test_to_json(self):
|
|
"""应正确序列化为 JSON。"""
|
|
report = ExecutionReport(
|
|
status="success",
|
|
affected_rows=10,
|
|
duration_ms=150,
|
|
before_snapshot="10 rows",
|
|
after_snapshot="15 rows",
|
|
validation_passed=True,
|
|
warnings=["UPDATE 缺少 WHERE"],
|
|
errors=[],
|
|
raw_output="UPDATE 10 rows",
|
|
)
|
|
j = report.to_json()
|
|
data = json.loads(j)
|
|
assert data["status"] == "success"
|
|
assert data["affected_rows"] == 10
|
|
assert data["duration_ms"] == 150
|
|
assert data["warnings"] == ["UPDATE 缺少 WHERE"]
|
|
|
|
def test_from_exception(self):
|
|
"""从异常创建失败的报告。"""
|
|
exc = RuntimeError("连接超时")
|
|
report = ExecutionReport.from_exception(exc, "执行")
|
|
assert report.status == "failed"
|
|
assert report.validation_passed is False
|
|
assert len(report.errors) == 1
|
|
assert "连接超时" in report.errors[0]
|
|
|
|
def test_to_json_chinese(self):
|
|
"""JSON 序列化应保留中文字符。"""
|
|
report = ExecutionReport(raw_output="更新了3行数据")
|
|
j = report.to_json()
|
|
assert "更新了3行数据" in j
|
|
|
|
|
|
# ===== TicketExecutor 初始化 =====
|
|
|
|
|
|
class TestTicketExecutorInit:
|
|
"""TicketExecutor 初始化测试。"""
|
|
|
|
def test_init_creates_analyzer(self):
|
|
"""初始化应创建 SQLAnalyzer。"""
|
|
executor = TicketExecutor()
|
|
assert executor._analyzer is not None
|
|
|
|
|
|
# ===== MySQL 语法预校验 =====
|
|
|
|
|
|
class TestMySQLSyntaxValidation:
|
|
"""MySQL 语法预校验测试。"""
|
|
|
|
def setup_method(self):
|
|
self.executor = TicketExecutor()
|
|
|
|
def test_empty_sql_rejected(self):
|
|
"""空 SQL 应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="不能为空"):
|
|
self.executor._validate_mysql_syntax("")
|
|
|
|
def test_whitespace_sql_rejected(self):
|
|
"""纯空格 SQL 应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="不能为空"):
|
|
self.executor._validate_mysql_syntax(" ")
|
|
|
|
def test_insert_pass(self):
|
|
"""INSERT 语句应通过校验。"""
|
|
self.executor._validate_mysql_syntax("INSERT INTO users (name) VALUES ('test')")
|
|
|
|
def test_update_pass(self):
|
|
"""UPDATE 语句应通过校验。"""
|
|
self.executor._validate_mysql_syntax("UPDATE users SET name='x' WHERE id=1")
|
|
|
|
def test_delete_pass(self):
|
|
"""DELETE 语句应通过校验。"""
|
|
self.executor._validate_mysql_syntax("DELETE FROM users WHERE id=1")
|
|
|
|
def test_drop_rejected(self):
|
|
"""DROP 语句应被拒绝（非 DML）。"""
|
|
with pytest.raises(SyntaxValidationError, match="不支持的 SQL 类型"):
|
|
self.executor._validate_mysql_syntax("DROP TABLE users")
|
|
|
|
def test_create_rejected(self):
|
|
"""CREATE 语句应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="不支持的 SQL 类型"):
|
|
self.executor._validate_mysql_syntax("CREATE TABLE test (id INT)")
|
|
|
|
def test_alter_rejected(self):
|
|
"""ALTER 语句应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="不支持的 SQL 类型"):
|
|
self.executor._validate_mysql_syntax("ALTER TABLE users ADD COLUMN age INT")
|
|
|
|
def test_truncate_rejected(self):
|
|
"""TRUNCATE 语句应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="不支持的 SQL 类型"):
|
|
self.executor._validate_mysql_syntax("TRUNCATE TABLE users")
|
|
|
|
|
|
# ===== PostgreSQL 语法预校验 =====
|
|
|
|
|
|
class TestPostgresSyntaxValidation:
|
|
"""PostgreSQL 语法预校验测试。"""
|
|
|
|
def setup_method(self):
|
|
self.executor = TicketExecutor()
|
|
|
|
def test_insert_pass(self):
|
|
"""INSERT 语句应通过校验。"""
|
|
self.executor._validate_postgres_syntax("INSERT INTO users (name) VALUES ('test')")
|
|
|
|
def test_update_pass(self):
|
|
"""UPDATE 语句应通过校验。"""
|
|
self.executor._validate_postgres_syntax("UPDATE users SET name='x' WHERE id=1")
|
|
|
|
def test_delete_pass(self):
|
|
"""DELETE 语句应通过校验。"""
|
|
self.executor._validate_postgres_syntax("DELETE FROM users WHERE id=1")
|
|
|
|
def test_drop_rejected(self):
|
|
"""DROP 语句应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError):
|
|
self.executor._validate_postgres_syntax("DROP TABLE users")
|
|
|
|
|
|
# ===== Redis 命令解析 =====
|
|
|
|
|
|
class TestRedisCommandParsing:
|
|
"""Redis 命令解析测试。"""
|
|
|
|
def setup_method(self):
|
|
self.executor = TicketExecutor()
|
|
|
|
def test_parse_json_commands(self):
|
|
"""应解析 JSON 格式的命令列表。"""
|
|
content = json.dumps({
|
|
"commands": [
|
|
{"command": "SET key1 value1", "db": 0},
|
|
{"command": "SET key2 value2", "db": 1},
|
|
]
|
|
})
|
|
commands = self.executor._parse_redis_commands(content)
|
|
assert len(commands) == 2
|
|
assert commands[0]["command"] == "SET key1 value1"
|
|
assert commands[0]["db"] == 0
|
|
assert commands[1]["db"] == 1
|
|
|
|
def test_parse_json_single_command(self):
|
|
"""应解析单条 JSON 命令。"""
|
|
content = json.dumps({"command": "SET key1 value1", "db": 2})
|
|
commands = self.executor._parse_redis_commands(content)
|
|
assert len(commands) == 1
|
|
assert commands[0]["command"] == "SET key1 value1"
|
|
assert commands[0]["db"] == 2
|
|
|
|
def test_parse_text_commands(self):
|
|
"""应解析文本格式命令（每行一个）。"""
|
|
content = "SET key1 value1\nSET key2 value2\n# this is a comment"
|
|
commands = self.executor._parse_redis_commands(content)
|
|
assert len(commands) == 2
|
|
assert commands[0]["command"] == "SET key1 value1"
|
|
assert commands[1]["command"] == "SET key2 value2"
|
|
|
|
def test_parse_empty_content(self):
|
|
"""空内容应返回空列表。"""
|
|
commands = self.executor._parse_redis_commands("")
|
|
assert commands == []
|
|
|
|
def test_parse_invalid_json_as_text(self):
|
|
"""无效 JSON 应作为文本解析。"""
|
|
content = "{not valid json"
|
|
commands = self.executor._parse_redis_commands(content)
|
|
# Should fall through to text parsing
|
|
assert len(commands) == 1
|
|
assert commands[0]["command"] == content
|
|
|
|
|
|
# ===== Redis 命令校验 =====
|
|
|
|
|
|
class TestRedisCommandValidation:
|
|
"""Redis 命令格式校验测试。"""
|
|
|
|
def setup_method(self):
|
|
self.executor = TicketExecutor()
|
|
|
|
def test_valid_set_command(self):
|
|
"""合法的 SET 命令应通过校验。"""
|
|
self.executor._validate_redis_commands([{"command": "SET key1 value1", "db": 0}])
|
|
|
|
def test_valid_hset_command(self):
|
|
"""合法的 HSET 命令应通过校验。"""
|
|
self.executor._validate_redis_commands([{"command": "HSET myhash field value", "db": 0}])
|
|
|
|
def test_empty_command_rejected(self):
|
|
"""空命令应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="不能为空"):
|
|
self.executor._validate_redis_commands([{"command": "", "db": 0}])
|
|
|
|
def test_no_args_command_with_ping(self):
|
|
"""PING 等无参数命令应通过。"""
|
|
self.executor._validate_redis_commands([{"command": "PING", "db": 0}])
|
|
|
|
def test_no_args_command_without_special(self):
|
|
"""需要参数的命令缺少参数应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="缺少参数"):
|
|
self.executor._validate_redis_commands([{"command": "GET", "db": 0}])
|
|
|
|
def test_invalid_command_name_rejected(self):
|
|
"""非法命令名应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="非法"):
|
|
self.executor._validate_redis_commands([{"command": "123BAD key", "db": 0}])
|
|
|
|
|
|
# ===== MongoDB 操作解析 =====
|
|
|
|
|
|
class TestMongoOperationParsing:
|
|
"""MongoDB 操作解析测试。"""
|
|
|
|
def setup_method(self):
|
|
self.executor = TicketExecutor()
|
|
|
|
def test_parse_valid_insert(self):
|
|
"""应解析有效的 insertOne 操作。"""
|
|
content = json.dumps({
|
|
"collection": "users",
|
|
"operation": "insertOne",
|
|
"document": {"name": "test", "age": 25},
|
|
})
|
|
result = self.executor._parse_mongo_operation(content)
|
|
assert result["collection"] == "users"
|
|
assert result["operation"] == "insertOne"
|
|
assert result["document"]["name"] == "test"
|
|
|
|
def test_parse_valid_update(self):
|
|
"""应解析有效的 updateMany 操作。"""
|
|
content = json.dumps({
|
|
"collection": "users",
|
|
"operation": "updateMany",
|
|
"filter": {"status": "inactive"},
|
|
"update": {"$set": {"status": "active"}},
|
|
})
|
|
result = self.executor._parse_mongo_operation(content)
|
|
assert result["operation"] == "updateMany"
|
|
|
|
def test_parse_empty_content_rejected(self):
|
|
"""空内容应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="不能为空"):
|
|
self.executor._parse_mongo_operation("")
|
|
|
|
def test_parse_invalid_json_rejected(self):
|
|
"""无效 JSON 应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="JSON"):
|
|
self.executor._parse_mongo_operation("not json")
|
|
|
|
def test_parse_non_dict_rejected(self):
|
|
"""非 JSON 对象应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="JSON 对象"):
|
|
self.executor._parse_mongo_operation("[1, 2, 3]")
|
|
|
|
|
|
# ===== MongoDB 操作校验 =====
|
|
|
|
|
|
class TestMongoOperationValidation:
|
|
"""MongoDB 操作格式校验测试。"""
|
|
|
|
def setup_method(self):
|
|
self.executor = TicketExecutor()
|
|
|
|
def test_insertone_valid(self):
|
|
"""insertOne 应通过校验。"""
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "insertOne",
|
|
"document": {"name": "test"},
|
|
})
|
|
|
|
def test_insertmany_valid(self):
|
|
"""insertMany 应通过校验。"""
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "insertMany",
|
|
"documents": [{"name": "a"}, {"name": "b"}],
|
|
})
|
|
|
|
def test_updateone_valid(self):
|
|
"""updateOne 应通过校验。"""
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "updateOne",
|
|
"filter": {"_id": 1},
|
|
"update": {"$set": {"name": "x"}},
|
|
})
|
|
|
|
def test_updatemany_valid(self):
|
|
"""updateMany 应通过校验。"""
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "updateMany",
|
|
"filter": {"status": "old"},
|
|
"update": {"$set": {"status": "new"}},
|
|
})
|
|
|
|
def test_deleteone_valid(self):
|
|
"""deleteOne 应通过校验。"""
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "deleteOne",
|
|
"filter": {"_id": 1},
|
|
})
|
|
|
|
def test_deletemany_valid(self):
|
|
"""deleteMany 应通过校验。"""
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "deleteMany",
|
|
"filter": {"status": "inactive"},
|
|
})
|
|
|
|
def test_replaceone_valid(self):
|
|
"""replaceOne 应通过校验。"""
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "replaceOne",
|
|
"filter": {"_id": 1},
|
|
"document": {"name": "new"},
|
|
})
|
|
|
|
def test_missing_collection_rejected(self):
|
|
"""缺少 collection 字段应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="collection"):
|
|
self.executor._validate_mongo_operation({
|
|
"operation": "insertOne",
|
|
"document": {"name": "test"},
|
|
})
|
|
|
|
def test_missing_operation_rejected(self):
|
|
"""缺少 operation 字段应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="操作类型"):
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
})
|
|
|
|
def test_unsupported_operation_rejected(self):
|
|
"""不支持的操作应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="不支持"):
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "createIndex",
|
|
"keys": {"name": 1},
|
|
})
|
|
|
|
def test_insertone_missing_document_rejected(self):
|
|
"""insertOne 缺少 document 应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="document"):
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "insertOne",
|
|
})
|
|
|
|
def test_insertmany_missing_documents_rejected(self):
|
|
"""insertMany 缺少 documents 应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="documents"):
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "insertMany",
|
|
})
|
|
|
|
def test_updateone_missing_filter_rejected(self):
|
|
"""updateOne 缺少 filter 应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="filter"):
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "updateOne",
|
|
"update": {"$set": {"x": 1}},
|
|
})
|
|
|
|
def test_updateone_missing_update_rejected(self):
|
|
"""updateOne 缺少 update 应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="update"):
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "updateOne",
|
|
"filter": {"_id": 1},
|
|
})
|
|
|
|
def test_replaceone_missing_document_rejected(self):
|
|
"""replaceOne 缺少 document 应被拒绝。"""
|
|
with pytest.raises(SyntaxValidationError, match="document"):
|
|
self.executor._validate_mongo_operation({
|
|
"collection": "users",
|
|
"operation": "replaceOne",
|
|
"filter": {"_id": 1},
|
|
})
|
|
|
|
|
|
# ===== SQL 工具方法 =====
|
|
|
|
|
|
class TestSQLUtilities:
|
|
"""SQL 工具方法测试。"""
|
|
|
|
def setup_method(self):
|
|
self.executor = TicketExecutor()
|
|
|
|
def test_split_statements_basic(self):
|
|
"""应按分号分割语句。"""
|
|
sql = "INSERT INTO t1 VALUES (1); INSERT INTO t2 VALUES (2);"
|
|
stmts = self.executor._split_statements(sql)
|
|
assert len(stmts) == 2
|
|
|
|
def test_split_statements_single(self):
|
|
"""单条语句无分号应返回一条。"""
|
|
sql = "INSERT INTO t1 VALUES (1)"
|
|
stmts = self.executor._split_statements(sql)
|
|
assert len(stmts) == 1
|
|
|
|
def test_split_statements_empty(self):
|
|
"""空内容应返回空列表。"""
|
|
stmts = self.executor._split_statements("")
|
|
assert stmts == []
|
|
|
|
def test_extract_table_name_insert(self):
|
|
"""应从 INSERT 语句中提取表名。"""
|
|
from app.services.sql_analyzer import SQLType
|
|
name = self.executor._extract_table_name(
|
|
"INSERT INTO users (name) VALUES ('test')", SQLType.INSERT
|
|
)
|
|
assert name == "USERS"
|
|
|
|
def test_extract_table_name_update(self):
|
|
"""应从 UPDATE 语句中提取表名。"""
|
|
from app.services.sql_analyzer import SQLType
|
|
name = self.executor._extract_table_name(
|
|
"UPDATE users SET name='x' WHERE id=1", SQLType.UPDATE
|
|
)
|
|
assert name == "USERS"
|
|
|
|
def test_extract_table_name_delete(self):
|
|
"""应从 DELETE 语句中提取表名。"""
|
|
from app.services.sql_analyzer import SQLType
|
|
name = self.executor._extract_table_name(
|
|
"DELETE FROM users WHERE id=1", SQLType.DELETE
|
|
)
|
|
assert name == "USERS"
|
|
|
|
def test_extract_where_clause_found(self):
|
|
"""应从带 WHERE 的 SQL 中提取条件。"""
|
|
where = self.executor._extract_where_clause_text(
|
|
"UPDATE users SET name='x' WHERE id=1 AND status='active'"
|
|
)
|
|
assert where is not None
|
|
assert "id=1" in where
|
|
|
|
def test_extract_where_clause_not_found(self):
|
|
"""没有 WHERE 的 SQL 应返回 None。"""
|
|
where = self.executor._extract_where_clause_text(
|
|
"INSERT INTO users (name) VALUES ('test')"
|
|
)
|
|
assert where is None
|
|
|
|
def test_format_redis_result_nil(self):
|
|
"""None 应格式化为 (nil)。"""
|
|
result = self.executor._format_redis_result(None)
|
|
assert result == "(nil)"
|
|
|
|
def test_format_redis_result_bool(self):
|
|
"""布尔值应格式化为 1/0。"""
|
|
assert self.executor._format_redis_result(True) == "1"
|
|
assert self.executor._format_redis_result(False) == "0"
|
|
|
|
def test_format_redis_result_list(self):
|
|
"""列表应格式化为方括号。"""
|
|
result = self.executor._format_redis_result([b"a", b"b", b"c"])
|
|
assert "[" in result
|
|
assert "a" in result
|
|
|
|
def test_format_redis_result_string(self):
|
|
"""字符串应原样返回。"""
|
|
result = self.executor._format_redis_result("OK")
|
|
assert result == "OK"
|
|
|
|
|
|
# ===== 路由分发 =====
|
|
|
|
|
|
class TestTicketExecutorRouting:
|
|
"""TicketExecutor 路由分发测试。"""
|
|
|
|
def setup_method(self):
|
|
self.executor = TicketExecutor()
|
|
|
|
def _make_ticket(self, sql_content, db_type="mysql", database="test_db"):
|
|
"""创建测试用 Ticket 对象。"""
|
|
ticket = MagicMock(spec=Ticket)
|
|
ticket.sql_content = sql_content
|
|
ticket.target_database = database
|
|
return ticket
|
|
|
|
def _make_instance(self, db_type="mysql"):
|
|
"""创建测试用 Instance 对象。"""
|
|
instance = MagicMock(spec=Instance)
|
|
instance.db_type = db_type
|
|
instance.host = "127.0.0.1"
|
|
instance.port = 3306
|
|
instance.username = "root"
|
|
instance.password = "test123"
|
|
instance.database = "test_db"
|
|
return instance
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unsupported_db_type_raises(self):
|
|
"""不支持的数据库类型应抛出 ValueError。"""
|
|
ticket = self._make_ticket("SELECT 1")
|
|
instance = self._make_instance("oracle")
|
|
with pytest.raises(ValueError, match="不支持的执行类型"):
|
|
await self.executor.execute(ticket, instance)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mysql_drop_rejected(self):
|
|
"""MySQL DROP 语句应被语法预校验拒绝。"""
|
|
ticket = self._make_ticket("DROP TABLE users")
|
|
instance = self._make_instance("mysql")
|
|
report = await self.executor.execute(ticket, instance)
|
|
assert report.status == "failed"
|
|
assert report.validation_passed is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_postgres_drop_rejected(self):
|
|
"""PostgreSQL DROP 语句应被语法预校验拒绝。"""
|
|
ticket = self._make_ticket("DROP TABLE users")
|
|
instance = self._make_instance("postgres")
|
|
report = await self.executor.execute(ticket, instance)
|
|
assert report.status == "failed"
|
|
assert report.validation_passed is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redis_empty_commands_rejected(self):
|
|
"""Redis 空命令应被拒绝。"""
|
|
ticket = self._make_ticket("")
|
|
instance = self._make_instance("redis")
|
|
report = await self.executor.execute(ticket, instance)
|
|
assert report.status == "failed"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mongo_invalid_json_rejected(self):
|
|
"""MongoDB 无效 JSON 应被拒绝。"""
|
|
ticket = self._make_ticket("not json")
|
|
instance = self._make_instance("mongo")
|
|
report = await self.executor.execute(ticket, instance)
|
|
assert report.status == "failed"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mongo_missing_collection_rejected(self):
|
|
"""MongoDB 缺少 collection 应被拒绝。"""
|
|
ticket = self._make_ticket(json.dumps({"operation": "insertOne"}))
|
|
instance = self._make_instance("mongo")
|
|
report = await self.executor.execute(ticket, instance)
|
|
assert report.status == "failed"
|