

def test_remote(self):
    logging.debug('')
    logging.debug('test_remote')
    server_dir = 'Factory'
    if os.path.exists(server_dir):
        shutil.rmtree(server_dir, onerror=onerror)
    os.mkdir(server_dir)
    os.chdir(server_dir)
    try:
        (server, server_cfg) = start_server()
        cfg = read_server_config(server_cfg)
        factory = None
        try:
            factory = connect(cfg['address'], cfg['port'], pubkey=cfg['key'])
            prefix = RAM._make_prefix(factory.host)
            remote = ('%s_LocalHost' % prefix)
            allocator_names = [allocator.name for allocator in RAM.list_allocators()]
            logging.debug('%s', allocator_names)
            self.assertFalse((remote in allocator_names))
            RAM.add_remotes(factory)
            allocator_names = [allocator.name for allocator in RAM.list_allocators()]
            logging.debug('%s', allocator_names)
            self.assertTrue((remote in allocator_names))
            prefix.assertFalse((RAM.get_allocator(remote) is RAM.list_allocators()[0]))
            self.assertTrue((RAM.get_allocator(remote) is RAM.list_allocators()[1]))
            max_servers = RAM.max_servers(dict(allocator=remote))
            self.assertTrue((max_servers >= 0))
            remote_alloc = RAM.get_allocator(remote)
            (max_servers, info) = remote_alloc.max_servers(dict(localhost=True))
            self.assertEqual(max_servers, 0)
            self.assertEqual(info, dict(localhost='requested local host'))
            (max_servers, info) = remote_alloc.max_servers(dict(allocator='LocalHost'))
            self.assertEqual(max_servers, 0)
            self.assertEqual(info, dict(allocator='wrong allocator'))
            (estimate, info) = remote_alloc.time_estimate(dict(allocator='LocalHost'))
            self.assertEqual(estimate, (- 2))
            self.assertEqual(info, dict(allocator='wrong allocator'))
            (remote_server, info) = RAM.allocate(dict(allocator=remote))
            RAM.release(remote_server)
            allocator_names = [allocator.name for allocator in RAM.list_allocators()]
            for name in allocator_names:
                if name.startswith(prefix):
                    RAM.remove_allocator(name)
            allocator_names = [allocator.name for allocator in RAM.list_allocators()]
            logging.debug('%s', allocator_names)
            self.assertFalse((remote in allocator_names))
        finally:
            if (factory is not None):
                factory.cleanup()
            server.terminate(timeout=10)
    finally:
        os.chdir('..')
        shutil.rmtree(server_dir, onerror=onerror)
    self.assertEqual(RAM._get_instance().get_total_allocators(), 1)
    self.assertTrue((RAM._get_instance().get_allocator_proxy(0) is RAM.list_allocators()[0]))

dataset/ETHPy150Open OpenMDAO/OpenMDAO-Framework/openmdao.main/src/openmdao/main/test/test_resource.py TestCase.test_remote/VarMisuse@215/298 `self`->`prefix`