本文整理汇总了Python中trigger.test.utils.create_trigger_rule函数的典型用法代码示例。如果您正苦于以下问题：Python create_trigger_rule函数的具体用法？Python create_trigger_rule怎么用？Python create_trigger_rule使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了create_trigger_rule函数的19个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_invalid_trigger_rule
def test_invalid_trigger_rule(self):
'''Tests calling RecipeTypeManager.edit_recipe_type() with a new invalid trigger rule'''
# Create recipe_type
name = 'test-recipe'
version = '1.0'
title = 'Test Recipe'
desc = 'Test description'
trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=recipe_test_utils.MOCK_TYPE,
configuration=self.trigger_config.get_dict())
trigger_rule_id = trigger_rule.id
new_trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=recipe_test_utils.MOCK_ERROR_TYPE,
configuration=self.new_trigger_config.get_dict())
recipe_type = RecipeType.objects.create_recipe_type(name, version, title, desc, self.recipe_def, trigger_rule)
with transaction.atomic():
recipe_type = RecipeType.objects.select_for_update().get(pk=recipe_type.id)
# Edit the recipe
self.assertRaises(InvalidRecipeConnection, RecipeType.objects.edit_recipe_type, recipe_type.id, None, None,
self.new_recipe_def, new_trigger_rule, False)
recipe_type = RecipeType.objects.select_related('trigger_rule').get(pk=recipe_type.id)
# Check results
self.assertEqual(recipe_type.title, title)
self.assertEqual(recipe_type.description, desc)
self.assertDictEqual(recipe_type.get_recipe_definition().get_dict(), self.recipe_def.get_dict())
self.assertEqual(recipe_type.revision_num, 1)
self.assertEqual(recipe_type.trigger_rule_id, trigger_rule_id)
trigger_rule = TriggerRule.objects.get(pk=trigger_rule_id)
self.assertTrue(trigger_rule.is_active)
num_of_revs = RecipeTypeRevision.objects.filter(recipe_type_id=recipe_type.id).count()
self.assertEqual(num_of_revs, 1)
开发者ID:cshamis，项目名称:scale，代码行数:31，代码来源:test_models.py
示例2: test_change_to_both
def test_change_to_both(self):
'''Tests calling JobTypeManager.edit_job_type() with a change to both the definition and the trigger rule
'''
name = 'my-job-type'
version = '1.0'
trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=job_test_utils.MOCK_TYPE,
configuration=self.trigger_config.get_dict())
new_trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=job_test_utils.MOCK_TYPE,
configuration=self.new_trigger_config.get_dict())
job_type = JobType.objects.create_job_type(name, version, self.job_interface, trigger_rule)
# Call test
JobType.objects.edit_job_type(job_type.id, self.new_job_interface, new_trigger_rule, False)
# Check results
job_type = JobType.objects.select_related('trigger_rule').get(pk=job_type.id)
self.assertDictEqual(job_type.get_job_interface().get_dict(), self.new_job_interface.get_dict())
self.assertEqual(job_type.revision_num, 2)
self.assertEqual(job_type.trigger_rule_id, new_trigger_rule.id)
trigger_rule = TriggerRule.objects.get(pk=trigger_rule.id)
self.assertFalse(trigger_rule.is_active)
new_trigger_rule = TriggerRule.objects.get(pk=new_trigger_rule.id)
self.assertTrue(new_trigger_rule.is_active)
# New revision due to definition change
num_of_revs = JobTypeRevision.objects.filter(job_type_id=job_type.id).count()
self.assertEqual(num_of_revs, 2)
开发者ID:cshamis，项目名称:scale，代码行数:27，代码来源:test_models.py
示例3: test_change_to_both
def test_change_to_both(self):
"""Tests calling RecipeTypeManager.edit_recipe_type() with a change to both the definition and trigger rule"""
# Create recipe_type
name = 'test-recipe'
version = '1.0'
title = 'Test Recipe'
desc = 'Test description'
trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=recipe_test_utils.MOCK_TYPE,
configuration=self.trigger_config.get_dict())
trigger_rule_id = trigger_rule.id
new_trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=recipe_test_utils.MOCK_TYPE,
configuration=self.new_trigger_config.get_dict())
new_trigger_rule_id = new_trigger_rule.id
recipe_type = RecipeType.objects.create_recipe_type(name, version, title, desc, self.recipe_def, trigger_rule)
with transaction.atomic():
recipe_type = RecipeType.objects.select_for_update().get(pk=recipe_type.id)
# Edit the recipe
RecipeType.objects.edit_recipe_type(recipe_type.id, None, None, self.new_recipe_def, new_trigger_rule,
False)
recipe_type = RecipeType.objects.select_related('trigger_rule').get(pk=recipe_type.id)
# Check results
self.assertEqual(recipe_type.title, title)
self.assertEqual(recipe_type.description, desc)
self.assertDictEqual(recipe_type.get_recipe_definition().get_dict(), self.new_recipe_def.get_dict())
self.assertEqual(recipe_type.revision_num, 2)
self.assertEqual(recipe_type.trigger_rule_id, new_trigger_rule_id)
trigger_rule = TriggerRule.objects.get(pk=trigger_rule_id)
self.assertFalse(trigger_rule.is_active)
new_trigger_rule = TriggerRule.objects.get(pk=new_trigger_rule_id)
self.assertTrue(new_trigger_rule.is_active)
# New revision due to definition change
num_of_revs = RecipeTypeRevision.objects.filter(recipe_type_id=recipe_type.id).count()
self.assertEqual(num_of_revs, 2)
开发者ID:AppliedIS，项目名称:scale，代码行数:35，代码来源:test_models.py
示例4: test_successful_recipe_creation
def test_successful_recipe_creation(self):
'''Tests successfully processing a parse that triggers recipe creation.'''
# Set up data
configuration = {
'version': '1.0',
'condition': {
'media_type': 'text/plain',
},
'data': {
'input_data_name': self.input_name,
'workspace_name': self.workspace.name
},
}
rule_model = trigger_test_utils.create_trigger_rule(trigger_type='PARSE', configuration=configuration)
self.recipe_type_1.trigger_rule = rule_model
self.recipe_type_1.save()
# Call method to test
ParseTriggerHandler().process_parsed_source_file(self.source_file)
# Check results...ensure first job is queued
queue_1 = Queue.objects.get(job_type=self.job_type_2.id)
job_exe_1 = JobExecution.objects.select_related().get(pk=queue_1.job_exe_id)
job_1 = job_exe_1.job
self.assertEqual(job_1.data['input_data'][0]['name'], self.input_name)
self.assertEqual(job_1.data['input_data'][0]['file_id'], self.source_file.id)
self.assertEqual(job_1.data['output_data'][0]['name'], self.output_name)
self.assertEqual(job_1.data['output_data'][0]['workspace_id'], self.workspace.id)
开发者ID:AppliedIS，项目名称:scale，代码行数:30，代码来源:test_parse_trigger_handler.py
示例5: test_change_simple_with_trigger
def test_change_simple_with_trigger(self):
"""Tests calling RecipeTypeManager.edit_recipe_type() with only basic attributes and a previous trigger rule"""
# Create recipe_type
name = 'test-recipe'
version = '1.0'
title = 'Test Recipe'
desc = 'Test description'
trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=recipe_test_utils.MOCK_TYPE,
configuration=self.trigger_config.get_dict())
trigger_rule_id = trigger_rule.id
recipe_type = RecipeType.objects.create_recipe_type(name, version, title, desc, self.recipe_def, trigger_rule)
with transaction.atomic():
recipe_type = RecipeType.objects.select_for_update().get(pk=recipe_type.id)
# Edit the recipe
new_title = 'New title'
new_desc = 'New description'
RecipeType.objects.edit_recipe_type(recipe_type.id, new_title, new_desc, None, None, False)
recipe_type = RecipeType.objects.select_related('trigger_rule').get(pk=recipe_type.id)
# Check results
self.assertEqual(recipe_type.title, new_title)
self.assertEqual(recipe_type.description, new_desc)
self.assertDictEqual(recipe_type.get_recipe_definition().get_dict(), self.recipe_def.get_dict())
self.assertEqual(recipe_type.revision_num, 1)
self.assertEqual(recipe_type.trigger_rule_id, trigger_rule_id)
num_of_revs = RecipeTypeRevision.objects.filter(recipe_type_id=recipe_type.id).count()
self.assertEqual(num_of_revs, 1)
开发者ID:AppliedIS，项目名称:scale，代码行数:28，代码来源:test_models.py
示例6: test_successful_job_creation
def test_successful_job_creation(self):
"""Tests successfully processing an ingest that triggers job creation."""
# Set up data
configuration = {
'version': '1.0',
'condition': {
'media_type': 'text/plain',
'data_types': ['type1', 'type2'],
},
'data': {
'input_data_name': self.input_name,
'workspace_name': self.workspace.name
},
}
rule_model = trigger_test_utils.create_trigger_rule(trigger_type='INGEST', configuration=configuration)
self.job_type_1.trigger_rule = rule_model
self.job_type_1.save()
# Call method to test
IngestTriggerHandler().process_ingested_source_file(self.source_file, now())
# Check results
queue_1 = Queue.objects.get(job_type=self.job_type_1.id)
job_exe_1 = JobExecution.objects.select_related().get(pk=queue_1.job_exe_id)
job_1 = job_exe_1.job
self.assertEqual(job_1.data['input_data'][0]['name'], self.input_name)
self.assertEqual(job_1.data['input_data'][0]['file_id'], self.source_file.id)
开发者ID:ngageoint，项目名称:scale，代码行数:28，代码来源:test_ingest_trigger_handler.py
示例7: setUp
def setUp(self):
django.setup()
self.workspace = storage_test_utils.create_workspace()
self.error = error_test_utils.create_error()
self.interface = {
'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_interface = JobInterface(self.interface)
self.error_mapping = ErrorInterface({
'version': '1.0',
'exit_codes': {
'1': self.error.name,
}
})
self.configuration = {
'version': '1.0',
'condition': {
'media_type': 'text/plain'
},
'data': {
'input_data_name': 'Test Input 1',
'workspace_name': self.workspace.name
}
}
self.trigger_config = job_test_utils.MockTriggerRuleConfiguration(job_test_utils.MOCK_TYPE, self.configuration)
self.trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=job_test_utils.MOCK_TYPE,
configuration=self.trigger_config.get_dict())
self.invalid_trigger_config = job_test_utils.MockErrorTriggerRuleConfiguration(job_test_utils.MOCK_ERROR_TYPE, self.configuration)
self.invalid_trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=job_test_utils.MOCK_ERROR_TYPE,
configuration=self.trigger_config.get_dict())
开发者ID:cshamis，项目名称:scale，代码行数:45，代码来源:test_models.py
示例8: create_job_type
def create_job_type(name=None, version=None, category=None, interface=None, priority=50, timeout=3600, max_tries=3,
max_scheduled=None, cpus=1.0, mem=1.0, disk=1.0, error_mapping=None, is_active=True,
is_operational=True, trigger_rule=None, configuration=None):
"""Creates a job type model for unit testing
:returns: The job type model
:rtype: :class:`job.models.JobType`
"""
if not name:
global JOB_TYPE_NAME_COUNTER
name = 'test-job-type-%i' % JOB_TYPE_NAME_COUNTER
JOB_TYPE_NAME_COUNTER += 1
if not version:
global JOB_TYPE_VERSION_COUNTER
version = '%i.0.0' % JOB_TYPE_VERSION_COUNTER
JOB_TYPE_VERSION_COUNTER += 1
if not category:
global JOB_TYPE_CATEGORY_COUNTER
category = 'test-category-%i' % JOB_TYPE_CATEGORY_COUNTER
JOB_TYPE_CATEGORY_COUNTER += 1
if not interface:
interface = {
'version': '1.2',
'command': 'test_cmd',
'command_arguments': 'test_arg',
'env_vars': [],
'settings': [],
'input_data': [],
'output_data': [],
'shared_resources': [],
}
if not error_mapping:
error_mapping = {
'version': '1.0',
'exit_codes': {}
}
if not trigger_rule:
trigger_rule = trigger_test_utils.create_trigger_rule()
if not configuration:
configuration = {
'version': '1.0',
'default_settings': {}
}
job_type = JobType.objects.create(name=name, version=version, category=category, interface=interface,
priority=priority, timeout=timeout, max_tries=max_tries,
max_scheduled=max_scheduled, cpus_required=cpus, mem_required=mem,
disk_out_const_required=disk, error_mapping=error_mapping, is_active=is_active,
is_operational=is_operational, trigger_rule=trigger_rule, configuration=configuration)
JobTypeRevision.objects.create_job_type_revision(job_type)
return job_type
开发者ID:ngageoint，项目名称:scale，代码行数:56，代码来源:utils.py
示例9: test_invalid_trigger_rule
def test_invalid_trigger_rule(self):
'''Tests calling JobTypeManager.create_job_type() with an invalid trigger rule'''
name = 'my-job-type'
version = '1.0'
trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=job_test_utils.MOCK_ERROR_TYPE,
configuration=self.trigger_config.get_dict())
# Call test
self.assertRaises(InvalidConnection, JobType.objects.create_job_type, name, version, self.job_interface,
trigger_rule, self.error_mapping)
开发者ID:cshamis，项目名称:scale，代码行数:11，代码来源:test_models.py
示例10: create_recipe_type
def create_recipe_type(name=None, version=None, title=None, description=None, definition=None, trigger_rule=None):
"""Creates a recipe type for unit testing
:returns: The RecipeType model
:rtype: :class:`recipe.models.RecipeType`
"""
if not name:
global NAME_COUNTER
name = 'test-recipe-type-%i' % NAME_COUNTER
NAME_COUNTER += 1
if not version:
global VERSION_COUNTER
version = '%i.0.0' % VERSION_COUNTER
VERSION_COUNTER += 1
if not title:
global TITLE_COUNTER
title = 'Test Recipe Type %i' % TITLE_COUNTER
TITLE_COUNTER += 1
if not description:
global DESCRIPTION_COUNTER
description = 'Test Description %i' % DESCRIPTION_COUNTER
DESCRIPTION_COUNTER += 1
if not definition:
definition = {
'version': '1.0',
'input_data': [],
'jobs': [],
}
if not trigger_rule:
trigger_rule = trigger_test_utils.create_trigger_rule()
recipe_type = RecipeType()
recipe_type.name = name
recipe_type.version = version
recipe_type.title = title
recipe_type.description = description
recipe_type.definition = definition
recipe_type.trigger_rule = trigger_rule
recipe_type.save()
RecipeTypeRevision.objects.create_recipe_type_revision(recipe_type)
return recipe_type
开发者ID:droessne，项目名称:scale，代码行数:49，代码来源:utils.py
示例11: test_uneditable_field
def test_uneditable_field(self):
'''Tests calling JobTypeManager.edit_job_type() to change an uneditable field'''
name = 'my-job-type'
version = '1.0'
title = 'my title'
new_title = 'my new title'
trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=job_test_utils.MOCK_TYPE,
configuration=self.trigger_config.get_dict())
job_type = JobType.objects.create_job_type(name, version, self.job_interface, trigger_rule, title=title)
# Call test
self.assertRaises(Exception, JobType.objects.edit_job_type, job_type.id, title=new_title, is_system=True)
# Check results
job_type = JobType.objects.select_related('trigger_rule').get(pk=job_type.id)
# No change
self.assertEqual(job_type.title, title)
开发者ID:cshamis，项目名称:scale，代码行数:18，代码来源:test_models.py
示例12: test_successful_with_trigger_rule
def test_successful_with_trigger_rule(self):
'''Tests calling JobTypeManager.create_job_type() successfully with a trigger rule and error mapping'''
name = 'my-job-type'
version = '1.0'
trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=job_test_utils.MOCK_TYPE,
configuration=self.trigger_config.get_dict())
# Call test
job_type = JobType.objects.create_job_type(name, version, self.job_interface, trigger_rule, self.error_mapping)
# Check results
job_type = JobType.objects.select_related('trigger_rule').get(pk=job_type.id)
self.assertDictEqual(job_type.get_job_interface().get_dict(), self.job_interface.get_dict())
self.assertEqual(job_type.revision_num, 1)
self.assertEqual(job_type.trigger_rule_id, trigger_rule.id)
trigger_rule = TriggerRule.objects.get(pk=trigger_rule.id)
self.assertTrue(trigger_rule.is_active)
self.assertDictEqual(job_type.get_error_interface().get_dict(), self.error_mapping.get_dict())
开发者ID:cshamis，项目名称:scale，代码行数:19，代码来源:test_models.py
示例13: test_change_general_fields
def test_change_general_fields(self):
'''Tests calling JobTypeManager.edit_job_type() with a change to some general fields'''
name = 'my-job-type'
version = '1.0'
title = 'my title'
priority = 12
error_mapping = ErrorInterface({
'version': '1.0',
'exit_codes': {
'-15': self.error.name,
}
})
new_title = 'my new title'
new_priority = 13
new_error_mapping = ErrorInterface({
'version': '1.0',
'exit_codes': {
'-16': self.error.name,
}
})
new_is_paused = True
trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=job_test_utils.MOCK_TYPE,
configuration=self.trigger_config.get_dict())
job_type = JobType.objects.create_job_type(name, version, self.job_interface, trigger_rule, title=title,
priority=priority, error_mapping=error_mapping)
# Call test
JobType.objects.edit_job_type(job_type.id, title=new_title, priority=new_priority,
error_mapping=new_error_mapping, is_paused=new_is_paused)
# Check results
job_type = JobType.objects.select_related('trigger_rule').get(pk=job_type.id)
self.assertDictEqual(job_type.get_job_interface().get_dict(), self.job_interface.get_dict())
self.assertEqual(job_type.revision_num, 1)
self.assertEqual(job_type.trigger_rule_id, trigger_rule.id)
trigger_rule = TriggerRule.objects.get(pk=trigger_rule.id)
self.assertTrue(trigger_rule.is_active)
self.assertEqual(job_type.title, new_title)
self.assertEqual(job_type.priority, new_priority)
self.assertDictEqual(job_type.get_error_interface().get_dict(), new_error_mapping.get_dict())
self.assertEqual(job_type.is_paused, new_is_paused)
self.assertIsNotNone(job_type.paused)
开发者ID:cshamis，项目名称:scale，代码行数:43，代码来源:test_models.py
示例14: create_clock_rule
def create_clock_rule(name=None, rule_type='CLOCK', event_type=None, schedule='PT1H0M0S', is_active=True):
'''Creates a scale clock trigger rule model for unit testing
:returns: The trigger rule model
:rtype: :class:`trigger.models.TriggerRule`
'''
if not event_type:
global RULE_EVENT_COUNTER
event_type = 'TEST_EVENT_%i' % RULE_EVENT_COUNTER
RULE_EVENT_COUNTER = RULE_EVENT_COUNTER + 1
config = {
'version': '1.0',
'event_type': event_type,
'schedule': schedule,
}
return trigger_test_utils.create_trigger_rule(name=name, trigger_type=rule_type, configuration=config,
is_active=is_active)
开发者ID:cuulee，项目名称:scale，代码行数:20，代码来源:utils.py
示例15: test_valid_feature_with_parse_rule
def test_valid_feature_with_parse_rule(self):
"""Tests calling save_parse_results with valid arguments and parse rules in place"""
# Setup parse rule
workspace = storage_utils.create_workspace()
configuration = {u'version': u'1.0',
u'condition': {u'media_type': u'text/plain', u'data_types': [u'type']},
u'data': {u'input_data_name': u'my_input', u'workspace_name': workspace.name}}
rule_model = trigger_utils.create_trigger_rule(trigger_type='PARSE', configuration=configuration)
interface = {'version': '1.0', 'command': '', 'command_arguments': '', 'input_data': [{'name': 'my_input', 'type': 'file'}]}
job_type = job_utils.create_job_type(interface=interface)
job_type.trigger_rule = rule_model
job_type.save()
# Call method to test
SourceFile.objects.save_parse_results(self.src_file.id, FEATURE_GEOJSON, self.started, self.ended, [], None, None)
# Ensure there's an event for the parse
evt = TriggerEvent.objects.first()
self.assertEqual(evt.description[u'version'], u'1.0')
self.assertEqual(evt.description[u'file_name'], u'text.txt')
开发者ID:cuulee，项目名称:scale，代码行数:21，代码来源:test_models.py
示例16: test_remove_trigger_rule
def test_remove_trigger_rule(self):
'''Tests calling JobTypeManager.edit_job_type() that removes the trigger rule'''
name = 'my-job-type'
version = '1.0'
trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type=job_test_utils.MOCK_TYPE,
configuration=self.trigger_config.get_dict())
job_type = JobType.objects.create_job_type(name, version, self.job_interface, trigger_rule)
# Call test
JobType.objects.edit_job_type(job_type.id, None, None, True)
# Check results
job_type = JobType.objects.select_related('trigger_rule').get(pk=job_type.id)
self.assertDictEqual(job_type.get_job_interface().get_dict(), self.job_interface.get_dict())
self.assertEqual(job_type.revision_num, 1)
self.assertIsNone(job_type.trigger_rule)
trigger_rule = TriggerRule.objects.get(pk=trigger_rule.id)
self.assertFalse(trigger_rule.is_active)
num_of_revs = JobTypeRevision.objects.filter(job_type_id=job_type.id).count()
self.assertEqual(num_of_revs, 1)
开发者ID:cshamis，项目名称:scale，代码行数:21，代码来源:test_models.py
示例17: setUp
def setUp(self):
django.setup()
self.job_type1 = job_test_utils.create_job_type()
self.job_type2 = job_test_utils.create_job_type()
self.workspace = storage_test_utils.create_workspace()
self.trigger_config = {
'version': '1.0',
'condition': {
'media_type': 'text/plain',
},
'data': {
'input_data_name': 'input_file',
'workspace_name': self.workspace.name,
}
}
self.trigger_rule = trigger_test_utils.create_trigger_rule(trigger_type='PARSE', is_active=True,
configuration=self.trigger_config)
self.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,
},
}],
}
self.recipe_type = recipe_test_utils.create_recipe_type(name='my-type', definition=self.definition,
trigger_rule=self.trigger_rule)
开发者ID:droessne，项目名称:scale，代码行数:39，代码来源:test_views.py
示例18: setUp
def setUp(self):
django.setup()
self.input_name = 'Test Input'
self.output_name = 'Test Output'
self.workspace = storage_test_utils.create_workspace()
# This job trigger should not match due to a different media type
job_trigger_config = {
'version': '1.0',
'condition': {
'media_type': 'image/png',
},
'data': {
'input_data_name': self.input_name,
'workspace_name': self.workspace.name
},
}
job_trigger_rule = trigger_test_utils.create_trigger_rule(configuration=job_trigger_config)
interface_1 = {
'version': '1.0',
'command': 'my_cmd',
'command_arguments': 'args',
'input_data': [{
'name': self.input_name,
'type': 'file',
}],
}
self.job_type_1 = job_test_utils.create_job_type(interface=interface_1, trigger_rule=job_trigger_rule)
interface_2 = {
'version': '1.0',
'command': 'my_cmd',
'command_arguments': 'args',
'input_data': [{
'name': self.input_name,
'type': 'file',
}],
'output_data': [{
'name': self.output_name,
'type': 'file',
}],
}
self.job_type_2 = job_test_utils.create_job_type(interface=interface_2, trigger_rule=job_trigger_rule)
# create a recipe that runs both jobs
definition_1 = {
'version': '1.0',
'input_data': [{
'name': self.input_name,
'type': 'file',
'required': True,
}],
'jobs': [{
'name': 'Job 2',
'job_type': {
'name': self.job_type_2.name,
'version': self.job_type_2.version,
},
'recipe_inputs': [{
'recipe_input': self.input_name,
'job_input': self.input_name,
}],
}, {
'name': 'Job 1',
'job_type': {
'name': self.job_type_1.name,
'version': self.job_type_1.version,
},
'dependencies': [{
'name': 'Job 2',
'connections': [{
'output': self.output_name,
'input': self.input_name,
}],
}],
}],
}
self.when_parsed = now()
self.file_name = 'my_file.txt'
self.data_type = 'test_file_type'
self.media_type = 'text/plain'
self.source_file = SourceFile.objects.create(file_name=self.file_name, media_type=self.media_type, file_size=10,
data_type=self.data_type, file_path='the_path',
workspace=self.workspace)
self.source_file.add_data_type_tag('type1')
self.source_file.add_data_type_tag('type2')
self.source_file.add_data_type_tag('type3')
self.source_file.parsed = now()
recipe_trigger_config = {
'version': '1.0',
'condition': {
'media_type': 'text/plain',
},
'data': {
'input_data_name': self.input_name,
'workspace_name': self.workspace.name
#.........这里部分代码省略.........
开发者ID:AppliedIS，项目名称:scale，代码行数:101，代码来源:test_parse_trigger_handler.py
示例19: setUp
def setUp(self):
django.setup()
self.workspace = storage_test_utils.create_workspace()
self.file = storage_test_utils.create_file()
configuration = {
'version': '1.0',
'condition': {
'media_type': 'text/plain',
},
'data': {
'input_data_name': 'Recipe Input',
'workspace_name': self.workspace.name,
},
}
self.rule = trigger_test_utils.create_trigger_rule(configuration=configuration)
self.event = trigger_test_utils.create_trigger_event(rule=self.rule)
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)
self.definition_1 = {
'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',
}],
}],
}
RecipeDefinition(self.definition_1).validate_job_interfaces()
self.recipe_type = recipe_test_utils.create_recipe_type(definition=self.definition_1, trigger_rule=self.rule)
self.interface_2 = {
'version': '1.0',
'command': 'my_command',
'command_arguments': 'args',
'input_data': [{
'name': 'Test Input 2',
'type': 'files',
'media_types': ['image/tiff'],
}],
'output_data': [{
'name': 'Test Output 2',
'type': 'file',
}],
}
self.job_type_2 = job_test_utils.create_job_type(interface=self.interface_2)
self.definition_2 = {
'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',
#.........这里部分代码省略.........
开发者ID:ngageoint，项目名称:scale，代码行数:101，代码来源:test_models.py
注：本文中的trigger.test.utils.create_trigger_rule函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论