IntegrationTestCase.tearDown() - Code Metrics - Inspection of "Merge pull request #3612 from StackStorm/more_v2.3..." - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — plexxi-v2.3.2 ( 5d46fe )

unknown

created 2017-08-22 17:03 UTC

IntegrationTestCase.tearDown() B

↳ Parent: IntegrationTestCase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
dl	0
loc	27
rs	7.5384
c	0
b	0
f	0

# Licensed to the StackStorm, Inc ('StackStorm') 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.

try:
    import simplejson as json
except ImportError:
    import json

import os
import os.path
import sys
import shutil
import logging

import six
import eventlet
import psutil
from oslo_config import cfg
from unittest2 import TestCase

import unittest2
from st2common.util.api import get_full_public_api_url
from st2common.constants.runners import COMMON_ACTION_ENV_VARIABLES
from st2common.constants.system import AUTH_TOKEN_ENV_VARIABLE_NAME
from st2common.exceptions.db import StackStormDBObjectConflictError
from st2common.models.db import db_setup, db_teardown, db_ensure_indexes
from st2common.bootstrap.base import ResourceRegistrar
from st2common.bootstrap.configsregistrar import ConfigsRegistrar
from st2common.content.utils import get_packs_base_paths
from st2common.exceptions.db import StackStormDBObjectNotFoundError
import st2common.models.db.rule as rule_model
import st2common.models.db.rule_enforcement as rule_enforcement_model
import st2common.models.db.sensor as sensor_model
import st2common.models.db.trigger as trigger_model
import st2common.models.db.action as action_model
import st2common.models.db.keyvalue as keyvalue_model
import st2common.models.db.runner as runner_model
import st2common.models.db.execution as execution_model
import st2common.models.db.executionstate as executionstate_model
import st2common.models.db.liveaction as liveaction_model
import st2common.models.db.actionalias as actionalias_model
import st2common.models.db.policy as policy_model
import st2tests.config

# Imports for backward compatibility (those classes have been moved to standalone modules)
from st2tests.actions import BaseActionTestCase
from st2tests.sensors import BaseSensorTestCase
from st2tests.action_aliases import BaseActionAliasTestCase


__all__ = [
    'EventletTestCase',
    'DbTestCase',
    'DbModelTestCase',
    'CleanDbTestCase',
    'CleanFilesTestCase',
    'IntegrationTestCase',
    'RunnerTestCase',

    # Pack test classes
    'BaseSensorTestCase',
    'BaseActionTestCase',
    'BaseActionAliasTestCase'
]

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

ALL_MODELS = []
ALL_MODELS.extend(rule_model.MODELS)
ALL_MODELS.extend(sensor_model.MODELS)
ALL_MODELS.extend(trigger_model.MODELS)
ALL_MODELS.extend(action_model.MODELS)
ALL_MODELS.extend(keyvalue_model.MODELS)
ALL_MODELS.extend(runner_model.MODELS)
ALL_MODELS.extend(execution_model.MODELS)
ALL_MODELS.extend(executionstate_model.MODELS)
ALL_MODELS.extend(liveaction_model.MODELS)
ALL_MODELS.extend(actionalias_model.MODELS)
ALL_MODELS.extend(policy_model.MODELS)
ALL_MODELS.extend(rule_enforcement_model.MODELS)

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
TESTS_CONFIG_PATH = os.path.join(BASE_DIR, '../conf/st2.conf')


class RunnerTestCase(unittest2.TestCase):
    def assertCommonSt2EnvVarsAvailableInEnv(self, env):
        """
        Method which asserts that the common ST2 environment variables are present in the provided
        environment.
        """
        for var_name in COMMON_ACTION_ENV_VARIABLES:
            self.assertTrue(var_name in env)

        self.assertEqual(env['ST2_ACTION_API_URL'], get_full_public_api_url())
        self.assertTrue(env[AUTH_TOKEN_ENV_VARIABLE_NAME] is not None)


class BaseTestCase(TestCase):

    @classmethod
    def _register_packs(self):

        """
        Register all the packs inside the fixtures directory.
        """

        registrar = ResourceRegistrar(use_pack_cache=False)
        registrar.register_packs(base_dirs=get_packs_base_paths())

    @classmethod
    def _register_pack_configs(self, validate_configs=False):

        """
        Register all the packs inside the fixtures directory.
        """
        registrar = ConfigsRegistrar(use_pack_cache=False, validate_configs=validate_configs)
        registrar.register_from_packs(base_dirs=get_packs_base_paths())


class EventletTestCase(TestCase):
    """
    Base test class which performs eventlet monkey patching before the tests run
    and un-patching after the tests have finished running.
    """

    @classmethod
    def setUpClass(cls):
        eventlet.monkey_patch(
            os=True,
            select=True,
            socket=True,
            thread=False if '--use-debugger' in sys.argv else True,
            time=True
        )

    @classmethod
    def tearDownClass(cls):
        eventlet.monkey_patch(
            os=False,
            select=False,
            socket=False,
            thread=False,
            time=False
        )


class BaseDbTestCase(BaseTestCase):

    # Set to True to enable printing of all the log messages to the console
    DISPLAY_LOG_MESSAGES = False

    @classmethod
    def setUpClass(cls):
        st2tests.config.parse_args()

        if cls.DISPLAY_LOG_MESSAGES:
            config_path = os.path.join(BASE_DIR, '../conf/logging.conf')
            logging.config.fileConfig(config_path,
                                      disable_existing_loggers=False)

    @classmethod
    def _establish_connection_and_re_create_db(cls):
        username = cfg.CONF.database.username if hasattr(cfg.CONF.database, 'username') else None
        password = cfg.CONF.database.password if hasattr(cfg.CONF.database, 'password') else None
        cls.db_connection = db_setup(
            cfg.CONF.database.db_name, cfg.CONF.database.host, cfg.CONF.database.port,
            username=username, password=password, ensure_indexes=False)
        cls._drop_collections()
        cls.db_connection.drop_database(cfg.CONF.database.db_name)

        # Explicity ensure indexes after we re-create the DB otherwise ensure_indexes could failure
        # inside db_setup if test inserted invalid data
        db_ensure_indexes()

    @classmethod
    def _drop_db(cls):
        cls._drop_collections()
        if cls.db_connection is not None:
            cls.db_connection.drop_database(cfg.CONF.database.db_name)
        db_teardown()
        cls.db_connection = None

    @classmethod
    def _drop_collections(cls):
        # XXX: Explicitly drop all the collection. Otherwise, artifacts are left over in
        # subsequent tests.
        # See: https://github.com/MongoEngine/mongoengine/issues/566
        # See: https://github.com/MongoEngine/mongoengine/issues/565
        global ALL_MODELS

        for model in ALL_MODELS:
            model.drop_collection()


class DbTestCase(BaseDbTestCase):
    """
    This class drops and re-creates the database once per TestCase run.

    This means database is only dropped once before all the tests from this class run. This means
    data is persited between different tests in this class.
    """

    db_connection = None
    current_result = None
    register_packs = False
    register_pack_configs = False

    @classmethod
    def setUpClass(cls):
        BaseDbTestCase.setUpClass()
        cls._establish_connection_and_re_create_db()

        if cls.register_packs:
            cls._register_packs()

        if cls.register_pack_configs:
            cls._register_pack_configs()

    @classmethod
    def tearDownClass(cls):
        drop_db = True

        if cls.current_result.errors or cls.current_result.failures:
            # Don't drop DB on test failure
            drop_db = False

        if drop_db:
            cls._drop_db()

    def run(self, result=None):
        # Remember result for use in tearDown and tearDownClass
        self.current_result = result
        self.__class__.current_result = result
        super(DbTestCase, self).run(result=result)


class DbModelTestCase(DbTestCase):
    access_type = None

    @classmethod
    def setUpClass(cls):
        super(DbModelTestCase, cls).setUpClass()
        cls.db_type = cls.access_type.impl.model

    def _assert_fields_equal(self, a, b, exclude=None):
        exclude = exclude or []
        fields = {k: v for k, v in six.iteritems(self.db_type._fields) if k not in exclude}

        assert_funcs = {
            'mongoengine.fields.DictField': self.assertDictEqual,
            'mongoengine.fields.ListField': self.assertListEqual,
            'mongoengine.fields.SortedListField': self.assertListEqual
        }

        for k, v in six.iteritems(fields):
            assert_func = assert_funcs.get(str(v), self.assertEqual)
            assert_func(getattr(a, k, None), getattr(b, k, None))

    def _assert_values_equal(self, a, values=None):
        values = values or {}

        assert_funcs = {
            'dict': self.assertDictEqual,
            'list': self.assertListEqual
        }

        for k, v in six.iteritems(values):
            assert_func = assert_funcs.get(type(v).__name__, self.assertEqual)
            assert_func(getattr(a, k, None), v)

    def _assert_crud(self, instance, defaults=None, updates=None):
        # Assert instance is not already in the database.
        self.assertIsNone(getattr(instance, 'id', None))

        # Assert default values are assigned.
        self._assert_values_equal(instance, values=defaults)

        # Assert instance is created in the datbaase.
        saved = self.access_type.add_or_update(instance)
        self.assertIsNotNone(saved.id)
        self._assert_fields_equal(instance, saved, exclude=['id'])
        retrieved = self.access_type.get_by_id(saved.id)
        self._assert_fields_equal(saved, retrieved)

        # Assert instance is updated in the database.
        for k, v in six.iteritems(updates or {}):
            setattr(instance, k, v)

        updated = self.access_type.add_or_update(instance)
        self._assert_fields_equal(instance, updated)

        # Assert instance is deleted from the database.
        retrieved = self.access_type.get_by_id(instance.id)
        retrieved.delete()
        self.assertRaises(StackStormDBObjectNotFoundError,
                          self.access_type.get_by_id, instance.id)

    def _assert_unique_key_constraint(self, instance):
        # Assert instance is not already in the database.
        self.assertIsNone(getattr(instance, 'id', None))

        # Assert instance is created in the datbaase.
        saved = self.access_type.add_or_update(instance)
        self.assertIsNotNone(saved.id)

        # Assert exception is thrown if try to create same instance again.
        delattr(instance, 'id')
        self.assertRaises(StackStormDBObjectConflictError,
                          self.access_type.add_or_update,
                          instance)


class CleanDbTestCase(BaseDbTestCase):
    """
    Class which ensures database is re-created before running each test method.

    This means each test inside this class is self-sustained and starts with a clean (empty)
    database.
    """

    register_packs = False
    register_pack_configs = False

    def setUp(self):
        self._establish_connection_and_re_create_db()

        if self.register_packs:
            self._register_packs()

        if self.register_pack_configs:
            self._register_pack_configs()


class CleanFilesTestCase(TestCase):
    """
    Base test class which deletes specified files and directories on setUp and `tearDown.
    """
    to_delete_files = []
    to_delete_directories = []

    def setUp(self):
        super(CleanFilesTestCase, self).setUp()
        self._delete_files()

    def tearDown(self):
        super(CleanFilesTestCase, self).tearDown()
        self._delete_files()

    def _delete_files(self):
        for file_path in self.to_delete_files:
            if not os.path.isfile(file_path):
                continue

            try:
                os.remove(file_path)
            except Exception:
                pass

        for file_path in self.to_delete_directories:
            if not os.path.isdir(file_path):
                continue

            try:
                shutil.rmtree(file_path)
            except Exception:
                pass


class IntegrationTestCase(TestCase):
    """
    Base test class for integration tests to inherit from.

    It includes various utility functions and assert methods for working with processes.
    """

    # Set to True to print process stdout and stderr in tearDown after killing the processes
    # which are still alive
    print_stdout_stderr_on_teardown = False

    processes = {}

    def tearDown(self):
        super(IntegrationTestCase, self).tearDown()

        # Make sure we kill all the processes on teardown so they don't linger around if an
        # exception was thrown.
        for pid, process in self.processes.items():

            try:
                process.kill()
            except OSError:
                # Process already exited or similar
                pass

            if self.print_stdout_stderr_on_teardown:
                try:
                    stdout = process.stdout.read()
                except:
                    stdout = None

                try:
                    stderr = process.stderr.read()
                except:
                    stderr = None

                print('Process "%s"' % (process.pid))
                print('Stdout: %s' % (stdout))
                print('Stderr: %s' % (stderr))

    def add_process(self, process):
        """
        Add a process to the local data structure to make sure it will get killed and cleaned up on
        tearDown.
        """
        self.processes[process.pid] = process

    def remove_process(self, process):
        """
        Remove process from a local data structure.
        """
        if process.pid in self.processes:
            del self.processes[process.pid]

    def assertProcessIsRunning(self, process):
        """
        Assert that a long running process provided Process object as returned by subprocess.Popen
        has succesfuly started and is running.
        """
        if not process:
            raise ValueError('process is None')

        return_code = process.poll()

        if return_code is not None:
            if process.stdout:
                stdout = process.stdout.read()
            else:
                stdout = ''

            if process.stderr:
                stderr = process.stderr.read()
            else:
                stderr = ''

            msg = ('Process exited with code=%s.\nStdout:\n%s\n\nStderr:\n%s' %
                   (return_code, stdout, stderr))
            self.fail(msg)

    def assertProcessExited(self, proc):
        try:
            status = proc.status()
        except psutil.NoSuchProcess:
            status = 'exited'

        if status not in ['exited', 'zombie']:
            self.fail('Process with pid "%s" is still running' % (proc.pid))


class FakeResponse(object):

    def __init__(self, text, status_code, reason):
        self.text = text
        self.status_code = status_code
        self.reason = reason

    def json(self):
        return json.loads(self.text)

    def raise_for_status(self):
        raise Exception(self.reason)


def get_fixtures_path():
    return os.path.join(os.path.dirname(__file__), 'fixtures')


def get_resources_path():
    return os.path.join(os.path.dirname(__file__), 'resources')


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			try:
17			import simplejson as json
18			except ImportError:
19			import json
20
21			import os
22			import os.path
23			import sys
24			import shutil
25			import logging
26
27			import six
28			import eventlet
29			import psutil
30			from oslo_config import cfg
31			from unittest2 import TestCase
32
33			import unittest2
34			from st2common.util.api import get_full_public_api_url
35			from st2common.constants.runners import COMMON_ACTION_ENV_VARIABLES
36			from st2common.constants.system import AUTH_TOKEN_ENV_VARIABLE_NAME
37			from st2common.exceptions.db import StackStormDBObjectConflictError
38			from st2common.models.db import db_setup, db_teardown, db_ensure_indexes
39			from st2common.bootstrap.base import ResourceRegistrar
40			from st2common.bootstrap.configsregistrar import ConfigsRegistrar
41			from st2common.content.utils import get_packs_base_paths
42			from st2common.exceptions.db import StackStormDBObjectNotFoundError
43			import st2common.models.db.rule as rule_model
44			import st2common.models.db.rule_enforcement as rule_enforcement_model
45			import st2common.models.db.sensor as sensor_model
46			import st2common.models.db.trigger as trigger_model
47			import st2common.models.db.action as action_model
48			import st2common.models.db.keyvalue as keyvalue_model
49			import st2common.models.db.runner as runner_model
50			import st2common.models.db.execution as execution_model
51			import st2common.models.db.executionstate as executionstate_model
52			import st2common.models.db.liveaction as liveaction_model
53			import st2common.models.db.actionalias as actionalias_model
54			import st2common.models.db.policy as policy_model
55			import st2tests.config
56
57			# Imports for backward compatibility (those classes have been moved to standalone modules)
58			from st2tests.actions import BaseActionTestCase
59			from st2tests.sensors import BaseSensorTestCase
60			from st2tests.action_aliases import BaseActionAliasTestCase
61
62
63			__all__ = [
64			'EventletTestCase',
65			'DbTestCase',
66			'DbModelTestCase',
67			'CleanDbTestCase',
68			'CleanFilesTestCase',
69			'IntegrationTestCase',
70			'RunnerTestCase',
71
72			# Pack test classes
73			'BaseSensorTestCase',
74			'BaseActionTestCase',
75			'BaseActionAliasTestCase'
76			]
77
78			BASE_DIR = os.path.dirname(os.path.abspath(__file__))
79
80			ALL_MODELS = []
81			ALL_MODELS.extend(rule_model.MODELS)
82			ALL_MODELS.extend(sensor_model.MODELS)
83			ALL_MODELS.extend(trigger_model.MODELS)
84			ALL_MODELS.extend(action_model.MODELS)
85			ALL_MODELS.extend(keyvalue_model.MODELS)
86			ALL_MODELS.extend(runner_model.MODELS)
87			ALL_MODELS.extend(execution_model.MODELS)
88			ALL_MODELS.extend(executionstate_model.MODELS)
89			ALL_MODELS.extend(liveaction_model.MODELS)
90			ALL_MODELS.extend(actionalias_model.MODELS)
91			ALL_MODELS.extend(policy_model.MODELS)
92			ALL_MODELS.extend(rule_enforcement_model.MODELS)
93
94			BASE_DIR = os.path.dirname(os.path.abspath(__file__))
95			TESTS_CONFIG_PATH = os.path.join(BASE_DIR, '../conf/st2.conf')
96
97
98			class RunnerTestCase(unittest2.TestCase):
99			def assertCommonSt2EnvVarsAvailableInEnv(self, env):
100			"""
101			Method which asserts that the common ST2 environment variables are present in the provided
102			environment.
103			"""
104			for var_name in COMMON_ACTION_ENV_VARIABLES:
105			self.assertTrue(var_name in env)
106
107			self.assertEqual(env['ST2_ACTION_API_URL'], get_full_public_api_url())
108			self.assertTrue(env[AUTH_TOKEN_ENV_VARIABLE_NAME] is not None)
109
110
111			class BaseTestCase(TestCase):
112
113			@classmethod
114			def _register_packs(self):
			0 ignored issues – show Coding Style Best Practice introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report The first argument of the class method `_register_packs` should be named `cls`. Loading history...
115			"""
116			Register all the packs inside the fixtures directory.
117			"""
118
119			registrar = ResourceRegistrar(use_pack_cache=False)
120			registrar.register_packs(base_dirs=get_packs_base_paths())
121
122			@classmethod
123			def _register_pack_configs(self, validate_configs=False):
			0 ignored issues – show Coding Style Best Practice introduced 2016-05-24 13:22 UTC by Report Bug Copy Issue Report The first argument of the class method `_register_pack_configs` should be named `cls`. Loading history...
124			"""
125			Register all the packs inside the fixtures directory.
126			"""
127			registrar = ConfigsRegistrar(use_pack_cache=False, validate_configs=validate_configs)
128			registrar.register_from_packs(base_dirs=get_packs_base_paths())
129
130
131			class EventletTestCase(TestCase):
132			"""
133			Base test class which performs eventlet monkey patching before the tests run
134			and un-patching after the tests have finished running.
135			"""
136
137			@classmethod
138			def setUpClass(cls):
139			eventlet.monkey_patch(
140			os=True,
141			select=True,
142			socket=True,
143			thread=False if '--use-debugger' in sys.argv else True,
144			time=True
145			)
146
147			@classmethod
148			def tearDownClass(cls):
149			eventlet.monkey_patch(
150			os=False,
151			select=False,
152			socket=False,
153			thread=False,
154			time=False
155			)
156
157
158			class BaseDbTestCase(BaseTestCase):
159
160			# Set to True to enable printing of all the log messages to the console
161			DISPLAY_LOG_MESSAGES = False
162
163			@classmethod
164			def setUpClass(cls):
165			st2tests.config.parse_args()
166
167			if cls.DISPLAY_LOG_MESSAGES:
168			config_path = os.path.join(BASE_DIR, '../conf/logging.conf')
169			logging.config.fileConfig(config_path,
170			disable_existing_loggers=False)
171
172			@classmethod
173			def _establish_connection_and_re_create_db(cls):
174			username = cfg.CONF.database.username if hasattr(cfg.CONF.database, 'username') else None
175			password = cfg.CONF.database.password if hasattr(cfg.CONF.database, 'password') else None
176			cls.db_connection = db_setup(
177			cfg.CONF.database.db_name, cfg.CONF.database.host, cfg.CONF.database.port,
178			username=username, password=password, ensure_indexes=False)
179			cls._drop_collections()
180			cls.db_connection.drop_database(cfg.CONF.database.db_name)
181
182			# Explicity ensure indexes after we re-create the DB otherwise ensure_indexes could failure
183			# inside db_setup if test inserted invalid data
184			db_ensure_indexes()
185
186			@classmethod
187			def _drop_db(cls):
188			cls._drop_collections()
189			if cls.db_connection is not None:
190			cls.db_connection.drop_database(cfg.CONF.database.db_name)
191			db_teardown()
192			cls.db_connection = None
193
194			@classmethod
195			def _drop_collections(cls):
196			# XXX: Explicitly drop all the collection. Otherwise, artifacts are left over in
197			# subsequent tests.
198			# See: https://github.com/MongoEngine/mongoengine/issues/566
199			# See: https://github.com/MongoEngine/mongoengine/issues/565
200			global ALL_MODELS
			0 ignored issues – show Unused Code introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report The variable `ALL_MODELS` was imported from global scope, but was never written to. Loading history...
201			for model in ALL_MODELS:
202			model.drop_collection()
203
204
205			class DbTestCase(BaseDbTestCase):
206			"""
207			This class drops and re-creates the database once per TestCase run.
208
209			This means database is only dropped once before all the tests from this class run. This means
210			data is persited between different tests in this class.
211			"""
212
213			db_connection = None
214			current_result = None
215			register_packs = False
216			register_pack_configs = False
217
218			@classmethod
219			def setUpClass(cls):
220			BaseDbTestCase.setUpClass()
221			cls._establish_connection_and_re_create_db()
222
223			if cls.register_packs:
224			cls._register_packs()
225
226			if cls.register_pack_configs:
227			cls._register_pack_configs()
228
229			@classmethod
230			def tearDownClass(cls):
231			drop_db = True
232
233			if cls.current_result.errors or cls.current_result.failures:
234			# Don't drop DB on test failure
235			drop_db = False
236
237			if drop_db:
238			cls._drop_db()
239
240			def run(self, result=None):
241			# Remember result for use in tearDown and tearDownClass
242			self.current_result = result
243			self.__class__.current_result = result
244			super(DbTestCase, self).run(result=result)
245
246
247			class DbModelTestCase(DbTestCase):
248			access_type = None
249
250			@classmethod
251			def setUpClass(cls):
252			super(DbModelTestCase, cls).setUpClass()
253			cls.db_type = cls.access_type.impl.model
254
255			def _assert_fields_equal(self, a, b, exclude=None):
256			exclude = exclude or []
257			fields = {k: v for k, v in six.iteritems(self.db_type._fields) if k not in exclude}
258
259			assert_funcs = {
260			'mongoengine.fields.DictField': self.assertDictEqual,
261			'mongoengine.fields.ListField': self.assertListEqual,
262			'mongoengine.fields.SortedListField': self.assertListEqual
263			}
264
265			for k, v in six.iteritems(fields):
266			assert_func = assert_funcs.get(str(v), self.assertEqual)
267			assert_func(getattr(a, k, None), getattr(b, k, None))
268
269			def _assert_values_equal(self, a, values=None):
270			values = values or {}
271
272			assert_funcs = {
273			'dict': self.assertDictEqual,
274			'list': self.assertListEqual
275			}
276
277			for k, v in six.iteritems(values):
278			assert_func = assert_funcs.get(type(v).__name__, self.assertEqual)
279			assert_func(getattr(a, k, None), v)
280
281			def _assert_crud(self, instance, defaults=None, updates=None):
282			# Assert instance is not already in the database.
283			self.assertIsNone(getattr(instance, 'id', None))
284
285			# Assert default values are assigned.
286			self._assert_values_equal(instance, values=defaults)
287
288			# Assert instance is created in the datbaase.
289			saved = self.access_type.add_or_update(instance)
290			self.assertIsNotNone(saved.id)
291			self._assert_fields_equal(instance, saved, exclude=['id'])
292			retrieved = self.access_type.get_by_id(saved.id)
293			self._assert_fields_equal(saved, retrieved)
294
295			# Assert instance is updated in the database.
296			for k, v in six.iteritems(updates or {}):
297			setattr(instance, k, v)
298
299			updated = self.access_type.add_or_update(instance)
300			self._assert_fields_equal(instance, updated)
301
302			# Assert instance is deleted from the database.
303			retrieved = self.access_type.get_by_id(instance.id)
304			retrieved.delete()
305			self.assertRaises(StackStormDBObjectNotFoundError,
306			self.access_type.get_by_id, instance.id)
307
308			def _assert_unique_key_constraint(self, instance):
309			# Assert instance is not already in the database.
310			self.assertIsNone(getattr(instance, 'id', None))
311
312			# Assert instance is created in the datbaase.
313			saved = self.access_type.add_or_update(instance)
314			self.assertIsNotNone(saved.id)
315
316			# Assert exception is thrown if try to create same instance again.
317			delattr(instance, 'id')
318			self.assertRaises(StackStormDBObjectConflictError,
319			self.access_type.add_or_update,
320			instance)
321
322
323			class CleanDbTestCase(BaseDbTestCase):
324			"""
325			Class which ensures database is re-created before running each test method.
326
327			This means each test inside this class is self-sustained and starts with a clean (empty)
328			database.
329			"""
330
331			register_packs = False
332			register_pack_configs = False
333
334			def setUp(self):
335			self._establish_connection_and_re_create_db()
336
337			if self.register_packs:
338			self._register_packs()
339
340			if self.register_pack_configs:
341			self._register_pack_configs()
342
343
344			class CleanFilesTestCase(TestCase):
345			"""
346			Base test class which deletes specified files and directories on setUp and `tearDown.
347			"""
348			to_delete_files = []
349			to_delete_directories = []
350
351			def setUp(self):
352			super(CleanFilesTestCase, self).setUp()
353			self._delete_files()
354
355			def tearDown(self):
356			super(CleanFilesTestCase, self).tearDown()
357			self._delete_files()
358
359			def _delete_files(self):
360			for file_path in self.to_delete_files:
361			if not os.path.isfile(file_path):
362			continue
363
364			try:
365			os.remove(file_path)
366			except Exception:
367			pass
368
369			for file_path in self.to_delete_directories:
370			if not os.path.isdir(file_path):
371			continue
372
373			try:
374			shutil.rmtree(file_path)
375			except Exception:
376			pass
377
378
379			class IntegrationTestCase(TestCase):
380			"""
381			Base test class for integration tests to inherit from.
382
383			It includes various utility functions and assert methods for working with processes.
384			"""
385
386			# Set to True to print process stdout and stderr in tearDown after killing the processes
387			# which are still alive
388			print_stdout_stderr_on_teardown = False
389
390			processes = {}
391
392			def tearDown(self):
393			super(IntegrationTestCase, self).tearDown()
394
395			# Make sure we kill all the processes on teardown so they don't linger around if an
396			# exception was thrown.
397			for pid, process in self.processes.items():
398
399			try:
400			process.kill()
401			except OSError:
402			# Process already exited or similar
403			pass
404
405			if self.print_stdout_stderr_on_teardown:
406			try:
407			stdout = process.stdout.read()
408			except:
409			stdout = None
410
411			try:
412			stderr = process.stderr.read()
413			except:
414			stderr = None
415
416			print('Process "%s"' % (process.pid))
417			print('Stdout: %s' % (stdout))
418			print('Stderr: %s' % (stderr))
419
420			def add_process(self, process):
421			"""
422			Add a process to the local data structure to make sure it will get killed and cleaned up on
423			tearDown.
424			"""
425			self.processes[process.pid] = process
426
427			def remove_process(self, process):
428			"""
429			Remove process from a local data structure.
430			"""
431			if process.pid in self.processes:
432			del self.processes[process.pid]
433
434			def assertProcessIsRunning(self, process):
435			"""
436			Assert that a long running process provided Process object as returned by subprocess.Popen
437			has succesfuly started and is running.
438			"""
439			if not process:
440			raise ValueError('process is None')
441
442			return_code = process.poll()
443
444			if return_code is not None:
445			if process.stdout:
446			stdout = process.stdout.read()
447			else:
448			stdout = ''
449
450			if process.stderr:
451			stderr = process.stderr.read()
452			else:
453			stderr = ''
454
455			msg = ('Process exited with code=%s.\nStdout:\n%s\n\nStderr:\n%s' %
456			(return_code, stdout, stderr))
457			self.fail(msg)
458
459			def assertProcessExited(self, proc):
460			try:
461			status = proc.status()
462			except psutil.NoSuchProcess:
463			status = 'exited'
464
465			if status not in ['exited', 'zombie']:
466			self.fail('Process with pid "%s" is still running' % (proc.pid))
467
468
469			class FakeResponse(object):
470
471			def __init__(self, text, status_code, reason):
472			self.text = text
473			self.status_code = status_code
474			self.reason = reason
475
476			def json(self):
477			return json.loads(self.text)
478
479			def raise_for_status(self):
480			raise Exception(self.reason)
481
482
483			def get_fixtures_path():
484			return os.path.join(os.path.dirname(__file__), 'fixtures')
485
486
487			def get_resources_path():
488			return os.path.join(os.path.dirname(__file__), 'resources')
489

Plexxi / st2

GitHub Access Token became invalid

Push — plexxi-v2.3.2 ( 5d46fe )

IntegrationTestCase.tearDown() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like