DataOps-platform/tests/knowledge/test_device_scope.py

from __future__ import annotations

from datetime import UTC, datetime
from uuid import uuid4

import pytest


def _source(**overrides):
    from app.core.data_research.sources import IngestionSourceRecord

    values = {
        "uid": str(uuid4()),
        "source_type": "database",
        "name": "设备源",
        "config": {"database_type": "postgresql"},
        "permission_scope": {"departments": ["equipment"]},
        "status": "active",
        "created_by": "admin-1",
        "created_at": datetime(2026, 7, 29, tzinfo=UTC),
        "updated_at": datetime(2026, 7, 29, tzinfo=UTC),
    }
    values.update(overrides)
    return IngestionSourceRecord(**values)


class Repository:
    def __init__(self, records=()):
        self.records = {record.uid: record for record in records}
        self.saved = []

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

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

    def list_device_sources(self):
        return tuple(self.records.values())


def test_device_source_scope_is_sorted_unique_and_preserves_other_keys():
    from app.core.knowledge.device_scope import DeviceSourceScopeService

    source = _source()
    repository = Repository((source,))
    service = DeviceSourceScopeService(
        repository,
        clock=lambda: datetime(2026, 7, 30, tzinfo=UTC),
    )
    domain_a = str(uuid4())
    domain_b = str(uuid4())

    updated = service.update(
        source.uid,
        {"business_domains": [domain_b, domain_a, domain_b]},
        actor_is_admin=True,
    )

    assert updated.permission_scope == {
        "departments": ["equipment"],
        "business_domains": sorted([domain_a, domain_b]),
    }
    assert repository.saved == [updated]
    assert updated.updated_at == datetime(2026, 7, 30, tzinfo=UTC)


def test_device_source_scope_rejects_invalid_uuid_and_more_than_100_domains():
    from app.core.knowledge.device_scope import (
        DeviceSourceScopeInvalid,
        DeviceSourceScopeService,
    )

    source = _source()
    service = DeviceSourceScopeService(Repository((source,)))

    with pytest.raises(DeviceSourceScopeInvalid, match="UUID"):
        service.update(
            source.uid,
            {"business_domains": ["not-a-uuid"]},
            actor_is_admin=True,
        )
    with pytest.raises(DeviceSourceScopeInvalid, match="100"):
        service.update(
            source.uid,
            {"business_domains": [str(uuid4()) for _ in range(101)]},
            actor_is_admin=True,
        )


def test_device_source_scope_missing_source_and_payload_are_rejected():
    from app.core.knowledge.device_scope import (
        DeviceSourceScopeInvalid,
        DeviceSourceScopeNotFound,
        DeviceSourceScopeService,
    )

    service = DeviceSourceScopeService(Repository())

    with pytest.raises(DeviceSourceScopeNotFound):
        service.update(
            str(uuid4()),
            {"business_domains": []},
            actor_is_admin=True,
        )
    with pytest.raises(DeviceSourceScopeInvalid, match="business_domains"):
        service.update(
            str(uuid4()),
            {},
            actor_is_admin=True,
        )


def test_empty_scope_is_explicitly_admin_only_and_updates_require_admin():
    from app.core.knowledge.device_scope import (
        DeviceSourceScopeForbidden,
        DeviceSourceScopeService,
        source_scope_access,
    )

    source = _source(permission_scope={"business_domains": []})
    service = DeviceSourceScopeService(Repository((source,)))

    assert source_scope_access(source.permission_scope) == {
        "business_domains": (),
        "admin_only": True,
    }
    with pytest.raises(DeviceSourceScopeForbidden):
        service.update(
            source.uid,
            {"business_domains": [str(uuid4())]},
            actor_is_admin=False,
        )


def test_list_exposes_scope_metadata_but_never_source_configuration():
    from app.core.knowledge.device_scope import DeviceSourceScopeService

    source = _source(
        config={"password": "must-not-leak"},
        permission_scope={"business_domains": [str(uuid4())]},
    )
    service = DeviceSourceScopeService(Repository((source,)))

    records = service.list()

    assert records[0]["uid"] == source.uid
    assert records[0]["business_domains"] == tuple(
        source.permission_scope["business_domains"]
    )
    assert "config" not in records[0]
    assert "permission_scope" not in records[0]