test_rolling_upgrades.py 文件源码

def test_rum_gets_list_of_instances_to_upgrade(
    rolling_upgrade_manager,
    mock_aws_manager
):
    Instance = namedtuple('Instance', ['id'])

    instances = [Instance('instance1'),
                 Instance('instance2'),
                 Instance('instance3')]
    rolling_upgrade_manager.compare_instance_to_config = mock.Mock()
    mock_aws_manager.get_instances_for_asg.return_value = instances

    rolling_upgrade_manager.compare_instance_to_config.side_effect = \
        ([], [], ['diff'])

    result = rolling_upgrade_manager.get_instances_to_upgrade("test-asg", {})

    assert len(result) == 1
    assert instances[2] in result

    rolling_upgrade_manager.compare_instance_to_config.side_effect = \
        (['diff1', 'diff2'], [], ['diff'])
    result = rolling_upgrade_manager.get_instances_to_upgrade("test-asg", {})

    assert len(result) == 2
    assert instances[0] in result
    assert instances[2] in result

    rolling_upgrade_manager.compare_instance_to_config.side_effect = \
        ([], [], [])
    result = rolling_upgrade_manager.get_instances_to_upgrade("test-asg", {})

    assert len(result) == 0

    mock_aws_manager.get_instances_for_asg.assert_has_calls((
        mock.call('test-asg'),
        mock.call('test-asg'),
        mock.call('test-asg'),
    ))