test(BA-6927): cover multi-project, multi-user, multi-kernel aggregation

One observation tick spanning two projects, three users and seven kernels,
with one user running in both projects.  The cross product compounded up
the hierarchy rather than applying a flat factor: this workload produced
1200 instead of 600 for a user (2x), 3600 instead of 1200 for a project
(3x) and 23100 instead of 3300 for the domain (7x), which is why the
domain figures looked most implausible in the report.

Also pins that user buckets stay keyed by (user, project) so a user
active in two projects does not collapse into a single bucket.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: jopemachine

tests/unit/manager/sokovan/scheduler/fair_share/test_aggregator_bucket_aggregation.py

@@ -31,6 +31,13 @@
 
 RESOURCE_GROUP_ID = ResourceGroupID(uuid4())
 
+_USER_A = UUID("11111111-1111-4111-8111-111111111111")
+_USER_B = UUID("22222222-2222-4222-8222-222222222222")
+_USER_C = UUID("33333333-3333-4333-8333-333333333333")
+_PROJECT_1 = UUID("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa")
+_PROJECT_2 = UUID("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb")
+_TICK_DAY = date(2026, 7, 17)
+
 
 def make_datetime(
     year: int, month: int, day: int, hour: int, minute: int, second: int = 0
@@ -830,3 +837,181 @@ def test_full_day_of_four_kernels(self, aggregator: FairShareAggregator) -> None
         )
         delta = result.user_usage_deltas[user_key]
         assert delta.resource_seconds["cuda.shares"] == Decimal("345600")
+
+
+@dataclass(frozen=True)
+class _Workload:
+    """A group of identical kernels one user runs inside one project."""
+
+    user_uuid: UUID
+    project_id: UUID
+    shares: Decimal
+    kernel_count: int
+
+
+@dataclass(frozen=True)
+class _UserBucketExpectation:
+    label: str
+    user_uuid: UUID
+    project_id: UUID
+    resource_seconds: Decimal
+    duration_seconds: int
+
+
+@dataclass(frozen=True)
+class _ProjectBucketExpectation:
+    label: str
+    project_id: UUID
+    resource_seconds: Decimal
+    duration_seconds: int
+
+
+class TestMultiTenantTick:
+    """One observation tick spanning two projects, three users and seven kernels.
+
+    The cross product does not merely double a number, it compounds up the
+    hierarchy: each level multiplies by the count of kernels *it* aggregates.
+    For this workload the old code produced 1200 instead of 600 for a user
+    (2x), 3600 instead of 1200 for a project (3x) and 23100 instead of 3300
+    for the domain (7x), which is why domain figures looked most absurd.
+
+    User A runs in both projects, so the user buckets must stay keyed by
+    (user, project) and must not collapse into one.
+    """
+
+    @pytest.fixture
+    def multi_tenant_specs(self) -> list[KernelUsageRecordCreatorSpec]:
+        """One 5-minute slice per kernel, all on the same day."""
+        workloads = [
+            _Workload(_USER_A, _PROJECT_1, Decimal("1"), 2),
+            _Workload(_USER_B, _PROJECT_1, Decimal("2"), 1),
+            _Workload(_USER_A, _PROJECT_2, Decimal("1"), 3),
+            _Workload(_USER_C, _PROJECT_2, Decimal("4"), 1),
+        ]
+        return [
+            make_spec(
+                period_start=make_datetime(2026, 7, 17, 10, 0, 0),
+                period_end=make_datetime(2026, 7, 17, 10, 5, 0),
+                resource_usage=ResourceSlot({"cuda.shares": workload.shares * 300}),
+                occupied_slots=ResourceSlot({"cuda.shares": workload.shares}),
+                user_uuid=workload.user_uuid,
+                project_id=workload.project_id,
+            )
+            for workload in workloads
+            for _ in range(workload.kernel_count)
+        ]
+
+    @pytest.mark.parametrize(
+        "case",
+        [
+            _UserBucketExpectation(
+                label="user-a-project-1",  # 2 kernels * 1 share * 300s
+                user_uuid=_USER_A,
+                project_id=_PROJECT_1,
+                resource_seconds=Decimal("600"),
+                duration_seconds=600,
+            ),
+            _UserBucketExpectation(
+                label="user-b-project-1",  # 1 kernel * 2 shares * 300s
+                user_uuid=_USER_B,
+                project_id=_PROJECT_1,
+                resource_seconds=Decimal("600"),
+                duration_seconds=300,
+            ),
+            _UserBucketExpectation(
+                label="user-a-project-2",  # 3 kernels * 1 share * 300s
+                user_uuid=_USER_A,
+                project_id=_PROJECT_2,
+                resource_seconds=Decimal("900"),
+                duration_seconds=900,
+            ),
+            _UserBucketExpectation(
+                label="user-c-project-2",  # 1 kernel * 4 shares * 300s
+                user_uuid=_USER_C,
+                project_id=_PROJECT_2,
+                resource_seconds=Decimal("1200"),
+                duration_seconds=300,
+            ),
+        ],
+        ids=lambda case: case.label,
+    )
+    def test_user_buckets_are_keyed_by_user_and_project(
+        self,
+        aggregator: FairShareAggregator,
+        multi_tenant_specs: list[KernelUsageRecordCreatorSpec],
+        case: _UserBucketExpectation,
+    ) -> None:
+        result = aggregator.aggregate_kernel_usage_to_buckets(multi_tenant_specs)
+
+        assert len(result.user_usage_deltas) == 4
+        delta = result.user_usage_deltas[
+            UserUsageBucketKey(
+                user_uuid=case.user_uuid,
+                project_id=case.project_id,
+                domain_name="default",
+                resource_group="default",
+                resource_group_id=RESOURCE_GROUP_ID,
+                period_date=_TICK_DAY,
+            )
+        ]
+        assert delta.resource_seconds["cuda.shares"] == case.resource_seconds
+        assert delta.duration_seconds == case.duration_seconds
+
+    @pytest.mark.parametrize(
+        "case",
+        [
+            _ProjectBucketExpectation(
+                label="project-1",  # user A 600 + user B 600
+                project_id=_PROJECT_1,
+                resource_seconds=Decimal("1200"),
+                duration_seconds=900,
+            ),
+            _ProjectBucketExpectation(
+                label="project-2",  # user A 900 + user C 1200
+                project_id=_PROJECT_2,
+                resource_seconds=Decimal("2100"),
+                duration_seconds=1200,
+            ),
+        ],
+        ids=lambda case: case.label,
+    )
+    def test_project_buckets_sum_their_users(
+        self,
+        aggregator: FairShareAggregator,
+        multi_tenant_specs: list[KernelUsageRecordCreatorSpec],
+        case: _ProjectBucketExpectation,
+    ) -> None:
+        result = aggregator.aggregate_kernel_usage_to_buckets(multi_tenant_specs)
+
+        assert len(result.project_usage_deltas) == 2
+        delta = result.project_usage_deltas[
+            ProjectUsageBucketKey(
+                project_id=case.project_id,
+                domain_name="default",
+                resource_group="default",
+                resource_group_id=RESOURCE_GROUP_ID,
+                period_date=_TICK_DAY,
+            )
+        ]
+        assert delta.resource_seconds["cuda.shares"] == case.resource_seconds
+        assert delta.duration_seconds == case.duration_seconds
+
+    def test_domain_bucket_sums_every_project(
+        self,
+        aggregator: FairShareAggregator,
+        multi_tenant_specs: list[KernelUsageRecordCreatorSpec],
+    ) -> None:
+        """Project 1 (1200) + project 2 (2100); the old code reported 23100."""
+        result = aggregator.aggregate_kernel_usage_to_buckets(multi_tenant_specs)
+
+        assert len(result.domain_usage_deltas) == 1
+        delta = result.domain_usage_deltas[
+            DomainUsageBucketKey(
+                domain_name="default",
+                resource_group="default",
+                resource_group_id=RESOURCE_GROUP_ID,
+                period_date=_TICK_DAY,
+            )
+        ]
+        assert delta.resource_seconds["cuda.shares"] == Decimal("3300")
+        assert delta.duration_seconds == 2100