govoplan-core/tests/test_tenant_summary_batching.py

from __future__ import annotations

import unittest
from collections.abc import Mapping, Sequence
from unittest.mock import patch

from govoplan_core.core.modules import ModuleManifest
from govoplan_core.core.registry import PlatformRegistry
from govoplan_core.tenancy.service import tenant_counts_many


EXPECTED_EMPTY_COUNTS = {
    "users": 0,
    "active_users": 0,
    "groups": 0,
    "campaigns": 0,
    "files": 0,
    "api_keys": 0,
    "active_api_keys": 0,
}


class _BatchAccessAdministration:
    def __init__(self) -> None:
        self.calls: list[tuple[str, ...]] = []

    def tenant_counts_many(
        self,
        session: object,
        tenant_ids: Sequence[str],
    ) -> Mapping[str, Mapping[str, int]]:
        del session
        ids = tuple(tenant_ids)
        self.calls.append(ids)
        return {
            tenant_id: {
                "users": index + 1,
                "active_users": index + 1,
                "groups": 1,
                "api_keys": 0,
                "active_api_keys": 0,
            }
            for index, tenant_id in enumerate(ids)
        }


class TenantSummaryBatchingTests(unittest.TestCase):
    def test_absent_providers_return_complete_zero_summaries(self) -> None:
        registry = PlatformRegistry()
        with (
            patch("govoplan_core.tenancy.service.get_registry", return_value=registry),
            patch("govoplan_core.tenancy.service._access_administration", return_value=None),
        ):
            result = tenant_counts_many(object(), ["tenant-1", "tenant-2"])

        self.assertEqual(
            {
                "tenant-1": EXPECTED_EMPTY_COUNTS,
                "tenant-2": EXPECTED_EMPTY_COUNTS,
            },
            result,
        )

    def test_legacy_single_tenant_provider_remains_supported(self) -> None:
        calls: list[str] = []

        def single_provider(session: object, tenant_id: str) -> Mapping[str, int]:
            del session
            calls.append(tenant_id)
            return {"campaigns": 1}

        registry = PlatformRegistry()
        registry.register(
            ModuleManifest(
                id="campaigns",
                name="Campaigns",
                version="test",
                tenant_summary_providers=(single_provider,),
            )
        )
        with (
            patch("govoplan_core.tenancy.service.get_registry", return_value=registry),
            patch("govoplan_core.tenancy.service._access_administration", return_value=None),
        ):
            result = tenant_counts_many(
                object(),
                ["tenant-1", "tenant-2"],
                module_ids=("campaigns",),
            )

        self.assertEqual(["tenant-1", "tenant-2"], calls)
        self.assertEqual(1, result["tenant-1"]["campaigns"])
        self.assertEqual(1, result["tenant-2"]["campaigns"])

    def test_batch_provider_handles_single_item_and_partial_results(self) -> None:
        calls: list[tuple[str, ...]] = []

        def single_provider(session: object, tenant_id: str) -> Mapping[str, int]:
            raise AssertionError(f"single provider should not run for {tenant_id}")

        def batch_provider(
            session: object,
            tenant_ids: Sequence[str],
        ) -> Mapping[str, Mapping[str, int]]:
            del session
            calls.append(tuple(tenant_ids))
            return {tenant_ids[0]: {"files": 7}}

        registry = PlatformRegistry()
        registry.register(
            ModuleManifest(
                id="files",
                name="Files",
                version="test",
                tenant_summary_providers=(single_provider,),
                tenant_summary_batch_providers=(batch_provider,),
            )
        )
        with (
            patch("govoplan_core.tenancy.service.get_registry", return_value=registry),
            patch("govoplan_core.tenancy.service._access_administration", return_value=None),
        ):
            single = tenant_counts_many(object(), ["tenant-1"], module_ids=("files",))
            partial = tenant_counts_many(
                object(),
                ["tenant-1", "tenant-2"],
                module_ids=("files",),
            )

        self.assertEqual([("tenant-1",), ("tenant-1", "tenant-2")], calls)
        self.assertEqual(7, single["tenant-1"]["files"])
        self.assertEqual(7, partial["tenant-1"]["files"])
        self.assertEqual(0, partial["tenant-2"]["files"])

    def test_batch_provider_failures_remain_visible(self) -> None:
        def single_provider(session: object, tenant_id: str) -> Mapping[str, int]:
            return {}

        def failing_provider(
            session: object,
            tenant_ids: Sequence[str],
        ) -> Mapping[str, Mapping[str, int]]:
            raise RuntimeError("summary unavailable")

        registry = PlatformRegistry()
        registry.register(
            ModuleManifest(
                id="files",
                name="Files",
                version="test",
                tenant_summary_providers=(single_provider,),
                tenant_summary_batch_providers=(failing_provider,),
            )
        )
        with (
            patch("govoplan_core.tenancy.service.get_registry", return_value=registry),
            patch("govoplan_core.tenancy.service._access_administration", return_value=None),
            self.assertRaisesRegex(RuntimeError, "summary unavailable"),
        ):
            tenant_counts_many(object(), ["tenant-1"], module_ids=("files",))

    def test_provider_calls_are_bounded_for_large_pages(self) -> None:
        provider_calls = {"campaigns": 0, "files": 0}

        def single_provider(session: object, tenant_id: str) -> Mapping[str, int]:
            raise AssertionError(f"single provider should not run for {tenant_id}")

        def batch_provider(module_id: str):
            def provide(
                session: object,
                tenant_ids: Sequence[str],
            ) -> Mapping[str, Mapping[str, int]]:
                del session
                provider_calls[module_id] += 1
                return {tenant_id: {} for tenant_id in tenant_ids}

            return provide

        registry = PlatformRegistry()
        for module_id in ("campaigns", "files"):
            registry.register(
                ModuleManifest(
                    id=module_id,
                    name=module_id.title(),
                    version="test",
                    tenant_summary_providers=(single_provider,),
                    tenant_summary_batch_providers=(batch_provider(module_id),),
                )
            )
        access = _BatchAccessAdministration()
        tenant_ids = [f"tenant-{index}" for index in range(100)]
        with (
            patch("govoplan_core.tenancy.service.get_registry", return_value=registry),
            patch("govoplan_core.tenancy.service._access_administration", return_value=access),
        ):
            result = tenant_counts_many(
                object(),
                tenant_ids,
                module_ids=("campaigns", "files"),
            )

        self.assertEqual({"campaigns": 1, "files": 1}, provider_calls)
        self.assertEqual([tuple(tenant_ids)], access.calls)
        self.assertEqual(100, len(result))
        self.assertEqual(100, result["tenant-99"]["users"])

    def test_registry_replacement_carries_batch_providers(self) -> None:
        def single_provider(session: object, tenant_id: str) -> Mapping[str, int]:
            return {}

        def batch_provider(
            session: object,
            tenant_ids: Sequence[str],
        ) -> Mapping[str, Mapping[str, int]]:
            return {}

        registry = PlatformRegistry()
        registry.replace(
            [
                ModuleManifest(
                    id="files",
                    name="Files",
                    version="test",
                    tenant_summary_providers=(single_provider,),
                    tenant_summary_batch_providers=(batch_provider,),
                )
            ]
        )

        self.assertIs(
            batch_provider,
            registry.tenant_summary_batch_providers()["files"],
        )


if __name__ == "__main__":
    unittest.main()