Source code for moe.tests.views.rest.gp_next_points_test

# -*- coding: utf-8 -*-
"""Test class for gp_next_points_epi view."""
import pyramid.testing

import simplejson as json

from moe.optimal_learning.python.constant import TEST_OPTIMIZER_MULTISTARTS, TEST_GRADIENT_DESCENT_PARAMETERS, TEST_LBFGSB_PARAMETERS, TEST_OPTIMIZER_NUM_RANDOM_SAMPLES, TEST_EXPECTED_IMPROVEMENT_MC_ITERATIONS, CONSTANT_LIAR_METHODS
from moe.tests.optimal_learning.python.gaussian_process_test_case import GaussianProcessTestCase
from moe.tests.views.rest_test_case import RestTestCase
from moe.views.constant import ALL_NEXT_POINTS_MOE_ROUTES, GP_NEXT_POINTS_CONSTANT_LIAR_ROUTE_NAME, GP_NEXT_POINTS_CONSTANT_LIAR_ENDPOINT, GP_NEXT_POINTS_EPI_ROUTE_NAME
from moe.views.gp_next_points_pretty_view import GpNextPointsPrettyView
from moe.views.schemas.gp_next_points_pretty_view import GpNextPointsResponse
from moe.views.utils import _make_optimizer_parameters_from_params


[docs]class TestGpNextPointsViews(GaussianProcessTestCase, RestTestCase): """Integration test for the /gp/next_points/* endpoints.""" precompute_gaussian_process_data = True num_sampled_list = (1, 2, 10) @staticmethod def _build_json_payload(domain, covariance, historical_data, num_to_sample, lie_value=None, lie_method=None, l_bfgs_b=False): """Create a json_payload to POST to the /gp/next_points/* endpoint with all needed info.""" if l_bfgs_b: dict_to_dump = { 'num_to_sample': num_to_sample, 'mc_iterations': TEST_EXPECTED_IMPROVEMENT_MC_ITERATIONS, 'gp_historical_info': historical_data.json_payload(), 'covariance_info': covariance.get_json_serializable_info(), 'domain_info': domain.get_json_serializable_info(), 'optimizer_info': { 'num_multistarts': TEST_OPTIMIZER_MULTISTARTS, 'num_random_samples': TEST_OPTIMIZER_NUM_RANDOM_SAMPLES, 'optimizer_parameters': dict(TEST_LBFGSB_PARAMETERS._asdict()), }, 'mvndst_parameters': { 'releps': 1.0, 'maxpts_per_dim': 200, }, } else: dict_to_dump = { 'num_to_sample': num_to_sample, 'mc_iterations': TEST_EXPECTED_IMPROVEMENT_MC_ITERATIONS, 'gp_historical_info': historical_data.json_payload(), 'covariance_info': covariance.get_json_serializable_info(), 'domain_info': domain.get_json_serializable_info(), 'optimizer_info': { 'num_multistarts': TEST_OPTIMIZER_MULTISTARTS, 'num_random_samples': TEST_OPTIMIZER_NUM_RANDOM_SAMPLES, 'optimizer_parameters': dict(TEST_GRADIENT_DESCENT_PARAMETERS._asdict()), }, } if lie_value is not None: dict_to_dump['lie_value'] = lie_value if lie_method is not None: dict_to_dump['lie_method'] = lie_method return json.dumps(dict_to_dump)
[docs] def test_optimizer_params_passed_through(self): """Test that the optimizer parameters get passed through to the endpoint.""" # TODO(GH-305): turn this into a unit test by going through OptimizableGpPrettyView # and mocking out dependencies (instead of awkwardly constructing a more complex object). test_case = self.gp_test_environments[0] num_to_sample = 1 python_domain, python_gp = test_case python_cov, historical_data = python_gp.get_core_data_copy() # Test default test parameters get passed through json_payload = json.loads(self._build_json_payload(python_domain, python_cov, historical_data, num_to_sample)) request = pyramid.testing.DummyRequest(post=json_payload) request.json_body = json_payload view = GpNextPointsPrettyView(request) # get_params_from_request() requires this field is set. value is arbitrary for now. # TODO(GH-305): mock out this and other members view._route_name = GP_NEXT_POINTS_CONSTANT_LIAR_ROUTE_NAME params = view.get_params_from_request() _, optimizer_parameters, num_random_samples = _make_optimizer_parameters_from_params(params) test_param_dict = TEST_GRADIENT_DESCENT_PARAMETERS._asdict() test_param_dict['num_multistarts'] = TEST_OPTIMIZER_MULTISTARTS assert optimizer_parameters._get_member_dict() == test_param_dict # Test arbitrary parameters get passed through for i, key in enumerate(test_param_dict.iterkeys()): test_param_dict[key] /= 2 test_num_multistarts = test_param_dict.pop('num_multistarts') json_payload['optimizer_info']['num_multistarts'] = test_num_multistarts json_payload['optimizer_info']['optimizer_parameters'] = test_param_dict request = pyramid.testing.DummyRequest(post=json_payload) request.json_body = json_payload view = GpNextPointsPrettyView(request) # get_params_from_request() requires this field is set. value is arbitrary for now. # TODO(GH-305): mock out this and other members view._route_name = GP_NEXT_POINTS_CONSTANT_LIAR_ROUTE_NAME params = view.get_params_from_request() _, optimizer_parameters, num_random_samples = _make_optimizer_parameters_from_params(params) test_param_dict['num_multistarts'] = test_num_multistarts assert optimizer_parameters._get_member_dict() == test_param_dict
[docs] def test_all_constant_liar_methods_function(self): """Test that each contant liar ``lie_method`` runs to completion. This is an integration test.""" for test_case in self.gp_test_environments: python_domain, python_gp = test_case python_cov, historical_data = python_gp.get_core_data_copy() for constant_liar_method in CONSTANT_LIAR_METHODS: json_payload = self._build_json_payload( python_domain, python_cov, historical_data, 2, # num_to_sample lie_method=constant_liar_method, ) resp = self.testapp.post(GP_NEXT_POINTS_CONSTANT_LIAR_ENDPOINT, json_payload) resp_schema = GpNextPointsResponse() resp_dict = resp_schema.deserialize(json.loads(resp.body)) assert 'points_to_sample' in resp_dict assert len(resp_dict['points_to_sample']) == 2 # num_to_sample assert len(resp_dict['points_to_sample'][0]) == python_gp.dim assert 'status' in resp_dict assert 'expected_improvement' in resp_dict['status'] assert resp_dict['status']['expected_improvement'] >= 0.0
[docs] def test_interface_returns_same_as_cpp(self): """Integration test for the /gp/next_points/* endpoints.""" for moe_route in ALL_NEXT_POINTS_MOE_ROUTES: for test_case in self.gp_test_environments: for num_to_sample in (1, 2, 4): python_domain, python_gp = test_case python_cov, historical_data = python_gp.get_core_data_copy() # Next point from REST if moe_route.route_name == GP_NEXT_POINTS_CONSTANT_LIAR_ROUTE_NAME: json_payload = self._build_json_payload(python_domain, python_cov, historical_data, num_to_sample, lie_value=0.0) elif moe_route.route_name == GP_NEXT_POINTS_EPI_ROUTE_NAME and num_to_sample > 1: json_payload = self._build_json_payload(python_domain, python_cov, historical_data, num_to_sample, l_bfgs_b=True) else: json_payload = self._build_json_payload(python_domain, python_cov, historical_data, num_to_sample) resp = self.testapp.post(moe_route.endpoint, json_payload) resp_schema = GpNextPointsResponse() resp_dict = resp_schema.deserialize(json.loads(resp.body)) assert 'points_to_sample' in resp_dict assert len(resp_dict['points_to_sample']) == num_to_sample assert len(resp_dict['points_to_sample'][0]) == python_gp.dim assert 'status' in resp_dict assert 'expected_improvement' in resp_dict['status'] assert resp_dict['status']['expected_improvement'] >= 0.0