code: def test_data_match_for_cf(self):
      tenant_col = AccessAnomalyConfig.default_tenant_col
      user_col = AccessAnomalyConfig.default_user_col
      res_col = AccessAnomalyConfig.default_res_col
  
      df1 = materialized_cache(
          without_ffa(data_set.training).select(
              f.col(tenant_col).alias("df1_tenant"),
              f.col(user_col).alias("df1_user"),
              f.col(res_col).alias("df1_res"),
          ),
      )
  
      df2 = materialized_cache(
          without_ffa(data_set.training).select(
              f.col(tenant_col).alias("df2_tenant"),
              f.col(user_col).alias("df2_user"),
              f.col(res_col).alias("df2_res"),
          ),
      )
  
      df_joined_same_department = materialized_cache(
          df1.join(
              df2,
              (df1.df1_tenant == df2.df2_tenant)
              & (df1.df1_user != df2.df2_user)
              & (
                  get_department(df1.df1_user)
                  == get_department(df2.df2_user)
              )
              & (df1.df1_res == df2.df2_res),
          )
          .groupBy("df1_tenant", df1.df1_user, df2.df2_user)
          .agg(f.count("*").alias("count")),
      )
  
      stats_same_map = create_stats(
          df_joined_same_department, "df1_tenant", "count"
      )
  
      for stats_same in stats_same_map.get_stats():
          assert stats_same.count > 1
          assert 1 <= stats_same.min <= 2
          assert stats_same.max >= 9
          assert stats_same.mean >= 5
  
      df_joined_diff_department = materialized_cache(
          df1.join(
              df2,
              (df1.df1_tenant == df2.df2_tenant)
              & (df1.df1_user != df2.df2_user)
              & (
                  get_department(df1.df1_user)
                  != get_department(df2.df2_user)
              )
              & (df1.df1_res == df2.df2_res),
          )
          .groupBy("df1_tenant", df1.df1_user, df2.df2_user)
          .agg(f.count("*").alias("count")),
      )
  
      assert df_joined_diff_department.count() == 0
  
      assert (
          without_ffa(data_set.intra_test)
          .filter(get_department(user_col) == get_department(res_col))
          .count()
          == without_ffa(data_set.intra_test).count()
      )
  
      assert (
          without_ffa(data_set.inter_test)
          .filter(get_department(user_col) != get_department(res_col))
          .count()
          == without_ffa(data_set.inter_test).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 4 for calling the TestAccessAnomaly obj's underlying member method code with example usage: obj.test_data_match_for_cf() without return types