code: def test(model, g, users_to_test, batch_test_flag=False):
      result = {
          "precision": np.zeros(len(Ks)),
          "recall": np.zeros(len(Ks)),
          "ndcg": np.zeros(len(Ks)),
          "hit_ratio": np.zeros(len(Ks)),
          "auc": 0.0,
      }
  
      pool = multiprocessing.Pool(cores)
  
      u_batch_size = 5000
      i_batch_size = BATCH_SIZE
  
      test_users = users_to_test
      n_test_users = len(test_users)
      n_user_batchs = n_test_users // u_batch_size + 1
  
      count = 0
  
      for u_batch_id in range(n_user_batchs):
          start = u_batch_id * u_batch_size
          end = (u_batch_id + 1) * u_batch_size
  
          user_batch = test_users[start:end]
  
          if batch_test_flag:
              # batch-item test
              n_item_batchs = ITEM_NUM // i_batch_size + 1
              rate_batch = np.zeros(shape=(len(user_batch), ITEM_NUM))
  
              i_count = 0
              for i_batch_id in range(n_item_batchs):
                  i_start = i_batch_id * i_batch_size
                  i_end = min((i_batch_id + 1) * i_batch_size, ITEM_NUM)
  
                  item_batch = range(i_start, i_end)
  
                  u_g_embeddings, pos_i_g_embeddings, _ = model(
                      g, "user", "item", user_batch, item_batch, []
                  )
                  i_rate_batch = (
                      model.rating(u_g_embeddings, pos_i_g_embeddings)
                      .detach()
                      .cpu()
                  )
  
                  rate_batch[:, i_start:i_end] = i_rate_batch
                  i_count += i_rate_batch.shape[1]
  
              assert i_count == ITEM_NUM
  
          else:
              # all-item test
              item_batch = range(ITEM_NUM)
              u_g_embeddings, pos_i_g_embeddings, _ = model(
                  g, "user", "item", user_batch, item_batch, []
              )
              rate_batch = (
                  model.rating(u_g_embeddings, pos_i_g_embeddings)
                  .detach()
                  .cpu()
              )
  
          user_batch_rating_uid = zip(rate_batch.numpy(), user_batch)
          batch_result = pool.map(test_one_user, user_batch_rating_uid)
          count += len(batch_result)
  
          for re in batch_result:
              result["precision"] += re["precision"] / n_test_users
              result["recall"] += re["recall"] / n_test_users
              result["ndcg"] += re["ndcg"] / n_test_users
              result["hit_ratio"] += re["hit_ratio"] / n_test_users
              result["auc"] += re["auc"] / n_test_users
  
      assert count == n_test_users
      pool.close()
      return result
  
file_path: dgl/examples/pytorch/NGCF/NGCF/utility/batch_test.py
imports: import heapq
  import multiprocessing
  import utility.metrics
  from utility.load_data import *
  from utility.parser import parse_args
label_desc: Use image node_id 6 for calling a global function with example usage: test(model, g, users_to_test, batch_test_flag) and returns: result