feat: 完成 issue #30 ② PostgreSQL 关系表 schema（模板/模型/用户/权限）

core/data-bus/tests/test_postgres_schema.py

@@ -7,11 +7,22 @@ import unittest
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 import _bootstrap  # noqa: F401
 
-from data_bus.postgres_schema import DEFAULT_TABLES, generate_grant_ddl, generate_schema_ddl
-from data_bus.templating import TemplateNaming
+from data_bus.postgres_schema import (  # noqa: E402
+    DEFAULT_TABLES,
+    generate_from_config,
+    generate_grant_ddl,
+    generate_schema_ddl,
+    load_postgres_config,
+)
+from data_bus.templating import TemplateNaming  # noqa: E402
 
 NAMING = TemplateNaming(template="ti-cl4")
 
+CONFIG = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+    "config", "postgres.template.yaml",
+)
+
 
 class SchemaDDLTest(unittest.TestCase):
     def test_schema_and_tables(self):
@@ -35,5 +46,30 @@ class GrantDDLTest(unittest.TestCase):
         self.assertIn("GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA tpl_ti_cl4 TO databus_rw;", text)
 
 
+class ConfigDrivenTest(unittest.TestCase):
+    """模板配置资产（issue #30）：换行业只改配置生成 schema/授权。"""
+
+    def test_config_parses(self):
+        raw = load_postgres_config(CONFIG)
+        self.assertEqual(raw["template"], "ti-cl4")
+        self.assertIn("databus", raw["postgres"]["roles"])
+        self.assertEqual(raw["postgres"]["schema_tables"], DEFAULT_TABLES)
+
+    def test_generate_from_config(self):
+        ddl, grants = generate_from_config(CONFIG)
+        # schema 按资产 template 推导
+        self.assertIn("CREATE SCHEMA IF NOT EXISTS tpl_ti_cl4;", ddl)
+        for table in DEFAULT_TABLES:
+            self.assertIn(f"CREATE TABLE IF NOT EXISTS tpl_ti_cl4.{table} (", ddl)
+        # 角色来自资产声明（databus / modeler → 4 个角色）
+        text = "\n".join(grants)
+        for role in ("databus_ro", "databus_rw", "modeler_ro", "modeler_rw"):
+            self.assertIn(f"TO {role};", text)
+
+    def test_generate_from_config_with_naming_override(self):
+        ddl, grants = generate_from_config(CONFIG, naming=TemplateNaming("resin"))
+        self.assertIn("CREATE SCHEMA IF NOT EXISTS tpl_resin;", ddl)
+
+
 if __name__ == "__main__":
     unittest.main()