|
11 | 11 | """ |
12 | 12 |
|
13 | 13 | import ast |
| 14 | +from pathlib import Path |
14 | 15 | from typing import Any |
15 | 16 |
|
16 | 17 | import pytest |
@@ -544,3 +545,150 @@ async def test_all_templates_pass_structural_gate_with_hostile_text() -> None: |
544 | 545 | assert_python_structure(real, ref, filename=name) |
545 | 546 | checked += 1 |
546 | 547 | assert checked == 13 # 10 主模板 + 3 子表模板 |
| 548 | + |
| 549 | +# ══════════════════════════════════════════════════════════════════════════════ |
| 550 | +# 方言类型与跨方言 JSON（issue #542） |
| 551 | +# 生成的 import 必须可解析；jsonb 必须走基类 JSONType 而非硬编码 JSONB。 |
| 552 | +# 注意：GenConstant.DB_TO_SQLALCHEMY 按 settings.DATABASE_TYPE 条件定义， |
| 553 | +# 因此 PG 专有类型用 DATABASE_TYPE=postgres 的子进程验证（本进程 conftest 设为 sqlite）。 |
| 554 | +# ══════════════════════════════════════════════════════════════════════════════ |
| 555 | + |
| 556 | +_BACKEND_DIR = Path(__file__).resolve().parents[1] |
| 557 | + |
| 558 | +_POSTGRES_PROBE = r''' |
| 559 | +import asyncio, json |
| 560 | +from app.modules.generator.gencode.schema import GenTableColumnOutSchema, GenTableOutSchema |
| 561 | +from app.modules.generator.gencode.jinja2_template_util import Jinja2TemplateUtil |
| 562 | +
|
| 563 | +
|
| 564 | +def _col(name, column_type, python_type, **kw): |
| 565 | + data = dict(table_id=1, column_name=name, column_comment="x", column_type=column_type, column_length=None, |
| 566 | + column_default="", is_pk=False, is_increment=False, is_nullable=True, is_unique=False, |
| 567 | + python_type=python_type, python_field=name, is_insert=True, is_edit=True, is_list=True, |
| 568 | + is_query=False, query_type=None, html_type="input", dict_type="", sort=1) |
| 569 | + data.update(kw) |
| 570 | + return GenTableColumnOutSchema(**data) |
| 571 | +
|
| 572 | +
|
| 573 | +def _table(cols): |
| 574 | + return GenTableOutSchema(table_name="gen_demo", table_comment="示例", class_name="Demo", |
| 575 | + package_name="module_gen", module_name="demo", business_name="demo", function_name="示例", |
| 576 | + columns=cols, parent_menu_id=None, sub_table_name=None, sub_table_fk_name=None, |
| 577 | + pk_column=cols[0], sub=False, sub_table=None) |
| 578 | +
|
| 579 | +
|
| 580 | +result = {} |
| 581 | +for ct in ("jsonb", "inet", "cidr", "macaddr", "bit", "tsvector", "tsquery", "point"): |
| 582 | + table = _table([_col("payload", ct, "dict")]) |
| 583 | + result[ct] = { |
| 584 | + "imports": Jinja2TemplateUtil.get_model_import_list(table), |
| 585 | + "type": Jinja2TemplateUtil.get_sqlalchemy_type(table.columns[0]), |
| 586 | + } |
| 587 | +
|
| 588 | +
|
| 589 | +async def _render_model(): |
| 590 | + table = _table([_col("id", "integer", "int", is_pk=True, is_increment=True, sort=1), |
| 591 | + _col("payload", "jsonb", "dict", sort=2)]) |
| 592 | + env = Jinja2TemplateUtil.get_env() |
| 593 | + ctx = Jinja2TemplateUtil.prepare_context(table) |
| 594 | + for tpl in Jinja2TemplateUtil.get_template_list(): |
| 595 | + src = await env.get_template(tpl).render_async(**ctx) |
| 596 | + if "class Demo" in src and "__tablename__" in src: |
| 597 | + return src |
| 598 | + return "" |
| 599 | +
|
| 600 | +
|
| 601 | +result["model_source"] = asyncio.run(_render_model()) |
| 602 | +print("__RESULT__" + json.dumps(result)) |
| 603 | +''' |
| 604 | + |
| 605 | + |
| 606 | +def _postgres_dialect_probe() -> dict[str, Any]: |
| 607 | + """在 DATABASE_TYPE=postgres 的子进程里跑生成器，返回各类型的 import 与渲染结果。""" |
| 608 | + import json |
| 609 | + import os |
| 610 | + import subprocess |
| 611 | + import sys |
| 612 | + |
| 613 | + env = {**os.environ, "DATABASE_TYPE": "postgres"} |
| 614 | + proc = subprocess.run( |
| 615 | + [sys.executable, "-c", _POSTGRES_PROBE], |
| 616 | + cwd=_BACKEND_DIR, |
| 617 | + env=env, |
| 618 | + capture_output=True, |
| 619 | + text=True, |
| 620 | + timeout=180, |
| 621 | + ) |
| 622 | + assert proc.returncode == 0, f"PG 方言子进程失败: {proc.stderr[-2000:]}" |
| 623 | + line = next(line for line in proc.stdout.splitlines() if line.startswith("__RESULT__")) |
| 624 | + return json.loads(line[len("__RESULT__") :]) |
| 625 | + |
| 626 | + |
| 627 | +def test_dialect_type_import_map_targets_resolve() -> None: |
| 628 | + """DIALECT_TYPE_IMPORTS 的每个 (模块, 类名) 都必须真实存在（几何类型是可选依赖，跳过）。""" |
| 629 | + import importlib |
| 630 | + |
| 631 | + from app.modules.generator.gencode.template_safety import DIALECT_TYPE_IMPORTS |
| 632 | + |
| 633 | + for type_name, (module, symbol) in DIALECT_TYPE_IMPORTS.items(): |
| 634 | + if module == "geoalchemy2": |
| 635 | + continue |
| 636 | + mod = importlib.import_module(module) |
| 637 | + assert hasattr(mod, symbol), f"{type_name}: {module}.{symbol} 不存在" |
| 638 | + |
| 639 | + |
| 640 | +def test_postgres_dialect_types_emit_valid_imports() -> None: |
| 641 | + """PG 方言下：方言类型从正确模块导入，且不产生 `from sqlalchemy import <方言类型>`。""" |
| 642 | + result = _postgres_dialect_probe() |
| 643 | + expected = { |
| 644 | + "jsonb": ("app.core.base_model", "JSONType"), |
| 645 | + "inet": ("sqlalchemy.dialects.postgresql", "INET"), |
| 646 | + "cidr": ("sqlalchemy.dialects.postgresql", "CIDR"), |
| 647 | + "macaddr": ("sqlalchemy.dialects.postgresql", "MACADDR"), |
| 648 | + "bit": ("sqlalchemy.dialects.postgresql", "BIT"), |
| 649 | + "tsvector": ("sqlalchemy.dialects.postgresql", "TSVECTOR"), |
| 650 | + "tsquery": ("sqlalchemy.dialects.postgresql", "TSQUERY"), |
| 651 | + "point": ("geoalchemy2", "Geometry"), |
| 652 | + } |
| 653 | + for column_type, (module, symbol) in expected.items(): |
| 654 | + imports = result[column_type]["imports"] |
| 655 | + assert f"from {module} import {symbol}" in imports, f"{column_type} 的 import 不正确: {imports}" |
| 656 | + assert f"from sqlalchemy import {symbol}" not in imports, f"{column_type} 仍产生非法顶层 import" |
| 657 | + |
| 658 | + |
| 659 | +def test_postgres_jsonb_column_renders_cross_dialect_type() -> None: |
| 660 | + """PG 方言下 jsonb 列渲染为 JSONType（而非 JSONB）。""" |
| 661 | + result = _postgres_dialect_probe() |
| 662 | + assert result["jsonb"]["type"] == "JSONType" |
| 663 | + assert result["bit"]["type"] == "BIT" |
| 664 | + |
| 665 | + |
| 666 | +def test_postgres_rendered_model_imports_cross_dialect_type() -> None: |
| 667 | + """端到端：含 jsonb 列的模型产物可 ast.parse，且从基类导入 JSONType。""" |
| 668 | + result = _postgres_dialect_probe() |
| 669 | + source = result["model_source"] |
| 670 | + assert source, "未渲染出模型文件" |
| 671 | + assert "from app.core.base_model import JSONType" in source |
| 672 | + assert "from sqlalchemy import JSONB" not in source |
| 673 | + ast.parse(source) |
| 674 | + |
| 675 | + |
| 676 | +def test_json_type_compiles_on_all_supported_dialects() -> None: |
| 677 | + """JSONType：PG 渲染 JSONB，SQLite/MySQL 渲染 JSON（issue #542 的核心承诺）。""" |
| 678 | + from sqlalchemy import Column, Integer, MetaData, Table |
| 679 | + from sqlalchemy.dialects import mysql, postgresql, sqlite |
| 680 | + from sqlalchemy.schema import CreateTable |
| 681 | + |
| 682 | + from app.core.base_model import JSONType |
| 683 | + |
| 684 | + metadata = MetaData() |
| 685 | + table = Table("t_json_type", metadata, Column("id", Integer, primary_key=True), Column("payload", JSONType)) |
| 686 | + |
| 687 | + def _ddl(dialect: Any) -> str: |
| 688 | + return " ".join(str(CreateTable(table).compile(dialect=dialect)).split()) |
| 689 | + |
| 690 | + assert "payload JSONB" in _ddl(postgresql.dialect()) |
| 691 | + for dialect in (sqlite.dialect(), mysql.dialect()): |
| 692 | + ddl = _ddl(dialect) |
| 693 | + assert "payload JSON" in ddl, ddl |
| 694 | + assert "JSONB" not in ddl, ddl |
0 commit comments