1# Copyright (c) 2020 Vestas Wind Systems A/S
2#
3# SPDX-License-Identifier: Apache-2.0
4
5import argparse
6import os
7from unittest.mock import patch, call
8
9import pytest
10
11from runners.canopen_program import CANopenBinaryRunner
12from conftest import RC_KERNEL_BIN
13
14#
15# Test values
16#
17
18TEST_DEF_CONTEXT = 'default'
19TEST_ALT_CONTEXT = 'alternate'
20
21#
22# Test cases
23#
24
25TEST_CASES = [(n, x, p, c, o, t, r, s, b)
26              for n in range(1, 3)
27              for x in (None, TEST_ALT_CONTEXT)
28              for p in range(1, 3)
29              for c in (False, True)
30              for o in (False, True)
31              for t in range(1, 3)
32              for r in range(1, 3)
33              for s in range(1, 3)
34              for b in range(False, True)]
35
36os_path_isfile = os.path.isfile
37
38def os_path_isfile_patch(filename):
39    if filename == RC_KERNEL_BIN:
40        return True
41    return os_path_isfile(filename)
42
43@pytest.mark.parametrize('test_case', TEST_CASES)
44@patch('runners.canopen_program.CANopenProgramDownloader')
45def test_canopen_program_create(cpd, test_case, runner_config):
46    '''Test CANopen runner created from command line parameters.'''
47    node_id, context, program_number, confirm, confirm_only, timeout, sdo_retries, sdo_timeout, block_transfer = test_case
48
49    args = ['--node-id', str(node_id)]
50    if context is not None:
51        args.extend(['--can-context', context])
52    if program_number:
53        args.extend(['--program-number', str(program_number)])
54    if not confirm:
55        args.append('--no-confirm')
56    if confirm_only:
57        args.append('--confirm-only')
58    if timeout:
59        args.extend(['--timeout', str(timeout)])
60    if sdo_retries:
61        args.extend(['--sdo-retries', str(sdo_retries)])
62    if sdo_timeout:
63        args.extend(['--sdo-timeout', str(sdo_timeout)])
64    if block_transfer:
65        args.append('--block_transfer')
66
67    mock = cpd.return_value
68    mock.flash_status.return_value = 0
69    mock.wait_for_flash_status_ok.return_value = 0
70    mock.swid.return_value = 0
71
72    parser = argparse.ArgumentParser(allow_abbrev=False)
73    CANopenBinaryRunner.add_parser(parser)
74    arg_namespace = parser.parse_args(args)
75    runner = CANopenBinaryRunner.create(runner_config, arg_namespace)
76    with patch('os.path.isfile', side_effect=os_path_isfile_patch):
77        runner.run('flash')
78
79    cpd.assert_called_once()
80    if context:
81        assert cpd.call_args == call(node_id=node_id,
82                                     can_context=context,
83                                     logger=runner.logger,
84                                     program_number=program_number,
85                                     sdo_retries=sdo_retries,
86                                     sdo_timeout=sdo_timeout,
87                                     block_transfer=block_transfer)
88    else:
89        assert cpd.call_args == call(node_id=node_id,
90                                     can_context=TEST_DEF_CONTEXT,
91                                     logger=runner.logger,
92                                     program_number=program_number,
93                                     sdo_retries=sdo_retries,
94                                     sdo_timeout=sdo_timeout,
95                                     block_transfer=block_transfer)
96
97    mock.connect.assert_called_once()
98
99    if confirm_only:
100        mock.wait_for_flash_status_ok.assert_called_with(timeout)
101        mock.swid.assert_called_once()
102        mock.enter_pre_operational.assert_called_once()
103        mock.zephyr_confirm_program.assert_called_once()
104        mock.clear_program.assert_not_called()
105        mock.stop_program.assert_not_called()
106        mock.download.assert_not_called()
107        mock.start_program.assert_not_called()
108        mock.wait_for_bootup.assert_not_called()
109    else:
110        mock.enter_pre_operational.assert_called()
111        mock.wait_for_flash_status_ok.assert_called_with(timeout)
112        mock.swid.assert_called()
113        mock.stop_program.assert_called_once()
114        mock.clear_program.assert_called_once()
115        mock.download.assert_called_once_with(RC_KERNEL_BIN)
116        mock.start_program.assert_called_once()
117        mock.wait_for_bootup.assert_called_once_with(timeout)
118        if confirm:
119            mock.zephyr_confirm_program.assert_called_once()
120        else:
121            mock.zephyr_confirm_program.assert_not_called()
122
123    mock.disconnect.assert_called_once()
124