Source code for catcher_modules.service.prepare
from catcher.steps.external_step import ExternalStep
import catcher_modules.database
from catcher.utils import misc
from catcher_modules.utils import module_utils
[docs]class Prepare(ExternalStep):
"""
Used for bulk actions to prepare test data. Is useful when you need to prepare a lot of data.
This step consists of 3 parts:
1. write sql ddl schema file (optional) - describe all tables/schemas/privileges needed to be created
2. prepare data in a csv file (optional)
3. call Catcher's prepare step to populate csv content into the database
Both sql schema and csv file supports templates.
**Important**:
* populate step is designed to be supported by all steps (in future). Currently it is supported only by
Postges/Oracle/MSSql/MySql/SQLite steps.
* to populate json as Postgres Json data type you need to use **use_json: true** flag
:Input:
:populate: Populate existing service with predefined data.
- <service_name>: See each own step's documentation for the parameters description and
information. Note, that not all steps are compatible with prepare step.
- variables: Variables, which will override state (only for this prepare step).
Please, keep it mind, that resources directory is used for all data and schema files.
Populate existing postgres with data from `pg_data_file`.
::
steps:
- prepare:
populate:
postgres:
conf: {{ pg_conf }}
schema: {{ pg_schema_file }}
data: {{ pg_data_file }}
Multiple populates and can be run at the same time. This will populate existing s3 with data, start local
salesforce and postgres in docker and populates them as well.
::
steps:
- prepare:
populate:
s3:
conf: {{ s3_url }}
path: {{ s3_path }}
data: {{ s3_data }}
postgres:
conf: {{ pg_conf }}
schema: {{ pg_schema_file }}
data: {{ pg_data_file }}
Prepare step with variables override.
::
- prepare:
populate:
postgres:
conf: '{{ postgres_conf }}'
schema: create_personal_data_customer.sql
variables:
email: '{{ random("email") }}'
"""
def action(self, includes: dict, variables: dict) -> dict or tuple:
input_data = self.simple_input(variables)
variables_override = misc.merge_two_dicts(variables, input_data['populate'].get('variables'))
db_modules = module_utils.list_modules_in_package(catcher_modules.database)
for service, data in input_data['populate'].items():
if service == 'variables':
continue
if service in db_modules: # database
found = module_utils.find_class_in_module('catcher_modules.database.' + service, service)
found(**{service: data}).populate(variables_override, **data)
# TODO cache
# TODO s3
# TODO http mock
return variables