st2tests.FixturesLoader._get_fixture_file_path_abs() - Code Metrics - Inspection of "Merge pull request #2246 from StackStorm/chatops/c..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d1f0a7...3b2ece )

by Edward

created 2015-11-30 10:28 UTC

_get_fixture_file_path_abs() A

↳ Parent: st2tests.FixturesLoader

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	2
rs	10
cc	1

# 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.

import copy
import os

import six

from st2common.content.loader import MetaLoader

from st2common.models.api.action import (ActionAPI, LiveActionAPI, ActionExecutionStateAPI,
                                         RunnerTypeAPI, ActionAliasAPI)
from st2common.models.api.auth import ApiKeyAPI, UserAPI
from st2common.models.api.execution import (ActionExecutionAPI)
from st2common.models.api.policy import (PolicyTypeAPI, PolicyAPI)
from st2common.models.api.rule import (RuleAPI)
from st2common.models.api.sensor import SensorTypeAPI
from st2common.models.api.trace import TraceAPI
from st2common.models.api.trigger import (TriggerAPI, TriggerTypeAPI, TriggerInstanceAPI)

from st2common.models.db.action import ActionDB
from st2common.models.db.actionalias import ActionAliasDB
from st2common.models.db.auth import ApiKeyDB, UserDB
from st2common.models.db.liveaction import LiveActionDB
from st2common.models.db.executionstate import ActionExecutionStateDB
from st2common.models.db.runner import RunnerTypeDB
from st2common.models.db.execution import (ActionExecutionDB)
from st2common.models.db.policy import (PolicyTypeDB, PolicyDB)
from st2common.models.db.rule import RuleDB
from st2common.models.db.sensor import SensorTypeDB
from st2common.models.db.trace import TraceDB
from st2common.models.db.trigger import (TriggerDB, TriggerTypeDB, TriggerInstanceDB)
from st2common.persistence.action import Action
from st2common.persistence.actionalias import ActionAlias
from st2common.persistence.execution import ActionExecution
from st2common.persistence.executionstate import ActionExecutionState
from st2common.persistence.auth import ApiKey, User
from st2common.persistence.liveaction import LiveAction
from st2common.persistence.runner import RunnerType
from st2common.persistence.policy import (PolicyType, Policy)
from st2common.persistence.rule import Rule
from st2common.persistence.sensor import SensorType
from st2common.persistence.trace import Trace
from st2common.persistence.trigger import (Trigger, TriggerType, TriggerInstance)


ALLOWED_DB_FIXTURES = ['actions', 'actionstates', 'aliases', 'executions', 'liveactions',
                       'policies', 'policytypes', 'rules', 'runners', 'sensors',
                       'triggertypes', 'triggers', 'triggerinstances', 'traces', 'apikeys',
                       'users']
ALLOWED_FIXTURES = copy.copy(ALLOWED_DB_FIXTURES)
ALLOWED_FIXTURES.extend(['actionchains', 'workflows'])

FIXTURE_DB_MODEL = {
    'actions': ActionDB,
    'aliases': ActionAliasDB,
    'actionstates': ActionExecutionStateDB,
    'apikeys': ApiKeyDB,
    'executions': ActionExecutionDB,
    'liveactions': LiveActionDB,
    'policies': PolicyDB,
    'policytypes': PolicyTypeDB,
    'rules': RuleDB,
    'runners': RunnerTypeDB,
    'sensors': SensorTypeDB,
    'traces': TraceDB,
    'triggertypes': TriggerTypeDB,
    'triggers': TriggerDB,
    'triggerinstances': TriggerInstanceDB,
    'users': UserDB
}

FIXTURE_API_MODEL = {
    'actions': ActionAPI,
    'aliases': ActionAliasAPI,
    'actionstates': ActionExecutionStateAPI,
    'apikeys': ApiKeyAPI,
    'executions': ActionExecutionAPI,
    'liveactions': LiveActionAPI,
    'policies': PolicyAPI,
    'policytypes': PolicyTypeAPI,
    'rules': RuleAPI,
    'runners': RunnerTypeAPI,
    'sensors': SensorTypeAPI,
    'traces': TraceAPI,
    'triggertypes': TriggerTypeAPI,
    'triggers': TriggerAPI,
    'triggerinstances': TriggerInstanceAPI,
    'users': UserAPI
}


FIXTURE_PERSISTENCE_MODEL = {
    'actions': Action,
    'aliases': ActionAlias,
    'actionstates': ActionExecutionState,
    'apikeys': ApiKey,
    'executions': ActionExecution,
    'liveactions': LiveAction,
    'policies': Policy,
    'policytypes': PolicyType,
    'rules': Rule,
    'runners': RunnerType,
    'sensors': SensorType,
    'traces': Trace,
    'triggertypes': TriggerType,
    'triggers': Trigger,
    'triggerinstances': TriggerInstance,
    'users': User
}


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


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


class FixturesLoader(object):
    def __init__(self):
        self.meta_loader = MetaLoader()

    def save_fixtures_to_db(self, fixtures_pack='generic', fixtures_dict=None):
        """
        Loads fixtures specified in fixtures_dict into the database
        and returns DB models for the fixtures.

        fixtures_dict should be of the form:
        {
            'actions': ['action-1.yaml', 'action-2.yaml'],
            'rules': ['rule-1.yaml'],
            'liveactions': ['execution-1.yaml']
        }

        :param fixtures_pack: Name of the pack to load fixtures from.
        :type fixtures_pack: ``str``

        :param fixtures_dict: Dictionary specifying the fixtures to load for each type.
        :type fixtures_dict: ``dict``

        :rtype: ``dict``
        """
        if fixtures_dict is None:
            fixtures_dict = {}
        fixtures_pack_path = self._validate_fixtures_pack(fixtures_pack)
        self._validate_fixture_dict(fixtures_dict, allowed=ALLOWED_DB_FIXTURES)

        db_models = {}
        for fixture_type, fixtures in six.iteritems(fixtures_dict):

            API_MODEL = FIXTURE_API_MODEL.get(fixture_type, None)
            PERSISTENCE_MODEL = FIXTURE_PERSISTENCE_MODEL.get(fixture_type, None)

            loaded_fixtures = {}
            for fixture in fixtures:
                fixture_dict = self.meta_loader.load(
                    self._get_fixture_file_path_abs(fixtures_pack_path, fixture_type, fixture))
                api_model = API_MODEL(**fixture_dict)
                db_model = API_MODEL.to_model(api_model)
                db_model = PERSISTENCE_MODEL.add_or_update(db_model)
                loaded_fixtures[fixture] = db_model

            db_models[fixture_type] = loaded_fixtures

        return db_models

    def load_fixtures(self, fixtures_pack='generic', fixtures_dict=None):
        """
        Loads fixtures specified in fixtures_dict. We
        simply want to load the meta into dict objects.

        fixtures_dict should be of the form:
        {
            'actionchains': ['actionchain1.yaml', 'actionchain2.yaml'],
            'workflows': ['workflow.yaml']
        }

        :param fixtures_pack: Name of the pack to load fixtures from.
        :type fixtures_pack: ``str``

        :param fixtures_dict: Dictionary specifying the fixtures to load for each type.
        :type fixtures_dict: ``dict``

        :rtype: ``dict``
        """
        if not fixtures_dict:
            return {}
        fixtures_pack_path = self._validate_fixtures_pack(fixtures_pack)
        self._validate_fixture_dict(fixtures_dict)

        all_fixtures = {}
        for fixture_type, fixtures in six.iteritems(fixtures_dict):
            loaded_fixtures = {}
            for fixture in fixtures:
                fixture_dict = self.meta_loader.load(
                    self._get_fixture_file_path_abs(fixtures_pack_path, fixture_type, fixture))
                loaded_fixtures[fixture] = fixture_dict
            all_fixtures[fixture_type] = loaded_fixtures

        return all_fixtures

    def load_models(self, fixtures_pack='generic', fixtures_dict=None):
        """
        Loads fixtures specified in fixtures_dict as db models. This method must be
        used for fixtures that have associated DB models. We simply want to load the
        meta as DB models but don't want to save them to db.

        fixtures_dict should be of the form:
        {
            'actions': ['action-1.yaml', 'action-2.yaml'],
            'rules': ['rule-1.yaml'],
            'liveactions': ['execution-1.yaml']
        }

        :param fixtures_pack: Name of the pack to load fixtures from.
        :type fixtures_pack: ``str``

        :param fixtures_dict: Dictionary specifying the fixtures to load for each type.
        :type fixtures_dict: ``dict``

        :rtype: ``dict``
        """
        if not fixtures_dict:
            return {}
        fixtures_pack_path = self._validate_fixtures_pack(fixtures_pack)
        self._validate_fixture_dict(fixtures_dict, allowed=ALLOWED_DB_FIXTURES)

        all_fixtures = {}
        for fixture_type, fixtures in six.iteritems(fixtures_dict):

            API_MODEL = FIXTURE_API_MODEL.get(fixture_type, None)

            loaded_models = {}
            for fixture in fixtures:
                fixture_dict = self.meta_loader.load(
                    self._get_fixture_file_path_abs(fixtures_pack_path, fixture_type, fixture))
                api_model = API_MODEL(**fixture_dict)
                db_model = API_MODEL.to_model(api_model)
                loaded_models[fixture] = db_model
            all_fixtures[fixture_type] = loaded_models

        return all_fixtures

    def delete_fixtures_from_db(self, fixtures_pack='generic', fixtures_dict=None,
                                raise_on_fail=False):
        """
        Deletes fixtures specified in fixtures_dict from the database.

        fixtures_dict should be of the form:
        {
            'actions': ['action-1.yaml', 'action-2.yaml'],
            'rules': ['rule-1.yaml'],
            'liveactions': ['execution-1.yaml']
        }

        :param fixtures_pack: Name of the pack to delete fixtures from.
        :type fixtures_pack: ``str``

        :param fixtures_dict: Dictionary specifying the fixtures to delete for each type.
        :type fixtures_dict: ``dict``

        :param raise_on_fail: Optional If True, raises exception if delete fails on any fixture.
        :type raise_on_fail: ``boolean``
        """
        if not fixtures_dict:
            return
        fixtures_pack_path = self._validate_fixtures_pack(fixtures_pack)
        self._validate_fixture_dict(fixtures_dict)

        for fixture_type, fixtures in six.iteritems(fixtures_dict):
            API_MODEL = FIXTURE_API_MODEL.get(fixture_type, None)
            PERSISTENCE_MODEL = FIXTURE_PERSISTENCE_MODEL.get(fixture_type, None)
            for fixture in fixtures:
                fixture_dict = self.meta_loader.load(
                    self._get_fixture_file_path_abs(fixtures_pack_path, fixture_type, fixture))
                # Note that when we have a reference mechanism consistent for
                # every model, we can just do a get and delete the object. Until
                # then, this model conversions are necessary.
                api_model = API_MODEL(**fixture_dict)
                db_model = API_MODEL.to_model(api_model)
                try:
                    PERSISTENCE_MODEL.delete(db_model)
                except:
                    if raise_on_fail:
                        raise

    def delete_models_from_db(self, models_dict, raise_on_fail=False):
        """
        Deletes models specified in models_dict from the database.

        models_dict should be of the form:
        {
            'actions': [ACTION1, ACTION2],
            'rules': [RULE1],
            'liveactions': [EXECUTION]
        }

        :param fixtures_dict: Dictionary specifying the fixtures to delete for each type.
        :type fixtures_dict: ``dict``.

        :param raise_on_fail: Optional If True, raises exception if delete fails on any model.
        :type raise_on_fail: ``boolean``
        """
        for model_type, models in six.iteritems(models_dict):
            PERSISTENCE_MODEL = FIXTURE_PERSISTENCE_MODEL.get(model_type, None)
            for model in models:
                try:
                    PERSISTENCE_MODEL.delete(model)
                except:
                    if raise_on_fail:
                        raise

    def _validate_fixtures_pack(self, fixtures_pack):
        fixtures_pack_path = self._get_fixtures_pack_path(fixtures_pack)

        if not self._is_fixture_pack_exists(fixtures_pack_path):
            raise Exception('Fixtures pack not found ' +
                            'in fixtures path %s.' % get_fixtures_base_path())
        return fixtures_pack_path

    def _validate_fixture_dict(self, fixtures_dict, allowed=ALLOWED_FIXTURES):
        fixture_types = fixtures_dict.keys()
        for fixture_type in fixture_types:
            if fixture_type not in allowed:
                raise Exception('Disallowed fixture type: %s' % fixture_type)

    def _is_fixture_pack_exists(self, fixtures_pack_path):
        return os.path.exists(fixtures_pack_path)

    def _get_fixture_file_path_abs(self, fixtures_pack_path, fixtures_type, fixture_name):
        return os.path.join(fixtures_pack_path, fixtures_type, fixture_name)

    def _get_fixtures_pack_path(self, fixtures_pack_name):
        return os.path.join(get_fixtures_base_path(), fixtures_pack_name)

    def get_fixture_file_path_abs(self, fixtures_pack, fixtures_type, fixture_name):
        return os.path.join(get_fixtures_base_path(), fixtures_pack, fixtures_type, fixture_name)


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			import copy
17			import os
18
19			import six
20
21			from st2common.content.loader import MetaLoader
22
23			from st2common.models.api.action import (ActionAPI, LiveActionAPI, ActionExecutionStateAPI,
24			RunnerTypeAPI, ActionAliasAPI)
25			from st2common.models.api.auth import ApiKeyAPI, UserAPI
26			from st2common.models.api.execution import (ActionExecutionAPI)
27			from st2common.models.api.policy import (PolicyTypeAPI, PolicyAPI)
28			from st2common.models.api.rule import (RuleAPI)
29			from st2common.models.api.sensor import SensorTypeAPI
30			from st2common.models.api.trace import TraceAPI
31			from st2common.models.api.trigger import (TriggerAPI, TriggerTypeAPI, TriggerInstanceAPI)
32
33			from st2common.models.db.action import ActionDB
34			from st2common.models.db.actionalias import ActionAliasDB
35			from st2common.models.db.auth import ApiKeyDB, UserDB
36			from st2common.models.db.liveaction import LiveActionDB
37			from st2common.models.db.executionstate import ActionExecutionStateDB
38			from st2common.models.db.runner import RunnerTypeDB
39			from st2common.models.db.execution import (ActionExecutionDB)
40			from st2common.models.db.policy import (PolicyTypeDB, PolicyDB)
41			from st2common.models.db.rule import RuleDB
42			from st2common.models.db.sensor import SensorTypeDB
43			from st2common.models.db.trace import TraceDB
44			from st2common.models.db.trigger import (TriggerDB, TriggerTypeDB, TriggerInstanceDB)
45			from st2common.persistence.action import Action
46			from st2common.persistence.actionalias import ActionAlias
47			from st2common.persistence.execution import ActionExecution
48			from st2common.persistence.executionstate import ActionExecutionState
49			from st2common.persistence.auth import ApiKey, User
50			from st2common.persistence.liveaction import LiveAction
51			from st2common.persistence.runner import RunnerType
52			from st2common.persistence.policy import (PolicyType, Policy)
53			from st2common.persistence.rule import Rule
54			from st2common.persistence.sensor import SensorType
55			from st2common.persistence.trace import Trace
56			from st2common.persistence.trigger import (Trigger, TriggerType, TriggerInstance)
57
58
59			ALLOWED_DB_FIXTURES = ['actions', 'actionstates', 'aliases', 'executions', 'liveactions',
60			'policies', 'policytypes', 'rules', 'runners', 'sensors',
61			'triggertypes', 'triggers', 'triggerinstances', 'traces', 'apikeys',
62			'users']
63			ALLOWED_FIXTURES = copy.copy(ALLOWED_DB_FIXTURES)
64			ALLOWED_FIXTURES.extend(['actionchains', 'workflows'])
65
66			FIXTURE_DB_MODEL = {
67			'actions': ActionDB,
68			'aliases': ActionAliasDB,
69			'actionstates': ActionExecutionStateDB,
70			'apikeys': ApiKeyDB,
71			'executions': ActionExecutionDB,
72			'liveactions': LiveActionDB,
73			'policies': PolicyDB,
74			'policytypes': PolicyTypeDB,
75			'rules': RuleDB,
76			'runners': RunnerTypeDB,
77			'sensors': SensorTypeDB,
78			'traces': TraceDB,
79			'triggertypes': TriggerTypeDB,
80			'triggers': TriggerDB,
81			'triggerinstances': TriggerInstanceDB,
82			'users': UserDB
83			}
84
85			FIXTURE_API_MODEL = {
86			'actions': ActionAPI,
87			'aliases': ActionAliasAPI,
88			'actionstates': ActionExecutionStateAPI,
89			'apikeys': ApiKeyAPI,
90			'executions': ActionExecutionAPI,
91			'liveactions': LiveActionAPI,
92			'policies': PolicyAPI,
93			'policytypes': PolicyTypeAPI,
94			'rules': RuleAPI,
95			'runners': RunnerTypeAPI,
96			'sensors': SensorTypeAPI,
97			'traces': TraceAPI,
98			'triggertypes': TriggerTypeAPI,
99			'triggers': TriggerAPI,
100			'triggerinstances': TriggerInstanceAPI,
101			'users': UserAPI
102			}
103
104
105			FIXTURE_PERSISTENCE_MODEL = {
106			'actions': Action,
107			'aliases': ActionAlias,
108			'actionstates': ActionExecutionState,
109			'apikeys': ApiKey,
110			'executions': ActionExecution,
111			'liveactions': LiveAction,
112			'policies': Policy,
113			'policytypes': PolicyType,
114			'rules': Rule,
115			'runners': RunnerType,
116			'sensors': SensorType,
117			'traces': Trace,
118			'triggertypes': TriggerType,
119			'triggers': Trigger,
120			'triggerinstances': TriggerInstance,
121			'users': User
122			}
123
124
125			def get_fixtures_base_path():
126			return os.path.join(os.path.dirname(__file__), 'fixtures')
127
128
129			def get_resources_base_path():
130			return os.path.join(os.path.dirname(__file__), 'resources')
131
132
133			class FixturesLoader(object):
134			def __init__(self):
135			self.meta_loader = MetaLoader()
136
137			def save_fixtures_to_db(self, fixtures_pack='generic', fixtures_dict=None):
138			"""
139			Loads fixtures specified in fixtures_dict into the database
140			and returns DB models for the fixtures.
141
142			fixtures_dict should be of the form:
143			{
144			'actions': ['action-1.yaml', 'action-2.yaml'],
145			'rules': ['rule-1.yaml'],
146			'liveactions': ['execution-1.yaml']
147			}
148
149			:param fixtures_pack: Name of the pack to load fixtures from.
150			:type fixtures_pack: ``str``
151
152			:param fixtures_dict: Dictionary specifying the fixtures to load for each type.
153			:type fixtures_dict: ``dict``
154
155			:rtype: ``dict``
156			"""
157			if fixtures_dict is None:
158			fixtures_dict = {}
159			fixtures_pack_path = self._validate_fixtures_pack(fixtures_pack)
160			self._validate_fixture_dict(fixtures_dict, allowed=ALLOWED_DB_FIXTURES)
161
162			db_models = {}
163			for fixture_type, fixtures in six.iteritems(fixtures_dict):
164
165			API_MODEL = FIXTURE_API_MODEL.get(fixture_type, None)
166			PERSISTENCE_MODEL = FIXTURE_PERSISTENCE_MODEL.get(fixture_type, None)
167
168			loaded_fixtures = {}
169			for fixture in fixtures:
170			fixture_dict = self.meta_loader.load(
171			self._get_fixture_file_path_abs(fixtures_pack_path, fixture_type, fixture))
172			api_model = API_MODEL(**fixture_dict)
173			db_model = API_MODEL.to_model(api_model)
174			db_model = PERSISTENCE_MODEL.add_or_update(db_model)
175			loaded_fixtures[fixture] = db_model
176
177			db_models[fixture_type] = loaded_fixtures
178
179			return db_models
180
181			def load_fixtures(self, fixtures_pack='generic', fixtures_dict=None):
182			"""
183			Loads fixtures specified in fixtures_dict. We
184			simply want to load the meta into dict objects.
185
186			fixtures_dict should be of the form:
187			{
188			'actionchains': ['actionchain1.yaml', 'actionchain2.yaml'],
189			'workflows': ['workflow.yaml']
190			}
191
192			:param fixtures_pack: Name of the pack to load fixtures from.
193			:type fixtures_pack: ``str``
194
195			:param fixtures_dict: Dictionary specifying the fixtures to load for each type.
196			:type fixtures_dict: ``dict``
197
198			:rtype: ``dict``
199			"""
200			if not fixtures_dict:
201			return {}
202			fixtures_pack_path = self._validate_fixtures_pack(fixtures_pack)
203			self._validate_fixture_dict(fixtures_dict)
204
205			all_fixtures = {}
206			for fixture_type, fixtures in six.iteritems(fixtures_dict):
207			loaded_fixtures = {}
208			for fixture in fixtures:
209			fixture_dict = self.meta_loader.load(
210			self._get_fixture_file_path_abs(fixtures_pack_path, fixture_type, fixture))
211			loaded_fixtures[fixture] = fixture_dict
212			all_fixtures[fixture_type] = loaded_fixtures
213
214			return all_fixtures
215
216			def load_models(self, fixtures_pack='generic', fixtures_dict=None):
217			"""
218			Loads fixtures specified in fixtures_dict as db models. This method must be
219			used for fixtures that have associated DB models. We simply want to load the
220			meta as DB models but don't want to save them to db.
221
222			fixtures_dict should be of the form:
223			{
224			'actions': ['action-1.yaml', 'action-2.yaml'],
225			'rules': ['rule-1.yaml'],
226			'liveactions': ['execution-1.yaml']
227			}
228
229			:param fixtures_pack: Name of the pack to load fixtures from.
230			:type fixtures_pack: ``str``
231
232			:param fixtures_dict: Dictionary specifying the fixtures to load for each type.
233			:type fixtures_dict: ``dict``
234
235			:rtype: ``dict``
236			"""
237			if not fixtures_dict:
238			return {}
239			fixtures_pack_path = self._validate_fixtures_pack(fixtures_pack)
240			self._validate_fixture_dict(fixtures_dict, allowed=ALLOWED_DB_FIXTURES)
241
242			all_fixtures = {}
243			for fixture_type, fixtures in six.iteritems(fixtures_dict):
244
245			API_MODEL = FIXTURE_API_MODEL.get(fixture_type, None)
246
247			loaded_models = {}
248			for fixture in fixtures:
249			fixture_dict = self.meta_loader.load(
250			self._get_fixture_file_path_abs(fixtures_pack_path, fixture_type, fixture))
251			api_model = API_MODEL(**fixture_dict)
252			db_model = API_MODEL.to_model(api_model)
253			loaded_models[fixture] = db_model
254			all_fixtures[fixture_type] = loaded_models
255
256			return all_fixtures
257
258			def delete_fixtures_from_db(self, fixtures_pack='generic', fixtures_dict=None,
259			raise_on_fail=False):
260			"""
261			Deletes fixtures specified in fixtures_dict from the database.
262
263			fixtures_dict should be of the form:
264			{
265			'actions': ['action-1.yaml', 'action-2.yaml'],
266			'rules': ['rule-1.yaml'],
267			'liveactions': ['execution-1.yaml']
268			}
269
270			:param fixtures_pack: Name of the pack to delete fixtures from.
271			:type fixtures_pack: ``str``
272
273			:param fixtures_dict: Dictionary specifying the fixtures to delete for each type.
274			:type fixtures_dict: ``dict``
275
276			:param raise_on_fail: Optional If True, raises exception if delete fails on any fixture.
277			:type raise_on_fail: ``boolean``
278			"""
279			if not fixtures_dict:
280			return
281			fixtures_pack_path = self._validate_fixtures_pack(fixtures_pack)
282			self._validate_fixture_dict(fixtures_dict)
283
284			for fixture_type, fixtures in six.iteritems(fixtures_dict):
285			API_MODEL = FIXTURE_API_MODEL.get(fixture_type, None)
286			PERSISTENCE_MODEL = FIXTURE_PERSISTENCE_MODEL.get(fixture_type, None)
287			for fixture in fixtures:
288			fixture_dict = self.meta_loader.load(
289			self._get_fixture_file_path_abs(fixtures_pack_path, fixture_type, fixture))
290			# Note that when we have a reference mechanism consistent for
291			# every model, we can just do a get and delete the object. Until
292			# then, this model conversions are necessary.
293			api_model = API_MODEL(**fixture_dict)
294			db_model = API_MODEL.to_model(api_model)
295			try:
296			PERSISTENCE_MODEL.delete(db_model)
297			except:
298			if raise_on_fail:
299			raise
300
301			def delete_models_from_db(self, models_dict, raise_on_fail=False):
302			"""
303			Deletes models specified in models_dict from the database.
304
305			models_dict should be of the form:
306			{
307			'actions': [ACTION1, ACTION2],
308			'rules': [RULE1],
309			'liveactions': [EXECUTION]
310			}
311
312			:param fixtures_dict: Dictionary specifying the fixtures to delete for each type.
313			:type fixtures_dict: ``dict``.
314
315			:param raise_on_fail: Optional If True, raises exception if delete fails on any model.
316			:type raise_on_fail: ``boolean``
317			"""
318			for model_type, models in six.iteritems(models_dict):
319			PERSISTENCE_MODEL = FIXTURE_PERSISTENCE_MODEL.get(model_type, None)
320			for model in models:
321			try:
322			PERSISTENCE_MODEL.delete(model)
323			except:
324			if raise_on_fail:
325			raise
326
327			def _validate_fixtures_pack(self, fixtures_pack):
328			fixtures_pack_path = self._get_fixtures_pack_path(fixtures_pack)
329
330			if not self._is_fixture_pack_exists(fixtures_pack_path):
331			raise Exception('Fixtures pack not found ' +
332			'in fixtures path %s.' % get_fixtures_base_path())
333			return fixtures_pack_path
334
335			def _validate_fixture_dict(self, fixtures_dict, allowed=ALLOWED_FIXTURES):
336			fixture_types = fixtures_dict.keys()
337			for fixture_type in fixture_types:
338			if fixture_type not in allowed:
339			raise Exception('Disallowed fixture type: %s' % fixture_type)
340
341			def _is_fixture_pack_exists(self, fixtures_pack_path):
342			return os.path.exists(fixtures_pack_path)
343
344			def _get_fixture_file_path_abs(self, fixtures_pack_path, fixtures_type, fixture_name):
345			return os.path.join(fixtures_pack_path, fixtures_type, fixture_name)
346
347			def _get_fixtures_pack_path(self, fixtures_pack_name):
348			return os.path.join(get_fixtures_base_path(), fixtures_pack_name)
349
350			def get_fixture_file_path_abs(self, fixtures_pack, fixtures_type, fixture_name):
351			return os.path.join(get_fixtures_base_path(), fixtures_pack, fixtures_type, fixture_name)
352

StackStorm / st2

Push — master ( d1f0a7...3b2ece )

_get_fixture_file_path_abs() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like