stackstorm-sql/tests/test_action_insert.py at b8ceadb0ec6d5e612a81ce1fae3e86b96ff165b5 · StackStorm-Exchange/stackstorm-sql · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
from sql_base_action_test_case import SqlBaseActionTestCase

from lib.base_action import BaseAction
from st2common.runners.base_action import Action
from insert import SQLInsertAction

import mock

__all__ = [
    'TestActionSQLInsertAction'
]


class TestActionSQLInsertAction(SqlBaseActionTestCase):
    __test__ = True
    action_cls = SQLInsertAction

    def test_init(self):
        action = self.get_action_instance({})
        self.assertIsInstance(action, SQLInsertAction)
        self.assertIsInstance(action, BaseAction)
        self.assertIsInstance(action, Action)

    @mock.patch('lib.base_action.BaseAction.db_connection')
    @mock.patch('insert.sqlalchemy')
    def test_run_object(self, mock_sqlalchemy, mock_connect_to_db):
        action = self.get_action_instance(self.config_good)
        connection_name = 'full'
        connection_config = self.config_good['connections'][connection_name]
        test_dict = {
            'table': 'Test_Table',
            'data': {
                'column_1': 'value_1',
                'column_2': 'value_2'
            }
        }
        test_dict.update(connection_config)
        action_meta = "MetaData"
        action_engine = "Engine Data"
        mock_sql_table = mock.Mock()
        insert_return = "Insert Statment"
        mock_sql_table.insert.return_value = insert_return
        mock_connection = mock.Mock()
        mock_connection.execute.return_value = "Successfully Executed Statement"
        mock_connection.close.return_value = "Successfully disconnected"
        mock_connect_to_db.return_value.__enter__.return_value = mock_connection

        action.conn = mock_connection
        action.meta = action_meta
        action.engine = action_engine
        mock_sqlalchemy.Table.return_value = mock_sql_table

        result = action.run(**test_dict)
        self.assertEqual(result, True)
        mock_connect_to_db.assert_called_once_with(test_dict)
        mock_sqlalchemy.Table.assert_called_once_with(test_dict['table'],
                                                      action_meta,
                                                      schema=None,
                                                      autoload=True,
                                                      autoload_with=action_engine)
        mock_connection.execute.assert_called_once_with(insert_return, [test_dict['data']])

    @mock.patch('lib.base_action.BaseAction.db_connection')
    @mock.patch('insert.sqlalchemy')
    def test_run_array(self, mock_sqlalchemy, mock_connect_to_db):
        action = self.get_action_instance(self.config_good)
        connection_name = 'full'
        test_dict = {
            'connection': connection_name,
            'table': 'Test_Table',
            'data': [{
                'column_1': 'value_1',
                'column_2': 'value_2'
            }, {
                'column_1': 'value_3',
                'column_2': 'value_4'
            }]
        }
        action_meta = "MetaData"
        action_engine = "Engine Data"
        mock_sql_table = mock.Mock()
        insert_return = "Insert Statment"
        mock_sql_table.insert.return_value = insert_return
        mock_connection = mock.Mock()
        mock_connection.execute.return_value = "Successfully Executed Statement"
        mock_connection.close.return_value = "Successfully disconnected"
        mock_connect_to_db.return_value.__enter__.return_value = mock_connection

        action.conn = mock_connection
        action.meta = action_meta
        action.engine = action_engine
        mock_sqlalchemy.Table.return_value = mock_sql_table

        result = action.run(**test_dict)
        self.assertEqual(result, True)
        mock_connect_to_db.assert_called_once_with(test_dict)
        mock_sqlalchemy.Table.assert_called_once_with(test_dict['table'],
                                                      action_meta,
                                                      schema=None,
                                                      autoload=True,
                                                      autoload_with=action_engine)
        mock_connection.execute.assert_called_once_with(insert_return, test_dict['data'])