stackstorm-sql/tests/test_action_delete.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
from sql_base_action_test_case import SqlBaseActionTestCase

from lib.base_action import BaseAction
from st2common.runners.base_action import Action
from delete import SQLDeleteAction

import mock

__all__ = [
    'TestActionSQLDeleteAction'
]


class TestActionSQLDeleteAction(SqlBaseActionTestCase):
    __test__ = True
    action_cls = SQLDeleteAction

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

    @mock.patch('lib.base_action.BaseAction.db_connection')
    @mock.patch('delete.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',
            'where': {
                'column_1': 'value_1'
            }
        }
        test_dict.update(connection_config)
        execute_dict = {'_column_1': 'value_1'}
        expected_result = {'affected_rows': 1}
        action_meta = 'MetaData'
        action_engine = "Engine Data"
        mock_where_return = "Where statment"
        mock_where = mock.Mock()
        mock_where.where.return_value = mock_where_return
        mock_sql_table = mock.Mock()
        mock_sql_table.delete.return_value = mock_where
        mock_query_results = mock.Mock(rowcount=1)
        mock_connection = mock.Mock()
        mock_connection.execute.return_value = mock_query_results
        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, expected_result)
        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(mock_where_return, execute_dict)

    @mock.patch('lib.base_action.BaseAction.db_connection')
    @mock.patch('delete.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'
        }
        expected_result = {'affected_rows': 20}
        action_meta = 'MetaData'
        action_engine = "Engine Data"
        mock_delete_return = "Delete Statement"
        mock_sql_table = mock.Mock()
        mock_sql_table.delete.return_value = mock_delete_return
        mock_query_results = mock.Mock(rowcount=20)
        mock_connection = mock.Mock()
        mock_connection.execute.return_value = mock_query_results
        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, expected_result)
        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(mock_delete_return, None)