src/ports/postgres/modules/utilities/test/unit_tests/test_control.py_in

# coding=utf-8
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

import sys
from os import path
# Add utilites module to the pythonpath.
sys.path.append(path.dirname(path.dirname(path.dirname(path.abspath(__file__)))))


import unittest
from mock import *
import sys
import plpy_mock as plpy

m4_changequote(`<!', `!>')


class ControlTestCase(unittest.TestCase):
    def setUp(self):
        patches = {
            'plpy': plpy
        }
        self.plpy_mock_execute = MagicMock()
        plpy.execute = self.plpy_mock_execute

        self.module_patcher = patch.dict('sys.modules', patches)
        self.module_patcher.start()

        import control
        self.subject = control

    def tearDown(self):
        self.module_patcher.stop()

    def test_ao_control_false(self):
        option = ('appendonly=true,blocksize=32768,compresstype=none,'
                  'checksum=true,orientation=row')
        self.plpy_mock_execute.return_value = [{'gp_default_storage_options': option}]
        with self.subject.AOControl(False) as C:
            self.assertFalse(C.storage_options_dict['appendonly'])
        self.plpy_mock_execute.assert_called_with(
            "set gp_default_storage_options='compresstype=none,blocksize=32768"
            ",appendonly=True,orientation=row,checksum=true'")

    def test_ao_control_true(self):
        option = ('appendonly=true,blocksize=32768,compresstype=none,'
                  'checksum=true,orientation=row')
        self.plpy_mock_execute.return_value = [{'gp_default_storage_options': option}]
        with self.subject.AOControl(True) as C:
            self.assertTrue(C.storage_options_dict['appendonly'])
        self.plpy_mock_execute.assert_called_with(
            "set gp_default_storage_options='compresstype=none,blocksize=32768"
            ",appendonly=True,orientation=row,checksum=true'")

    def test_ao_control_missing(self):
        option = ('appendonly=true,blocksize=32768,compresstype=none,'
                  'checksum=true,orientation=row')
        self.plpy_mock_execute.side_effect = plpy.SPIError(
            'Unrecognized configuration parameter "gp_default_storage_options"')
        with self.subject.AOControl(True) as C:
            self.assertFalse(C.guc_exists)

class SetGUCTestCase(unittest.TestCase):
    def setUp(self):
        patches = {
            'plpy': plpy
        }
        self.plpy_mock_execute = MagicMock()
        plpy.execute = self.plpy_mock_execute

        self.module_patcher = patch.dict('sys.modules', patches)
        self.module_patcher.start()

        import control
        self.subject = control

    def tearDown(self):
        self.module_patcher.stop()

    def test_set_guc_sets_new_value(self):
        self.plpy_mock_execute.return_value = [{'foo': 'old_bar'}]
        with self.subject.SetGUC("foo", "new_bar") as C:
            self.assertTrue("new_bar", C.new_guc_value)
        self.assertEqual(3, self.plpy_mock_execute.call_count)
        self.plpy_mock_execute.assert_has_calls([call('show foo'),
                                                 call("set foo='new_bar'"),
                                                 call("set foo='old_bar'")])


    def test_set_guc_missing(self):
        self.plpy_mock_execute.side_effect = plpy.SPIError(
            'Unrecognized configuration parameter "foo"')
        with self.subject.SetGUC("foo", "new_bar") as C:
            self.assertFalse(C.guc_exists)
        self.assertEqual(1, self.plpy_mock_execute.call_count)
        self.plpy_mock_execute.assert_has_calls([call('show foo')])

if __name__ == '__main__':
    unittest.main()