code: def test_model_end2end(self):
      num_users = 10
      num_resources = 25
  
      tenant_col = AccessAnomalyConfig.default_tenant_col
      user_col = AccessAnomalyConfig.default_user_col
      user_vec_col = AccessAnomalyConfig.default_user_col + "_vec"
      res_col = AccessAnomalyConfig.default_res_col
      res_vec_col = AccessAnomalyConfig.default_res_col + "_vec"
      likelihood_col = AccessAnomalyConfig.default_likelihood_col
      output_col = AccessAnomalyConfig.default_output_col
  
      user_model_schema = t.StructType(
          [
              t.StructField(tenant_col, t.StringType(), False),
              t.StructField(user_col, t.StringType(), False),
              t.StructField(
                  user_vec_col, t.ArrayType(t.DoubleType()), False
              ),
          ],
      )
  
      res_model_schema = t.StructType(
          [
              t.StructField(tenant_col, t.StringType(), False),
              t.StructField(res_col, t.StringType(), False),
              t.StructField(
                  res_vec_col, t.ArrayType(t.DoubleType()), False
              ),
          ],
      )
  
      user_mapping_df = materialized_cache(
          spark.createDataFrame(
              [
                  [
                      0,
                      "roy_{0}".format(i),
                      [
                          float(i % 10),
                          float((i + 1) % (num_users / 2)),
                          0.0,
                          1.0,
                      ],
                  ]
                  for i in range(num_users)
              ],
              user_model_schema,
          ),
      )
  
      res_mapping_df = materialized_cache(
          spark.createDataFrame(
              [
                  [
                      0,
                      "res_{0}".format(i),
                      [
                          float(i % 10),
                          float((i + 1) % num_resources / 2),
                          1.0,
                          0.0,
                      ],
                  ]
                  for i in range(num_resources)
              ],
              res_model_schema,
          ),
      )
  
      df = (
          user_mapping_df.select(tenant_col, user_col)
          .distinct()
          .join(
              res_mapping_df.select(tenant_col, res_col).distinct(),
              tenant_col,
          )
          .withColumn(likelihood_col, f.lit(0.0))
      )
  
      assert df.count() == num_users * num_resources
  
      user_mapping_df = materialized_cache(
          spark.createDataFrame(
              [
                  [
                      0,
                      "roy_{0}".format(i),
                      [
                          float(i % 10),
                          float((i + 1) % (num_users / 2)),
                          0.0,
                          1.0,
                      ],
                  ]
                  for i in range(num_users)
              ],
              user_model_schema,
          ),
      )
  
      res_mapping_df = materialized_cache(
          spark.createDataFrame(
              [
                  [
                      0,
                      "res_{0}".format(i),
                      [
                          float(i % 10),
                          float((i + 1) % num_resources / 2),
                          1.0,
                          0.0,
                      ],
                  ]
                  for i in range(num_resources)
              ],
              res_model_schema,
          ),
      )
  
      df = (
          user_mapping_df.select(tenant_col, user_col)
          .distinct()
          .join(
              res_mapping_df.select(tenant_col, res_col).distinct(),
              tenant_col,
          )
          .withColumn(likelihood_col, f.lit(0.0))
      )
  
      assert df.count() == num_users * num_resources
  
      user_res_cf_mapping = UserResourceFeatureVectorMapping(
          tenant_col,
          user_col,
          user_vec_col,
          res_col,
          res_vec_col,
          df.select(tenant_col, user_col, res_col),
          None,
          None,
          user_mapping_df,
          res_mapping_df,
      )
  
      assert user_res_cf_mapping.check()
  
      model_normalizer = ModelNormalizeTransformer(df, 2)
  
      user_res_norm_cf_mapping = model_normalizer.transform(
          user_res_cf_mapping
      )
  
      assert user_res_cf_mapping.check()
      assert (
          user_res_norm_cf_mapping.user_feature_vector_mapping_df.count()
          == user_mapping_df.count()
      )
      assert (
          user_res_norm_cf_mapping.res_feature_vector_mapping_df.count()
          == res_mapping_df.count()
      )
  
      assert (
          user_res_norm_cf_mapping.user_feature_vector_mapping_df.filter(
              f.size(f.col(user_vec_col)) == 4,
          ).count()
          == user_mapping_df.count()
      )
  
      assert (
          user_res_norm_cf_mapping.res_feature_vector_mapping_df.filter(
              f.size(f.col(res_vec_col)) == 4,
          ).count()
          == res_mapping_df.count()
      )
  
      for preserve_history in [False, True]:
          an_model = AccessAnomalyModel(
              user_res_norm_cf_mapping, output_col
          )
          an_model.preserve_history = preserve_history
  
          fixed_df = an_model.transform(df)
          assert fixed_df is not None
          assert fixed_df.count() == df.count()
  
          stats_map = create_stats(fixed_df, tenant_col, output_col)
  
          for stats in stats_map.get_stats():
              if not preserve_history:
                  assert stats.min < -epsilon, stats
                  assert stats.max > epsilon, stats
                  assert abs(stats.mean) < epsilon, stats
                  assert abs(stats.std - 1.0) < epsilon, stats
              else:
                  assert stats.min == 0.0 and stats.max == 0.0
  
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 TestModelNormalizeTransformer obj's underlying member method code with example usage: obj.test_model_end2end() without return types