458 lines
18 KiB
Python
458 lines
18 KiB
Python
"""
|
||
代码生成器模块单元测试
|
||
=============================================================
|
||
|
||
验证代码生成器 generator 模块的核心纯函数（无需数据库）：
|
||
- 表分页：_paginate_tables（总数/页数/切片/越界回退）
|
||
- 表项解析：_parse_table_item（管道分隔/纯表名/字典三种格式）
|
||
- 生成入口校验：generate_tables（空/非数组/超限抛 BizError，单项失败继续）
|
||
- 字段类型映射：TableSchemaParser（DB 类型 -> SQLAlchemy 类型、长度、choices）
|
||
- 代码块构建：_build_column_block / _build_form_field_block / _build_filters_literal
|
||
- 自动注册：_insert_module_import / _insert_blueprint（插入与幂等）
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from src.core.exceptions import BizError
|
||
from src.modules.generator import service as gen
|
||
from src.modules.generator.code_generator import (
|
||
CodeGenerator,
|
||
_build_column_block,
|
||
_build_filters_literal,
|
||
_build_form_field_block,
|
||
_choice_tag_type,
|
||
normalize_config,
|
||
)
|
||
from src.modules.generator.parser import TableSchemaParser
|
||
|
||
|
||
# ============================================================
|
||
# 表分页
|
||
# ============================================================
|
||
|
||
class TestPaginateTables:
|
||
"""内存分页：切片 + 页数计算 + 越界回退"""
|
||
|
||
def test_basic(self):
|
||
tables = [{'tableName': f't{i}'} for i in range(25)]
|
||
result = gen._paginate_tables(tables, 1, 10)
|
||
assert result['total'] == 25
|
||
assert result['pages'] == 3
|
||
assert result['size'] == 10
|
||
assert result['current'] == 1
|
||
assert len(result['records']) == 10
|
||
assert result['records'][0]['tableName'] == 't0'
|
||
|
||
def test_last_page(self):
|
||
tables = [{'tableName': f't{i}'} for i in range(25)]
|
||
result = gen._paginate_tables(tables, 3, 10)
|
||
assert result['current'] == 3
|
||
assert len(result['records']) == 5
|
||
|
||
def test_page_out_of_range_falls_back_to_first(self):
|
||
tables = [{'tableName': 't0'}] * 5
|
||
result = gen._paginate_tables(tables, 99, 10)
|
||
assert result['current'] == 1
|
||
assert result['pages'] == 1
|
||
|
||
def test_empty(self):
|
||
result = gen._paginate_tables([], 1, 10)
|
||
assert result['total'] == 0
|
||
assert result['pages'] == 0
|
||
assert result['records'] == []
|
||
|
||
|
||
# ============================================================
|
||
# 表项解析
|
||
# ============================================================
|
||
|
||
class TestParseTableItem:
|
||
"""tableNames 项三种格式解析"""
|
||
|
||
def test_pipe_format(self):
|
||
assert gen._parse_table_item('flask_article|文章管理表') == ('flask_article', '文章管理表')
|
||
|
||
def test_plain_name(self):
|
||
assert gen._parse_table_item('flask_article') == ('flask_article', '')
|
||
|
||
def test_pipe_with_empty_comment(self):
|
||
assert gen._parse_table_item('flask_article|') == ('flask_article', '')
|
||
|
||
def test_dict_format(self):
|
||
assert gen._parse_table_item({'name': 'flask_user', 'comment': '用户表'}) == ('flask_user', '用户表')
|
||
|
||
|
||
# ============================================================
|
||
# 生成入口校验
|
||
# ============================================================
|
||
|
||
class TestGenerateTables:
|
||
"""参数校验与批量容错"""
|
||
|
||
def test_missing_key_raises(self):
|
||
with pytest.raises(BizError, match='表名不能为空'):
|
||
gen.generate_tables({})
|
||
|
||
def test_empty_list_raises(self):
|
||
with pytest.raises(BizError, match='表名不能为空'):
|
||
gen.generate_tables({'tableNames': []})
|
||
|
||
def test_not_list_raises(self):
|
||
with pytest.raises(BizError, match='必须是数组'):
|
||
gen.generate_tables({'tableNames': 'flask_article'})
|
||
|
||
def test_too_many_raises(self):
|
||
names = [f't{i}' for i in range(51)]
|
||
with pytest.raises(BizError, match='最多支持'):
|
||
gen.generate_tables({'tableNames': names})
|
||
|
||
def test_success_returns_count(self, monkeypatch):
|
||
monkeypatch.setattr(gen, '_generate_one', lambda name, comment: None)
|
||
count = gen.generate_tables({'tableNames': ['flask_a|A表', 'flask_b']})
|
||
assert count == 2
|
||
|
||
def test_partial_failure_continues(self, monkeypatch):
|
||
def fake_generate(table_name, table_comment):
|
||
if table_name == 'bad':
|
||
raise RuntimeError('boom')
|
||
|
||
monkeypatch.setattr(gen, '_generate_one', fake_generate)
|
||
count = gen.generate_tables({'tableNames': ['good|A', 'bad', 'ok']})
|
||
assert count == 2
|
||
|
||
|
||
# ============================================================
|
||
# 字段类型映射（不查库，仅构造列字典）
|
||
# ============================================================
|
||
|
||
def _col(**overrides):
|
||
"""构造 information_schema 列字典"""
|
||
base = {
|
||
'DATA_TYPE': 'varchar',
|
||
'COLUMN_NAME': 'title',
|
||
'COLUMN_COMMENT': '标题',
|
||
'IS_NULLABLE': 'NO',
|
||
'CHARACTER_MAXIMUM_LENGTH': 255,
|
||
'NUMERIC_PRECISION': None,
|
||
'NUMERIC_SCALE': None,
|
||
'COLUMN_DEFAULT': None,
|
||
'EXTRA': '',
|
||
}
|
||
base.update(overrides)
|
||
return base
|
||
|
||
|
||
class TestFieldTypeMapping:
|
||
"""DB 数据类型 -> SQLAlchemy 类型 / 构造参数 / choices"""
|
||
|
||
def test_varchar_to_string(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
assert parser._get_field_type(_col(DATA_TYPE='varchar')) == 'String'
|
||
|
||
def test_text_to_text(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
assert parser._get_field_type(_col(DATA_TYPE='longtext')) == 'Text'
|
||
|
||
def test_bigint_to_biginteger(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
assert parser._get_field_type(_col(DATA_TYPE='bigint')) == 'BigInteger'
|
||
|
||
def test_decimal_to_numeric(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
assert parser._get_field_type(_col(DATA_TYPE='decimal')) == 'Numeric'
|
||
|
||
def test_db_args_string_length(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
assert parser._get_db_args(_col(DATA_TYPE='varchar', CHARACTER_MAXIMUM_LENGTH=100)) == '100'
|
||
|
||
def test_db_args_numeric_precision(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
assert parser._get_db_args(_col(DATA_TYPE='decimal', NUMERIC_PRECISION=10, NUMERIC_SCALE=2)) == '10, 2'
|
||
|
||
def test_db_args_none_for_text(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
assert parser._get_db_args(_col(DATA_TYPE='text')) is None
|
||
|
||
def test_form_type_integer(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
assert parser._get_form_type(_col(DATA_TYPE='int')) == 'IntegerField'
|
||
|
||
def test_choices_parse(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
choices, min_value, max_value = parser._get_choices(_col(COLUMN_COMMENT='0-正常 1-下架'))
|
||
assert choices == [(0, '正常'), (1, '下架')]
|
||
assert (min_value, max_value) == (0, 1)
|
||
|
||
def test_choices_none(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
choices, min_value, max_value = parser._get_choices(_col(COLUMN_COMMENT='标题'))
|
||
assert choices is None
|
||
assert min_value is None
|
||
|
||
def test_filter_type_string_fuzzy(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
assert parser._get_filter_type(_col(COLUMN_NAME='title', DATA_TYPE='varchar')) == '模糊查询'
|
||
|
||
def test_filter_type_code_exact(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
assert parser._get_filter_type(_col(COLUMN_NAME='code', DATA_TYPE='varchar')) == '精确匹配'
|
||
|
||
def test_numeric_default_normalized(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
col = _col(DATA_TYPE='int', COLUMN_NAME='sort', COLUMN_DEFAULT='0')
|
||
fields = parser._build_fields([col], set())
|
||
# MySQL 默认值以字符串返回，整数列应转为 int（生成 default=0 而非 default='0'）
|
||
assert fields[0]['default_repr'] == '0'
|
||
assert isinstance(fields[0]['default'], int)
|
||
|
||
def test_string_default_kept_as_string(self):
|
||
parser = TableSchemaParser('flask_t')
|
||
col = _col(DATA_TYPE='varchar', COLUMN_NAME='name', COLUMN_DEFAULT="'normal'")
|
||
fields = parser._build_fields([col], set())
|
||
assert fields[0]['default'] == 'normal'
|
||
assert fields[0]['default_repr'] == "'normal'"
|
||
|
||
|
||
# ============================================================
|
||
# 代码块构建
|
||
# ============================================================
|
||
|
||
class TestBlockBuilders:
|
||
"""models/schema/repository 代码块预构建"""
|
||
|
||
def test_column_block(self):
|
||
field = dict(name='title', comment='标题', py_comment='标题', db_type='String', db_args='200',
|
||
nullable=False, default_repr=None, is_unique=False, db_index=False)
|
||
block = _build_column_block(field)
|
||
assert 'title = db.Column(' in block
|
||
assert 'db.String(200)' in block
|
||
assert 'nullable=False' in block
|
||
assert 'comment="标题"' in block
|
||
|
||
def test_column_block_with_default_unique_index(self):
|
||
field = dict(name='status', comment='状态', py_comment='状态', db_type='Integer', db_args=None,
|
||
nullable=False, default_repr='1', is_unique=True, db_index=True)
|
||
block = _build_column_block(field)
|
||
assert 'db.Integer' in block
|
||
assert 'default=1' in block
|
||
assert 'unique=True' in block
|
||
assert 'index=True' in block
|
||
|
||
def test_form_field_block(self):
|
||
field = dict(camel_name='title', form_type='StringField', comment='标题', py_comment='标题',
|
||
required=True, max_length=200, min_value=None, max_value=None)
|
||
block = _build_form_field_block(field)
|
||
assert 'title = StringField(' in block
|
||
assert "DataRequired(message='标题不能为空')" in block
|
||
assert 'Length(max=200' in block
|
||
|
||
def test_filters_literal_empty(self):
|
||
assert _build_filters_literal([]) == '{}'
|
||
|
||
def test_filters_literal(self):
|
||
literal = _build_filters_literal([('status', 'status'), ('name', 'name')])
|
||
assert "'status': params.get('status')," in literal
|
||
assert "'name': params.get('name')," in literal
|
||
|
||
|
||
# ============================================================
|
||
# 自动注册辅助
|
||
# ============================================================
|
||
|
||
# ============================================================
|
||
# 配置标准化（兼容 config.json 参考配置）
|
||
# ============================================================
|
||
|
||
class TestNormalizeConfig:
|
||
"""config.json 参考字段类型 -> SQLAlchemy 类型转换"""
|
||
|
||
def test_django_config_normalized(self):
|
||
config = {
|
||
'app_name': 'example',
|
||
'fields': [
|
||
{'name': 'name', 'type': 'CharField', 'max_length': 100, 'comment': '名称', 'default': None},
|
||
{'name': 'status', 'type': 'IntegerField', 'comment': '状态', 'default': '1'},
|
||
{'name': 'sort', 'type': 'IntegerField', 'comment': '排序', 'default': '0'},
|
||
],
|
||
}
|
||
out = normalize_config(config)
|
||
|
||
name_field = out['fields'][0]
|
||
assert name_field['db_type'] == 'String'
|
||
assert name_field['db_args'] == '100'
|
||
assert name_field['form_type'] == 'StringField'
|
||
assert name_field['camel_name'] == 'name'
|
||
|
||
status_field = out['fields'][1]
|
||
assert status_field['db_type'] == 'Integer'
|
||
assert status_field['form_type'] == 'IntegerField'
|
||
assert status_field['default'] == 1 # 字符串默认值转 int
|
||
assert status_field['default_repr'] == '1'
|
||
assert status_field['is_status'] is True
|
||
|
||
sort_field = out['fields'][2]
|
||
assert sort_field['is_sort'] is True
|
||
|
||
def test_flask_config_passthrough(self):
|
||
# 已含 db_type 的 Flask 配置原样返回（幂等）
|
||
config = {'app_name': 'x', 'fields': [{'name': 'a', 'db_type': 'String'}]}
|
||
assert normalize_config(config) is config
|
||
|
||
def test_empty_fields_passthrough(self):
|
||
config = {'app_name': 'x', 'fields': []}
|
||
assert normalize_config(config) is config
|
||
|
||
|
||
# ============================================================
|
||
# 前端 Vue 上下文
|
||
# ============================================================
|
||
|
||
def _gen_config(fields, **extra):
|
||
"""构造最小生成配置"""
|
||
base = {
|
||
'app_name': 'example', 'module_comment': '案例', 'model_class_name': 'Example',
|
||
'permission_prefix': 'sys:example', 'route_prefix': 'example', 'primary_key': 'id',
|
||
'display_field': 'name', 'display_field_camel': 'name',
|
||
'has_sort': True, 'has_status': True, 'has_unique_code': False, 'has_export': False,
|
||
'fields': fields,
|
||
}
|
||
base.update(extra)
|
||
return base
|
||
|
||
|
||
def _vf(name, db_type='String', comment='字段', choices=None, filterable=False, **kwargs):
|
||
"""构造最小字段配置"""
|
||
field = {
|
||
'name': name, 'camel_name': name, 'comment': comment, 'py_comment': comment,
|
||
'db_type': db_type, 'db_args': '100' if db_type == 'String' else None,
|
||
'form_type': 'StringField', 'type_display': '字符串',
|
||
'max_length': 100 if db_type == 'String' else None,
|
||
'nullable': False, 'required': True, 'default_repr': None, 'is_unique': False,
|
||
'db_index': False, 'default': None, 'filterable': filterable,
|
||
'filter_type': '模糊查询' if filterable else '',
|
||
'in_form': True, 'in_list': True, 'editable': True, 'searchable': filterable,
|
||
'is_image': False, 'is_rich_text': False, 'is_sort': False, 'is_status': False,
|
||
'choices': choices, 'min_value': None, 'max_value': None,
|
||
}
|
||
field.update(kwargs)
|
||
return field
|
||
|
||
|
||
class TestCleanFieldComment:
|
||
"""清洗注释中的选项说明"""
|
||
|
||
def test_strip_choices(self):
|
||
gen = CodeGenerator(_gen_config([]))
|
||
assert gen._clean_field_comment('案例类型：1-类型1 2-类型2') == '案例类型'
|
||
assert gen._clean_field_comment('案例状态：1-正常 2-禁用') == '案例状态'
|
||
|
||
def test_plain(self):
|
||
gen = CodeGenerator(_gen_config([]))
|
||
assert gen._clean_field_comment('标题') == '标题'
|
||
assert gen._clean_field_comment('') == ''
|
||
|
||
|
||
class TestBuildChoices:
|
||
"""选项元组 -> 前端字典列表"""
|
||
|
||
def test_transformed(self):
|
||
gen = CodeGenerator(_gen_config([]))
|
||
assert gen._build_choices([(1, '正常'), (2, '禁用')]) == [
|
||
{'value': 1, 'label': '正常', 'type': 'success'},
|
||
{'value': 2, 'label': '禁用', 'type': 'danger'},
|
||
]
|
||
|
||
def test_none(self):
|
||
gen = CodeGenerator(_gen_config([]))
|
||
assert gen._build_choices(None) is None
|
||
|
||
|
||
class TestChoiceTagType:
|
||
"""选项值 -> 标签颜色"""
|
||
|
||
def test_values(self):
|
||
assert _choice_tag_type(1) == 'success'
|
||
assert _choice_tag_type(2) == 'danger'
|
||
assert _choice_tag_type(3) == 'warning'
|
||
assert _choice_tag_type(9) == 'info'
|
||
|
||
|
||
class TestVueContext:
|
||
"""前端模板上下文：普通列表与树状列表"""
|
||
|
||
def test_normal_context(self):
|
||
fields = [
|
||
_vf('name', comment='案例名称', filterable=True),
|
||
_vf('status', db_type='Integer', comment='案例状态：1-正常 2-禁用',
|
||
choices=[(1, '正常'), (2, '禁用')], filterable=True),
|
||
]
|
||
gen = CodeGenerator(_gen_config(fields))
|
||
vctx = gen._prepare_vue_context(gen._prepare_context())
|
||
|
||
assert vctx['is_tree_structure'] is False
|
||
assert vctx['route_prefix'] == 'example' # 覆盖为 app_name，与后端蓝图一致
|
||
assert vctx['model_class_name_camel'] == 'example'
|
||
assert vctx['module_name'] == 'example'
|
||
assert vctx['api_path'] == 'tool/example'
|
||
|
||
# 字段重塑：type 转配置字段类型名、label 清洗、choices 转字典
|
||
name_field = next(f for f in vctx['fields'] if f['name'] == 'name')
|
||
assert name_field['type'] == 'CharField'
|
||
assert name_field['label'] == '案例名称'
|
||
status_field = next(f for f in vctx['fields'] if f['name'] == 'status')
|
||
assert status_field['type'] == 'IntegerField'
|
||
assert status_field['label'] == '案例状态'
|
||
assert status_field['choices'][0] == {'value': 1, 'label': '正常', 'type': 'success'}
|
||
|
||
def test_tree_context(self):
|
||
fields = [_vf('name', comment='部门名称'), _vf('parent_id', db_type='Integer', comment='上级部门')]
|
||
gen = CodeGenerator(_gen_config(fields))
|
||
vctx = gen._prepare_vue_context(gen._prepare_context())
|
||
|
||
assert vctx['is_tree_structure'] is True
|
||
assert vctx['parent_id_field'] == 'parentId'
|
||
assert vctx['has_search_form'] is False # 树形默认不显示搜索表单
|
||
|
||
def test_module_name_underscore_stripped(self):
|
||
gen = CodeGenerator(_gen_config([], app_name='order_item', model_class_name='OrderItem'))
|
||
vctx = gen._prepare_vue_context(gen._prepare_context())
|
||
assert vctx['module_name'] == 'orderitem'
|
||
assert vctx['model_class_name_camel'] == 'orderItem'
|
||
|
||
def test_filter_fields(self):
|
||
fields = [_vf('name', filterable=True), _vf('type', db_type='Integer', filterable=True)]
|
||
gen = CodeGenerator(_gen_config(fields))
|
||
vctx = gen._prepare_vue_context(gen._prepare_context())
|
||
assert [f['name'] for f in vctx['filter_fields']] == ['name', 'type']
|
||
assert vctx['has_search_form'] is True
|
||
|
||
|
||
class TestRegisterHelpers:
|
||
"""_insert_module_import / _insert_blueprint 插入与幂等"""
|
||
|
||
def test_insert_module_import(self):
|
||
content = "from src.modules.auth.router import login_blue\nfrom src.modules.city.router import city_blue\n"
|
||
out = CodeGenerator._insert_module_import(
|
||
content, 'from src.modules.generator.router import generator_blue'
|
||
)
|
||
assert 'from src.modules.generator.router import generator_blue\n' in out
|
||
# 插入位置应在最后一个 src.modules 导入之后
|
||
assert out.index('generator') > out.index('city_blue')
|
||
|
||
def test_insert_module_import_idempotent(self):
|
||
content = "from src.modules.city.router import city_blue\n"
|
||
out = CodeGenerator._insert_module_import(content, 'from src.modules.city.router import city_blue')
|
||
assert out == content
|
||
|
||
def test_insert_blueprint(self):
|
||
content = "BLUEPRINTS = [\n login_blue,\n city_blue,\n]\n"
|
||
out = CodeGenerator._insert_blueprint(content, 'generator_blue')
|
||
assert 'generator_blue' in out
|
||
assert out.index('generator_blue') > out.index('city_blue')
|
||
assert out.endswith(' generator_blue,\n]\n')
|
||
|
||
def test_insert_blueprint_idempotent(self):
|
||
content = "BLUEPRINTS = [\n generator_blue,\n]\n"
|
||
out = CodeGenerator._insert_blueprint(content, 'generator_blue')
|
||
assert out == content
|