DataOps-platform/tests/data_research/test_database_source_registration.py

from __future__ import annotations

from types import SimpleNamespace

import pytest


class MemorySourceRepository:
    def __init__(self):
        self.records = {}

    def get(self, uid):
        return self.records.get(str(uid))

    def save(self, record):
        self.records[str(record.uid)] = record
        return record


def definition(**overrides):
    values = {
        "uid": "00000000-0000-0000-0000-000000000001",
        "name_en": "equipment_registry",
        "name_zh": "设备台账库",
        "database_type": "postgresql",
        "database": "equipment",
        "schema": "asset",
        "status": True,
        "host": "db.internal.example",
        "credential_ref": "vault://must-not-copy",
    }
    values.update(overrides)
    return SimpleNamespace(**values)


def test_database_source_registration_is_idempotent_and_secret_free():
    from app.core.data_research.sources import DatabaseSourceRegistrationService

    repository = MemorySourceRepository()
    current = definition()
    commits = []
    service = DatabaseSourceRegistrationService(
        repository,
        definition_resolver=lambda _uid: current,
        commit=lambda: commits.append("commit"),
    )

    first, first_created = service.ensure(current.uid, actor_uid="editor-1")
    second, second_created = service.ensure(current.uid, actor_uid="editor-2")

    assert first_created is True
    assert second_created is False
    assert second.uid == current.uid
    assert second.name == "设备台账库"
    assert second.config == {
        "database_type": "postgresql",
        "database": "equipment",
        "schema": "asset",
    }
    assert "host" not in second.config
    assert "credential" not in str(second.config).lower()
    assert commits == ["commit", "commit"]


@pytest.mark.parametrize(
    ("candidate", "message"),
    [
        (None, "not found"),
        (definition(status=False), "disabled"),
        (definition(database_type="oracle"), "not supported"),
    ],
)
def test_database_source_registration_rejects_unusable_definitions(
    candidate,
    message,
):
    from app.core.data_research.errors import IngestionSourceInvalid
    from app.core.data_research.sources import DatabaseSourceRegistrationService

    repository = MemorySourceRepository()
    service = DatabaseSourceRegistrationService(
        repository,
        definition_resolver=lambda _uid: candidate,
    )

    with pytest.raises(IngestionSourceInvalid, match=message):
        service.ensure(
            "00000000-0000-0000-0000-000000000001",
            actor_uid="editor-1",
        )

    assert repository.records == {}