100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
from sqlalchemy import create_mock_engine
|
|
from sqlalchemy.dialects import postgresql
|
|
|
|
|
|
class ExternalSystemMigrationTests(unittest.TestCase):
|
|
def test_0027_upgrade_compiles_as_postgresql_ddl(self):
|
|
statements: list[str] = []
|
|
|
|
def capture(sql, *multiparams, **params):
|
|
statements.append(str(sql.compile(dialect=postgresql.dialect())))
|
|
|
|
engine = create_mock_engine("postgresql+psycopg://", capture)
|
|
connection = engine.connect()
|
|
operations = Operations(MigrationContext.configure(connection))
|
|
migration = importlib.import_module(
|
|
"db.migrations.versions.20260807_1000_0027_external_system_governance"
|
|
)
|
|
with patch.object(migration, "op", operations):
|
|
migration.upgrade()
|
|
|
|
rendered = "\n".join(statements)
|
|
self.assertIn("external_system_grants", rendered)
|
|
self.assertIn("external_system_audits", rendered)
|
|
self.assertIn("operation_policies", rendered)
|
|
self.assertIn("operation_mode", rendered)
|
|
|
|
|
|
class FactoryDefinitionDataMigrationTests(unittest.TestCase):
|
|
def test_materializes_factory_defaults_as_generic_query_config(self):
|
|
from scripts.migrate_factory_mes_definitions import migrated_config
|
|
|
|
config = migrated_config(
|
|
{
|
|
"base_url": "https://factory.invalid",
|
|
"openapi_url": "https://factory.invalid/swagger.json",
|
|
"allowed_post_operations": ["quality_report"],
|
|
}
|
|
)
|
|
|
|
self.assertEqual(config["auth_type"], "password_jwt")
|
|
self.assertEqual(config["operation_mode"], "query")
|
|
self.assertEqual(
|
|
config["operation_policies"],
|
|
{"bi_dataset_exec": "read", "quality_report": "read"},
|
|
)
|
|
self.assertNotIn("allowed_post_operations", config)
|
|
self.assertEqual(
|
|
config["recommended_operation_ids"],
|
|
["bi_dataset_list", "bi_dataset_exec"],
|
|
)
|
|
|
|
def test_explicit_factory_config_is_preserved_except_execution_mode(self):
|
|
from scripts.migrate_factory_mes_definitions import migrated_config
|
|
|
|
config = migrated_config(
|
|
{
|
|
"base_url": "https://factory.invalid/api",
|
|
"openapi_url": "https://factory.invalid/openapi.json",
|
|
"login_path": "/login",
|
|
"token_field": "data.token",
|
|
"query_guidance": "使用质量聚合接口",
|
|
"recommended_operation_ids": ["quality_summary"],
|
|
"operation_mode": "upstream_managed",
|
|
"operation_policies": {"quality_summary": "export"},
|
|
}
|
|
)
|
|
|
|
self.assertEqual(config["base_url"], "https://factory.invalid/api")
|
|
self.assertEqual(config["login_path"], "/login")
|
|
self.assertEqual(config["token_field"], "data.token")
|
|
self.assertEqual(config["query_guidance"], "使用质量聚合接口")
|
|
self.assertEqual(config["recommended_operation_ids"], ["quality_summary"])
|
|
self.assertEqual(config["operation_mode"], "query")
|
|
self.assertEqual(
|
|
config["operation_policies"],
|
|
{"bi_dataset_exec": "read", "quality_summary": "export"},
|
|
)
|
|
|
|
def test_runtime_provider_registry_has_no_factory_preset(self):
|
|
from core.external_systems.registry import get_provider, provider_specs
|
|
|
|
self.assertEqual(
|
|
{spec.provider for spec in provider_specs()},
|
|
{"generic_openapi", "generic_mcp"},
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "不支持"):
|
|
get_provider("factory_mes")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|