code: def test_enrich_and_normalize(self):
      training = Dataset.create_new_training(1.0).cache()
  
      access_anomaly = AccessAnomaly(
          tenantCol=AccessAnomalyConfig.default_tenant_col,
          maxIter=10,
          applyImplicitCf=False,
      )
  
      tenant_col = access_anomaly.tenant_col
      user_col = access_anomaly.user_col
      indexed_user_col = access_anomaly.indexed_user_col
      res_col = access_anomaly.res_col
      indexed_res_col = access_anomaly.indexed_res_col
      scaled_likelihood_col = access_anomaly.scaled_likelihood_col
  
      assert training.filter(f.col(user_col).isNull()).count() == 0
      assert training.filter(f.col(res_col).isNull()).count() == 0
  
      the_indexer = indexers.MultiIndexer(
          indexers=[
              indexers.IdIndexer(
                  input_col=user_col,
                  partition_key=tenant_col,
                  output_col=indexed_user_col,
                  reset_per_partition=False,
              ),
              indexers.IdIndexer(
                  input_col=res_col,
                  partition_key=tenant_col,
                  output_col=indexed_res_col,
                  reset_per_partition=False,
              ),
          ],
      )
  
      the_indexer_model = the_indexer.fit(training)
      indexed_df = materialized_cache(
          the_indexer_model.transform(training)
      )
  
      assert (
          indexed_df.filter(f.col(indexed_user_col).isNull()).count()
          == 0
      )
      assert (
          indexed_df.filter(f.col(indexed_res_col).isNull()).count()
          == 0
      )
      assert (
          indexed_df.filter(f.col(indexed_user_col) <= 0).count() == 0
      )
      assert indexed_df.filter(f.col(indexed_res_col) <= 0).count() == 0
  
      unindexed_df = materialized_cache(
          the_indexer_model.undo_transform(indexed_df)
      )
      assert unindexed_df.filter(f.col(user_col).isNull()).count() == 0
      assert unindexed_df.filter(f.col(res_col).isNull()).count() == 0
  
      enriched_indexed_df = materialized_cache(
          access_anomaly._enrich_and_normalize(indexed_df),
      )
      enriched_df = materialized_cache(
          without_ffa(
              the_indexer_model.undo_transform(enriched_indexed_df)
          ),
      )
  
      assert enriched_df.filter(f.col(user_col).isNull()).count() == 0
      assert enriched_df.filter(f.col(res_col).isNull()).count() == 0
  
      assert (
          enriched_df.filter(
              (get_department(user_col) == get_department(res_col))
              & (f.col(scaled_likelihood_col) == 1.0),
          ).count()
          == 0
      )
  
      assert (
          enriched_df.filter(
              (get_department(user_col) != get_department(res_col))
              & (f.col(scaled_likelihood_col) != 1.0),
          ).count()
          == 0
      )
  
      assert (
          enriched_df.filter(
              (get_department(user_col) != get_department(res_col)),
          ).count()
          == enriched_df.filter(
              f.col(scaled_likelihood_col) == 1.0
          ).count()
      )
  
      assert (
          enriched_df.filter(
              (get_department(user_col) == get_department(res_col)),
          ).count()
          == enriched_df.filter(
              f.col(scaled_likelihood_col) != 1.0
          ).count()
      )
  
      low_value = access_anomaly.low_value
      high_value = access_anomaly.high_value
  
      assert enriched_df.count() > training.count()
      assert (
          enriched_df.filter(
              (
                  (f.col(scaled_likelihood_col) >= low_value)
                  & (f.col(scaled_likelihood_col) <= high_value)
              )
              | (f.col(scaled_likelihood_col) == 1.0),
          ).count()
          == enriched_df.count()
      )
  
imports: import os
  import tempfile
  import unittest
  from typing import Dict, Optional, Set, Type, Union
  from pandas.testing import assert_frame_equal
  from pyspark.sql import DataFrame, types, functions
  from synapse.ml.cyber.feature import indexers
  from synapse.ml.cyber import DataFactory
  from synapse.ml.cyber.anomaly.collaborative_filtering import AccessAnomaly, AccessAnomalyModel, AccessAnomalyConfig, ConnectedComponents, ModelNormalizeTransformer, _UserResourceFeatureVectorMapping
  from synapsemltest.cyber.explain_tester import ExplainTester
  from synapsemltest.spark import *
label_desc: Use image node_id 2 for calling the TestAccessAnomaly obj's underlying member method code with example usage: obj.test_enrich_and_normalize() without return types