Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_create_experiment_new_default(self):
"""Test creating a new experiment with all defaults"""
name = 'all_default'
space = {'x': 'uniform(0, 10)'}
with OrionState():
experiment = create_experiment(name='all_default', space=space)
assert experiment.name == name
assert experiment.space.configuration == space
assert experiment.max_trials == orion.core.config.experiment.max_trials
assert experiment.working_dir == orion.core.config.experiment.working_dir
assert experiment.algorithms.configuration == {'random': {'seed': None}}
assert experiment.configuration['producer'] == {'strategy': 'MaxParallelStrategy'}
def test_hierarchical_space(self, new_config):
"""Verify space can have hierarchical structure"""
space = {'a': {'x': 'uniform(0, 10, discrete=True)'},
'b': {'y': 'loguniform(1e-08, 1)',
'z': 'choices([\'voici\', \'voila\', 2])'}}
with OrionState(experiments=[], trials=[]):
exp = experiment_builder.build('hierarchy', space=space)
exp2 = experiment_builder.build('hierarchy')
assert 'a.x' in exp.space
assert 'b.y' in exp.space
assert 'b.z' in exp.space
# Make sure it can be fetched properly from db as well
assert 'a.x' in exp2.space
assert 'b.y' in exp2.space
assert 'b.z' in exp2.space
def test_create_experiment_hit_no_config(self):
"""Test creating an existing experiment by specifying the name only."""
with OrionState(experiments=[config]):
experiment = create_experiment(config['name'])
assert experiment.name == config['name']
assert experiment.version == 1
assert experiment.space.configuration == config['space']
assert experiment.algorithms.configuration == config['algorithms']
assert experiment.max_trials == config['max_trials']
assert experiment.working_dir == config['working_dir']
assert experiment.producer['strategy'].configuration == config['producer']['strategy']
def test_register_lie_fail(self, storage):
"""Test register lie"""
with OrionState(experiments=[base_experiment], lies=[base_trial], database=storage) as cfg:
storage = cfg.storage()
with pytest.raises(DuplicateKeyError):
storage.register_lie(Trial(**cfg.lies[0]))
def test_reserve_success(self, exp_config_file, random_dt):
"""Successfully find new trials in db and reserve the first one"""
with OrionState(from_yaml=exp_config_file) as cfg:
exp = cfg.get_experiment('supernaedo2-dendi')
trial = exp.reserve_trial()
cfg.trials[1]['status'] = 'reserved'
cfg.trials[1]['start_time'] = random_dt
cfg.trials[1]['heartbeat'] = random_dt
assert trial.to_dict() == cfg.trials[1]
def test_push_trial_results(self, storage):
"""Successfully push a completed trial into database."""
with OrionState(experiments=[], trials=[base_trial], database=storage) as cfg:
storage = cfg.storage()
trial = storage.get_trial(Trial(**base_trial))
results = [
Trial.Result(name='loss', type='objective', value=2)
]
trial.results = results
assert storage.push_trial_results(trial), 'should update successfully'
trial2 = storage.get_trial(trial)
assert trial2.results == results
def test_old_experiment_w_version(self, parent_version_config, child_version_config):
"""Create an already existing experiment with a version."""
with OrionState(experiments=[parent_version_config, child_version_config]):
exp = experiment_builder.build(name=parent_version_config["name"], version=1)
assert exp.version == 1
def test_fetch_trials(self, storage):
"""Test fetch experiment trials"""
with OrionState(
experiments=[base_experiment], trials=generate_trials(), database=storage) as cfg:
storage = cfg.storage()
experiment = cfg.get_experiment('default_name', 'default_user', version=None)
trials1 = storage.fetch_trials(experiment=experiment)
trials2 = storage.fetch_trials(uid=experiment._id)
with pytest.raises(MissingArguments):
storage.fetch_trials()
with pytest.raises(AssertionError):
storage.fetch_trials(experiment=experiment, uid='123')
assert len(trials1) == len(cfg.trials), 'trial count should match'
assert len(trials2) == len(cfg.trials), 'trial count should match'
def retrieve_result(self, storage, generated_result):
"""Test retrieve result"""
results_file = tempfile.NamedTemporaryFile(
mode='w', prefix='results_', suffix='.log', dir='.', delete=True
)
# Generate fake result
with open(results_file.name, 'w') as file:
json.dump([generated_result], file)
# --
with OrionState(experiments=[], trials=[], database=storage) as cfg:
storage = cfg.storage()
trial = Trial(**base_trial)
trial = storage.retrieve_result(trial, results_file)
results = trial.results
assert len(results) == 1
assert results[0].to_dict() == generated_result
def test_register_duplicate_trial(self, storage):
"""Test register trial"""
with OrionState(
experiments=[base_experiment], trials=[base_trial], database=storage) as cfg:
storage = cfg.storage()
with pytest.raises(DuplicateKeyError):
storage.register_trial(Trial(**base_trial))