DownloadGitRepoActionTestCase.test_run_pack_download_multiple_packs() - 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

test_run_pack_download_multiple_packs() A

↳ Parent: DownloadGitRepoActionTestCase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
c	0
b	0
f	0
dl	0
loc	16
rs	9.4285

#!/usr/bin/env python

# 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 os
import mock
import shutil
import tempfile
import hashlib

from lockfile import LockFile
from lockfile import LockTimeout
from git.repo import Repo
from gitdb.exc import BadName
from st2common.services import packs as pack_service
from st2tests.base import BaseActionTestCase

import pack_mgmt.download
from pack_mgmt.download import DownloadGitRepoAction

PACK_INDEX = {
    "test": {
        "version": "0.4.0",
        "name": "test",
        "repo_url": "https://github.com/StackStorm-Exchange/stackstorm-test",
        "author": "st2-dev",
        "keywords": ["some", "search", "another", "terms"],
        "email": "[email protected]",
        "description": "st2 pack to test package management pipeline"
    },
    "test2": {
        "version": "0.5.0",
        "name": "test2",
        "repo_url": "https://github.com/StackStorm-Exchange/stackstorm-test2",
        "author": "stanley",
        "keywords": ["some", "special", "terms"],
        "email": "[email protected]",
        "description": "another st2 pack to test package management pipeline"
    },
    "test3": {
        "version": "0.5.0",
        "stackstorm_version": ">=1.6.0, <2.2.0",
        "name": "test3",
        "repo_url": "https://github.com/StackStorm-Exchange/stackstorm-test3",
        "author": "stanley",
        "keywords": ["some", "special", "terms"],
        "email": "[email protected]",
        "description": "another st2 pack to test package management pipeline"
    }
}


@mock.patch.object(pack_service, 'fetch_pack_index', mock.MagicMock(return_value=(PACK_INDEX, {})))
class DownloadGitRepoActionTestCase(BaseActionTestCase):
    action_cls = DownloadGitRepoAction

    def setUp(self):
        super(DownloadGitRepoActionTestCase, self).setUp()

        clone_from = mock.patch.object(Repo, 'clone_from')

        self.addCleanup(clone_from.stop)
        self.clone_from = clone_from.start()

        expand_user = mock.patch.object(os.path, 'expanduser',
                                        mock.MagicMock(return_value=tempfile.mkdtemp()))

        self.addCleanup(expand_user.stop)
        self.expand_user = expand_user.start()

        self.repo_base = tempfile.mkdtemp()

        self.repo_instance = mock.MagicMock()

        def side_effect(url, to_path, **kwargs):
            # Since we have no way to pass pack name here, we would have to derive it from repo url
            fixture_name = url.split('/')[-1]
            fixture_path = os.path.join(self._get_base_pack_path(), 'tests/fixtures', fixture_name)
            shutil.copytree(fixture_path, to_path)
            return self.repo_instance

        self.clone_from.side_effect = side_effect

    def tearDown(self):
        shutil.rmtree(self.repo_base)
        shutil.rmtree(self.expand_user())

    def test_run_pack_download(self):
        action = self.get_action_instance()
        result = action.run(packs=['test'], abs_repo_base=self.repo_base)
        temp_dir = hashlib.md5(PACK_INDEX['test']['repo_url']).hexdigest()

        self.assertEqual(result, {'test': 'Success.'})
        self.clone_from.assert_called_once_with(PACK_INDEX['test']['repo_url'],
                                                os.path.join(os.path.expanduser('~'), temp_dir))
        self.assertTrue(os.path.isfile(os.path.join(self.repo_base, 'test/pack.yaml')))

    def test_run_pack_download_existing_pack(self):
        action = self.get_action_instance()
        action.run(packs=['test'], abs_repo_base=self.repo_base)
        self.assertTrue(os.path.isfile(os.path.join(self.repo_base, 'test/pack.yaml')))

        result = action.run(packs=['test'], abs_repo_base=self.repo_base)

        self.assertEqual(result, {'test': 'Success.'})

    def test_run_pack_download_multiple_packs(self):
        action = self.get_action_instance()
        result = action.run(packs=['test', 'test2'], abs_repo_base=self.repo_base)
        temp_dirs = [
            hashlib.md5(PACK_INDEX['test']['repo_url']).hexdigest(),
            hashlib.md5(PACK_INDEX['test2']['repo_url']).hexdigest()
        ]

        self.assertEqual(result, {'test': 'Success.', 'test2': 'Success.'})
        self.clone_from.assert_any_call(PACK_INDEX['test']['repo_url'],
                                        os.path.join(os.path.expanduser('~'), temp_dirs[0]))
        self.clone_from.assert_any_call(PACK_INDEX['test2']['repo_url'],
                                        os.path.join(os.path.expanduser('~'), temp_dirs[1]))
        self.assertEqual(self.clone_from.call_count, 2)
        self.assertTrue(os.path.isfile(os.path.join(self.repo_base, 'test/pack.yaml')))
        self.assertTrue(os.path.isfile(os.path.join(self.repo_base, 'test2/pack.yaml')))

    @mock.patch.object(Repo, 'clone_from')
    def test_run_pack_download_error(self, clone_from):
        clone_from.side_effect = Exception('Something went terribly wrong during the clone')

        action = self.get_action_instance()
        self.assertRaises(Exception, action.run, packs=['test'], abs_repo_base=self.repo_base)

    def test_run_pack_download_no_tag(self):
        self.repo_instance.commit.side_effect = BadName

        action = self.get_action_instance()
        self.assertRaises(ValueError, action.run, packs=['test=1.2.3'],
                          abs_repo_base=self.repo_base)

    def test_run_pack_lock_is_already_acquired(self):
        action = self.get_action_instance()
        temp_dir = hashlib.md5(PACK_INDEX['test']['repo_url']).hexdigest()

        original_acquire = LockFile.acquire

        def mock_acquire(self, timeout=None):
            original_acquire(self, timeout=0.1)

        LockFile.acquire = mock_acquire

        try:
            lock_file = LockFile('/tmp/%s' % (temp_dir))

            # Acquire a lock (file) so acquire inside download will fail
            with open(lock_file.lock_file, 'w') as fp:
                fp.write('')

            expected_msg = 'Timeout waiting to acquire lock for'
            self.assertRaisesRegexp(LockTimeout, expected_msg, action.run, packs=['test'],
                                    abs_repo_base=self.repo_base)
        finally:
            os.unlink(lock_file.lock_file)
            LockFile.acquire = original_acquire

    def test_run_pack_lock_is_already_acquired_force_flag(self):
        # Lock is already acquired but force is true so it should be deleted and released
        action = self.get_action_instance()
        temp_dir = hashlib.md5(PACK_INDEX['test']['repo_url']).hexdigest()

        original_acquire = LockFile.acquire

        def mock_acquire(self, timeout=None):
            original_acquire(self, timeout=0.1)

        LockFile.acquire = mock_acquire

        try:
            lock_file = LockFile('/tmp/%s' % (temp_dir))

            # Acquire a lock (file) so acquire inside download will fail
            with open(lock_file.lock_file, 'w') as fp:
                fp.write('')

            result = action.run(packs=['test'], abs_repo_base=self.repo_base, force=True)
        finally:
            LockFile.acquire = original_acquire

        self.assertEqual(result, {'test': 'Success.'})

    def test_run_pack_download_v_tag(self):
        def side_effect(ref):
            if ref[0] != 'v':
                raise BadName()
            return mock.MagicMock(hexsha='abcdef')

        self.repo_instance.commit.side_effect = side_effect
        self.repo_instance.git = mock.MagicMock(
            branch=(lambda *args: 'master'),
            checkout=(lambda *args: True)
        )

        action = self.get_action_instance()
        result = action.run(packs=['test=1.2.3'], abs_repo_base=self.repo_base)

        self.assertEqual(result, {'test': 'Success.'})

    @mock.patch.object(DownloadGitRepoAction, '_get_valid_versions_for_repo',
                       mock.Mock(return_value=['1.0.0', '2.0.0']))
    def test_run_pack_download_invalid_version(self):
        self.repo_instance.commit.side_effect = lambda ref: None

        action = self.get_action_instance()

        expected_msg = ('is not a valid version, hash, tag or branch.*?'
                        'Available versions are: 1.0.0, 2.0.0.')
        self.assertRaisesRegexp(ValueError, expected_msg, action.run,
                                packs=['test=2.2.3'], abs_repo_base=self.repo_base)

    def test_download_pack_stackstorm_version_identifier_check(self):
        action = self.get_action_instance()

        # Version is satisfied
        pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '2.0.0'

        result = action.run(packs=['test3'], abs_repo_base=self.repo_base)
        self.assertEqual(result['test3'], 'Success.')

        # Pack requires a version which is not satisfied by current StackStorm version
        pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '2.2.0'
        expected_msg = ('Pack "test3" requires StackStorm ">=1.6.0, <2.2.0", but '
                        'current version is "2.2.0"')
        self.assertRaisesRegexp(ValueError, expected_msg, action.run, packs=['test3'],
                                abs_repo_base=self.repo_base)

        pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '2.3.0'
        expected_msg = ('Pack "test3" requires StackStorm ">=1.6.0, <2.2.0", but '
                        'current version is "2.3.0"')
        self.assertRaisesRegexp(ValueError, expected_msg, action.run, packs=['test3'],
                                abs_repo_base=self.repo_base)

        pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '1.5.9'
        expected_msg = ('Pack "test3" requires StackStorm ">=1.6.0, <2.2.0", but '
                        'current version is "1.5.9"')
        self.assertRaisesRegexp(ValueError, expected_msg, action.run, packs=['test3'],
                                abs_repo_base=self.repo_base)

        pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '1.5.0'
        expected_msg = ('Pack "test3" requires StackStorm ">=1.6.0, <2.2.0", but '
                        'current version is "1.5.0"')
        self.assertRaisesRegexp(ValueError, expected_msg, action.run, packs=['test3'],
                                abs_repo_base=self.repo_base)

        # Version is not met, but force=true parameter is provided
        pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '1.5.0'
        result = action.run(packs=['test3'], abs_repo_base=self.repo_base, force=True)
        self.assertEqual(result['test3'], 'Success.')

    def test_resolve_urls(self):
        url = DownloadGitRepoAction._eval_repo_url(
            "https://github.com/StackStorm-Exchange/stackstorm-test")
        self.assertEqual(url, "https://github.com/StackStorm-Exchange/stackstorm-test")

        url = DownloadGitRepoAction._eval_repo_url(
            "https://github.com/StackStorm-Exchange/stackstorm-test.git")
        self.assertEqual(url, "https://github.com/StackStorm-Exchange/stackstorm-test.git")

        url = DownloadGitRepoAction._eval_repo_url("StackStorm-Exchange/stackstorm-test")
        self.assertEqual(url, "https://github.com/StackStorm-Exchange/stackstorm-test")

        url = DownloadGitRepoAction._eval_repo_url("git://StackStorm-Exchange/stackstorm-test")
        self.assertEqual(url, "git://StackStorm-Exchange/stackstorm-test")

        url = DownloadGitRepoAction._eval_repo_url("git://StackStorm-Exchange/stackstorm-test.git")
        self.assertEqual(url, "git://StackStorm-Exchange/stackstorm-test.git")

        url = DownloadGitRepoAction._eval_repo_url("[email protected]:foo/bar.git")
        self.assertEqual(url, "[email protected]:foo/bar.git")

        url = DownloadGitRepoAction._eval_repo_url("file:///home/vagrant/stackstorm-test")
        self.assertEqual(url, "file:///home/vagrant/stackstorm-test")

        url = DownloadGitRepoAction._eval_repo_url('ssh://<user@host>/AutomationStackStorm')
        self.assertEqual(url, 'ssh://<user@host>/AutomationStackStorm')

        url = DownloadGitRepoAction._eval_repo_url('ssh://joe@local/AutomationStackStorm')
        self.assertEqual(url, 'ssh://joe@local/AutomationStackStorm')

    def test_run_pack_download_edge_cases(self):
        """
        Edge cases to test:

        default branch is master, ref is pack version
        default branch is master, ref is branch name
        default branch is master, ref is default branch name
        default branch is not master, ref is pack version
        default branch is not master, ref is branch name
        default branch is not master, ref is default branch name
        """

        def side_effect(ref):
            if ref[0] != 'v':
                raise BadName()
            return mock.MagicMock(hexsha='abcdeF')

        self.repo_instance.commit.side_effect = side_effect

        edge_cases = {
            'master': '1.2.3',
            'master': 'some-branch',
            'master': 'default-branch',
            'master': None,
            'default-branch': '1.2.3',
            'default-branch': 'some-branch',
            'default-branch': 'default-branch',
            'default-branch': None
        }

        for default_branch, ref in edge_cases.items():
            self.repo_instance.git = mock.MagicMock(
                branch=(lambda *args: default_branch),
                checkout=(lambda *args: True)
            )

            # Set default branch
            self.repo_instance.active_branch.name = default_branch
            self.repo_instance.active_branch.object = 'aBcdef'
            self.repo_instance.head.commit = 'aBcdef'

            # Fake gitref object
            gitref = mock.MagicMock(hexsha='abcDef')

            # Fool _get_gitref into working when its ref == our ref
            def fake_commit(arg_ref):
                if arg_ref == ref:
                    return gitref
                else:
                    raise BadName()
            self.repo_instance.commit = fake_commit
            self.repo_instance.active_branch.object = gitref

            action = self.get_action_instance()

            if ref:
                packs = ['test=%s' % (ref)]
            else:
                packs = ['test']

            result = action.run(packs=packs, abs_repo_base=self.repo_base)
            self.assertEqual(result, {'test': 'Success.'})


1			#!/usr/bin/env python
2
3			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
4			# contributor license agreements. See the NOTICE file distributed with
5			# this work for additional information regarding copyright ownership.
6			# The ASF licenses this file to You under the Apache License, Version 2.0
7			# (the "License"); you may not use this file except in compliance with
8			# the License. You may obtain a copy of the License at
9			#
10			# http://www.apache.org/licenses/LICENSE-2.0
11			#
12			# Unless required by applicable law or agreed to in writing, software
13			# distributed under the License is distributed on an "AS IS" BASIS,
14			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15			# See the License for the specific language governing permissions and
16			# limitations under the License.
17
18			import os
19			import mock
20			import shutil
21			import tempfile
22			import hashlib
23
24			from lockfile import LockFile
25			from lockfile import LockTimeout
26			from git.repo import Repo
27			from gitdb.exc import BadName
28			from st2common.services import packs as pack_service
29			from st2tests.base import BaseActionTestCase
30
31			import pack_mgmt.download
32			from pack_mgmt.download import DownloadGitRepoAction
33
34			PACK_INDEX = {
35			"test": {
36			"version": "0.4.0",
37			"name": "test",
38			"repo_url": "https://github.com/StackStorm-Exchange/stackstorm-test",
39			"author": "st2-dev",
40			"keywords": ["some", "search", "another", "terms"],
41			"email": "[email protected]",
42			"description": "st2 pack to test package management pipeline"
43			},
44			"test2": {
45			"version": "0.5.0",
46			"name": "test2",
47			"repo_url": "https://github.com/StackStorm-Exchange/stackstorm-test2",
48			"author": "stanley",
49			"keywords": ["some", "special", "terms"],
50			"email": "[email protected]",
51			"description": "another st2 pack to test package management pipeline"
52			},
53			"test3": {
54			"version": "0.5.0",
55			"stackstorm_version": ">=1.6.0, <2.2.0",
56			"name": "test3",
57			"repo_url": "https://github.com/StackStorm-Exchange/stackstorm-test3",
58			"author": "stanley",
59			"keywords": ["some", "special", "terms"],
60			"email": "[email protected]",
61			"description": "another st2 pack to test package management pipeline"
62			}
63			}
64
65
66			@mock.patch.object(pack_service, 'fetch_pack_index', mock.MagicMock(return_value=(PACK_INDEX, {})))
67			class DownloadGitRepoActionTestCase(BaseActionTestCase):
68			action_cls = DownloadGitRepoAction
69
70			def setUp(self):
71			super(DownloadGitRepoActionTestCase, self).setUp()
72
73			clone_from = mock.patch.object(Repo, 'clone_from')
74
75			self.addCleanup(clone_from.stop)
76			self.clone_from = clone_from.start()
77
78			expand_user = mock.patch.object(os.path, 'expanduser',
79			mock.MagicMock(return_value=tempfile.mkdtemp()))
80
81			self.addCleanup(expand_user.stop)
82			self.expand_user = expand_user.start()
83
84			self.repo_base = tempfile.mkdtemp()
85
86			self.repo_instance = mock.MagicMock()
87
88			def side_effect(url, to_path, **kwargs):
89			# Since we have no way to pass pack name here, we would have to derive it from repo url
90			fixture_name = url.split('/')[-1]
91			fixture_path = os.path.join(self._get_base_pack_path(), 'tests/fixtures', fixture_name)
92			shutil.copytree(fixture_path, to_path)
93			return self.repo_instance
94
95			self.clone_from.side_effect = side_effect
96
97			def tearDown(self):
98			shutil.rmtree(self.repo_base)
99			shutil.rmtree(self.expand_user())
100
101			def test_run_pack_download(self):
102			action = self.get_action_instance()
103			result = action.run(packs=['test'], abs_repo_base=self.repo_base)
104			temp_dir = hashlib.md5(PACK_INDEX['test']['repo_url']).hexdigest()
105
106			self.assertEqual(result, {'test': 'Success.'})
107			self.clone_from.assert_called_once_with(PACK_INDEX['test']['repo_url'],
108			os.path.join(os.path.expanduser('~'), temp_dir))
109			self.assertTrue(os.path.isfile(os.path.join(self.repo_base, 'test/pack.yaml')))
110
111			def test_run_pack_download_existing_pack(self):
112			action = self.get_action_instance()
113			action.run(packs=['test'], abs_repo_base=self.repo_base)
114			self.assertTrue(os.path.isfile(os.path.join(self.repo_base, 'test/pack.yaml')))
115
116			result = action.run(packs=['test'], abs_repo_base=self.repo_base)
117
118			self.assertEqual(result, {'test': 'Success.'})
119
120			def test_run_pack_download_multiple_packs(self):
121			action = self.get_action_instance()
122			result = action.run(packs=['test', 'test2'], abs_repo_base=self.repo_base)
123			temp_dirs = [
124			hashlib.md5(PACK_INDEX['test']['repo_url']).hexdigest(),
125			hashlib.md5(PACK_INDEX['test2']['repo_url']).hexdigest()
126			]
127
128			self.assertEqual(result, {'test': 'Success.', 'test2': 'Success.'})
129			self.clone_from.assert_any_call(PACK_INDEX['test']['repo_url'],
130			os.path.join(os.path.expanduser('~'), temp_dirs[0]))
131			self.clone_from.assert_any_call(PACK_INDEX['test2']['repo_url'],
132			os.path.join(os.path.expanduser('~'), temp_dirs[1]))
133			self.assertEqual(self.clone_from.call_count, 2)
134			self.assertTrue(os.path.isfile(os.path.join(self.repo_base, 'test/pack.yaml')))
135			self.assertTrue(os.path.isfile(os.path.join(self.repo_base, 'test2/pack.yaml')))
136
137			@mock.patch.object(Repo, 'clone_from')
138			def test_run_pack_download_error(self, clone_from):
139			clone_from.side_effect = Exception('Something went terribly wrong during the clone')
140
141			action = self.get_action_instance()
142			self.assertRaises(Exception, action.run, packs=['test'], abs_repo_base=self.repo_base)
143
144			def test_run_pack_download_no_tag(self):
145			self.repo_instance.commit.side_effect = BadName
146
147			action = self.get_action_instance()
148			self.assertRaises(ValueError, action.run, packs=['test=1.2.3'],
149			abs_repo_base=self.repo_base)
150
151			def test_run_pack_lock_is_already_acquired(self):
152			action = self.get_action_instance()
153			temp_dir = hashlib.md5(PACK_INDEX['test']['repo_url']).hexdigest()
154
155			original_acquire = LockFile.acquire
156
157			def mock_acquire(self, timeout=None):
158			original_acquire(self, timeout=0.1)
159
160			LockFile.acquire = mock_acquire
161
162			try:
163			lock_file = LockFile('/tmp/%s' % (temp_dir))
164
165			# Acquire a lock (file) so acquire inside download will fail
166			with open(lock_file.lock_file, 'w') as fp:
167			fp.write('')
168
169			expected_msg = 'Timeout waiting to acquire lock for'
170			self.assertRaisesRegexp(LockTimeout, expected_msg, action.run, packs=['test'],
171			abs_repo_base=self.repo_base)
172			finally:
173			os.unlink(lock_file.lock_file)
174			LockFile.acquire = original_acquire
175
176			def test_run_pack_lock_is_already_acquired_force_flag(self):
177			# Lock is already acquired but force is true so it should be deleted and released
178			action = self.get_action_instance()
179			temp_dir = hashlib.md5(PACK_INDEX['test']['repo_url']).hexdigest()
180
181			original_acquire = LockFile.acquire
182
183			def mock_acquire(self, timeout=None):
184			original_acquire(self, timeout=0.1)
185
186			LockFile.acquire = mock_acquire
187
188			try:
189			lock_file = LockFile('/tmp/%s' % (temp_dir))
190
191			# Acquire a lock (file) so acquire inside download will fail
192			with open(lock_file.lock_file, 'w') as fp:
193			fp.write('')
194
195			result = action.run(packs=['test'], abs_repo_base=self.repo_base, force=True)
196			finally:
197			LockFile.acquire = original_acquire
198
199			self.assertEqual(result, {'test': 'Success.'})
200
201			def test_run_pack_download_v_tag(self):
202			def side_effect(ref):
203			if ref[0] != 'v':
204			raise BadName()
205			return mock.MagicMock(hexsha='abcdef')
206
207			self.repo_instance.commit.side_effect = side_effect
208			self.repo_instance.git = mock.MagicMock(
209			branch=(lambda *args: 'master'),
210			checkout=(lambda *args: True)
211			)
212
213			action = self.get_action_instance()
214			result = action.run(packs=['test=1.2.3'], abs_repo_base=self.repo_base)
215
216			self.assertEqual(result, {'test': 'Success.'})
217
218			@mock.patch.object(DownloadGitRepoAction, '_get_valid_versions_for_repo',
219			mock.Mock(return_value=['1.0.0', '2.0.0']))
220			def test_run_pack_download_invalid_version(self):
221			self.repo_instance.commit.side_effect = lambda ref: None
222
223			action = self.get_action_instance()
224
225			expected_msg = ('is not a valid version, hash, tag or branch.*?'
226			'Available versions are: 1.0.0, 2.0.0.')
227			self.assertRaisesRegexp(ValueError, expected_msg, action.run,
228			packs=['test=2.2.3'], abs_repo_base=self.repo_base)
229
230			def test_download_pack_stackstorm_version_identifier_check(self):
231			action = self.get_action_instance()
232
233			# Version is satisfied
234			pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '2.0.0'
235
236			result = action.run(packs=['test3'], abs_repo_base=self.repo_base)
237			self.assertEqual(result['test3'], 'Success.')
238
239			# Pack requires a version which is not satisfied by current StackStorm version
240			pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '2.2.0'
241			expected_msg = ('Pack "test3" requires StackStorm ">=1.6.0, <2.2.0", but '
242			'current version is "2.2.0"')
243			self.assertRaisesRegexp(ValueError, expected_msg, action.run, packs=['test3'],
244			abs_repo_base=self.repo_base)
245
246			pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '2.3.0'
247			expected_msg = ('Pack "test3" requires StackStorm ">=1.6.0, <2.2.0", but '
248			'current version is "2.3.0"')
249			self.assertRaisesRegexp(ValueError, expected_msg, action.run, packs=['test3'],
250			abs_repo_base=self.repo_base)
251
252			pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '1.5.9'
253			expected_msg = ('Pack "test3" requires StackStorm ">=1.6.0, <2.2.0", but '
254			'current version is "1.5.9"')
255			self.assertRaisesRegexp(ValueError, expected_msg, action.run, packs=['test3'],
256			abs_repo_base=self.repo_base)
257
258			pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '1.5.0'
259			expected_msg = ('Pack "test3" requires StackStorm ">=1.6.0, <2.2.0", but '
260			'current version is "1.5.0"')
261			self.assertRaisesRegexp(ValueError, expected_msg, action.run, packs=['test3'],
262			abs_repo_base=self.repo_base)
263
264			# Version is not met, but force=true parameter is provided
265			pack_mgmt.download.CURRENT_STACKSTROM_VERSION = '1.5.0'
266			result = action.run(packs=['test3'], abs_repo_base=self.repo_base, force=True)
267			self.assertEqual(result['test3'], 'Success.')
268
269			def test_resolve_urls(self):
270			url = DownloadGitRepoAction._eval_repo_url(
271			"https://github.com/StackStorm-Exchange/stackstorm-test")
272			self.assertEqual(url, "https://github.com/StackStorm-Exchange/stackstorm-test")
273
274			url = DownloadGitRepoAction._eval_repo_url(
275			"https://github.com/StackStorm-Exchange/stackstorm-test.git")
276			self.assertEqual(url, "https://github.com/StackStorm-Exchange/stackstorm-test.git")
277
278			url = DownloadGitRepoAction._eval_repo_url("StackStorm-Exchange/stackstorm-test")
279			self.assertEqual(url, "https://github.com/StackStorm-Exchange/stackstorm-test")
280
281			url = DownloadGitRepoAction._eval_repo_url("git://StackStorm-Exchange/stackstorm-test")
282			self.assertEqual(url, "git://StackStorm-Exchange/stackstorm-test")
283
284			url = DownloadGitRepoAction._eval_repo_url("git://StackStorm-Exchange/stackstorm-test.git")
285			self.assertEqual(url, "git://StackStorm-Exchange/stackstorm-test.git")
286
287			url = DownloadGitRepoAction._eval_repo_url("[email protected]:foo/bar.git")
288			self.assertEqual(url, "[email protected]:foo/bar.git")
289
290			url = DownloadGitRepoAction._eval_repo_url("file:///home/vagrant/stackstorm-test")
291			self.assertEqual(url, "file:///home/vagrant/stackstorm-test")
292
293			url = DownloadGitRepoAction._eval_repo_url('ssh://<user@host>/AutomationStackStorm')
294			self.assertEqual(url, 'ssh://<user@host>/AutomationStackStorm')
295
296			url = DownloadGitRepoAction._eval_repo_url('ssh://joe@local/AutomationStackStorm')
297			self.assertEqual(url, 'ssh://joe@local/AutomationStackStorm')
298
299			def test_run_pack_download_edge_cases(self):
300			"""
301			Edge cases to test:
302
303			default branch is master, ref is pack version
304			default branch is master, ref is branch name
305			default branch is master, ref is default branch name
306			default branch is not master, ref is pack version
307			default branch is not master, ref is branch name
308			default branch is not master, ref is default branch name
309			"""
310
311			def side_effect(ref):
312			if ref[0] != 'v':
313			raise BadName()
314			return mock.MagicMock(hexsha='abcdeF')
315
316			self.repo_instance.commit.side_effect = side_effect
317
318			edge_cases = {
319			'master': '1.2.3',
320			'master': 'some-branch',
321			'master': 'default-branch',
322			'master': None,
323			'default-branch': '1.2.3',
324			'default-branch': 'some-branch',
325			'default-branch': 'default-branch',
326			'default-branch': None
327			}
328
329			for default_branch, ref in edge_cases.items():
330			self.repo_instance.git = mock.MagicMock(
331			branch=(lambda *args: default_branch),
332			checkout=(lambda *args: True)
333			)
334
335			# Set default branch
336			self.repo_instance.active_branch.name = default_branch
337			self.repo_instance.active_branch.object = 'aBcdef'
338			self.repo_instance.head.commit = 'aBcdef'
339
340			# Fake gitref object
341			gitref = mock.MagicMock(hexsha='abcDef')
342
343			# Fool _get_gitref into working when its ref == our ref
344			def fake_commit(arg_ref):
345			if arg_ref == ref:
346			return gitref
347			else:
348			raise BadName()
349			self.repo_instance.commit = fake_commit
350			self.repo_instance.active_branch.object = gitref
351
352			action = self.get_action_instance()
353
354			if ref:
355			packs = ['test=%s' % (ref)]
356			else:
357			packs = ['test']
358
359			result = action.run(packs=packs, abs_repo_base=self.repo_base)
360			self.assertEqual(result, {'test': 'Success.'})
361

Plexxi / st2

GitHub Access Token became invalid

Push — plexxi-v2.3.2 ( 5d46fe )

test_run_pack_download_multiple_packs() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like