code: def test_per_partition_error_metrics(self, pre_aggregated: bool):
      # Arrange
      aggregator_params = pipeline_dp.AggregateParams(
          noise_kind=pipeline_dp.NoiseKind.GAUSSIAN,
          metrics=[pipeline_dp.Metrics.COUNT],
          max_partitions_contributed=1,
          max_contributions_per_partition=2,
      )
  
      budget_accountant = pipeline_dp.NaiveBudgetAccountant(
          total_epsilon=2, total_delta=1e-10
      )
      data_extractors = pipeline_dp.DataExtractors(
          privacy_id_extractor=lambda x: x[0],
          partition_extractor=lambda x: f"pk{x[1]}",
          value_extractor=lambda x: 0,
      )
  
      # Input collection has 10 privacy ids where each privacy id
      # contributes to the same 10 partitions, three times in each partition.
      input = [(i, j) for i in range(10) for j in range(10)] * 3
      if pre_aggregated:
          input = analysis.pre_aggregation.preaggregate(
              input, pipeline_dp.LocalBackend(), data_extractors
          )
          data_extractors = (
              self._get_default_pre_aggregated_extractors()
          )
  
      engine = utility_analysis_engine.UtilityAnalysisEngine(
          budget_accountant=budget_accountant,
          backend=pipeline_dp.LocalBackend(),
      )
  
      options = analysis.UtilityAnalysisOptions(
          epsilon=1,
          delta=0,
          aggregate_params=aggregator_params,
          pre_aggregated_data=pre_aggregated,
      )
      output = engine.analyze(
          col=input, options=options, data_extractors=data_extractors
      )
      budget_accountant.compute_budgets()
  
      output = list(output)
  
      # Assert
      self.assertLen(output, 10)
      # Assert count metrics are correct.
      [
          self.assertEqual(v[1][2].clipping_to_max_error, -10)
          for v in output
      ]
      [
          self.assertAlmostEqual(
              v[1][2].expected_l0_bounding_error, -18.0, delta=1e-5
          )
          for v in output
      ]
      [
          self.assertAlmostEqual(
              v[1][2].std_l0_bounding_error, 1.89736, delta=1e-5
          )
          for v in output
      ]
      [
          self.assertAlmostEqual(
              v[1][2].std_noise, 11.95312, delta=1e-5
          )
          for v in output
      ]
  
file_path: pipelinedp/analysis/tests/utility_analysis_engine_test.py
imports: from absl.testing import absltest
  from absl.testing import parameterized
  from unittest.mock import patch
  import copy
  import pipeline_dp
  from pipeline_dp import budget_accounting
  from analysis import utility_analysis_engine
  from analysis import metrics
  import analysis
label_desc: Use image node_id 6 for calling the UtilityAnalysisEngineTest obj's underlying member method code with example usage: obj.test_per_partition_error_metrics(pre_aggregated) without return types