airship
/
pegleg
Code Issues Proposed changes

Restructure usage of test fixtures 

Pytest includes a fixture that can be used to generate temporary
directories. Previously Pegleg had implemented a hombrewed version of a
temporary directory fixture. This change removes the homebrewed version
and replaces it with the tmpdir fixture.

Implement tmpdir fixture in tests

Upgrade all testing packages to use the latest features

Removes unused imports and organizes import lists

Removes mock package requirement and uses unittest.mock, included in
python >3.3

Implements a slightly cleaner method to get proxy info

Change-Id: If66e1cfba858d5fb8948529deb8fb2d32345f630
This commit is contained in:
Ian H. Pittwood 2019-07-22 15:33:12 -05:00 committed by Ian H Pittwood
parent 1c8d92ef6b
commit 4480ab5574
31 changed files with 160 additions and 246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pegleg/engine/catalog/pki_generator.py
View File

@ -41,7 +41,6 @@ class PKIGenerator(object):
``<site>/secrets/<subpath>``.
"""
def __init__(
self, sitename, block_strings=True, author=None, duration=365):
"""Constructor for ``PKIGenerator``.

1
pegleg/engine/catalog/pki_utility.py
View File

@ -47,7 +47,6 @@ class PKIUtility(object):
of ``pegleg/PeglegManagedDocument/v1``.
"""
@staticmethod
def cfssl_exists():
"""Checks whether cfssl command exists. Useful for testing."""

1
pegleg/engine/catalogs/base_catalog.py
View File

@ -28,7 +28,6 @@ __all__ = ['BaseCatalog']
class BaseCatalog(ABC):
"""Abstract Base Class for all site catalogs."""
def __init__(self, kind, sitename, documents=None):
"""
Search for site catalog of the specified ``kind`` among the site

1
pegleg/engine/catalogs/passphrase_catalog.py
View File

@ -39,7 +39,6 @@ class PassphraseCatalog(BaseCatalog):
passphrase catalog documents.
"""
def __init__(self, sitename, documents=None):
"""
Parse the site passphrase catalog documents and capture the

1
pegleg/engine/generators/base_generator.py
View File

@ -28,7 +28,6 @@ class BaseGenerator(ABC):
Abstract Base Class, providing the common data and methods for all
generator classes
"""
def __init__(self, sitename, save_location, author=None):
"""Constructor for ``BaseGenerator``.

1
pegleg/engine/generators/passphrase_generator.py
View File

@ -35,7 +35,6 @@ class PassphraseGenerator(BaseGenerator):
Generates passphrases for a given environment, specified in a
passphrase catalog.
"""
def __init__(self, sitename, save_location, author):
"""Constructor for ``PassphraseGenerator``.

1
pegleg/engine/util/pegleg_managed_document.py
View File

@ -30,7 +30,6 @@ __all__ = ['PeglegManagedSecretsDocument']
class PeglegManagedSecretsDocument(object):
"""Object representing one Pegleg managed secret document."""
def __init__(self, document, generated=False, catalog=None, author=None):
"""
Parse and wrap an externally generated document in a

1
pegleg/engine/util/pegleg_secret_management.py
View File

@ -29,7 +29,6 @@ LOG = logging.getLogger(__name__)
class PeglegSecretManagement(object):
"""An object to handle operations on of a pegleg managed file."""
def __init__(
self,
file_path=None,

3
pegleg/engine/util/shipyard_helper.py
View File

@ -30,14 +30,12 @@ LOG = logging.getLogger(__name__)
class AuthValuesError(exceptions.PeglegBaseException):
"""Shipyard authentication failed. """
def __init__(self, *, diagnostic):
self.diagnostic = diagnostic
class DocumentUploadError(exceptions.PeglegBaseException):
"""Exception occurs while uploading documents"""
def __init__(self, message):
self.message = message
@ -50,7 +48,6 @@ class ShipyardHelper(object):
3. Commits the document
4. Formats response from Shipyard api_client
"""
def __init__(self, context, buffer_mode='replace'):
"""
Initializes params to be used by Shipyard

19
test-requirements.txt
View File

@ -1,17 +1,16 @@
# Testing
pytest==3.2.1
pytest-cov==2.5.1
testfixtures==6.8.2
pytest-xdist==1.23.2
mock==2.0.0
pytest>=5.0.1
pytest-cov>=2.5.1
testfixtures>=6.8.2
pytest-xdist>=1.23.2
# Formatting
yapf==0.27.0
yapf>=0.27.0
# Linting
hacking==1.1.0
flake8-import-order==0.18.1
hacking>=1.1.0
flake8-import-order>=0.18.1
# Security
bandit==1.6.0
safety==1.8.5
bandit>=1.6.0
safety>=1.8.5

30
tests/unit/fixtures.py → tests/conftest.py
View File

@ -15,8 +15,6 @@
from __future__ import absolute_import
import copy
import os
import shutil
import tempfile
import pytest
import yaml
@ -39,6 +37,18 @@ data: %(name)s-password
"""
@pytest.fixture(autouse=True)
def restore_config():
"""Used for ensuring the original global context is reset in memory
following each test execution.
"""
original_global_context = copy.deepcopy(config.GLOBAL_CONTEXT)
try:
yield
finally:
config.GLOBAL_CONTEXT = original_global_context
def _gen_document(**kwargs):
if "storagePolicy" not in kwargs:
kwargs["storagePolicy"] = "cleartext"
@ -46,8 +56,8 @@ def _gen_document(**kwargs):
return yaml.safe_load(test_document)
@pytest.fixture()
def create_tmp_deployment_files(tmpdir):
@pytest.fixture
def temp_deployment_files(tmpdir):
"""Fixture that creates a temporary directory structure."""
sitenames = ['cicd', 'lab']
@ -154,14 +164,4 @@ schema: pegleg/SiteDefinition/v1
cicd_path = os.path.join(str(p), files._site_path(site))
files._create_tree(cicd_path, tree=test_structure)
yield tmpdir
@pytest.fixture()
def temp_path():
temp_folder = tempfile.mkdtemp()
try:
yield temp_folder
finally:
if os.path.exists(temp_folder):
shutil.rmtree(temp_folder, ignore_errors=True)
return tmpdir

59
tests/unit/conftest.py
View File

@ -1,59 +0,0 @@
# 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 atexit
import copy
import os
import shutil
import tempfile
import pytest
from pegleg import config
"""Fixtures that are applied to all unit tests."""
@pytest.fixture(autouse=True)
def restore_config():
"""Used for ensuring the original global context is reset in memory
following each test execution.
"""
original_global_context = copy.deepcopy(config.GLOBAL_CONTEXT)
try:
yield
finally:
config.GLOBAL_CONTEXT = original_global_context
# NOTE(felipemonteiro): This uses `atexit` rather than a `pytest.fixture`
# decorator because 1) this only needs to be run exactly once and 2) this
# works across multiple test executors via `pytest -n <num_cores>`
@atexit.register
def clean_temporary_git_repos():
"""Iterates through all temporarily created directories and deletes each
one that was created for testing.
"""
def temporary_git_repos():
root_tempdir = tempfile.gettempdir()
tempdirs = os.listdir(root_tempdir)
for tempdir in tempdirs:
path = os.path.join(root_tempdir, tempdir)
if os.path.isdir(path) and os.access(path, os.R_OK):
if any(p.startswith('airship') for p in os.listdir(path)):
yield path
for tempdir in temporary_git_repos():
shutil.rmtree(tempdir, ignore_errors=True)

4
tests/unit/engine/catalog/test_pki_generator.py
View File

@ -16,9 +16,8 @@ import copy
import os
import shutil
import textwrap
from unittest import mock
import click
import mock
import pytest
import yaml
@ -173,7 +172,6 @@ def create_tmp_pki_structure(tmpdir):
structure with pki/ subfolder.
"""
def _create_tmp_folder_system(sitename, pki_catalog):
"""Creates a temporary site folder system.

3
tests/unit/engine/catalog/test_pki_utility.py
View File

@ -14,9 +14,8 @@
import json
import time
from unittest import mock
import click
import mock
import pytest
from pegleg import config

16
tests/unit/engine/test_build_genesis_bundle.py
View File

@ -14,8 +14,8 @@
import logging
import os
from unittest import mock
import mock
import pytest
import yaml
@ -25,8 +25,6 @@ from pegleg.engine.exceptions import GenesisBundleEncryptionException
from pegleg.engine.exceptions import GenesisBundleGenerateException
from pegleg.engine.util import files
from tests.unit.fixtures import temp_path
SITE_DEFINITION = """
---
# High-level pegleg site definition file
@ -92,15 +90,15 @@ data: ABAgagajajkb839215387
'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC',
'PEGLEG_SALT': 'MySecretSalt1234567890]['
})
def test_no_encryption_key(temp_path):
def test_no_encryption_key(tmpdir):
# Write the test data to temp file
config_data = list(yaml.safe_load_all(SITE_CONFIG_DATA))
base_config_dir = os.path.join(temp_path, 'config_dir')
base_config_dir = os.path.join(tmpdir, 'config_dir')
config.set_site_repo(base_config_dir)
config_dir = os.path.join(base_config_dir, 'site', 'test_site')
config_path = os.path.join(config_dir, 'config_file.yaml')
build_dir = os.path.join(temp_path, 'build_dir')
build_dir = os.path.join(tmpdir, 'build_dir')
os.makedirs(config_dir)
files.write(config_data, config_path)
@ -123,15 +121,15 @@ def test_no_encryption_key(temp_path):
'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC',
'PEGLEG_SALT': 'MySecretSalt1234567890]['
})
def test_failed_deckhand_validation(temp_path):
def test_failed_deckhand_validation(tmpdir):
# Write the test data to temp file
config_data = list(yaml.safe_load_all(SITE_CONFIG_DATA))
base_config_dir = os.path.join(temp_path, 'config_dir')
base_config_dir = os.path.join(tmpdir, 'config_dir')
config.set_site_repo(base_config_dir)
config_dir = os.path.join(base_config_dir, 'site', 'test_site')
config_path = os.path.join(config_dir, 'config_file.yaml')
build_dir = os.path.join(temp_path, 'build_dir')
build_dir = os.path.join(tmpdir, 'build_dir')
os.makedirs(config_dir)
files.write(config_data, config_path)
files.write(

2
tests/unit/engine/test_generate_passphrases.py
View File

@ -15,10 +15,10 @@
import base64
import os
import tempfile
from unittest import mock
import uuid
from cryptography import fernet
import mock
import pytest
from testfixtures import log_capture
import yaml

5
tests/unit/engine/test_lint.py
View File

@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import mock
from unittest import mock
from pegleg.engine import lint
from pegleg.engine.errorcodes import DECKHAND_DUPLICATE_SCHEMA
@ -21,11 +21,10 @@ from pegleg.engine.util import deckhand
from pegleg.engine.util import files
from pegleg.engine.util.pegleg_managed_document \
import PeglegManagedSecretsDocument
from tests.unit.fixtures import create_tmp_deployment_files
def test_verify_deckhand_render_site_documents_separately(
create_tmp_deployment_files):
temp_deployment_files):
expected_documents = {
'cicd': [
'global-common', 'global-v1.0', 'cicd-type-common',

35
tests/unit/engine/test_secrets.py
View File

@ -14,11 +14,9 @@
import os
from os import listdir
from unittest import mock
import click
import mock
import pytest
import tempfile
import yaml
from pegleg import config
@ -26,16 +24,13 @@ from pegleg.engine.catalog.pki_generator import PKIGenerator
from pegleg.engine.catalog import pki_utility
from pegleg.engine import exceptions
from pegleg.engine import secrets
from pegleg.engine.util import encryption as crypt, catalog, git
from pegleg.engine.util import encryption as crypt, git
from pegleg.engine.util import files
from pegleg.engine.util.pegleg_managed_document import \
PeglegManagedSecretsDocument
from pegleg.engine.util.pegleg_secret_management import PeglegSecretManagement
from tests.unit import test_utils
from tests.unit.fixtures import temp_path, create_tmp_deployment_files, \
_gen_document
from tests.unit.test_cli import TestSiteSecretsActions, BaseCLIActionTest, \
TEST_PARAMS
from tests.unit.test_cli import TEST_PARAMS
TEST_DATA = """
---
@ -150,7 +145,7 @@ def test_short_salt():
'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC',
'PEGLEG_SALT': 'MySecretSalt1234567890]['
})
def test_secret_encrypt_and_decrypt(create_tmp_deployment_files, tmpdir):
def test_secret_encrypt_and_decrypt(temp_deployment_files, tmpdir):
site_dir = tmpdir.join("deployment_files", "site", "cicd")
passphrase_doc = """---
schema: deckhand/Passphrase/v1
@ -250,12 +245,12 @@ def test_pegleg_secret_management_double_encrypt():
'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC',
'PEGLEG_SALT': 'MySecretSalt1234567890]['
})
def test_encrypt_decrypt_using_file_path(temp_path):
def test_encrypt_decrypt_using_file_path(tmpdir):
# write the test data to temp file
test_data = list(yaml.safe_load_all(TEST_DATA))
file_path = os.path.join(temp_path, 'secrets_file.yaml')
file_path = os.path.join(tmpdir, 'secrets_file.yaml')
files.write(test_data, file_path)
save_path = os.path.join(temp_path, 'encrypted_secrets_file.yaml')
save_path = os.path.join(tmpdir, 'encrypted_secrets_file.yaml')
# encrypt documents and validate that they were encrypted
doc_mgr = PeglegSecretManagement(file_path=file_path, author='test_author')
@ -279,10 +274,10 @@ def test_encrypt_decrypt_using_file_path(temp_path):
'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC',
'PEGLEG_SALT': 'MySecretSalt1234567890]['
})
def test_encrypt_decrypt_using_docs(temp_path):
def test_encrypt_decrypt_using_docs(tmpdir):
# write the test data to temp file
test_data = list(yaml.safe_load_all(TEST_DATA))
save_path = os.path.join(temp_path, 'encrypted_secrets_file.yaml')
save_path = os.path.join(tmpdir, 'encrypted_secrets_file.yaml')
# encrypt documents and validate that they were encrypted
doc_mgr = PeglegSecretManagement(docs=test_data, author='test_author')
@ -314,7 +309,7 @@ def test_encrypt_decrypt_using_docs(temp_path):
'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC',
'PEGLEG_SALT': 'MySecretSalt1234567890]['
})
def test_generate_pki_using_local_repo_path(create_tmp_deployment_files):
def test_generate_pki_using_local_repo_path(temp_deployment_files):
"""Validates ``generate-pki`` action using local repo path."""
# Scenario:
#
@ -342,7 +337,7 @@ def test_generate_pki_using_local_repo_path(create_tmp_deployment_files):
'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC',
'PEGLEG_SALT': 'MySecretSalt1234567890]['
})
def test_check_expiry(create_tmp_deployment_files):
def test_check_expiry(temp_deployment_files):
""" Validates check_expiry """
repo_path = str(
git.git_handler(TEST_PARAMS["repo_url"], ref=TEST_PARAMS["repo_rev"]))
@ -376,7 +371,7 @@ def test_check_expiry(create_tmp_deployment_files):
'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC',
'PEGLEG_SALT': 'MySecretSalt1234567890]['
})
def test_get_global_creds_missing_creds(create_tmp_deployment_files, tmpdir):
def test_get_global_creds_missing_creds(temp_deployment_files, tmpdir):
# Create site files
site_dir = tmpdir.join("deployment_files", "site", "cicd")
@ -395,7 +390,7 @@ def test_get_global_creds_missing_creds(create_tmp_deployment_files, tmpdir):
'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC',
'PEGLEG_SALT': 'MySecretSalt1234567890]['
})
def test_get_global_creds_missing_pass(create_tmp_deployment_files, tmpdir):
def test_get_global_creds_missing_pass(temp_deployment_files, tmpdir):
# Create site files
site_dir = tmpdir.join("deployment_files", "site", "cicd")
@ -419,7 +414,7 @@ def test_get_global_creds_missing_pass(create_tmp_deployment_files, tmpdir):
'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC',
'PEGLEG_SALT': 'MySecretSalt1234567890]['
})
def test_get_global_creds(create_tmp_deployment_files, tmpdir):
def test_get_global_creds(temp_deployment_files, tmpdir):
# Create site files
site_dir = tmpdir.join("deployment_files", "site", "cicd")
@ -449,7 +444,7 @@ def test_get_global_creds(create_tmp_deployment_files, tmpdir):
'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC',
'PEGLEG_SALT': 'MySecretSalt1234567890]['
})
def test_global_encrypt_decrypt(create_tmp_deployment_files, tmpdir):
def test_global_encrypt_decrypt(temp_deployment_files, tmpdir):
# Create site files
site_dir = tmpdir.join("deployment_files", "site", "cicd")

9
tests/unit/engine/test_selectable_linting.py
View File

@ -13,18 +13,16 @@
# limitations under the License.
import os
from unittest import mock
import click
import mock
import pytest
from deckhand.engine import layering
from deckhand import errors as dh_errors
from pegleg import config
from pegleg.engine import errorcodes
from pegleg.engine import lint
from tests.unit.fixtures import create_tmp_deployment_files
_SKIP_P003_REASON = """Currently broken with revisioned repositories
directory layout changes. The old pseudo-revision folders like 'v4.0' are
@ -224,12 +222,11 @@ class TestSelectableLinting(object):
class TestSelectableLintingHelperFunctions(object):
"""The fixture ``create_tmp_deployment_files`` produces many linting errors
"""The fixture ``temp_deployment_files`` produces many linting errors
by default.
"""
def test_verify_file_contents(self, create_tmp_deployment_files):
def test_verify_file_contents(self, temp_deployment_files):
"""Validate that linting by a specific site ("cicd") produces a subset
of all the linting errors produced for all sites.

7
tests/unit/engine/test_site_collect.py
View File

@ -12,16 +12,15 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import mock
import os
import shutil
import yaml
from unittest import mock
import click
from pegleg.engine import site
from pegleg.engine.util import deckhand
from tests.unit.fixtures import create_tmp_deployment_files
def _site_definition(site_name):
@ -85,7 +84,7 @@ def _test_site_collect_to_file(tmpdir, site_name, collection_path):
shutil.rmtree(collection_str_path, ignore_errors=True)
def test_site_collect_to_file(tmpdir, create_tmp_deployment_files):
def test_site_collect_to_file(tmpdir, temp_deployment_files):
_test_site_collect_to_file(tmpdir, "cicd", "cicd_path")
_test_site_collect_to_file(tmpdir, "lab", "lab_path")
@ -104,7 +103,7 @@ def _test_site_collect_to_stdout(site_name):
assert 'name: {}'.format(expected) in all_lines
def test_site_collect_to_stdout(create_tmp_deployment_files):
def test_site_collect_to_stdout(temp_deployment_files):
_test_site_collect_to_stdout("cicd")
_test_site_collect_to_stdout("lab")

3
tests/unit/engine/test_site_repository.py
View File

@ -12,8 +12,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import mock
import os
from unittest import mock
import click
import pytest
@ -120,7 +120,6 @@ def _test_process_repositories(
All params above are mutually exclusive. Can only test one at a time.
"""
@mock.patch.object(
util.definition,
'load_as_params',

5
tests/unit/engine/test_util_files.py
View File

@ -12,11 +12,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import mock
from unittest import mock
from pegleg import config
from pegleg.engine.util import files
from tests.unit.fixtures import create_tmp_deployment_files
TEST_DATA = [('/tmp/test_repo', 'test_file.yaml')]
TEST_DATA_2 = [{'schema': 'pegleg/SiteDefinition/v1', 'data': 'test'}]
@ -36,7 +35,7 @@ def test_no_non_yamls(tmpdir):
assert i.endswith('.yaml')
def test_list_all_files(create_tmp_deployment_files):
def test_list_all_files(temp_deployment_files):
expected_files = sorted(
[
'deployment_files/global/common/global-common.yaml',

3
tests/unit/engine/util/test_cryptostring.py
View File

@ -13,7 +13,6 @@
# limitations under the License.
from pegleg.engine.util.cryptostring import CryptoString
import string
def test_cryptostring_default_len():
@ -74,7 +73,7 @@ def test_cryptostring_has_symbol():
s_util = CryptoString()
crypto_string = 'Th1sP@sswordH4sSymbols!'
assert s_util.has_symbol(crypto_string) is True
crypto_string = '!@#$%^&*()[]\}{|<>?,./~`'
crypto_string = r'!@#$%^&*()[]\}{|<>?,./~`'
assert s_util.has_symbol(crypto_string) is True
crypto_string = 'ThisPasswordH4sNoSymbols'
assert s_util.has_symbol(crypto_string) is False

6
tests/unit/engine/util/test_definition.py
View File

@ -11,9 +11,9 @@
# 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.
from pytest import mark
from pegleg.engine.util import definition
from tests.unit.fixtures import create_tmp_deployment_files
class TestSiteDefinitionHelpers(object):
@ -43,11 +43,11 @@ class TestSiteDefinitionHelpers(object):
return global_documents + site_documents
def test_documents_for_site(self, create_tmp_deployment_files):
def test_documents_for_site(self, temp_deployment_files):
self._test_documents_for_site("cicd")
self._test_documents_for_site("lab")
def test_documents_for_each_site(self, create_tmp_deployment_files):
def test_documents_for_each_site(self, temp_deployment_files):
documents_by_site = definition.documents_for_each_site()
sort_func = lambda x: x['metadata']['name']

22
tests/unit/engine/util/test_files.py
View File

@ -19,22 +19,20 @@ import yaml
from pegleg import config
from pegleg.engine.util import files
from tests.unit.fixtures import create_tmp_deployment_files
from tests.unit.fixtures import temp_path
EXPECTED_FILE_PERM = '0o640'
EXPECTED_DIR_PERM = '0o750'
class TestFileHelpers(object):
def test_read_compatible_file(self, create_tmp_deployment_files):
def test_read_compatible_file(self, temp_deployment_files):
path = os.path.join(
config.get_site_repo(), 'site', 'cicd', 'secrets', 'passphrases',
'cicd-passphrase.yaml')
documents = files.read(path)
assert 1 == len(documents)
def test_read_incompatible_file(self, create_tmp_deployment_files):
def test_read_incompatible_file(self, temp_deployment_files):
# NOTE(felipemonteiro): The Pegleg site-definition.yaml is a
# Deckhand-formatted document currently but probably shouldn't be,
# because it has no business being in Deckhand. As such, validate that
@ -46,7 +44,7 @@ class TestFileHelpers(object):
"Documents returned should be empty for "
"site-definition.yaml")
def test_write(self, create_tmp_deployment_files):
def test_write(self, temp_deployment_files):
path = os.path.join(
config.get_site_repo(), 'site', 'cicd', 'test_out.yaml')
files.write("test text", path)
@ -64,13 +62,13 @@ class TestFileHelpers(object):
with pytest.raises(ValueError) as _:
files.write(object(), path)
def test_file_permissions(self, create_tmp_deployment_files):
def test_file_permissions(self, temp_deployment_files):
path = os.path.join(
config.get_site_repo(), 'site', 'cicd', 'test_out.yaml')
files.write("test text", path)
assert oct(os.stat(path).st_mode & 0o777) == EXPECTED_FILE_PERM
def test_dir_permissions(self, create_tmp_deployment_files):
def test_dir_permissions(self, temp_deployment_files):
path = os.path.join(config.get_site_repo(), 'site', 'cicd', 'test_dir')
os.makedirs(path)
assert oct(os.stat(path).st_mode & 0o777) == EXPECTED_DIR_PERM
@ -84,16 +82,16 @@ def test_file_in_subdir():
assert not files.file_in_subdir("aaa/bbb/../ccc.txt", "bbb")
def test_read(temp_path):
def test_read(tmpdir):
# This will throw an error if yaml attempts to read the tag.
with open(os.path.join(temp_path, "invalid.yaml"), "w") as invalid_yaml:
with open(os.path.join(tmpdir, "invalid.yaml"), "w") as invalid_yaml:
invalid_yaml.write("!!python/name:fake_class''\n")
files.read(os.path.join(temp_path, "invalid.yaml"))
files.read(os.path.join(tmpdir, "invalid.yaml"))
# Under PyYAML's default behavior, the tag !!python/name:builtins.int
# will be parsed into the method int. files.read should ignore this tag.
with open(os.path.join(temp_path, "valid.yaml"), "w") as valid_yaml:
with open(os.path.join(tmpdir, "valid.yaml"), "w") as valid_yaml:
valid_yaml.write("!!python/name:builtins.int ''\n")
read_files = files.read(os.path.join(temp_path, "valid.yaml"))
read_files = files.read(os.path.join(tmpdir, "valid.yaml"))
# Assert that the tag was not parsed into the method int
assert int not in read_files

12
tests/unit/engine/util/test_git.py
View File

@ -14,14 +14,13 @@
import os
import shutil
from unittest import mock
from git import Repo
import mock
import pytest
from pegleg.engine import exceptions
from pegleg.engine.util import git
from tests.unit.fixtures import temp_path
from tests.unit import test_utils
@ -99,14 +98,15 @@ def test_git_clone_with_patch_ref():
@pytest.mark.skipif(
not test_utils.is_connected_behind_proxy(),
not test_utils.get_proxies()[0]
or not test_utils.is_connected_behind_proxy(),
reason='git clone requires proxy connectivity.')
@mock.patch.object(git, 'LOG', autospec=True)
def test_git_clone_behind_proxy(mock_log):
url = 'https://review.opendev.org/airship/armada'
commit = 'cba78d1d03e4910f6ab1691bae633c5bddce893d'
for proxy_server in test_utils._PROXY_SERVERS.values():
for proxy_server in test_utils.get_proxies()[1].values():
git_dir = git.git_handler(url, commit, proxy_server=proxy_server)
_validate_git_clone(git_dir, commit)
@ -508,8 +508,8 @@ def test_is_repository():
subpath='deployment_files/site')
def test_is_repository_negative(temp_path):
assert not git.is_repository(temp_path)
def test_is_repository_negative(tmpdir):
assert not git.is_repository(tmpdir)
@pytest.mark.skipif(

2
tests/unit/engine/util/test_shipyard_helper.py
View File

@ -13,8 +13,8 @@
# limitations under the License.
import os
from unittest import mock
import mock
import pytest
import yaml

104
tests/unit/test_cli.py
View File

@ -13,18 +13,17 @@
# limitations under the License.
import os
from unittest import mock
from click.testing import CliRunner
import mock
import pytest
import yaml
from pegleg import cli
from pegleg.engine.catalog import pki_utility
from pegleg.engine import errorcodes
from pegleg.engine.catalog import pki_utility
from pegleg.engine.util import git
from tests.unit import test_utils
from tests.unit.fixtures import temp_path
TEST_PARAMS = {
"site_name": "airship-seaworthy",
@ -34,7 +33,7 @@ TEST_PARAMS = {
"repo_url": "https://opendev.org/airship/treasuremap.git",
}
test_cert = """
TEST_CERT = """
-----BEGIN CERTIFICATE-----
DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF
@ -83,8 +82,7 @@ class TestSiteCLIOptions(BaseCLIActionTest):
### clone_path tests ###
def test_list_sites_using_remote_repo_and_clone_path_option(
self, temp_path):
def test_list_sites_using_remote_repo_and_clone_path_option(self, tmpdir):
"""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
@ -99,15 +97,14 @@ class TestSiteCLIOptions(BaseCLIActionTest):
# 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'])
cli.site, ['-p', tmpdir, '-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))
assert os.path.exists(os.path.join(tmpdir, self.repo_name))
assert git.is_repository(os.path.join(tmpdir, self.repo_name))
def test_list_sites_using_local_repo_and_clone_path_option(
self, temp_path):
def test_list_sites_using_local_repo_and_clone_path_option(self, tmpdir):
"""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
@ -121,11 +118,11 @@ class TestSiteCLIOptions(BaseCLIActionTest):
# 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'])
cli.site, ['-p', tmpdir, '-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))
assert not os.path.exists(os.path.join(tmpdir, self.repo_name))
class TestSiteCLIOptionsNegative(BaseCLIActionTest):
@ -134,7 +131,7 @@ class TestSiteCLIOptionsNegative(BaseCLIActionTest):
### Negative clone_path tests ###
def test_list_sites_using_remote_repo_and_reuse_clone_path_option(
self, temp_path):
self, tmpdir):
"""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
@ -149,20 +146,17 @@ class TestSiteCLIOptionsNegative(BaseCLIActionTest):
# 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'])
cli.site, ['-p', tmpdir, '-r', repo_url, 'list'])
assert git.is_repository(os.path.join(temp_path, self.repo_name))
assert git.is_repository(os.path.join(tmpdir, 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'])
cli.site, ['-p', tmpdir, '-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
assert 'File exists' in site_list.output
class TestSiteCliActions(BaseCLIActionTest):
@ -185,7 +179,7 @@ class TestSiteCliActions(BaseCLIActionTest):
# 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):
def test_collect_using_remote_repo_url(self, tmpdir):
"""Validates collect action using a remote URL."""
# Scenario:
#
@ -195,10 +189,9 @@ class TestSiteCliActions(BaseCLIActionTest):
repo_url = 'https://opendev.org/airship/%s@%s' % (
self.repo_name, self.repo_rev)
self._validate_collect_site_action(repo_url, temp_path)
self._validate_collect_site_action(repo_url, tmpdir)
def test_collect_using_remote_repo_url_ending_with_dot_git(
self, temp_path):
def test_collect_using_remote_repo_url_ending_with_dot_git(self, tmpdir):
"""Validates collect action using a remote URL ending in .git."""
# Scenario:
#
@ -208,9 +201,9 @@ class TestSiteCliActions(BaseCLIActionTest):
repo_url = 'https://opendev.org/airship/%s@%s.git' % (
self.repo_name, self.repo_rev)
self._validate_collect_site_action(repo_url, temp_path)
self._validate_collect_site_action(repo_url, tmpdir)
def test_collect_using_local_path(self, temp_path):
def test_collect_using_local_path(self, tmpdir):
"""Validates collect action using a path to a local repo."""
# Scenario:
#
@ -219,7 +212,7 @@ class TestSiteCliActions(BaseCLIActionTest):
# 3) Check that expected file name is there
repo_path = self.treasuremap_path
self._validate_collect_site_action(repo_path, temp_path)
self._validate_collect_site_action(repo_path, tmpdir)
### Lint tests ###
@ -279,8 +272,8 @@ class TestSiteCliActions(BaseCLIActionTest):
### List tests ###
def _validate_list_site_action(self, repo_path_or_url, temp_path):
mock_output = os.path.join(temp_path, 'output')
def _validate_list_site_action(self, repo_path_or_url, tmpdir):
mock_output = os.path.join(tmpdir, 'output')
result = self.runner.invoke(
cli.site, ['-r', repo_path_or_url, 'list', '-o', mock_output])
@ -290,7 +283,7 @@ class TestSiteCliActions(BaseCLIActionTest):
assert self.site_name in table_output
assert self.site_type in table_output
def test_list_sites_using_remote_repo_url(self, temp_path):
def test_list_sites_using_remote_repo_url(self, tmpdir):
"""Validates list action using remote repo URL."""
# Scenario:
#
@ -299,21 +292,21 @@ class TestSiteCliActions(BaseCLIActionTest):
repo_url = 'https://opendev.org/airship/%s@%s' % (
self.repo_name, self.repo_rev)
self._validate_list_site_action(repo_url, temp_path)
self._validate_list_site_action(repo_url, tmpdir)
def test_list_sites_using_local_path(self, temp_path):
def test_list_sites_using_local_path(self, tmpdir):
"""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, temp_path)
self._validate_list_site_action(repo_path, tmpdir)
### Show tests ###
def _validate_site_show_action(self, repo_path_or_url, temp_path):
mock_output = os.path.join(temp_path, 'output')
def _validate_site_show_action(self, repo_path_or_url, tmpdir):
mock_output = os.path.join(tmpdir, 'output')
result = self.runner.invoke(
cli.site, [
'-r', repo_path_or_url, 'show', self.site_name, '-o',
@ -325,7 +318,7 @@ class TestSiteCliActions(BaseCLIActionTest):
table_output = f.read()
assert self.site_name in table_output
def test_show_site_using_remote_repo_url(self, temp_path):
def test_show_site_using_remote_repo_url(self, tmpdir):
"""Validates show action using remote repo URL."""
# Scenario:
#
@ -333,16 +326,16 @@ class TestSiteCliActions(BaseCLIActionTest):
repo_url = 'https://opendev.org/airship/%s@%s' % (
self.repo_name, self.repo_rev)
self._validate_site_show_action(repo_url, temp_path)
self._validate_site_show_action(repo_url, tmpdir)
def test_show_site_using_local_path(self, temp_path):
def test_show_site_using_local_path(self, tmpdir):
"""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, temp_path)
self._validate_site_show_action(repo_path, tmpdir)
### Render tests ###
@ -486,7 +479,6 @@ class TestRepoCliActions(BaseCLIActionTest):
class TestSiteSecretsActions(BaseCLIActionTest):
"""Tests site secrets-related CLI actions."""
@classmethod
def setup_class(cls):
super(TestSiteSecretsActions, cls).setup_class()
@ -615,7 +607,7 @@ class TestSiteSecretsActions(BaseCLIActionTest):
output_path = os.path.join(file_dir, "test.yaml")
with open(file_path, "w") as test_crt_fi:
test_crt_fi.write(test_cert)
test_crt_fi.write(TEST_CERT)
secrets_opts = [
'secrets', 'wrap', "-a", "lm734y", "--filename", file_path, "-s",
"deckhand/Certificate/v1", "-n", "test-certificate", "-l", "site",
@ -626,7 +618,7 @@ class TestSiteSecretsActions(BaseCLIActionTest):
with open(output_path, "r") as output_fi:
doc = yaml.safe_load(output_fi)
assert doc["data"]["managedDocument"]["data"] == test_cert
assert doc["data"]["managedDocument"]["data"] == TEST_CERT
assert doc["data"]["managedDocument"][
"schema"] == "deckhand/Certificate/v1"
assert doc["data"]["managedDocument"]["metadata"][
@ -653,7 +645,6 @@ class TestSiteSecretsActions(BaseCLIActionTest):
class TestTypeCliActions(BaseCLIActionTest):
"""Tests type-level CLI actions."""
def setup(self):
self.expected_types = ['foundry']
@ -661,8 +652,8 @@ class TestTypeCliActions(BaseCLIActionTest):
for expected_type in self.expected_types:
assert expected_type in table_output
def _validate_type_list_action(self, repo_path_or_url, temp_path):
mock_output = os.path.join(temp_path, 'output')
def _validate_type_list_action(self, repo_path_or_url, tmpdir):
mock_output = os.path.join(tmpdir, 'output')
result = self.runner.invoke(
cli.type, ['-r', repo_path_or_url, 'list', '-o', mock_output])
with open(mock_output, 'r') as f:
@ -671,7 +662,7 @@ class TestTypeCliActions(BaseCLIActionTest):
assert result.exit_code == 0, result.output
self._assert_table_has_expected_sites(table_output)
def test_list_types_using_remote_repo_url(self, temp_path):
def test_list_types_using_remote_repo_url(self, tmpdir):
"""Validates list types action using remote repo URL."""
# Scenario:
#
@ -679,21 +670,20 @@ class TestTypeCliActions(BaseCLIActionTest):
repo_url = 'https://opendev.org/airship/%s@%s' % (
self.repo_name, self.repo_rev)
self._validate_type_list_action(repo_url, temp_path)
self._validate_type_list_action(repo_url, tmpdir)
def test_list_types_using_local_repo_path(self, temp_path):
def test_list_types_using_local_repo_path(self, tmpdir):
"""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, temp_path)
self._validate_type_list_action(repo_path, tmpdir)
class TestSiteCliActionsWithSubdirectory(BaseCLIActionTest):
"""Tests site CLI actions with subdirectories in repository paths."""
def setup(self):
self.expected_sites = ['demo', 'gate-multinode', 'dev', 'dev-proxy']
@ -701,8 +691,8 @@ class TestSiteCliActionsWithSubdirectory(BaseCLIActionTest):
for expected_site in self.expected_sites:
assert expected_site in table_output
def _validate_list_site_action(self, repo_path_or_url, temp_path):
mock_output = os.path.join(temp_path, 'output')
def _validate_list_site_action(self, repo_path_or_url, tmpdir):
mock_output = os.path.join(tmpdir, 'output')
result = self.runner.invoke(
cli.site, ['-r', repo_path_or_url, 'list', '-o', mock_output])
@ -712,7 +702,7 @@ class TestSiteCliActionsWithSubdirectory(BaseCLIActionTest):
assert result.exit_code == 0, result.output
self._assert_table_has_expected_sites(table_output)
def test_site_action_with_subpath_in_remote_url(self, temp_path):
def test_site_action_with_subpath_in_remote_url(self, tmpdir):
"""Validates list action with subpath in remote URL."""
# Scenario:
#
@ -725,9 +715,9 @@ class TestSiteCliActionsWithSubdirectory(BaseCLIActionTest):
repo_url = 'https://opendev.org/airship/%s/deployment_files@%s' % (
repo_name, repo_rev)
self._validate_list_site_action(repo_url, temp_path)
self._validate_list_site_action(repo_url, tmpdir)
def test_site_action_with_subpath_in_local_repo_path(self, temp_path):
def test_site_action_with_subpath_in_local_repo_path(self, tmpdir):
"""Validates list action with subpath in local repo path."""
# Scenario:
#
@ -741,4 +731,4 @@ class TestSiteCliActionsWithSubdirectory(BaseCLIActionTest):
_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, temp_path)
self._validate_list_site_action(repo_path, tmpdir)

6
tests/unit/test_exceptions.py
View File

@ -15,12 +15,12 @@ import logging
import pytest
from testfixtures import log_capture
from pegleg.engine import exceptions as exc
from pegleg.engine import exceptions
@log_capture()
def test_exception_with_missing_kwargs(capture):
message = 'Testing missing kwargs exception with {text}'
with pytest.raises(exc.PeglegBaseException):
raise exc.PeglegBaseException(message=message, key="value")
with pytest.raises(exceptions.PeglegBaseException):
raise exceptions.PeglegBaseException(message=message, key="value")
capture.check(('pegleg.engine.exceptions', 'WARNING', 'Missing kwargs'))

31
tests/unit/test_utils.py
View File

@ -21,13 +21,6 @@ import random
import requests
import uuid
_PROXY_SERVERS = {
'http': os.getenv(
'HTTP_PROXY', os.getenv('http_proxy', 'http://proxy.example.com')),
'https': os.getenv(
'HTTPS_PROXY', os.getenv('https_proxy', 'https://proxy.example.com'))
}
def rand_name(name='', prefix='pegleg'):
"""Generate a random name that includes a random number
@ -48,6 +41,28 @@ def rand_name(name='', prefix='pegleg'):
return rand_name
def get_proxies():
use_proxy = False
http_proxy = None
https_proxy = None
if 'http_proxy' in os.environ:
http_proxy = os.environ['http_proxy']
use_proxy = True
elif 'HTTP_PROXY' in os.environ:
http_proxy = os.environ['HTTP_PROXY']
use_proxy = True
if 'https_proxy' in os.environ:
https_proxy = os.environ['https_proxy']
use_proxy = True
elif 'HTTPS_PROXY' in os.environ:
https_proxy = os.environ['HTTPS_PROXY']
use_proxy = True
return use_proxy, {'http': http_proxy, 'https_proxy': https_proxy}
def is_connected():
"""Verifies whether network connectivity is up.
@ -71,7 +86,7 @@ def is_connected_behind_proxy():
for _ in range(3):
try:
r = requests.get(
"http://www.github.com/", proxies=_PROXY_SERVERS, timeout=3)
"http://www.github.com/", proxies=get_proxies()[1], timeout=3)
r.raise_for_status()
return True
except requests.exceptions.RequestException:

12
tox.ini
View File

@ -24,9 +24,9 @@ commands =
[testenv:fmt]
basepython = python3
deps =
-r{toxinidir}/test-requirements.txt
-r{toxinidir}/test-requirements.txt
commands =
yapf -ir {toxinidir}/pegleg {toxinidir}/tests
yapf -ir {toxinidir}/pegleg {toxinidir}/tests
[testenv:pep8]
basepython = python3
@ -57,11 +57,11 @@ commands = bandit -r pegleg -n 5
[testenv:safety]
basepython = python3
deps =
safety
safety
commands =
safety check -r {toxinidir}/requirements.txt --full-report
safety check -r {toxinidir}/test-requirements.txt --full-report
safety check -r {toxinidir}/doc/requirements.txt --full-report
safety check -r {toxinidir}/requirements.txt --full-report
safety check -r {toxinidir}/test-requirements.txt --full-report
safety check -r {toxinidir}/doc/requirements.txt --full-report
[testenv:cover]
basepython = python3