get_fixtures_packs_base_path() - Code Metrics - Inspection of "Merge pull request #2795 from StackStorm/enhanceme..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( d23be3...e1e109 )

by Plexxi

created 2016-07-08 06:45 UTC

get_fixtures_packs_base_path() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	2
rs	10

# 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.rule_enforcement import RuleEnforcementAPI
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.rule_enforcement import RuleEnforcementDB
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.rule_enforcement import RuleEnforcement
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', 'enforcements']
ALLOWED_FIXTURES = copy.copy(ALLOWED_DB_FIXTURES)
ALLOWED_FIXTURES.extend(['actionchains', 'workflows'])

FIXTURE_DB_MODEL = {
    'actions': ActionDB,
    'aliases': ActionAliasDB,
    'actionstates': ActionExecutionStateDB,
    'apikeys': ApiKeyDB,
    'enforcements': RuleEnforcementDB,
    '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,
    'enforcements': RuleEnforcementAPI,
    '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,
    'enforcements': RuleEnforcement,
    '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_fixtures_packs_base_path():
    return os.path.join(os.path.dirname(__file__), 'fixtures/packs')


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:
                # Guard against copy and type and similar typos
                if fixture in loaded_fixtures:
                    msg = 'Fixture "%s" is specified twice, probably a typo.' % (fixture)
                    raise ValueError(msg)

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

StackStorm / st2

Push — develop ( d23be3...e1e109 )

get_fixtures_packs_base_path() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like