code: def test_user_affinity(spark, demo_usage_data, sar_settings, header):
      time_now = demo_usage_data[header["col_timestamp"]].max()
  
      model = SARPlus(
          spark,
          **header,
          timedecay_formula=True,
          time_decay_coefficient=30,
          time_now=time_now,
          similarity_type="cooccurrence",
      )
  
      df = spark.createDataFrame(demo_usage_data)
      model.fit(df)
  
      user_affinity_ref = pd.read_csv(
          sar_settings["FILE_DIR"] + "user_aff.csv"
      )
      user_affinity_ref = pd.melt(
          user_affinity_ref,
          user_affinity_ref.columns[0],
          user_affinity_ref.columns[1:],
          "ItemId",
          "Rating",
      )
      user_affinity_ref = user_affinity_ref[
          user_affinity_ref.Rating > 0
      ].reset_index(drop=True)
  
      # construct dataframe with test user id we'd like to get the affinity for
      df_test = spark.createDataFrame(
          pd.DataFrame(
              {header["col_user"]: [sar_settings["TEST_USER_ID"]]}
          )
      )
      user_affinity = (
          model.get_user_affinity(df_test)
          .toPandas()
          .reset_index(drop=True)
      )
  
      # verify the that item ids are the same
      assert (
          user_affinity[header["col_item"]] == user_affinity_ref.ItemId
      ).all()
  
      assert np.allclose(
          user_affinity_ref[header["col_rating"]].values,
          user_affinity["Rating"].values,
          atol=sar_settings["ATOL"],
      )
  
      # Set time_now to 60 days later
      user_affinity_ref = (
          pd.read_csv(
              sar_settings["FILE_DIR"] + "user_aff_2_months_later.csv"
          )
          .iloc[:, 1:]
          .squeeze()
      )
      user_affinity_ref = user_affinity_ref[user_affinity_ref > 0]
  
      two_months = 2 * 30 * (24 * 60 * 60)
      model = SARPlus(
          spark,
          **header,
          timedecay_formula=True,
          time_decay_coefficient=30,
          time_now=demo_usage_data[header["col_timestamp"]].max()
          + two_months,
          similarity_type="cooccurrence",
      )
      model.fit(spark.createDataFrame(demo_usage_data))
      df_test = pd.DataFrame(
          {header["col_user"]: [sar_settings["TEST_USER_ID"]]}
      )
      df_test = spark.createDataFrame(df_test)
      user_affinity = model.get_user_affinity(df_test).toPandas()
      user_affinity = user_affinity.set_index(header["col_item"])[
          header["col_rating"]
      ]
      user_affinity = user_affinity[user_affinity_ref.index]
  
      assert np.allclose(
          user_affinity_ref, user_affinity, atol=sar_settings["ATOL"]
      )
  
imports: import math
  from pathlib import Path
  import numpy
  import pandas
  from pyspark.sql import SparkSession
  import pytest
  from pandas.testing import assert_frame_equal
  from pysarplus import SARPlus, SARModel
label_desc: Use image node_id 13 for calling a global function with example usage: test_user_affinity(spark, demo_usage_data, sar_settings, header) without return types