airship
/
pegleg
Code Issues Proposed changes
pegleg/tests/unit/test_cli.py

519 lines
20 KiB
Python
Raw Blame History

# Copyright 2018 AT&T Intellectual Property. All other rights reserved.
#
# Licensed 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 shutil
from click.testing import CliRunner
from mock import ANY
import mock
import pytest
from pegleg import cli
from pegleg.engine import errorcodes
from pegleg.engine.util import git
from tests.unit import test_utils
from tests.unit.fixtures import temp_path
@pytest.mark.skipif(
not test_utils.is_connected(),
reason='git clone requires network connectivity.')
class BaseCLIActionTest(object):
"""Tests end-to-end flows for all Pegleg CLI actions, with minimal mocking.
General pattern should be to include exactly one test that uses a remote
repo URL and as many other tests that are required that use a local repo
path for runtime optimization.
All tests should validate that the ``exit_code`` from the CLI is 0 (for
positive tests).
"""
# TODO(felipemonteiro): Need tests that validate repository overrides. Also
# need to write tests that use a site-defintion.yaml with repositories key.
@classmethod
def setup_class(cls):
cls.runner = CliRunner()
# Pin so we know that airship-seaworthy is a valid site.
cls.site_name = "airship-seaworthy"
cls.site_type = "foundry"
cls.repo_rev = '6b183e148b9bb7ba6f75c98dd13451088255c60b'
cls.repo_name = "airship-treasuremap"
repo_url = "https://github.com/openstack/%s.git" % cls.repo_name
cls.treasuremap_path = git.git_handler(repo_url, ref=cls.repo_rev)
class TestSiteCLIOptions(BaseCLIActionTest):
"""Tests site-level CLI options."""
### clone_path tests ###
def test_list_sites_using_remote_repo_and_clone_path_option(
self, temp_path):
"""Validates clone_path (-p) option is working properly with site list
action when using remote repo. Verify that the repo was cloned in the
clone_path
"""
# Scenario:
#
# 1) List sites (should clone repo automatically to `clone_path`
# location if `clone_path` is set)
repo_url = 'https://github.com/openstack/%s@%s' % (self.repo_name,
self.repo_rev)
# Note that the -p option is used to specify the clone_folder
site_list = self.runner.invoke(
cli.site, ['-p', temp_path, '-r', repo_url, 'list'])
assert site_list.exit_code == 0
# Verify that the repo was cloned into the clone_path
assert os.path.exists(os.path.join(temp_path, self.repo_name))
assert git.is_repository(os.path.join(temp_path, self.repo_name))
def test_list_sites_using_local_repo_and_clone_path_option(
self, temp_path):
"""Validates clone_path (-p) option is working properly with site list
action when using a local repo. Verify that the clone_path has NO
effect when using a local repo
"""
# Scenario:
#
# 1) List sites (when using local repo there should be not cloning
# even if the clone_path is passed in)
repo_path = self.treasuremap_path
# Note that the -p option is used to specify the clone_folder
site_list = self.runner.invoke(
cli.site, ['-p', temp_path, '-r', repo_path, 'list'])
assert site_list.exit_code == 0
# Verify that passing in clone_path when using local repo has no effect
assert not os.path.exists(os.path.join(temp_path, self.repo_name))
class TestSiteCLIOptionsNegative(BaseCLIActionTest):
"""Negative Tests for site-level CLI options."""
### Negative clone_path tests ###
def test_list_sites_using_remote_repo_and_reuse_clone_path_option(
self, temp_path):
"""Validates clone_path (-p) option is working properly with site list
action when using remote repo. Verify that the same repo can't be
cloned in the same clone_path if it already exists
"""
# Scenario:
#
# 1) List sites (should clone repo automatically to `clone_path`
# location if `clone_path` is set)
repo_url = 'https://github.com/openstack/%s@%s' % (self.repo_name,
self.repo_rev)
# Note that the -p option is used to specify the clone_folder
site_list = self.runner.invoke(
cli.site, ['-p', temp_path, '-r', repo_url, 'list'])
assert git.is_repository(os.path.join(temp_path, self.repo_name))
# Run site list for a second time to validate that the repo can't be
# cloned twice in the same clone_path
site_list = self.runner.invoke(
cli.site, ['-p', temp_path, '-r', repo_url, 'list'])
assert site_list.exit_code == 1
msg = "The repository already exists in the given path. Either " \
"provide a new clone path or pass in the path of the local " \
"repository as the site repository (-r)."
assert msg in site_list.output
class TestSiteCliActions(BaseCLIActionTest):
"""Tests site-level CLI actions."""
### Collect tests ###
def _validate_collect_site_action(self, repo_path_or_url, save_location):
result = self.runner.invoke(cli.site, [
'-r', repo_path_or_url, 'collect', self.site_name, '-s',
save_location
])
collected_files = os.listdir(save_location)
assert result.exit_code == 0, result.output
assert len(collected_files) == 1
# Validates that site manifests collected from cloned repositories
# are written out to sensibly named files like airship-treasuremap.yaml
assert collected_files[0] == ("%s.yaml" % self.repo_name)
def test_collect_using_remote_repo_url(self, temp_path):
"""Validates collect action using a remote URL."""
# Scenario:
#
# 1) Create temporary save location
# 2) Collect into save location (should clone repo automatically)
# 3) Check that expected file name is there
repo_url = 'https://github.com/openstack/%s@%s' % (self.repo_name,
self.repo_rev)
self._validate_collect_site_action(repo_url, temp_path)
def test_collect_using_remote_repo_url_ending_with_dot_git(
self, temp_path):
"""Validates collect action using a remote URL ending in .git."""
# Scenario:
#
# 1) Create temporary save location
# 2) Collect into save location (should clone repo automatically)
# 3) Check that expected file name is there
repo_url = 'https://github.com/openstack/%s@%s.git' % (self.repo_name,
self.repo_rev)
self._validate_collect_site_action(repo_url, temp_path)
def test_collect_using_local_path(self, temp_path):
"""Validates collect action using a path to a local repo."""
# Scenario:
#
# 1) Create temporary save location
# 2) Collect into save location (should skip clone repo)
# 3) Check that expected file name is there
repo_path = self.treasuremap_path
self._validate_collect_site_action(repo_path, temp_path)
### Lint tests ###
def _test_lint_site_action(self, repo_path_or_url, exclude=True):
flag = '-x' if exclude else '-w'
lint_command = ['-r', repo_path_or_url, 'lint', self.site_name]
exclude_lint_command = [
flag, errorcodes.SCHEMA_STORAGE_POLICY_MISMATCH_FLAG, flag,
errorcodes.SECRET_NOT_ENCRYPTED_POLICY
]
with mock.patch('pegleg.engine.site.util.deckhand') as mock_deckhand:
mock_deckhand.deckhand_render.return_value = ([], [])
result = self.runner.invoke(cli.site,
lint_command + exclude_lint_command)
assert result.exit_code == 0, result.output
if exclude:
# A successful result (while setting lint checks to exclude) should
# output nothing.
assert not result.output
else:
assert result.output
def test_lint_site_using_remote_repo_url_with_exclude(self):
"""Validates site lint action using remote repo URL."""
# Scenario:
#
# 1) Mock out Deckhand render (so we can ignore P005 issues)
# 2) Lint site with exclude flags (should clone repo automatically)
repo_url = 'https://github.com/openstack/%s@%s' % (self.repo_name,
self.repo_rev)
self._test_lint_site_action(repo_url, exclude=True)
def test_lint_site_using_local_path_with_exclude(self):
"""Validates site lint action using local repo path."""
# Scenario:
#
# 1) Mock out Deckhand render (so we can ignore P005 issues)
# 2) Lint site with exclude flags (should skip clone repo)
repo_path = self.treasuremap_path
self._test_lint_site_action(repo_path, exclude=True)
def test_lint_site_using_local_path_with_warn(self):
"""Validates site lint action using local repo path."""
# Scenario:
#
# 1) Mock out Deckhand render (so we can ignore P005 issues)
# 2) Lint site with warn flags (should skip clone repo)
repo_path = self.treasuremap_path
self._test_lint_site_action(repo_path, exclude=False)
### List tests ###
def _validate_list_site_action(self, repo_path_or_url):
mock_output = mock.Mock()
result = self.runner.invoke(
cli.site, ['-r', repo_path_or_url, 'list', '-o', mock_output])
assert result.exit_code == 0, result.output
table_output = mock_output.write.mock_calls[0][1][0]
assert self.site_name in table_output
assert self.site_type in table_output
def test_list_sites_using_remote_repo_url(self):
"""Validates list action using remote repo URL."""
# Scenario:
#
# 1) List sites (should clone repo automatically)
repo_url = 'https://github.com/openstack/%s@%s' % (self.repo_name,
self.repo_rev)
self._validate_list_site_action(repo_url)
def test_list_sites_using_local_path(self):
"""Validates list action using local repo path."""
# Scenario:
#
# 1) List sites (should skip clone repo)
repo_path = self.treasuremap_path
self._validate_list_site_action(repo_path)
### Show tests ###
def _validate_site_show_action(self, repo_path_or_url):
mock_output = mock.Mock()
result = self.runner.invoke(cli.site, [
'-r', repo_path_or_url, 'show', self.site_name, '-o', mock_output
])
assert result.exit_code == 0, result.output
table_output = mock_output.write.mock_calls[0][1][0]
assert self.site_name in table_output
def test_show_site_using_remote_repo_url(self):
"""Validates show action using remote repo URL."""
# Scenario:
#
# 1) Show site (should clone repo automatically)
repo_url = 'https://github.com/openstack/%s@%s' % (self.repo_name,
self.repo_rev)
self._validate_site_show_action(repo_url)
def test_show_site_using_local_path(self):
"""Validates show action using local repo path."""
# Scenario:
#
# 1) Show site (should skip clone repo)
repo_path = self.treasuremap_path
self._validate_site_show_action(repo_path)
### Render tests ###
def _validate_render_site_action(self, repo_path_or_url):
render_command = ['-r', repo_path_or_url, 'render', self.site_name]
with mock.patch('pegleg.engine.site.yaml') as mock_yaml:
with mock.patch(
'pegleg.engine.site.util.deckhand') as mock_deckhand:
mock_deckhand.deckhand_render.return_value = ([], [])
result = self.runner.invoke(cli.site, render_command)
assert result.exit_code == 0
mock_yaml.dump_all.assert_called_once()
def test_render_site_using_remote_repo_url(self):
"""Validates render action using remote repo URL."""
# Scenario:
#
# 1) Mock out Deckhand render (so we can ignore P005 issues)
# 2) Render site (should clone repo automatically)
repo_url = 'https://github.com/openstack/%s@%s' % (self.repo_name,
self.repo_rev)
self._validate_render_site_action(repo_url)
def test_render_site_using_local_path(self):
"""Validates render action using local repo path."""
# Scenario:
#
# 1) Mock out Deckhand render (so we can ignore P005 issues)
# 2) Render site (should skip clone repo)
repo_path = self.treasuremap_path
self._validate_render_site_action(repo_path)
def test_upload_documents_shipyard_using_local_repo_path(self):
"""Validates ShipyardHelper is called with correct arguments."""
# Scenario:
#
# 1) Mock out ShipyardHelper
# 2) Check ShipyardHelper was called with correct arguments
repo_path = self.treasuremap_path
with mock.patch('pegleg.cli.ShipyardHelper') as mock_obj:
result = self.runner.invoke(cli.site,
['-r', repo_path, 'upload', self.site_name])
assert result.exit_code == 0
mock_obj.assert_called_once()
class TestRepoCliActions(BaseCLIActionTest):
"""Tests repo-level CLI actions."""
### Lint tests ###
def test_lint_repo_using_remote_repo_url_with_exclude(self):
"""Validates repo lint action using remote repo URL."""
# Scenario:
#
# 1) Mock out Deckhand render (so we can ignore P005 issues)
# 2) Lint repo with exclude flags (should clone repo automatically)
repo_url = 'https://github.com/openstack/%s@%s' % (self.repo_name,
self.repo_rev)
lint_command = ['-r', repo_url, 'lint']
exclude_lint_command = [
'-x', errorcodes.SCHEMA_STORAGE_POLICY_MISMATCH_FLAG, '-x',
errorcodes.SECRET_NOT_ENCRYPTED_POLICY
]
with mock.patch('pegleg.engine.site.util.deckhand') as mock_deckhand:
mock_deckhand.deckhand_render.return_value = ([], [])
result = self.runner.invoke(cli.repo,
lint_command + exclude_lint_command)
assert result.exit_code == 0, result.output
# A successful result (while setting lint checks to exclude) should
# output nothing.
assert not result.output
def test_lint_repo_using_local_path_with_exclude(self):
"""Validates repo lint action using local repo path."""
# Scenario:
#
# 1) Mock out Deckhand render (so we can ignore P005 issues)
# 2) Lint repo with exclude flags (should skip clone repo)
repo_path = self.treasuremap_path
lint_command = ['-r', repo_path, 'lint']
exclude_lint_command = [
'-x', errorcodes.SCHEMA_STORAGE_POLICY_MISMATCH_FLAG, '-x',
errorcodes.SECRET_NOT_ENCRYPTED_POLICY
]
with mock.patch('pegleg.engine.site.util.deckhand') as mock_deckhand:
mock_deckhand.deckhand_render.return_value = ([], [])
result = self.runner.invoke(cli.repo,
lint_command + exclude_lint_command)
assert result.exit_code == 0, result.output
# A successful result (while setting lint checks to exclude) should
# output nothing.
assert not result.output
class TestTypeCliActions(BaseCLIActionTest):
"""Tests type-level CLI actions."""
def setup(self):
self.expected_types = ['foundry']
def _assert_table_has_expected_sites(self, mock_output):
table_output = mock_output.write.mock_calls[0][1][0]
for expected_type in self.expected_types:
assert expected_type in table_output
def _validate_type_list_action(self, repo_path_or_url):
mock_output = mock.Mock()
result = self.runner.invoke(
cli.type, ['-r', repo_path_or_url, 'list', '-o', mock_output])
assert result.exit_code == 0, result.output
self._assert_table_has_expected_sites(mock_output)
def test_list_types_using_remote_repo_url(self):
"""Validates list types action using remote repo URL."""
# Scenario:
#
# 1) List types (should clone repo automatically)
repo_url = 'https://github.com/openstack/%s@%s' % (self.repo_name,
self.repo_rev)
self._validate_type_list_action(repo_url)
def test_list_types_using_local_repo_path(self):
"""Validates list types action using local repo path."""
# Scenario:
#
# 1) List types for local repo path
repo_path = self.treasuremap_path
self._validate_type_list_action(repo_path)
class TestSiteCliActionsWithSubdirectory(BaseCLIActionTest):
"""Tests site CLI actions with subdirectories in repository paths."""
def setup(self):
self.expected_sites = ['demo', 'gate-multinode', 'dev', 'dev-proxy']
def _assert_table_has_expected_sites(self, mock_output):
table_output = mock_output.write.mock_calls[0][1][0]
for expected_site in self.expected_sites:
assert expected_site in table_output
def _validate_list_site_action(self, repo_path_or_url):
mock_output = mock.Mock()
result = self.runner.invoke(
cli.site, ['-r', repo_path_or_url, 'list', '-o', mock_output])
assert result.exit_code == 0, result.output
self._assert_table_has_expected_sites(mock_output)
def test_site_action_with_subpath_in_remote_url(self):
"""Validates list action with subpath in remote URL."""
# Scenario:
#
# 1) List sites for https://github.com/airship-in-a-bottle/
# deployment_files (subpath in remote URL)
# Perform site action using remote URL.
repo_name = 'airship-in-a-bottle'
repo_rev = '7a0717adc68261c7adb3a3db74a9326d6103519f'
repo_url = 'https://github.com/openstack/%s/deployment_files@%s' % (
repo_name, repo_rev)
self._validate_list_site_action(repo_url)
def test_site_action_with_subpath_in_local_repo_path(self):
"""Validates list action with subpath in local repo path."""
# Scenario:
#
# 1) List sites for local repo at /tmp/.../airship-in-a-bottle/
# deployment_files
# Perform site action using local repo path.
repo_name = 'airship-in-a-bottle'
repo_rev = '7a0717adc68261c7adb3a3db74a9326d6103519f'
repo_url = 'https://github.com/openstack/%s' % repo_name
_repo_path = git.git_handler(repo_url, ref=repo_rev)
repo_path = os.path.join(_repo_path, 'deployment_files')
self._validate_list_site_action(repo_path)
View Git Blame Copy Permalink