Example: Upgrade safely

Upgrading to a new major release of Engine brings you new features, fixes bugs and can make your life easier. You should, however, make sure that your Engine processes are not affected by any of the breaking changes. Before upgrading, always review the Release Notes to familiarize yourself with what you may expect.

Another great way to make sure that the functionality that you need remains intact, is to write a flow script that tests critical parts of your processes before and after an update. To help you set up your own tests, we created an example for you that illustrates ways you can go about writing a comprehensive test.

Example test flow

Feel free to take inspiration from the example below.

If you manage your own workspace, you could run your own tests prior and after an update to ensure functionality. If we manage your installation (and therefore the update process), you can set up a webhook that we can call to know if your tests run successfully after we update your workspace.

example

import flow_api

def handler(system: flow_api.System, this: flow_api.Execution, inputs: dict):
    # contents
    #   1 test connections to external systems
    #   2 test self built scheduler
    #   3 test critical processes (with flag test=True)

    # prerequisites for this test:
    #   my-test-file.txt exists on the FTP server
    #   connetors 'my-ftp-connector', 'my-scp-conector', 'my-mssql-connector'
    #   and 'my-smtp-connector' are set up
    #   my-critical-flow-1 and my-critical-flow-2 exist and are written in a way
    #   that they have a test mode (input_value[test]=True)
    
    errors = []

    # 1 test connections to external systems

    # make sure the file doesn't exists from previous tests
    if system.file('my-test-file.txt').exists():
        system.file('my-test-file.txt').delete(permanently=True)
    
    # 1.1 download test file from FTP server
    try:
        this.connect(
            'my-ftp-connector',
            name='download from FTP server',
            src='my/test/file.txt',
            dst=f'cloudomation:my-test-file.txt',
        )
    except flow_api.DependencyFailedError:
        errors.append('my-ftp-connector failed')
    
    # 1.2 move test file via SCP connector
    try:
        this.connect(
            'my-scp-conector',
            name='move to remote',
            src=f'cloudomation:my-test-file.txt',
            dst=f'/path/to/remote/my-test-file.txt',
        )
    except flow_api.DependencyFailedError:
        errors.append('my-scp-connector failed')

    # delete test file
    system.file('my-test-file.txt').delete(permanently=True)

    # 1.3 execute MSSQL query
    try:
        this.connect(
            'my-mssql-connector',
            name='execute MSSQL query',
            fetchval='SELECT 1 FROM tbl_my_table',
        )
    except flow_api.DependencyFailedError:
        errors.append('my-mssql-connector failed')

    # 1.4 send mail
    try:
        this.connect(
            'my-smtp-connector',
            name='send mail',
            from_='my-user@my-domain.com',
            to=['other-user@my-domain.com'],
            subject=f'smtp connection test',
            text='This is a test.',
        )
    except flow_api.DependencyFailedError:
        errors.append('my-smtp-connector failed')

    # 2 test self built scheduler

    # 2.1 create schedule with my-self-build-scheduler
    my_schedule = system.schedule(f'my-test-schedule').save(
        scheduler_id='my-self-build-scheduler-id', # get id from self built scheduler
        setting_id='my-test-setting-id', # get id from setting to be used
        flow_id='my-test-flow-id', # get id from flow to be used
        project_id='my-test-project-id', # get id from project
        is_enabled=True,
    )

    # 2.2 check if schedule was successfully created
    this.sleep(delay_sec=5, message=f'give the system time to create schedule')
    found_schedule = (
        next(
            system.executions(
                limit=1,
                order='-created_at',
                filter_={
                    'and': [
                        {
                            'field': 'type',
                            'op': 'eq',
                            'value': 'SCHEDULE',
                        },
                        {
                            'field': 'schedule_id',
                            'op': 'eq',
                            'value': my_schedule.get('id'),
                        },
                        {
                            'field': 'created_at',
                            'op': 'gt',
                            'value': this.get('created_at'),
                        },
                        {
                            'field': 'status',
                            'op': 'eq',
                            'value': 'SCHEDULED',
                        },
                    ],
                },
            ),
            None
        )
    )
    if not found_schedule:
        errors.append('my-self-build-scheduler failed')

    # disable schedule again
    my_schedule.save(
        is_enabled=False,
    )

    # 3 test critical processes (with flag test=True)

    # 3.1 test my-critical-flow-1
    try:
        this.flow(
            'my-critical-flow-1',
            test=True,
        )
    except flow_api.DependencyFailedError:
        errors.append('my-critical-flow-1 failed')
    
    # 3.1 test my-critical-flow-2
    try:
        this.flow(
            'my-critical-flow-2',
            test=True,
        )
    except flow_api.DependencyFailedError:
        errors.append('my-critical-flow-2 failed')


    # 4 return the result of the test
    if errors:
        body=f'Test unsuccessful. Errors: {chr(10) + chr(10).join(errors)}'
        status=500
    else:
        body='Test successful.'
        status=200

    return this.webhook_response(
        body=body,
        status=status,
    )