code: def test_wo_public_partitions(self, pre_aggregated: bool):
      # Arrange
      aggregate_params = pipeline_dp.AggregateParams(
          noise_kind=pipeline_dp.NoiseKind.GAUSSIAN,
          metrics=[
              pipeline_dp.Metrics.COUNT,
              pipeline_dp.Metrics.PRIVACY_ID_COUNT,
          ],
          max_partitions_contributed=1,
          max_contributions_per_partition=2,
      )
  
      # Input collection has 10 privacy ids where each privacy id
      # contributes to the same 10 partitions, three times in each partition.
      if not pre_aggregated:
          col = [(i, j) for i in range(10) for j in range(10)] * 3
      else:
          # This is a pre-aggregated dataset, namely each element has a format
          # (partition_key, (count, sum, num_partition_contributed).
          # And each element is in one-to-one correspondence to pairs
          # (privacy_id, partition_key) from the dataset.
          col = [(i, (3, 1, 10)) for i in range(10)] * 10
  
      if not pre_aggregated:
          data_extractors = pipeline_dp.DataExtractors(
              privacy_id_extractor=lambda x: x[0],
              partition_extractor=lambda x: f"pk{x[1]}",
              value_extractor=lambda x: 1,
          )
      else:
          data_extractors = pipeline_dp.PreAggregateExtractors(
              partition_extractor=lambda x: f"pk{x[0]}",
              preaggregate_extractor=lambda x: x[1],
          )
  
      col, per_partition_result = analysis.perform_utility_analysis(
          col=col,
          backend=pipeline_dp.LocalBackend(),
          options=analysis.UtilityAnalysisOptions(
              epsilon=3,
              delta=0.9,
              aggregate_params=aggregate_params,
              pre_aggregated_data=pre_aggregated,
          ),
          data_extractors=data_extractors,
      )
  
      col = list(col)
      per_partition_result = list(per_partition_result)
  
      # Assert
      self.assertLen(col, 1)
      report = col[0]
      self.assertIsInstance(report, metrics.UtilityReport)
      expected = metrics.UtilityReport(
          configuration_index=0,
          partitions_info=metrics.PartitionsInfo(
              public_partitions=False,
              num_dataset_partitions=10,
              num_non_public_partitions=None,
              num_empty_partitions=None,
              strategy=pipeline_dp.PartitionSelectionStrategy.TRUNCATED_GEOMETRIC,
              kept_partitions=metrics.MeanVariance(
                  mean=3.51622411, var=2.2798409
              ),
          ),
          metric_errors=[
              metrics.MetricUtility(
                  metric=pipeline_dp.Metrics.COUNT,
                  noise_std=1.380859375,
                  noise_kind=pipeline_dp.NoiseKind.GAUSSIAN,
                  ratio_data_dropped=metrics.DataDropInfo(
                      l0=0.6,
                      linf=0.333333333,
                      partition_selection=0.04322517259988915,
                  ),
                  absolute_error=metrics.ValueErrors(
                      bounding_errors=metrics.ContributionBoundingErrors(
                          l0=metrics.MeanVariance(mean=-18, var=3.6),
                          linf_min=0.0,
                          linf_max=-10,
                      ),
                      mean=-28,
                      variance=5.5067726,
                      rmse=28.098163153,
                      l1=0.0,
                      rmse_with_dropped_partitions=29.331271542782087,
                      l1_with_dropped_partitions=0.0,
                  ),
                  relative_error=metrics.ValueErrors(
                      bounding_errors=metrics.ContributionBoundingErrors(
                          l0=metrics.MeanVariance(mean=-0.6, var=0.004),
                          linf_min=0.0,
                          linf_max=-0.33333333,
                      ),
                      mean=-0.93333333,
                      variance=0.006118636237250433,
                      rmse=0.9366054384576044,
                      l1=0.0,
                      rmse_with_dropped_partitions=0.9777090514260699,
                      l1_with_dropped_partitions=0.0,
                  ),
              ),
              metrics.MetricUtility(
                  metric=pipeline_dp.Metrics.PRIVACY_ID_COUNT,
                  noise_std=0.6904296875,
                  noise_kind=pipeline_dp.NoiseKind.GAUSSIAN,
                  ratio_data_dropped=metrics.DataDropInfo(
                      l0=0.9,
                      linf=-0.0,
                      partition_selection=0.06483775889983372,
                  ),
                  absolute_error=metrics.ValueErrors(
                      bounding_errors=metrics.ContributionBoundingErrors(
                          l0=metrics.MeanVariance(mean=-9, var=0.9),
                          linf_min=0.0,
                          linf_max=0.0,
                      ),
                      mean=-9,
                      variance=1.37669315,
                      rmse=9.07616070,
                      l1=0.0,
                      rmse_with_dropped_partitions=9.67515739991,
                      l1_with_dropped_partitions=0.0,
                  ),
                  relative_error=metrics.ValueErrors(
                      bounding_errors=metrics.ContributionBoundingErrors(
                          l0=metrics.MeanVariance(mean=-0.9, var=0.009),
                          linf_min=0.0,
                          linf_max=0.0,
                      ),
                      mean=-0.9,
                      variance=0.013766931533,
                      rmse=0.90761607055726,
                      l1=0.0,
                      rmse_with_dropped_partitions=0.9675157399915,
                      l1_with_dropped_partitions=0.0,
                  ),
              ),
          ],
      )
      expected_copy = copy.deepcopy(expected)
      expected.utility_report_histogram = [
          metrics.UtilityReportBin(
              partition_size_from=20,
              partition_size_to=50,
              report=expected_copy,
          )
      ]
      common.assert_dataclasses_are_equal(self, report, expected)
      self.assertLen(per_partition_result, 10)
  
imports: import copy
  from absl.testing import absltest
  from absl.testing import parameterized
  import pipeline_dp
  import analysis
  from analysis import metrics
  from analysis import utility_analysis
  from analysis.tests import common
  from typing import Optional
label_desc: Use image node_id 4 for calling the UtilityAnalysis obj's underlying member method code with example usage: obj.test_wo_public_partitions(pre_aggregated) without return types