tests/test_connector_readiness_boundaries.py

Rama main · réplica de lectura del repositorio original.

from types import SimpleNamespace

import pytest

from dm.connect.readiness import build_readiness_report


def _dataset(name="orders", *, columns=None, primary_key=None):
    return SimpleNamespace(
        name=name,
        columns=["id"] if columns is None else columns,
        primary_key=["id"] if primary_key is None else primary_key,
    )


def test_readiness_rejects_scalar_collection_inputs():
    for datasets in (None, "orders", {"orders": _dataset()}):
        with pytest.raises(ValueError):
            build_readiness_report("erp", datasets, [])
    for known in (None, "orders", {"orders": True}):
        with pytest.raises(ValueError):
            build_readiness_report("erp", [_dataset()], known)


def test_readiness_rejects_malformed_known_object_names():
    for known in ([" orders"], [""], ["orders\n"], [7]):
        with pytest.raises(ValueError):
            build_readiness_report("erp", [_dataset()], known)


def test_readiness_rejects_scalar_columns_and_primary_keys():
    with pytest.raises(ValueError, match="columns"):
        build_readiness_report("erp", [_dataset(columns="id")], ["orders"])
    with pytest.raises(ValueError, match="primary_key"):
        build_readiness_report("erp", [_dataset(primary_key="id")], ["orders"])


def test_readiness_rejects_duplicate_or_malformed_columns():
    with pytest.raises(ValueError, match="duplicate"):
        build_readiness_report("erp", [_dataset(columns=["id", "id"])], ["orders"])
    for columns in ([" id"], [""], ["name\n"], [7]):
        with pytest.raises(ValueError, match="column"):
            build_readiness_report("erp", [_dataset(columns=columns, primary_key=[])], ["orders"])


def test_readiness_rejects_duplicate_or_malformed_primary_key_columns():
    with pytest.raises(ValueError, match="duplicate"):
        build_readiness_report("erp", [_dataset(primary_key=["id", "id"])], ["orders"])
    for primary_key in ([" id"], [""], [7]):
        with pytest.raises(ValueError):
            build_readiness_report("erp", [_dataset(primary_key=primary_key)], ["orders"])


def test_readiness_rejects_primary_keys_outside_dataset_columns():
    with pytest.raises(ValueError, match="unknown columns"):
        build_readiness_report(
            "erp",
            [_dataset(columns=["id", "name"], primary_key=["missing_id"])],
            ["orders"],
        )


def test_readiness_keeps_stable_mapping_summary():
    report = build_readiness_report(
        "erp",
        [_dataset("orders"), _dataset("customers", primary_key=[])],
        (name for name in ["orders"]),
    )
    assert report["mapped_object_types"] == ["orders"]
    assert report["unmapped_tables"] == ["customers"]
    assert report["datasets"][0]["pk"] == ["id"]
    assert report["readiness"] == "1/2 表已有对象模型，1 表待建模"