poppy.core.test module¶

class poppy.core.test.CommandTestCase[source]¶

Bases: object

The CommandTestCase class is designed to be overridden in derived classes to create unit tests for commands.

Example

class TestMyPluginCommands(CommandTestCase):

def test_sub_command1(self):

# — create some fake data —

value1 = ‘value1’ value2 = ‘value2’

# — initialize the command —

command = [‘pop’,

‘my_command’, ‘my_sub_command’, ‘–my_option1’, value1, ‘–my_option2’, value2, ‘–dry-run’]

# — run the command —

self.run_command(command)

# — make assertions —

# test the result assert sorted([‘file1’, ‘file2’]) == sorted(os.listdir(‘my_result_path’))

# (...)

config_path = None¶

database_identifier = None¶

generate_url(**kwargs)[source]¶

Generate the URL of the database from the parameters.

load_configuration()[source]¶

load_settings()[source]¶

mock_configure()[source]¶

mock_configure_settings(module=None, dictionary={}, ignore_pipeline_settings=False)[source]¶

mock_create_connection(database, *args, **kwargs)[source]¶

mock_create_engine(database, *args, **kwargs)[source]¶

mock_session_factory()[source]¶

run_command(command)[source]¶

settings¶

settings_path = None¶

setup_database()[source]¶

Setup the database before the generation of the tested command

Returns:

setup_method(method)[source]¶

Setup the database before each test :return:

teardown_method(method)[source]¶

class poppy.core.test.TaskTestCase[source]¶

Bases: object

The TaskTestCase class is designed to be overridden in derived classes to create unit tests for tasks.

Example

class TestMyPluginTasks(TaskTestCase):

def test_task1(self):

# — initialize the task — from my_plugin import task1

self.task = task1()

# — create some fake data — # (you can use directly pipeline attributes)

self.task.pipeline.properties.input_dir = ‘my_input_dir’ self.task.pipeline.properties.output = ‘my_output_dir’

# (...)

# — run the task —

self.run_task()

# — make assertions —

# test the result assert self.task.pipeline.properties.result == ‘my_result’

# (...)

def test_task2(self):

# — initialize the task — from my_plugin import task2

self.task = task2()

# (...)

run_task()[source]¶

setUp()[source]¶

Setup the pipeline before each test :return:

task¶