本文整理汇总了Python中recipe.configuration.definition.recipe_definition.RecipeDefinition类的典型用法代码示例。如果您正苦于以下问题：Python RecipeDefinition类的具体用法？Python RecipeDefinition怎么用？Python RecipeDefinition使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了RecipeDefinition类的17个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_get_job_type_keys_unique
def test_get_job_type_keys_unique(self):
"""Tests getting job type keys without duplicates."""
definition = {
'version': '1.0',
'input_data': [],
'jobs': [{
'name': 'Job 1a',
'job_type': {
'name': self.job_type1.name,
'version': self.job_type1.version,
},
}, {
'name': 'Job 2',
'job_type': {
'name': self.job_type2.name,
'version': self.job_type2.version,
},
}, {
'name': 'Job 1b',
'job_type': {
'name': self.job_type1.name,
'version': self.job_type1.version,
},
}],
}
recipe_definition = RecipeDefinition(definition)
results = recipe_definition.get_job_type_keys()
self.assertSetEqual(results, {(self.job_type1.name, self.job_type1.version),
(self.job_type2.name, self.job_type2.version)})
开发者ID:AppliedIS，项目名称:scale，代码行数:31，代码来源:test_recipe_definition.py
示例2: test_successful
def test_successful(self):
'''Tests calling RecipeDefinition.get_unqueued_job_statuses() successfully.'''
recipe_definition = RecipeDefinition(self.definition)
recipe_definition.validate_job_interfaces()
recipe_jobs = {
'job_failed': self.job_failed,
'job_completed': self.job_completed,
'job_running': self.job_running,
'job_queued': self.job_queued,
'job_canceled': self.job_canceled,
'job_fa_co_a': self.job_fa_co_a,
'job_fa_co_b': self.job_fa_co_b,
'job_co_ru_qu_a': self.job_co_ru_qu_a,
'job_co_ru_qu_b': self.job_co_ru_qu_b,
'job_qu_ca_a': self.job_qu_ca_a,
'job_qu_ca_b': self.job_qu_ca_b,
}
results = recipe_definition.get_unqueued_job_statuses(recipe_jobs)
expected_results = {
self.job_fa_co_a.id: 'BLOCKED',
self.job_fa_co_b.id: 'BLOCKED',
self.job_co_ru_qu_a.id: 'PENDING',
self.job_co_ru_qu_b.id: 'PENDING',
self.job_qu_ca_a.id: 'BLOCKED',
self.job_qu_ca_b.id: 'BLOCKED',
}
self.assertDictEqual(results, expected_results)
开发者ID:cshamis，项目名称:scale，代码行数:31，代码来源:test_recipe_definition.py
示例3: test_get_job_type_keys_multi
def test_get_job_type_keys_multi(self):
'''Tests getting job type keys from the definition.'''
definition = {
'version': '1.0',
'input_data': [],
'jobs': [{
'name': 'Job 1',
'job_type': {
'name': self.job_type1.name,
'version': self.job_type1.version,
},
}, {
'name': 'Job 2',
'job_type': {
'name': self.job_type2.name,
'version': self.job_type2.version,
},
}],
}
recipe_definition = RecipeDefinition(definition)
results = recipe_definition.get_job_type_keys()
self.assertSetEqual(results, {(self.job_type1.name, self.job_type1.version),
(self.job_type2.name, self.job_type2.version)})
开发者ID:cshamis，项目名称:scale，代码行数:25，代码来源:test_recipe_definition.py
示例4: setUp
def setUp(self):
django.setup()
workspace = storage_test_utils.create_workspace()
source_file = source_test_utils.create_source(workspace=workspace)
self.event = trigger_test_utils.create_trigger_event()
interface_1 = {
"version": "1.0",
"command": "test_command",
"command_arguments": "test_arg",
"input_data": [{"name": "Test Input 1", "type": "file", "media_types": ["text/plain"]}],
"output_data": [{"name": "Test Output 1", "type": "files", "media_type": "image/png"}],
}
self.job_type_1 = job_test_utils.create_job_type(interface=interface_1)
interface_2 = {
"version": "1.0",
"command": "test_command",
"command_arguments": "test_arg",
"input_data": [{"name": "Test Input 2", "type": "files", "media_types": ["image/png", "image/tiff"]}],
"output_data": [{"name": "Test Output 2", "type": "file"}],
}
self.job_type_2 = job_test_utils.create_job_type(interface=interface_2)
definition = {
"version": "1.0",
"input_data": [{"name": "Recipe Input", "type": "file", "media_types": ["text/plain"]}],
"jobs": [
{
"name": "Job 1",
"job_type": {"name": self.job_type_1.name, "version": self.job_type_1.version},
"recipe_inputs": [{"recipe_input": "Recipe Input", "job_input": "Test Input 1"}],
},
{
"name": "Job 2",
"job_type": {"name": self.job_type_2.name, "version": self.job_type_2.version},
"dependencies": [
{"name": "Job 1", "connections": [{"output": "Test Output 1", "input": "Test Input 2"}]}
],
},
],
}
recipe_definition = RecipeDefinition(definition)
recipe_definition.validate_job_interfaces()
self.recipe_type = recipe_test_utils.create_recipe_type(definition=definition)
self.data = {
"version": "1.0",
"input_data": [{"name": "Recipe Input", "file_id": source_file.id}],
"workspace_id": workspace.id,
}
# Register a fake processor
self.mock_processor = MagicMock(QueueEventProcessor)
Queue.objects.register_processor(lambda: self.mock_processor)
开发者ID:Carl4，项目名称:scale，代码行数:58，代码来源:test_models.py
示例5: test_get_job_type_keys_empty
def test_get_job_type_keys_empty(self):
"""Tests getting job type keys when there are no jobs defined."""
definition = {
'version': '1.0',
'input_data': [],
'jobs': [],
}
recipe_definition = RecipeDefinition(definition)
results = recipe_definition.get_job_type_keys()
self.assertSetEqual(results, set())
开发者ID:AppliedIS，项目名称:scale，代码行数:12，代码来源:test_recipe_definition.py
示例6: test_successful
def test_successful(self, mock_store):
"""Tests calling RecipeDefinition.validate_data() successfully."""
definition = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'type': 'file',
'media_types': ['text/plain'],
}],
'jobs': [{
'name': 'Job 1',
'job_type': {
'name': self.job_type_1.name,
'version': self.job_type_1.version,
},
'recipe_inputs': [{
'recipe_input': 'Recipe Input',
'job_input': self.input_name_1,
}],
}, {
'name': 'Job 2',
'job_type': {
'name': self.job_type_2.name,
'version': self.job_type_2.version,
},
'dependencies': [{
'name': 'Job 1',
'connections': [{
'output': self.output_name_1,
'input': self.input_name_2,
}],
}],
}],
}
recipe = RecipeDefinition(definition)
recipe.validate_job_interfaces()
data = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'file_id': self.file_1.id,
}],
'workspace_id': 1,
}
recipe_data = RecipeData(data)
# No exception is success
recipe.validate_data(recipe_data)
开发者ID:AppliedIS，项目名称:scale，代码行数:50，代码来源:test_recipe_definition.py
示例7: test_missing_workspace
def test_missing_workspace(self, mock_store):
"""Tests calling RecipeDefinition.validate_data() with a missing required workspace."""
definition = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'type': 'file',
'media_types': ['text/plain'],
}],
'jobs': [{
'name': 'Job 1',
'job_type': {
'name': self.job_type_1.name,
'version': self.job_type_1.version,
},
'recipe_inputs': [{
'recipe_input': 'Recipe Input',
'job_input': self.input_name_1,
}],
}, {
'name': 'Job 2',
'job_type': {
'name': self.job_type_2.name,
'version': self.job_type_2.version,
},
'dependencies': [{
'name': 'Job 1',
'connections': [{
'output': self.output_name_1,
'input': self.input_name_2,
}],
}],
}],
}
recipe = RecipeDefinition(definition)
recipe.validate_job_interfaces()
data = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'file_id': self.file_1.id,
}],
}
recipe_data = RecipeData(data)
self.assertRaises(InvalidRecipeData, recipe.validate_data, recipe_data)
开发者ID:AppliedIS，项目名称:scale，代码行数:48，代码来源:test_recipe_definition.py
示例8: test_get_job_type_keys_one
def test_get_job_type_keys_one(self):
"""Tests getting a job type key from the definition."""
definition = {
'version': '1.0',
'input_data': [],
'jobs': [{
'name': 'Job 1',
'job_type': {
'name': self.job_type1.name,
'version': self.job_type1.version,
},
}],
}
recipe_definition = RecipeDefinition(definition)
results = recipe_definition.get_job_type_keys()
self.assertSetEqual(results, {(self.job_type1.name, self.job_type1.version)})
开发者ID:AppliedIS，项目名称:scale，代码行数:18，代码来源:test_recipe_definition.py
示例9: test_successful_new_recipe
def test_successful_new_recipe(self):
"""Tests calling RecipeDefinition.get_job_type_map() successfully."""
definition = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'type': 'file',
'media_types': ['text/plain'],
}],
'jobs': [{
'name': 'Job 1',
'job_type': {
'name': self.job_type1.name,
'version': self.job_type1.version,
},
'recipe_inputs': [{
'recipe_input': 'Recipe Input',
'job_input': 'Input 1',
}]
}, {
'name': 'Job 2',
'job_type': {
'name': self.job_type2.name,
'version': self.job_type2.version,
},
'dependencies': [{
'name': 'Job 1',
'connections': [{
'output': 'Output 1',
'input': 'Input 2',
}]
}]
}]
}
recipe_definition = RecipeDefinition(definition)
results = recipe_definition.get_job_type_map()
self.assertDictEqual(results, {'Job 1': self.job_type1, 'Job 2': self.job_type2})
开发者ID:AppliedIS，项目名称:scale，代码行数:39，代码来源:test_recipe_definition.py
示例10: test_successful_no_workspace
def test_successful_no_workspace(self, mock_store):
"""Tests calling RecipeDefinition.validate_data() successfully with no workspace."""
definition = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'type': 'file',
'media_types': ['text/plain'],
}],
'jobs': [{
'name': 'Job 3',
'job_type': {
'name': self.job_type_3.name,
'version': self.job_type_3.version,
},
'recipe_inputs': [{
'recipe_input': 'Recipe Input',
'job_input': self.input_name_3,
}],
}],
}
recipe = RecipeDefinition(definition)
recipe.validate_job_interfaces()
data = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'file_id': self.file_1.id,
}],
}
recipe_data = RecipeData(data)
# No exception is success
recipe.validate_data(recipe_data)
开发者ID:AppliedIS，项目名称:scale，代码行数:36，代码来源:test_recipe_definition.py
示例11: setUp
def setUp(self):
django.setup()
self.workspace = storage_test_utils.create_workspace()
interface_1 = {
'version': '1.0',
'command': 'my_command',
'command_arguments': 'args',
'input_data': [{
'name': 'Test Input 1',
'type': 'file',
'media_types': ['text/plain'],
}],
'output_data': [{
'name': 'Test Output 1',
'type': 'files',
'media_type': 'image/png',
}]}
self.job_type_1 = job_test_utils.create_job_type(interface=interface_1)
interface_2 = {
'version': '1.0',
'command': 'my_command',
'command_arguments': 'args',
'input_data': [{
'name': 'Test Input 2',
'type': 'files',
'media_types': ['image/png', 'image/tiff'],
}],
'output_data': [{
'name': 'Test Output 2',
'type': 'file',
}]}
self.job_type_2 = job_test_utils.create_job_type(interface=interface_2)
self.definition = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'type': 'file',
'media_types': ['text/plain'],
}],
'jobs': [{
'name': 'Job 1',
'job_type': {
'name': self.job_type_1.name,
'version': self.job_type_1.version,
},
'recipe_inputs': [{
'recipe_input': 'Recipe Input',
'job_input': 'Test Input 1',
}]
}, {
'name': 'Job 2',
'job_type': {
'name': self.job_type_2.name,
'version': self.job_type_2.version,
},
'dependencies': [{
'name': 'Job 1',
'connections': [{
'output': 'Test Output 1',
'input': 'Test Input 2',
}]
}]
}]
}
self.recipe_def = RecipeDefinition(self.definition)
self.recipe_def.validate_job_interfaces()
开发者ID:AppliedIS，项目名称:scale，代码行数:70，代码来源:test_models.py
示例12: test_successful_job_1_completed
def test_successful_job_1_completed(self, mock_store):
'''Tests calling RecipeDefinition.get_next_jobs_to_queue() successfully when job 1 has been completed.'''
definition = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'type': 'file',
'media_types': ['text/plain'],
}],
'jobs': [{
'name': 'Job 1',
'job_type': {
'name': self.job_type_1.name,
'version': self.job_type_1.version,
},
'recipe_inputs': [{
'recipe_input': 'Recipe Input',
'job_input': self.input_name_1,
}]
}, {
'name': 'Job 2',
'job_type': {
'name': self.job_type_2.name,
'version': self.job_type_2.version,
},
'dependencies': [{
'name': 'Job 1',
'connections': [{
'output': self.output_name_1,
'input': self.input_name_2,
}],
}],
}],
}
recipe_definition = RecipeDefinition(definition)
recipe_definition.validate_job_interfaces()
data = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'file_id': self.file_1.id,
}],
'workspace_id': 1,
}
recipe_data = RecipeData(data)
recipe_definition.validate_data(recipe_data)
png_file_ids = [98, 99, 100]
job_results = JobResults()
job_results.add_file_list_parameter(self.output_name_1, png_file_ids)
job_1 = Job.objects.select_related('job_type').get(pk=self.job_1.id)
job_1.results = job_results.get_dict()
job_1.save()
job_2 = Job.objects.select_related('job_type').get(pk=self.job_2.id)
results = recipe_definition.get_next_jobs_to_queue(recipe_data, {'Job 2': job_2}, {'Job 1': job_1})
# Make sure only Job 2 is returned and that its job data is correct
self.assertListEqual([self.job_2.id], results.keys())
self.assertDictEqual(results[self.job_2.id].get_dict(), {
'version': '1.0',
'input_data': [{
'name': self.input_name_2,
'file_ids': png_file_ids,
}],
'output_data': [{
'name': self.output_name_2,
'workspace_id': 1,
}],
})
开发者ID:cshamis，项目名称:scale，代码行数:72，代码来源:test_recipe_definition.py
示例13: _import_recipe_type
def _import_recipe_type(recipe_type_dict, recipe_type=None):
'''Attempts to apply the given recipe types configuration to the system.
Note that proper model locking must be performed before calling this method.
:param recipe_type_dict: A dictionary of recipe type configuration changes to import.
:type recipe_type_dict: dict
:param recipe_type: The existing recipe type model to update if applicable.
:type recipe_type: :class:`recipe.models.RecipeType`
:returns: A list of warnings discovered during import.
:rtype: list[:class:`port.schema.ValidationWarning`]
:raises :class:`port.schema.InvalidConfiguration`: If any part of the configuration violates the specification.
'''
warnings = []
# Parse the JSON content into validated model fields
recipe_type_serializer = serializers.ConfigurationRecipeTypeSerializer(recipe_type, data=recipe_type_dict)
if not recipe_type_serializer.is_valid():
raise InvalidConfiguration('Invalid recipe type schema: %s -> %s' % (recipe_type_dict['name'],
recipe_type_serializer.errors))
result = recipe_type_serializer.validated_data
# Validate the recipe definition
try:
definition_dict = None
if 'definition' in result:
definition_dict = result.get('definition')
elif recipe_type:
definition_dict = recipe_type.definition
definition = RecipeDefinition(definition_dict)
warnings.extend(definition.validate_job_interfaces())
except (InvalidDefinition, InvalidRecipeConnection) as ex:
raise InvalidConfiguration('Recipe type definition invalid: %s -> %s' % (result.get('name'), unicode(ex)))
# Validate the trigger rule
trigger_rule = None
if 'trigger_rule' in result and result.get('trigger_rule'):
trigger_rule = TriggerRule(**result.get('trigger_rule'))
if trigger_rule:
trigger_config = trigger_rule.get_configuration()
if not isinstance(trigger_config, RecipeTriggerRuleConfiguration):
logger.exception('Recipe type trigger rule type invalid')
raise InvalidConfiguration('Recipe type trigger type invalid: %s -> %s' % (result.get('name'),
trigger_rule.type))
try:
warnings.extend(trigger_config.validate_trigger_for_recipe(definition))
# Create a new rule when the trigger content was provided
if recipe_type_dict.get('trigger_rule'):
rule_handler = trigger_handler.get_trigger_rule_handler(trigger_rule.type)
trigger_rule = rule_handler.create_trigger_rule(trigger_rule.configuration, trigger_rule.name,
trigger_rule.is_active)
except (InvalidDefinition, InvalidTriggerType, InvalidTriggerRule, InvalidRecipeConnection) as ex:
logger.exception('Recipe type trigger rule invalid')
raise InvalidConfiguration('Recipe type trigger rule invalid: %s -> %s' % (result.get('name'), unicode(ex)))
remove_trigger_rule = 'trigger_rule' in recipe_type_dict and not recipe_type_dict['trigger_rule']
# Edit or create the associated recipe type model
if recipe_type:
try:
RecipeType.objects.edit_recipe_type(recipe_type.id, result.get('title'), result.get('description'),
definition, trigger_rule, remove_trigger_rule)
except (InvalidDefinition, InvalidTriggerType, InvalidTriggerRule, InvalidRecipeConnection) as ex:
logger.exception('Recipe type edit failed')
raise InvalidConfiguration('Unable to edit recipe type: %s -> %s' % (result.get('name'), unicode(ex)))
else:
try:
RecipeType.objects.create_recipe_type(result.get('name'), result.get('version'), result.get('title'),
result.get('description'), definition, trigger_rule)
except (InvalidDefinition, InvalidTriggerType, InvalidTriggerRule, InvalidRecipeConnection) as ex:
logger.exception('Recipe type create failed')
raise InvalidConfiguration('Unable to create new recipe type: %s -> %s' % (result.get('name'), unicode(ex)))
return warnings
开发者ID:AppliedIS，项目名称:scale，代码行数:74，代码来源:importer.py
示例14: setUp
def setUp(self):
django.setup()
workspace = storage_test_utils.create_workspace()
source_file = source_test_utils.create_source(workspace=workspace)
self.event = trigger_test_utils.create_trigger_event()
interface_1 = {
'version': '1.0',
'command': 'test_command',
'command_arguments': 'test_arg',
'input_data': [{
'name': 'Test Input 1',
'type': 'file',
'media_types': ['text/plain'],
}],
'output_data': [{
'name': 'Test Output 1',
'type': 'files',
'media_type': 'image/png',
}]
}
self.job_type_1 = job_test_utils.create_job_type(interface=interface_1)
interface_2 = {
'version': '1.0',
'command': 'test_command',
'command_arguments': 'test_arg',
'input_data': [{
'name': 'Test Input 2',
'type': 'files',
'media_types': ['image/png', 'image/tiff'],
}],
'output_data': [{
'name': 'Test Output 2',
'type': 'file',
}]
}
self.job_type_2 = job_test_utils.create_job_type(interface=interface_2)
definition = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'type': 'file',
'media_types': ['text/plain'],
}],
'jobs': [{
'name': 'Job 1',
'job_type': {
'name': self.job_type_1.name,
'version': self.job_type_1.version,
},
'recipe_inputs': [{
'recipe_input': 'Recipe Input',
'job_input': 'Test Input 1',
}]
}, {
'name': 'Job 2',
'job_type': {
'name': self.job_type_2.name,
'version': self.job_type_2.version,
},
'dependencies': [{
'name': 'Job 1',
'connections': [{
'output': 'Test Output 1',
'input': 'Test Input 2',
}]
}]
}]
}
recipe_definition = RecipeDefinition(definition)
recipe_definition.validate_job_interfaces()
self.recipe_type = recipe_test_utils.create_recipe_type(definition=definition)
data = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'file_id': source_file.id,
}],
'workspace_id': workspace.id,
}
self.data = RecipeData(data)
# Register a fake processor
self.mock_processor = MagicMock(QueueEventProcessor)
Queue.objects.register_processor(lambda: self.mock_processor)
开发者ID:AppliedIS，项目名称:scale，代码行数:91，代码来源:test_models.py
示例15: TestRecipeTypeManagerCreateRecipeType
class TestRecipeTypeManagerCreateRecipeType(TransactionTestCase):
def setUp(self):
django.setup()
self.workspace = storage_test_utils.create_workspace()
interface_1 = {
'version': '1.0',
'command': 'my_command',
'command_arguments': 'args',
'input_data': [{
'name': 'Test Input 1',
'type': 'file',
'media_types': ['text/plain'],
}],
'output_data': [{
'name': 'Test Output 1',
'type': 'files',
'media_type': 'image/png',
}]}
self.job_type_1 = job_test_utils.create_job_type(interface=interface_1)
interface_2 = {
'version': '1.0',
'command': 'my_command',
'command_arguments': 'args',
'input_data': [{
'name': 'Test Input 2',
'type': 'files',
'media_types': ['image/png', 'image/tiff'],
}],
'output_data': [{
'name': 'Test Output 2',
'type': 'file',
}]}
self.job_type_2 = job_test_utils.create_job_type(interface=interface_2)
self.definition = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'type': 'file',
'media_types': ['text/plain'],
}],
'jobs': [{
'name': 'Job 1',
'job_type': {
'name': self.job_type_1.name,
'version': self.job_type_1.version,
},
'recipe_inputs': [{
'recipe_input': 'Recipe Input',
'job_input': 'Test Input 1',
}]
}, {
'name': 'Job 2',
'job_type': {
'name': self.job_type_2.name,
'version': self.job_type_2.version,
},
'dependencies': [{
'name': 'Job 1',
'connections': [{
'output': 'Test Output 1',
'input': 'Test Input 2',
}]
}]
}]
}
self.recipe_def = RecipeDefinition(self.definition)
self.recipe_def.validate_job_interfaces()
def test_successful(self):
"""Tests calling RecipeTypeManager.create_recipe_type() successfully."""
name = 'test-recipe'
version = '1.0'
title = 'Test Recipe'
desc = 'Test description'
recipe_type = RecipeType.objects.create_recipe_type(name, version, title, desc, self.recipe_def, None)
results_recipe_type = RecipeType.objects.get(pk=recipe_type.id)
self.assertEqual(results_recipe_type.name, name)
self.assertEqual(results_recipe_type.version, version)
self.assertEqual(results_recipe_type.title, title)
self.assertEqual(results_recipe_type.description, desc)
self.assertDictEqual(results_recipe_type.definition, self.definition)
results_recipe_type_rev = RecipeTypeRevision.objects.get(recipe_type_id=recipe_type.id, revision_num=1)
self.assertDictEqual(results_recipe_type_rev.definition, self.definition)
开发者ID:AppliedIS，项目名称:scale，代码行数:91，代码来源:test_models.py
示例16: TestRecipeTypeManagerEditRecipeType
class TestRecipeTypeManagerEditRecipeType(TransactionTestCase):
def setUp(self):
django.setup()
self.workspace = storage_test_utils.create_workspace()
interface_1 = {
'version': '1.0',
'command': 'my_command',
'command_arguments': 'args',
'input_data': [{
'name': 'Test Input 1',
'type': 'file',
'media_types': ['text/plain'],
}],
'output_data': [{
'name': 'Test Output 1',
'type': 'files',
'media_type': 'image/png',
}]}
self.job_type_1 = job_test_utils.create_job_type(interface=interface_1)
interface_2 = {
'version': '1.0',
'command': 'my_command',
'command_arguments': 'args',
'input_data': [{
'name': 'Test Input 2',
'type': 'files',
'media_types': ['image/png', 'image/tiff'],
}],
'output_data': [{
'name': 'Test Output 2',
'type': 'file',
}]}
self.job_type_2 = job_test_utils.create_job_type(interface=interface_2)
self.definition = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'type': 'file',
'media_types': ['text/plain'],
}],
'jobs': [{
'name': 'Job 1',
'job_type': {
'name': self.job_type_1.name,
'version': self.job_type_1.version,
},
'recipe_inputs': [{
'recipe_input': 'Recipe Input',
'job_input': 'Test Input 1',
}]
}, {
'name': 'Job 2',
'job_type': {
'name': self.job_type_2.name,
'version': self.job_type_2.version,
},
'dependencies': [{
'name': 'Job 1',
'connections': [{
'output': 'Test Output 1',
'input': 'Test Input 2',
}]
}]
}]
}
self.recipe_def = RecipeDefinition(self.definition)
self.recipe_def.validate_job_interfaces()
self.new_definition = {
'version': '1.0',
'input_data': [{
'name': 'Recipe Input',
'type': 'file',
'media_types': ['text/plain'],
}],
'jobs': [{
'name': 'Job 1',
'job_type': {
'name': self.job_type_1.name,
'version': self.job_type_1.version,
},
'recipe_inputs': [{
'recipe_input': 'Recipe Input',
'job_input': 'Test Input 1',
}]
}]
}
self.new_recipe_def = RecipeDefinition(self.new_definition)
self.new_recipe_def.validate_job_interfaces()
self.configuration = {
'version': '1.0',
'condition': {
'media_type': 'text/plain'
},
#.........这里部分代码省略.........
开发者ID:AppliedIS，项目名称:scale，代码行数:101，代码来源:test_models.py
示例17: setUp
def setUp(self):
django.setup()
self.job_a = job_test_utils.create_job()
self.job_b = job_test_utils.create_job()
self.job_c = job_test_utils.create_job()
self.job_d = job_test_utils.create_job()
self.job_e = job_test_utils.create_job()
self.job_f = job_test_utils.create_job()
self.job_g = job_test_utils.create_job()
self.job_h = job_test_utils.create_job()
definition = {
"version": "1.0",
"input_data": [
{"name": "Recipe Input 1", "type": "file", "media_types": ["text/plain"]},
{"name": "Recipe Input 2", "type": "property"},
],
"jobs": [
{
"name": "Job A",
"job_type": {"name": self.job_a.job_type.name, "version": self.job_a.job_type.version},
"recipe_inputs": [{"recipe_input": "Recipe Input 1", "job_input": "Job Input 1"}],
},
{
"name": "Job B",
"job_type": {"name": self.job_b.job_type.name, "version": self.job_b.job_type.version},
"recipe_inputs": [{"recipe_input": "Recipe Input 2", "job_input": "Job Input 1"}],
},
{
"name": "Job C",
"job_type": {"name": self.job_c.job_type.name, "version": self.job_c.job_type.version},
"recipe_inputs": [{"recipe_input": "Recipe Input 2", "job_input": "Job Input 1"}],
},
{
"name": "Job D",
"job_type": {"name": self.job_d.job_type.name, "version": self.job_d.job_type.version},
"dependencies": [
{"name": "Job A", "connections": [{"output": "Job Output 1", "input": "Job Input 1"}]},
{"name": "Job B", "connections": [{"output": "Job Output 1", "input": "Job Input 2"}]},
],
},
{
"name": "Job E",
"job_type": {"name": self.job_e.job_type.name, "version": self.job_e.job_type.version},
"dependencies": [
{"name": "Job B", "connections": [{"output": "Job Output 1", "input": "Job Input 1"}]}
],
},
{
"name": "Job F",
"job_type": {"name": self.job_f.job_type.name, "version": self.job_f.job_type.version},
"dependencies": [
{"name": "Job D", "connections": [{"output": "Job Output 1", "input": "Job Input 1"}]}
],
},
{
"name": "Job G",
"job_type": {"name": self.job_g.job_type.name, "version": self.job_g.job_type.version},
"dependencies": [
{"name": "Job D", "connections": [{"output": "Job Output 1", "input": "Job Input 1"}]},
{"name": "Job E", "connections": [{"output": "Job Output 1", "input": "Job Input 2"}]},
],
},
{
"name": "Job H",
"job_type": {"name": self.job_h.job_type.name, "version": self.job_h.job_type.version},
"dependencies": [
{"name": "Job C", "connections": [{"output": "Job Output 1", "input": "Job Input 1"}]},
{"name": "Job D", "connections": [{"output": "Job Output 1", "input": "Job Input 2"}]},
],
},
],
}
recipe_definition = RecipeDefinition(definition)
self.graph = recipe_definition.get_graph()
开发者ID:ngageoint，项目名称:scale，代码行数:77，代码来源:test_graph.py
注：本文中的recipe.configuration.definition.recipe_definition.RecipeDefinition类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论