class_docstr: This class tests all member functions of crypten package
class_docstr_tok: ["This","class","tests","all","member","functions","of","crypten","package"]
docstr: Test that crypten.save_from_party and crypten.load_from_party
  properly save and load plaintext tensors
code: def test_plaintext_save_load_from_party(self):
      """Test that crypten.save_from_party and crypten.load_from_party
      properly save and load plaintext tensors"""
      import tempfile
  
      import numpy as np
  
      def custom_load_function(f):
          np_arr = np.load(f)
          tensor = torch.from_numpy(np_arr)
          return tensor
  
      def custom_save_function(obj, f):
          np_arr = obj.numpy()
          np.save(f, np_arr)
  
      comm = crypten.communicator
      filename = tempfile.NamedTemporaryFile(delete=True).name
      all_save_fns = [torch.save, custom_save_function]
      all_load_fns = [torch.load, custom_load_function]
      all_file_completions = [".pth", ".npy"]
      all_test_load_fns = [torch.load, np.load]
      for dimensions in range(1, 5):
          # Create tensors with different sizes on each rank
          size = [self.rank + 1] * dimensions
          size = tuple(size)
          tensor = torch.randn(size=size)
  
          for i, save_closure in enumerate(all_save_fns):
              load_closure = all_load_fns[i]
              test_load_fn = all_test_load_fns[i]
              complete_file = filename + all_file_completions[i]
              for src in range(comm.get().get_world_size()):
                  crypten.save_from_party(
                      tensor,
                      complete_file,
                      src=src,
                      save_closure=save_closure,
                  )
  
                  # the following line will throw an error if an object saved with
                  # torch.save is attempted to be loaded with np.load
                  if self.rank == src:
                      test_load_fn(complete_file)
  
                  encrypted_load = crypten.load_from_party(
                      complete_file, src=src, load_closure=load_closure
                  )
  
                  reference_size = tuple([src + 1] * dimensions)
                  self.assertEqual(
                      encrypted_load.size(), reference_size
                  )
  
                  size_out = [src + 1] * dimensions
                  reference = (
                      tensor
                      if self.rank == src
                      else torch.empty(size=size_out)
                  )
                  comm.get().broadcast(reference, src)
                  self._check(
                      encrypted_load, reference, "crypten.load() failed"
                  )
  
                  # test for invalid load_closure
                  with self.assertRaises(TypeError):
                      crypten.load_from_party(
                          complete_file,
                          src=src,
                          load_closure=(lambda f: None),
                      )
  
                  # test pre-loaded
                  encrypted_preloaded = crypten.load_from_party(
                      src=src, preloaded=tensor
                  )
                  self._check(
                      encrypted_preloaded,
                      reference,
                      "crypten.load() failed using preloaded",
                  )
  
imports: import itertools
  import logging
  import math
  import random
  import unittest
  from collections import defaultdict
  import crypten
  import crypten.communicator
  import torch
  import torch.nn.functional
  from crypten.common import serial
  from crypten.common.tensor_types import is_float_tensor
  from crypten.config import cfg
  from test.multiprocess_test_case import get_random_test_tensor, MultiProcessTestCase
  from torch import nn
label_desc: Use image node_id 13 for calling the TestCrypten obj's underlying member method code with example usage: obj.test_plaintext_save_load_from_party() and returns: tensor