Enable minimal testing

This commit is contained in:
Luke Murphy 2019-10-23 13:04:57 +02:00
parent e96a71866d
commit 76ef7ff97e
No known key found for this signature in database
GPG Key ID: 5E2EF5A63E3718CC
42 changed files with 1773 additions and 472 deletions

4
.gitignore vendored Normal file
View File

@ -0,0 +1,4 @@
.tox
.coverage
*.xml
*egg*

31
.pre-commit-config.yaml Normal file
View File

@ -0,0 +1,31 @@
---
repos:
- repo: https://github.com/python/black.git
rev: 19.3b0
hooks:
- id: black
language_version: python3
- repo: https://github.com/pre-commit/pre-commit-hooks.git
rev: v2.2.3
hooks:
- id: end-of-file-fixer
- id: trailing-whitespace
- id: mixed-line-ending
- id: check-byte-order-marker
- id: check-merge-conflict
- id: debug-statements
- repo: https://gitlab.com/pycqa/flake8.git
rev: 3.7.8
hooks:
- id: flake8
language_version: python3
- repo: https://github.com/adrienverge/yamllint.git
rev: v1.17.0
hooks:
- id: yamllint
entry: yamllint -c .yamllint.yml --strict
language: python

27
.travis.yml Normal file
View File

@ -0,0 +1,27 @@
---
language: python
cache:
pip: true
directories:
- $HOME/.cache/pre-commit
- $HOME/.pre-commit
jobs:
fast_finish: true
include:
- python: "3.7"
name: linting
env:
- TOXENV=lint
- python: "3.7"
name: unit tests
env:
- TOXENV=ansible28
before_script:
- pip install tox==3.14.0
script:
- tox

11
.yamllint.yml Normal file
View File

@ -0,0 +1,11 @@
---
extends: default
ignore: molecule_hetznercloud/cookiecutter/scenario/driver/hetznercloud
rules:
line-length: disable
braces:
max-spaces-inside: 1
level: error

View File

@ -1,23 +0,0 @@
***************************************
Hetzner Cloud driver installation guide
***************************************
Requirements
============
* Ansible >= 2.8
* ``HCLOUD_TOKEN`` exposed in your environment
Install
=======
Please refer to the `Virtual environment`_ documentation for installation best
practices. If not using a virtual environment, please consider passing the
widely recommended `'--user' flag`_ when invoking ``pip``.
.. _Virtual environment: https://virtualenv.pypa.io/en/latest/
.. _'--user' flag: https://packaging.python.org/tutorials/installing-packages/#installing-to-the-user-site
.. code-block:: bash
$ pip install 'molecule[hetznercloud]'

View File

@ -0,0 +1,16 @@
***************************************
Hetzner Cloud driver installation guide
***************************************
Requirements
============
* Ansible >= 2.8
* ``HCLOUD_TOKEN`` exposed in your environment
Install
=======
.. code-block:: bash
$ pip install 'molecule[hetznercloud]'

View File

@ -0,0 +1,88 @@
---
{% raw -%}
- name: Create
hosts: localhost
connection: local
gather_facts: false
no_log: "{{ molecule_no_log }}"
vars:
ssh_port: 22
ssh_user: root
ssh_path: "{{ lookup('env', 'MOLECULE_EPHEMERAL_DIRECTORY') }}/ssh_key"
tasks:
- name: Create SSH key
user:
name: "{{ lookup('env', 'USER') }}"
generate_ssh_key: true
ssh_key_file: "{{ ssh_path }}"
force: true
register: generated_ssh_key
- name: Register the SSH key name
set_fact:
ssh_key_name: "molecule-generated-{{ 12345 | random | to_uuid }}"
- name: Register SSH key for test instance(s)
hcloud_ssh_key:
name: "{{ ssh_key_name }}"
public_key: "{{ generated_ssh_key.ssh_public_key }}"
state: present
- name: Create molecule instance(s)
hcloud_server:
name: "{{ item.name }}"
server_type: "{{ item.server_type }}"
ssh_keys:
- "{{ ssh_key_name }}"
volumes: "{{ item.volumes | default(omit) }}"
image: "{{ item.image }}"
location: "{{ item.location | default(omit) }}"
datacenter: "{{ item.datacenter | default(omit) }}"
user_data: "{{ item.user_data | default(omit) }}"
api_token: "{{ lookup('env', 'HCLOUD_TOKEN') }}"
state: present
register: server
with_items: "{{ molecule_yml.platforms }}"
async: 7200
poll: 0
- name: Wait for instance(s) creation to complete
async_status:
jid: "{{ item.ansible_job_id }}"
register: hetzner_jobs
until: hetzner_jobs.finished
retries: 300
with_items: "{{ server.results }}"
- name: Populate instance config dict
set_fact:
instance_conf_dict: {
'instance': "{{ item.hcloud_server.name }}",
'ssh_key_name': "{{ ssh_key_name }}",
'address': "{{ item.hcloud_server.ipv4_address }}",
'user': "{{ ssh_user }}",
'port': "{{ ssh_port }}",
'identity_file': "{{ ssh_path }}", }
with_items: "{{ hetzner_jobs.results }}"
register: instance_config_dict
when: server.changed | bool
- name: Convert instance config dict to a list
set_fact:
instance_conf: "{{ instance_config_dict.results | map(attribute='ansible_facts.instance_conf_dict') | list }}"
when: server.changed | bool
- name: Dump instance config
copy:
content: "{{ instance_conf | to_json | from_json | molecule_to_yaml | molecule_header }}"
dest: "{{ molecule_instance_config }}"
when: server.changed | bool
- name: Wait for SSH
wait_for:
port: "{{ ssh_port }}"
host: "{{ item.address }}"
search_regex: SSH
delay: 10
with_items: "{{ lookup('file', molecule_instance_config) | molecule_from_yaml }}"
{%- endraw %}

View File

@ -0,0 +1,57 @@
---
{% raw -%}
- name: Destroy
hosts: localhost
connection: local
gather_facts: false
no_log: "{{ molecule_no_log }}"
tasks:
- name: Populate the instance config
block:
- name: Populate instance config from file
set_fact:
instance_conf: "{{ lookup('file', molecule_instance_config) | molecule_from_yaml }}"
skip_instances: false
rescue:
- name: Populate instance config when file missing
set_fact:
instance_conf: {}
skip_instances: true
- name: Destroy molecule instance(s)
hcloud_server:
name: "{{ item.instance }}"
api_token: "{{ lookup('env', 'HCLOUD_TOKEN') }}"
state: absent
register: server
with_items: "{{ instance_conf }}"
when: not skip_instances
async: 7200
poll: 0
- name: Wait for instance(s) deletion to complete
async_status:
jid: "{{ item.ansible_job_id }}"
register: hetzner_jobs
until: hetzner_jobs.finished
retries: 300
with_items: "{{ server.results }}"
- name: Remove registered SSH key
hcloud_ssh_key:
name: "{{ instance_conf[0].ssh_key_name }}"
state: absent
when:
- not skip_instances
- instance_conf | length # must contain at least one instance
- name: Populate instance config
set_fact:
instance_conf: {}
- name: Dump instance config
copy:
content: "{{ instance_conf | molecule_to_yaml | molecule_header }}"
dest: "{{ molecule_instance_config }}"
when: server.changed | bool
{%- endraw %}

View File

@ -22,8 +22,7 @@ import os
from molecule import logger, util from molecule import logger, util
from molecule.api import Driver from molecule.api import Driver
from molecule.util import lru_cache from molecule.util import lru_cache, sysexit_with_message
from molecule.util import sysexit_with_message
log = logger.get_logger(__name__) log = logger.get_logger(__name__)
@ -88,7 +87,7 @@ class HetznerCloud(Driver):
def __init__(self, config=None): def __init__(self, config=None):
super(HetznerCloud, self).__init__(config) super(HetznerCloud, self).__init__(config)
self._name = 'hetznercloud' self._name = "hetznercloud"
@property @property
def name(self): def name(self):
@ -100,14 +99,14 @@ class HetznerCloud(Driver):
@property @property
def login_cmd_template(self): def login_cmd_template(self):
connection_options = ' '.join(self.ssh_connection_options) connection_options = " ".join(self.ssh_connection_options)
return ( return (
'ssh {{address}} ' "ssh {{address}} "
'-l {{user}} ' "-l {{user}} "
'-p {{port}} ' "-p {{port}} "
'-i {{identity_file}} ' "-i {{identity_file}} "
'{}' "{}"
).format(connection_options) ).format(connection_options)
@property @property
@ -119,7 +118,7 @@ class HetznerCloud(Driver):
return self._get_ssh_connection_options() return self._get_ssh_connection_options()
def login_options(self, instance_name): def login_options(self, instance_name):
d = {'instance': instance_name} d = {"instance": instance_name}
return util.merge_dicts(d, self._get_instance_config(instance_name)) return util.merge_dicts(d, self._get_instance_config(instance_name))
@ -128,12 +127,12 @@ class HetznerCloud(Driver):
d = self._get_instance_config(instance_name) d = self._get_instance_config(instance_name)
return { return {
'ansible_user': d['user'], "ansible_user": d["user"],
'ansible_host': d['address'], "ansible_host": d["address"],
'ansible_port': d['port'], "ansible_port": d["port"],
'ansible_private_key_file': d['identity_file'], "ansible_private_key_file": d["identity_file"],
'connection': 'ssh', "connection": "ssh",
'ansible_ssh_common_args': ' '.join(self.ssh_connection_options), "ansible_ssh_common_args": " ".join(self.ssh_connection_options),
} }
except StopIteration: except StopIteration:
return {} return {}
@ -146,7 +145,7 @@ class HetznerCloud(Driver):
instance_config_dict = util.safe_load_file(self._config.driver.instance_config) instance_config_dict = util.safe_load_file(self._config.driver.instance_config)
return next( return next(
item for item in instance_config_dict if item['instance'] == instance_name item for item in instance_config_dict if item["instance"] == instance_name
) )
@lru_cache() @lru_cache()
@ -159,16 +158,16 @@ class HetznerCloud(Driver):
import hcloud # noqa import hcloud # noqa
except ImportError: except ImportError:
msg = ( msg = (
'Missing Hetzner Cloud driver dependency. Please ' "Missing Hetzner Cloud driver dependency. Please "
"install via 'molecule[hetznercloud]' or refer to " "install via 'molecule[hetznercloud]' or refer to "
'your INSTALL.rst driver documentation file' "your INSTALL.rst driver documentation file"
) )
sysexit_with_message(msg) sysexit_with_message(msg)
if 'HCLOUD_TOKEN' not in os.environ: if "HCLOUD_TOKEN" not in os.environ:
msg = ( msg = (
'Missing Hetzner Cloud API token. Please expose ' "Missing Hetzner Cloud API token. Please expose "
'the HCLOUD_TOKEN environment variable with your ' "the HCLOUD_TOKEN environment variable with your "
'account API token value' "account API token value"
) )
sysexit_with_message(msg) sysexit_with_message(msg)

View File

View File

@ -0,0 +1,104 @@
# Copyright (c) 2015-2018 Cisco Systems, Inc.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
import contextlib
import os
import random
import string
import pytest
from molecule import config, logger, util
from molecule.scenario import ephemeral_directory
LOG = logger.get_logger(__name__)
@pytest.helpers.register
def run_command(cmd, env=os.environ, log=True):
if log:
cmd = _rebake_command(cmd, env)
cmd = cmd.bake(_truncate_exc=False)
return util.run_command(cmd)
def _rebake_command(cmd, env, out=LOG.out, err=LOG.error):
return cmd.bake(_env=env, _out=out, _err=err)
@pytest.fixture
def random_string(length=5):
return "".join((random.choice(string.ascii_uppercase) for _ in range(length)))
@contextlib.contextmanager
def change_dir_to(dir_name):
cwd = os.getcwd()
os.chdir(dir_name)
yield
os.chdir(cwd)
@pytest.fixture
def temp_dir(tmpdir, random_string, request):
directory = tmpdir.mkdir(random_string)
with change_dir_to(directory.strpath):
yield directory
@pytest.fixture
def resources_folder_path():
resources_folder_path = os.path.join(os.path.dirname(__file__), "resources")
return resources_folder_path
@pytest.helpers.register
def molecule_project_directory():
return os.getcwd()
@pytest.helpers.register
def molecule_directory():
return config.molecule_directory(molecule_project_directory())
@pytest.helpers.register
def molecule_scenario_directory():
return os.path.join(molecule_directory(), "default")
@pytest.helpers.register
def molecule_file():
return get_molecule_file(molecule_scenario_directory())
@pytest.helpers.register
def get_molecule_file(path):
return config.molecule_file(path)
@pytest.helpers.register
def molecule_ephemeral_directory(_fixture_uuid):
project_directory = "test-project-{}".format(_fixture_uuid)
scenario_name = "test-instance"
return ephemeral_directory(
os.path.join("molecule_test", project_directory, scenario_name)
)

View File

@ -0,0 +1,9 @@
# ansible-lint config for functional testing, used to bypass expected metadata
# errors in molecule-generated roles. Loaded via the metadata_lint_update
# pytest helper. For reference, see "E7xx - metadata" in:
# https://docs.ansible.com/ansible-lint/rules/default_rules.html
skip_list:
# metadata/701 - Role info should contain platforms
- '701'
# metadata/703 - Should change default metadata: <field>"
- '703'

View File

@ -0,0 +1,246 @@
# Copyright (c) 2015-2018 Cisco Systems, Inc.
# Copyright (c) 2018 Red Hat, Inc.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
import os
import pkg_resources
import shutil
import pexpect
import pytest
import sh
from molecule import logger
from molecule import util
from ..conftest import change_dir_to
LOG = logger.get_logger(__name__)
IS_TRAVIS = os.getenv("TRAVIS") and os.getenv("CI")
def _env_vars_exposed(env_vars, env=os.environ):
"""Check if environment variables are exposed and populated."""
for env_var in env_vars:
if env_var not in os.environ:
return False
return os.environ[env_var] != ""
@pytest.fixture
def with_scenario(request, scenario_to_test, driver_name, scenario_name, skip_test):
scenario_directory = os.path.join(
os.path.dirname(util.abs_path(__file__)),
os.path.pardir,
"scenarios",
scenario_to_test,
)
with change_dir_to(scenario_directory):
yield
if scenario_name:
msg = "CLEANUP: Destroying instances for all scenario(s)"
LOG.out(msg)
options = {"driver_name": driver_name, "all": True}
cmd = sh.molecule.bake("destroy", **options)
pytest.helpers.run_command(cmd)
@pytest.fixture
def skip_test(request, driver_name):
msg_tmpl = (
"Ignoring '{}' tests for now"
if driver_name == "delegated"
else "Skipped '{}' not supported"
)
support_checks_map = {
"hetznercloud": lambda: min_ansible("2.8") and supports_hetznercloud()
}
try:
check_func = support_checks_map[driver_name]
if not check_func():
pytest.skip(msg_tmpl.format(driver_name))
except KeyError:
pass
@pytest.helpers.register
def idempotence(scenario_name):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("create", **options)
pytest.helpers.run_command(cmd)
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("converge", **options)
pytest.helpers.run_command(cmd)
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("idempotence", **options)
pytest.helpers.run_command(cmd)
@pytest.helpers.register
def init_role(temp_dir, driver_name):
role_directory = os.path.join(temp_dir.strpath, "test-init")
cmd = sh.molecule.bake(
"init", "role", {"driver-name": driver_name, "role-name": "test-init"}
)
pytest.helpers.run_command(cmd)
pytest.helpers.metadata_lint_update(role_directory)
with change_dir_to(role_directory):
options = {"all": True}
cmd = sh.molecule.bake("test", **options)
pytest.helpers.run_command(cmd)
@pytest.helpers.register
def init_scenario(temp_dir, driver_name):
role_directory = os.path.join(temp_dir.strpath, "test-init")
cmd = sh.molecule.bake(
"init", "role", {"driver-name": driver_name, "role-name": "test-init"}
)
pytest.helpers.run_command(cmd)
pytest.helpers.metadata_lint_update(role_directory)
with change_dir_to(role_directory):
molecule_directory = pytest.helpers.molecule_directory()
scenario_directory = os.path.join(molecule_directory, "test-scenario")
options = {"scenario_name": "test-scenario", "role_name": "test-init"}
cmd = sh.molecule.bake("init", "scenario", **options)
pytest.helpers.run_command(cmd)
assert os.path.isdir(scenario_directory)
options = {"scenario_name": "test-scenario", "all": True}
cmd = sh.molecule.bake("test", **options)
pytest.helpers.run_command(cmd)
@pytest.helpers.register
def metadata_lint_update(role_directory):
ansible_lint_src = os.path.join(
os.path.dirname(util.abs_path(__file__)), ".ansible-lint"
)
shutil.copy(ansible_lint_src, role_directory)
with change_dir_to(role_directory):
cmd = sh.ansible_lint.bake(".")
pytest.helpers.run_command(cmd)
@pytest.helpers.register
def list(x):
cmd = sh.molecule.bake("list")
out = pytest.helpers.run_command(cmd, log=False)
out = out.stdout.decode("utf-8")
out = util.strip_ansi_color(out)
for l in x.splitlines():
assert l in out
@pytest.helpers.register
def list_with_format_plain(x):
cmd = sh.molecule.bake("list", {"format": "plain"})
out = pytest.helpers.run_command(cmd, log=False)
out = out.stdout.decode("utf-8")
out = util.strip_ansi_color(out)
for l in x.splitlines():
assert l in out
@pytest.helpers.register
def login(login_args, scenario_name="default"):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("destroy", **options)
pytest.helpers.run_command(cmd)
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("create", **options)
pytest.helpers.run_command(cmd)
for instance, regexp in login_args:
if len(login_args) > 1:
child_cmd = "molecule login --host {} --scenario-name {}".format(
instance, scenario_name
)
else:
child_cmd = "molecule login --scenario-name {}".format(scenario_name)
child = pexpect.spawn(child_cmd)
child.expect(regexp)
child.sendline("exit")
@pytest.helpers.register
def test(driver_name, scenario_name="default", parallel=False):
options = {
"scenario_name": scenario_name,
"all": scenario_name is None,
"parallel": parallel,
}
if driver_name == "delegated":
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("test", **options)
pytest.helpers.run_command(cmd)
@pytest.helpers.register
def verify(scenario_name="default"):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("create", **options)
pytest.helpers.run_command(cmd)
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("converge", **options)
pytest.helpers.run_command(cmd)
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("verify", **options)
pytest.helpers.run_command(cmd)
def min_ansible(version):
"""Ensure current Ansible is newer than a given a minimal one."""
try:
from ansible.release import __version__
return pkg_resources.parse_version(__version__) >= pkg_resources.parse_version(
version
)
except ImportError as exception:
LOG.error("Unable to parse Ansible version", exc_info=exception)
return False
@pytest.helpers.register
def supports_hetznercloud():
pytest.importorskip("hcloud")
env_vars = ("HCLOUD_TOKEN",)
return _env_vars_exposed(env_vars)

View File

@ -0,0 +1,312 @@
# Copyright (c) 2015-2018 Cisco Systems, Inc.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
import os
import pytest
import sh
from molecule.scenario import ephemeral_directory
@pytest.fixture
def scenario_to_test(request):
return request.param
@pytest.fixture
def scenario_name(request):
try:
return request.param
except AttributeError:
return None
@pytest.fixture
def driver_name(request):
return request.param
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", "default")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_check(scenario_to_test, with_scenario, scenario_name):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("check", **options)
pytest.helpers.run_command(cmd)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", "default")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_cleanup(scenario_to_test, with_scenario, scenario_name):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("cleanup", **options)
pytest.helpers.run_command(cmd)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", "default")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_converge(scenario_to_test, with_scenario, scenario_name):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("converge", **options)
pytest.helpers.run_command(cmd)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", "default")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_create(scenario_to_test, with_scenario, scenario_name):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("create", **options)
pytest.helpers.run_command(cmd)
@pytest.mark.skip(
reason="Disabled due to https://github.com/ansible/galaxy/issues/2030"
)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("dependency", "hetznercloud", "ansible-galaxy")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_dependency_ansible_galaxy(
request, scenario_to_test, with_scenario, scenario_name
):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("dependency", **options)
pytest.helpers.run_command(cmd)
dependency_role = os.path.join(
ephemeral_directory("molecule"),
"dependency",
"ansible-galaxy",
"roles",
"timezone",
)
assert os.path.isdir(dependency_role)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("dependency", "hetznercloud", "gilt")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_dependency_gilt(
request, scenario_to_test, with_scenario, scenario_name
):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("dependency", **options)
pytest.helpers.run_command(cmd)
dependency_role = os.path.join(
ephemeral_directory("molecule"), "dependency", "gilt", "roles", "timezone"
)
assert os.path.isdir(dependency_role)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("dependency", "hetznercloud", "shell")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_dependency_shell(
request, scenario_to_test, with_scenario, scenario_name
):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("dependency", **options)
pytest.helpers.run_command(cmd)
dependency_role = os.path.join(
ephemeral_directory("molecule"), "dependency", "shell", "roles", "timezone"
)
assert os.path.isdir(dependency_role)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", "default")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_destroy(scenario_to_test, with_scenario, scenario_name):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("destroy", **options)
pytest.helpers.run_command(cmd)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", "default")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_idempotence(scenario_to_test, with_scenario, scenario_name):
pytest.helpers.idempotence(scenario_name)
@pytest.mark.parametrize(
"driver_name",
[
("digitalocean"),
("docker"),
("ec2"),
("gce"),
("linode"),
("openstack"),
("vagrant"),
],
indirect=["driver_name"],
)
def test_command_init_role(temp_dir, driver_name, skip_test):
pytest.helpers.init_role(temp_dir, driver_name)
@pytest.mark.parametrize("driver_name", [("hetznercloud")], indirect=["driver_name"])
def test_command_init_scenario(temp_dir, driver_name, skip_test):
pytest.helpers.init_scenario(temp_dir, driver_name)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", "default")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_lint(scenario_to_test, with_scenario, scenario_name):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("lint", **options)
pytest.helpers.run_command(cmd)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, expected",
[
(
"driver/hetznercloud",
"hetznercloud",
"""
Instance Name Driver Name Provisioner Name Scenario Name Created Converged
--------------- ------------- ------------------ --------------- --------- -----------
instance hetznercloud ansible default false false
instance-1 hetznercloud ansible multi-node false false
instance-2 hetznercloud ansible multi-node false false
""".strip(), # noqa
)
],
indirect=["scenario_to_test", "driver_name"],
)
def test_command_list(scenario_to_test, with_scenario, expected):
pytest.helpers.list(expected)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, expected",
[
(
"driver/hetznercloud",
"hetznercloud",
"""
instance hetznercloud ansible default false false
instance-1 hetznercloud ansible multi-node false false
instance-2 hetznercloud ansible multi-node false false
""".strip(),
) # noqa
],
indirect=["scenario_to_test", "driver_name"],
)
def test_command_list_with_format_plain(scenario_to_test, with_scenario, expected):
pytest.helpers.list_with_format_plain(expected)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, login_args, scenario_name",
[
(
"driver/hetznercloud",
"hetznercloud",
[["instance-1", ".*instance-1.*"], ["instance-2", ".*instance-2.*"]],
"multi-node",
)
],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_login(scenario_to_test, with_scenario, login_args, scenario_name):
pytest.helpers.login(login_args, scenario_name)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", "default")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_prepare(scenario_to_test, with_scenario, scenario_name):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("create", **options)
pytest.helpers.run_command(cmd)
cmd = sh.molecule.bake("prepare", **options)
pytest.helpers.run_command(cmd)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", "default")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_side_effect(scenario_to_test, with_scenario, scenario_name):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("side-effect", **options)
pytest.helpers.run_command(cmd)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", "default")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_syntax(scenario_to_test, with_scenario, scenario_name):
options = {"scenario_name": scenario_name}
cmd = sh.molecule.bake("syntax", **options)
pytest.helpers.run_command(cmd)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", None)],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_test(scenario_to_test, with_scenario, scenario_name, driver_name):
pytest.helpers.test(driver_name, scenario_name)
@pytest.mark.parametrize(
"scenario_to_test, driver_name, scenario_name",
[("driver/hetznercloud", "hetznercloud", "default")],
indirect=["scenario_to_test", "driver_name", "scenario_name"],
)
def test_command_verify(scenario_to_test, with_scenario, scenario_name):
pytest.helpers.verify(scenario_name)

View File

@ -0,0 +1,13 @@
---
extends: default
rules:
braces:
max-spaces-inside: 1
level: error
brackets:
max-spaces-inside: 1
level: error
line-length: disable
truthy: disable

View File

@ -0,0 +1,20 @@
---
dependency:
name: galaxy
driver:
name: docker
lint:
name: yamllint
platforms:
- name: instance
image: centos:${TEST_CENTOS_VERSION}
provisioner:
name: ansible
lint:
name: ansible-lint
scenario:
name: ansible-galaxy
verifier:
name: testinfra
lint:
name: flake8

View File

@ -1,5 +1,4 @@
--- ---
{% raw -%}
- name: Create - name: Create
hosts: localhost hosts: localhost
connection: local connection: local
@ -13,8 +12,8 @@
- name: Create SSH key - name: Create SSH key
user: user:
name: "{{ lookup('env', 'USER') }}" name: "{{ lookup('env', 'USER') }}"
generate_ssh_key: true
ssh_key_file: "{{ ssh_path }}" ssh_key_file: "{{ ssh_path }}"
generate_ssh_key: true
force: true force: true
register: generated_ssh_key register: generated_ssh_key
@ -87,4 +86,3 @@
search_regex: SSH search_regex: SSH
delay: 10 delay: 10
with_items: "{{ lookup('file', molecule_instance_config) | molecule_from_yaml }}" with_items: "{{ lookup('file', molecule_instance_config) | molecule_from_yaml }}"
{%- endraw %}

View File

@ -1,5 +1,4 @@
--- ---
{% raw -%}
- name: Destroy - name: Destroy
hosts: localhost hosts: localhost
connection: local connection: local
@ -56,4 +55,3 @@
content: "{{ instance_conf | molecule_to_yaml | molecule_header }}" content: "{{ instance_conf | molecule_to_yaml | molecule_header }}"
dest: "{{ molecule_instance_config }}" dest: "{{ molecule_instance_config }}"
when: server.changed | bool when: server.changed | bool
{%- endraw %}

View File

@ -0,0 +1,40 @@
---
dependency:
name: galaxy
driver:
name: hetznercloud
lint:
name: yamllint
options:
config-file: ../../../resources/.yamllint
platforms:
- name: instance
server_type: cx11
image: debian-9
provisioner:
name: ansible
playbooks:
create: ../../../../../resources/playbooks/hetznercloud/create.yml
destroy: ../../../../../resources/playbooks/hetznercloud/destroy.yml
env:
ANSIBLE_ROLES_PATH: ../../../../../resources/roles/
lint:
name: ansible-lint
config_options:
defaults:
timeout: 100
ssh_connection:
scp_if_ssh: true
scenario:
name: default
verifier:
name: testinfra
lint:
name: flake8
enabled: false

View File

@ -0,0 +1,6 @@
---
- name: Converge
hosts: all
gather_facts: false
roles:
- molecule

View File

@ -0,0 +1,50 @@
---
dependency:
name: galaxy
driver:
name: hetznercloud
lint:
name: yamllint
options:
config-file: ../../../resources/.yamllint
platforms:
- name: instance-1
server_type: cx11
image: debian-9
groups:
- foo
- bar
- name: instance-2
server_type: cx11
image: debian-9
groups:
- foo
- baz
provisioner:
name: ansible
playbooks:
create: ../../../../../resources/playbooks/hetznercloud/create.yml
destroy: ../../../../../resources/playbooks/hetznercloud/destroy.yml
env:
ANSIBLE_ROLES_PATH: ../../../../../resources/roles/
lint:
name: ansible-lint
config_options:
defaults:
timeout: 100
ssh_connection:
scp_if_ssh: true
scenario:
name: multi-node
verifier:
name: testinfra
lint:
name: flake8
enabled: false

View File

@ -0,0 +1,24 @@
---
- name: Converge
hosts: all
gather_facts: false
roles:
- molecule
- name: Converge
hosts: bar
gather_facts: false
roles:
- molecule
- name: Converge
hosts: foo
gather_facts: false
roles:
- molecule
- name: Converge
hosts: baz
gather_facts: false
roles:
- molecule

View File

@ -0,0 +1,252 @@
# Copyright (c) 2015-2018 Cisco Systems, Inc.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
from uuid import uuid4
import copy
import functools
import os
import shutil
try:
from pathlib import Path
except ImportError:
from pathlib2 import Path
import pytest
from molecule import util
from molecule import config
@pytest.helpers.register
def write_molecule_file(filename, data):
util.write_file(filename, util.safe_dump(data))
@pytest.helpers.register
def os_split(s):
rest, tail = os.path.split(s)
if rest in ("", os.path.sep):
return (tail,)
return os_split(rest) + (tail,)
@pytest.fixture
def _molecule_dependency_galaxy_section_data():
return {"dependency": {"name": "galaxy"}}
@pytest.fixture
def _molecule_driver_section_data():
return {"driver": {"name": "docker", "options": {"managed": True}}}
@pytest.fixture
def _molecule_lint_section_data():
return {"lint": {"name": "yamllint"}}
@pytest.fixture
def _molecule_platforms_section_data():
return {
"platforms": [
{"name": "instance-1", "groups": ["foo", "bar"], "children": ["child1"]},
{"name": "instance-2", "groups": ["baz", "foo"], "children": ["child2"]},
]
}
@pytest.fixture
def _molecule_provisioner_section_data():
return {
"provisioner": {
"name": "ansible",
"options": {"become": True},
"lint": {"name": "ansible-lint"},
"config_options": {},
}
}
@pytest.fixture
def _molecule_scenario_section_data():
return {"scenario": {"name": "default"}}
@pytest.fixture
def _molecule_verifier_section_data():
return {"verifier": {"name": "testinfra", "lint": {"name": "flake8"}}}
@pytest.fixture
def molecule_data(
_molecule_dependency_galaxy_section_data,
_molecule_driver_section_data,
_molecule_lint_section_data,
_molecule_platforms_section_data,
_molecule_provisioner_section_data,
_molecule_scenario_section_data,
_molecule_verifier_section_data,
):
fixtures = [
_molecule_dependency_galaxy_section_data,
_molecule_driver_section_data,
_molecule_lint_section_data,
_molecule_platforms_section_data,
_molecule_provisioner_section_data,
_molecule_scenario_section_data,
_molecule_verifier_section_data,
]
return functools.reduce(lambda x, y: util.merge_dicts(x, y), fixtures)
@pytest.fixture
def molecule_directory_fixture(temp_dir):
return pytest.helpers.molecule_directory()
@pytest.fixture
def molecule_scenario_directory_fixture(molecule_directory_fixture):
path = pytest.helpers.molecule_scenario_directory()
if not os.path.isdir(path):
os.makedirs(path)
return path
@pytest.fixture
def molecule_ephemeral_directory_fixture(molecule_scenario_directory_fixture):
path = pytest.helpers.molecule_ephemeral_directory(str(uuid4()))
if not os.path.isdir(path):
os.makedirs(path)
yield
shutil.rmtree(str(Path(path).parent))
@pytest.fixture
def molecule_file_fixture(
molecule_scenario_directory_fixture, molecule_ephemeral_directory_fixture
):
return pytest.helpers.molecule_file()
@pytest.fixture
def config_instance(molecule_file_fixture, molecule_data, request):
mdc = copy.deepcopy(molecule_data)
if hasattr(request, "param"):
util.merge_dicts(mdc, request.getfixturevalue(request.param))
pytest.helpers.write_molecule_file(molecule_file_fixture, mdc)
c = config.Config(molecule_file_fixture)
c.command_args = {"subcommand": "test"}
return c
# Mocks
@pytest.fixture
def patched_print_debug(mocker):
return mocker.patch("molecule.util.print_debug")
@pytest.fixture
def patched_logger_info(mocker):
return mocker.patch("logging.Logger.info")
@pytest.fixture
def patched_logger_out(mocker):
return mocker.patch("molecule.logger.CustomLogger.out")
@pytest.fixture
def patched_logger_warn(mocker):
return mocker.patch("logging.Logger.warn")
@pytest.fixture
def patched_logger_error(mocker):
return mocker.patch("logging.Logger.error")
@pytest.fixture
def patched_logger_critical(mocker):
return mocker.patch("logging.Logger.critical")
@pytest.fixture
def patched_logger_success(mocker):
return mocker.patch("molecule.logger.CustomLogger.success")
@pytest.fixture
def patched_run_command(mocker):
m = mocker.patch("molecule.util.run_command")
m.return_value = mocker.Mock(stdout=b"patched-run-command-stdout")
return m
@pytest.fixture
def patched_ansible_converge(mocker):
m = mocker.patch("molecule.provisioner.ansible.Ansible.converge")
m.return_value = "patched-ansible-converge-stdout"
return m
@pytest.fixture
def patched_add_or_update_vars(mocker):
return mocker.patch("molecule.provisioner.ansible.Ansible._add_or_update_vars")
@pytest.fixture
def patched_yamllint(mocker):
return mocker.patch("molecule.lint.yamllint.Yamllint.execute")
@pytest.fixture
def patched_flake8(mocker):
return mocker.patch("molecule.verifier.lint.flake8.Flake8.execute")
@pytest.fixture
def patched_ansible_galaxy(mocker):
return mocker.patch("molecule.dependency.ansible_galaxy.AnsibleGalaxy.execute")
@pytest.fixture
def patched_testinfra(mocker):
return mocker.patch("molecule.verifier.testinfra.Testinfra.execute")
@pytest.fixture
def patched_scenario_setup(mocker):
mocker.patch("molecule.config.Config.env")
return mocker.patch("molecule.scenario.Scenario._setup")
@pytest.fixture
def patched_config_validate(mocker):
return mocker.patch("molecule.config.Config._validate")

View File

@ -0,0 +1,82 @@
# Copyright (c) 2015-2018 Cisco Systems, Inc.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
import os
import pytest
import sh
from molecule import util
from molecule.command.init import base
from molecule.model import schema_v2
class CommandBase(base.Base):
pass
@pytest.fixture
def _base_class():
return CommandBase
@pytest.fixture
def _instance(_base_class):
return _base_class()
@pytest.fixture
def _command_args():
return {
"dependency_name": "galaxy",
"driver_name": "docker",
"lint_name": "yamllint",
"provisioner_name": "ansible",
"scenario_name": "default",
"role_name": "test-role",
"verifier_name": "testinfra",
}
@pytest.fixture
def _role_directory():
return "."
@pytest.fixture
def _molecule_file(_role_directory):
return os.path.join(
_role_directory, "test-role", "molecule", "default", "molecule.yml"
)
@pytest.mark.parametrize("driver", [("hetznercloud")])
def test_drivers(
driver, temp_dir, _molecule_file, _role_directory, _command_args, _instance
):
_command_args["driver_name"] = driver
_instance._process_templates("molecule", _command_args, _role_directory)
data = util.safe_load_file(_molecule_file)
assert {} == schema_v2.validate(data)
cmd = sh.yamllint.bake("-s", _molecule_file)
pytest.helpers.run_command(cmd)

View File

@ -0,0 +1,200 @@
# Copyright (c) 2019 Red Hat, Inc.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
import os
import pytest
from molecule import config
from molecule.driver import hetznercloud
@pytest.fixture
def hetznercloud_instance(patched_config_validate, config_instance):
return hetznercloud.HetznerCloud(config_instance)
def test_hetznercloud_config_gives_config_object(hetznercloud_instance):
assert isinstance(hetznercloud_instance._config, config.Config)
def test_hetznercloud_testinfra_options_property(hetznercloud_instance):
assert {
"connection": "ansible",
"ansible-inventory": hetznercloud_instance._config.provisioner.inventory_file,
} == hetznercloud_instance.testinfra_options
def test_hetznercloud_name_property(hetznercloud_instance):
assert "hetznercloud" == hetznercloud_instance.name
def test_hetznercloud_options_property(hetznercloud_instance):
assert {"managed": True} == hetznercloud_instance.options
def test_hetznercloud_login_cmd_template_property(hetznercloud_instance):
template = "ssh {address} -l {user} -p {port}"
assert template in hetznercloud_instance.login_cmd_template
def test_hetznercloud_safe_files_property(hetznercloud_instance):
expected_safe_files = [
os.path.join(
hetznercloud_instance._config.scenario.ephemeral_directory,
"instance_config.yml",
)
]
assert expected_safe_files == hetznercloud_instance.safe_files
def test_hetznercloud_default_safe_files_property(hetznercloud_instance):
expected_default_safe_files = [
os.path.join(
hetznercloud_instance._config.scenario.ephemeral_directory,
"instance_config.yml",
)
]
assert expected_default_safe_files == hetznercloud_instance.default_safe_files
def test_hetznercloud_delegated_property(hetznercloud_instance):
assert not hetznercloud_instance.delegated
def test_hetznercloud_managed_property(hetznercloud_instance):
assert hetznercloud_instance.managed
def test_hetznercloud_default_ssh_connection_options_property(hetznercloud_instance):
expected_options = [
"-o UserKnownHostsFile=/dev/null",
"-o ControlMaster=auto",
"-o ControlPersist=60s",
"-o IdentitiesOnly=yes",
"-o StrictHostKeyChecking=no",
]
assert expected_options == (hetznercloud_instance.default_ssh_connection_options)
def test_hetznercloud_login_options(hetznercloud_instance, mocker):
target = "molecule.driver.hetznercloud.HetznerCloud._get_instance_config"
get_instance_config_patch = mocker.patch(target)
get_instance_config_patch.return_value = {
"instance": "hetznercloud",
"address": "172.16.0.2",
"user": "hetzner-admin",
"port": 22,
}
get_instance_config_patch = {
"instance": "hetznercloud",
"address": "172.16.0.2",
"user": "hetzner-admin",
"port": 22,
}
assert get_instance_config_patch == hetznercloud_instance.login_options(
"hetznercloud"
)
def test_hetznercloud_ansible_connection_opts(hetznercloud_instance, mocker):
target = "molecule.driver.hetznercloud.HetznerCloud._get_instance_config"
get_instance_config_patch = mocker.patch(target)
get_instance_config_patch.return_value = {
"instance": "hetznercloud",
"address": "172.16.0.2",
"user": "hetzner-admin",
"port": 22,
"identity_file": "/foo/bar",
}
get_instance_config_patch = {
"ansible_host": "172.16.0.2",
"ansible_port": 22,
"ansible_user": "hetzner-admin",
"ansible_private_key_file": "/foo/bar",
"connection": "ssh",
"ansible_ssh_common_args": (
"-o UserKnownHostsFile=/dev/null "
"-o ControlMaster=auto "
"-o ControlPersist=60s "
"-o IdentitiesOnly=yes "
"-o StrictHostKeyChecking=no"
),
}
connection_options = hetznercloud_instance.ansible_connection_options(
"hetznercloud"
)
assert get_instance_config_patch == connection_options
def test_hetznercloud_instance_config_property(hetznercloud_instance):
instance_config_path = os.path.join(
hetznercloud_instance._config.scenario.ephemeral_directory,
"instance_config.yml",
)
assert instance_config_path == hetznercloud_instance.instance_config
def test_hetznercloud_ssh_connection_options_property(hetznercloud_instance):
expected_options = [
"-o UserKnownHostsFile=/dev/null",
"-o ControlMaster=auto",
"-o ControlPersist=60s",
"-o IdentitiesOnly=yes",
"-o StrictHostKeyChecking=no",
]
assert expected_options == hetznercloud_instance.ssh_connection_options
def test_hetznercloud_status(mocker, hetznercloud_instance):
hetzner_status = hetznercloud_instance.status()
assert 2 == len(hetzner_status)
assert hetzner_status[0].instance_name == "instance-1"
assert hetzner_status[0].driver_name == "hetznercloud"
assert hetzner_status[0].provisioner_name == "ansible"
assert hetzner_status[0].scenario_name == "default"
assert hetzner_status[0].created == "false"
assert hetzner_status[0].converged == "false"
assert hetzner_status[1].instance_name == "instance-2"
assert hetzner_status[1].driver_name == "hetznercloud"
assert hetzner_status[1].provisioner_name == "ansible"
assert hetzner_status[1].scenario_name == "default"
assert hetzner_status[1].created == "false"
assert hetzner_status[1].converged == "false"
def test_created(hetznercloud_instance):
assert "false" == hetznercloud_instance._created()
def test_converged(hetznercloud_instance):
assert "false" == hetznercloud_instance._converged()

View File

@ -0,0 +1,46 @@
# Copyright (c) 2015-2018 Cisco Systems, Inc.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
import os
import pytest
from molecule import util
@pytest.fixture
def _molecule_file():
return os.path.join(
os.path.dirname(__file__),
os.path.pardir,
os.path.pardir,
os.path.pardir,
"resources",
"molecule_docker.yml",
)
@pytest.fixture
def _config(_molecule_file, request):
d = util.safe_load(open(_molecule_file))
if hasattr(request, "param"):
d = util.merge_dicts(d, request.getfixturevalue(request.param))
return d

View File

@ -0,0 +1,101 @@
# Copyright (c) 2015-2018 Cisco Systems, Inc.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
import pytest
from molecule.model import schema_v2
@pytest.fixture
def _model_platform_hetznercloud_section_data():
return {
"driver": {"name": "hetznercloud"},
"platforms": [
{
"name": "instance",
"server_type": "",
"volumes": [""],
"image": "",
"location": "",
"datacenter": "",
"user_data": "",
}
],
}
@pytest.mark.parametrize(
"_config", ["_model_platform_hetznercloud_section_data"], indirect=True
)
def test_platforms_hetznercloud(_config):
assert {} == schema_v2.validate(_config)
@pytest.fixture
def _model_platforms_hetznercloud_errors_section_data():
return {
"driver": {"name": "hetznercloud"},
"platforms": [
{
"name": 0,
"server_type": 0,
"volumes": {},
"image": 0,
"location": 0,
"datacenter": 0,
"user_data": 0,
}
],
}
@pytest.mark.parametrize(
"_config", ["_model_platforms_hetznercloud_errors_section_data"], indirect=True
)
def test_platforms_hetznercloud_has_errors(_config):
expected_config = {
"platforms": [
{
0: [
{
"name": ["must be of string type"],
"server_type": ["must be of string type"],
"volumes": ["must be of list type"],
"image": ["must be of string type"],
"location": ["must be of string type"],
"datacenter": ["must be of string type"],
"user_data": ["must be of string type"],
}
]
}
]
}
assert expected_config == schema_v2.validate(_config)
@pytest.mark.parametrize(
"_config", ["_model_platform_hetznercloud_section_data"], indirect=True
)
@pytest.mark.parametrize("_required_field", ("server_type", "image"))
def test_platforms_hetznercloud_fields_required(_config, _required_field):
del _config["platforms"][0][_required_field]
expected_config = {"platforms": [{0: [{_required_field: ["required field"]}]}]}
assert expected_config == schema_v2.validate(_config)

View File

@ -6,6 +6,3 @@ requires = [
"wheel", "wheel",
] ]
build-backend = "setuptools.build_meta" build-backend = "setuptools.build_meta"
[tool.black]
skip-string-normalization = true

View File

@ -1,12 +1,5 @@
[pytest] [pytest]
addopts = -v -rxXs --doctest-modules --durations 10 --cov=molecule --cov-report term-missing:skip-covered --cov-report xml -m "not molecule" addopts = -v -rxXs --doctest-modules --durations 10 --cov=molecule_hetznercloud --cov-report term-missing:skip-covered --cov-report xml -m "not molecule"
doctest_optionflags = ALLOW_UNICODE ELLIPSIS doctest_optionflags = ALLOW_UNICODE ELLIPSIS
junit_suite_name = molecule_test_suite junit_suite_name = molecule_test_suite
norecursedirs = dist doc build .tox .eggs norecursedirs = dist doc build .tox .eggs
filterwarnings =
# remove once https://github.com/cookiecutter/cookiecutter/pull/1127 is released
ignore::DeprecationWarning:cookiecutter
# remove once https://github.com/pytest-dev/pytest-cov/issues/327 is released
ignore::pytest.PytestWarning:pytest_cov
# remove once https://bitbucket.org/astanin/python-tabulate/issues/174/invalid-escape-sequence
ignore::DeprecationWarning:tabulate

View File

@ -12,7 +12,7 @@ project_urls =
Release Management = https://github.com/pycontribs/molecule-hetznercloud/projects Release Management = https://github.com/pycontribs/molecule-hetznercloud/projects
CI: Travis = https://travis-ci.com/pycontribs/molecule-hetznercloud CI: Travis = https://travis-ci.com/pycontribs/molecule-hetznercloud
Source Code = https://github.com/pycontribs/molecule-hetznercloud Source Code = https://github.com/pycontribs/molecule-hetznercloud
description = Hetznercloud Molecule Plugin :: run molecule tests with hetznercloud as verifier description = Molecule Hetzner Cloud Plugin :: run molecule tests with hetzner cloud as a provisioner
long_description = file: README.rst long_description = file: README.rst
long_description_content_type = text/x-rst long_description_content_type = text/x-rst
author = Luke Murphy author = Luke Murphy
@ -70,19 +70,17 @@ install_requires =
[options.extras_require] [options.extras_require]
test = test =
flake8>=3.6.0, < 4 hcloud>=1.6.2
mock>=3.0.5, < 4 mock>=3.0.5, < 4
pytest>=4.6.3, < 5
pytest-cov>=2.7.1, < 3 pytest-cov>=2.7.1, < 3
pytest-helpers-namespace>=2019.1.8, < 2020 pytest-helpers-namespace>=2019.1.8, < 2020
pytest-mock>=1.10.4, < 2 pytest-mock>=1.10.4, < 2
pytest-verbose-parametrize>=1.7.0, < 2 pytest-verbose-parametrize>=1.7.0, < 2
pytest-xdist>=1.29.0, < 2 pytest-xdist>=1.29.0, < 2
hcloud>=1.2.1 pytest>=4.6.3, < 5
[options.entry_points]
molecule.verifier =
hetznercloud = molecule_hetznercloud.hetznercloud:hetznercloud
[options.packages.find] [options.packages.find]
where = . where = .
[flake8]
max-line-length = 88

327
setup.py
View File

@ -1,330 +1,5 @@
#!/usr/bin/env python #!/usr/bin/env python
# -*- coding: utf-8 -*-
import setuptools import setuptools
HAS_DIST_INFO_CMD = False setuptools.setup()
try:
import setuptools.command.dist_info
HAS_DIST_INFO_CMD = True
except ImportError:
"""Setuptools version is too old."""
ALL_STRING_TYPES = tuple(map(type, ("", b"", u"")))
MIN_NATIVE_SETUPTOOLS_VERSION = 34, 4, 0
"""Minimal setuptools having good read_configuration implementation."""
RUNTIME_SETUPTOOLS_VERSION = tuple(map(int, setuptools.__version__.split(".")))
"""Setuptools imported now."""
READ_CONFIG_SHIM_NEEDED = RUNTIME_SETUPTOOLS_VERSION < MIN_NATIVE_SETUPTOOLS_VERSION
def str_if_nested_or_str(s):
"""Turn input into a native string if possible."""
if isinstance(s, ALL_STRING_TYPES):
return str(s)
if isinstance(s, (list, tuple)):
return type(s)(map(str_if_nested_or_str, s))
if isinstance(s, (dict,)):
return stringify_dict_contents(s)
return s
def stringify_dict_contents(dct):
"""Turn dict keys and values into native strings."""
return {str_if_nested_or_str(k): str_if_nested_or_str(v) for k, v in dct.items()}
if not READ_CONFIG_SHIM_NEEDED:
from setuptools.config import read_configuration, ConfigOptionsHandler
import setuptools.config
import setuptools.dist
# Set default value for 'use_scm_version'
setattr(setuptools.dist.Distribution, "use_scm_version", False)
# Attach bool parser to 'use_scm_version' option
class ShimConfigOptionsHandler(ConfigOptionsHandler):
"""Extension class for ConfigOptionsHandler."""
@property
def parsers(self):
"""Return an option mapping with default data type parsers."""
_orig_parsers = super(ShimConfigOptionsHandler, self).parsers
return dict(use_scm_version=self._parse_bool, **_orig_parsers)
def parse_section_packages__find(self, section_options):
find_kwargs = super(
ShimConfigOptionsHandler, self
).parse_section_packages__find(section_options)
return stringify_dict_contents(find_kwargs)
setuptools.config.ConfigOptionsHandler = ShimConfigOptionsHandler
else:
"""This is a shim for setuptools<required."""
import functools
import io
import json
import sys
import warnings
try:
import setuptools.config
def filter_out_unknown_section(i):
def chi(self, *args, **kwargs):
i(self, *args, **kwargs)
self.sections = {
s: v for s, v in self.sections.items() if s != "packages.find"
}
return chi
setuptools.config.ConfigHandler.__init__ = filter_out_unknown_section(
setuptools.config.ConfigHandler.__init__
)
except ImportError:
pass
def ignore_unknown_options(s):
@functools.wraps(s)
def sw(**attrs):
try:
ignore_warning_regex = (
r"Unknown distribution option: "
r"'(license_file|project_urls|python_requires)'"
)
warnings.filterwarnings(
"ignore",
message=ignore_warning_regex,
category=UserWarning,
module="distutils.dist",
)
return s(**attrs)
finally:
warnings.resetwarnings()
return sw
def parse_predicates(python_requires):
import itertools
import operator
sorted_operators_map = tuple(
sorted(
{
">": operator.gt,
"<": operator.lt,
">=": operator.ge,
"<=": operator.le,
"==": operator.eq,
"!=": operator.ne,
"": operator.eq,
}.items(),
key=lambda i: len(i[0]),
reverse=True,
)
)
def is_decimal(s):
return type(u"")(s).isdecimal()
conditions = map(str.strip, python_requires.split(","))
for c in conditions:
for op_sign, op_func in sorted_operators_map:
if not c.startswith(op_sign):
continue
raw_ver = itertools.takewhile(
is_decimal, c[len(op_sign) :].strip().split(".")
)
ver = tuple(map(int, raw_ver))
yield op_func, ver
break
def validate_required_python_or_fail(python_requires=None):
if python_requires is None:
return
python_version = sys.version_info
preds = parse_predicates(python_requires)
for op, v in preds:
py_ver_slug = python_version[: max(len(v), 3)]
condition_matches = op(py_ver_slug, v)
if not condition_matches:
raise RuntimeError(
"requires Python '{}' but the running Python is {}".format(
python_requires, ".".join(map(str, python_version[:3]))
)
)
def verify_required_python_runtime(s):
@functools.wraps(s)
def sw(**attrs):
try:
validate_required_python_or_fail(attrs.get("python_requires"))
except RuntimeError as re:
sys.exit("{} {!s}".format(attrs["name"], re))
return s(**attrs)
return sw
setuptools.setup = ignore_unknown_options(setuptools.setup)
setuptools.setup = verify_required_python_runtime(setuptools.setup)
try:
from configparser import ConfigParser, NoSectionError
except ImportError:
from ConfigParser import ConfigParser, NoSectionError
ConfigParser.read_file = ConfigParser.readfp
def maybe_read_files(d):
"""Read files if the string starts with `file:` marker."""
FILE_FUNC_MARKER = "file:"
d = d.strip()
if not d.startswith(FILE_FUNC_MARKER):
return d
descs = []
for fname in map(str.strip, str(d[len(FILE_FUNC_MARKER) :]).split(",")):
with io.open(fname, encoding="utf-8") as f:
descs.append(f.read())
return "".join(descs)
def cfg_val_to_list(v):
"""Turn config val to list and filter out empty lines."""
return list(filter(bool, map(str.strip, str(v).strip().splitlines())))
def cfg_val_to_dict(v):
"""Turn config val to dict and filter out empty lines."""
return dict(
map(
lambda l: list(map(str.strip, l.split("=", 1))),
filter(bool, map(str.strip, str(v).strip().splitlines())),
)
)
def cfg_val_to_primitive(v):
"""Parse primitive config val to appropriate data type."""
return json.loads(v.strip().lower())
def read_configuration(filepath):
"""Read metadata and options from setup.cfg located at filepath."""
cfg = ConfigParser()
with io.open(filepath, encoding="utf-8") as f:
cfg.read_file(f)
md = dict(cfg.items("metadata"))
for list_key in "classifiers", "keywords", "project_urls":
try:
md[list_key] = cfg_val_to_list(md[list_key])
except KeyError:
pass
try:
md["long_description"] = maybe_read_files(md["long_description"])
except KeyError:
pass
opt = dict(cfg.items("options"))
for list_key in "include_package_data", "use_scm_version", "zip_safe":
try:
opt[list_key] = cfg_val_to_primitive(opt[list_key])
except KeyError:
pass
for list_key in "scripts", "install_requires", "setup_requires":
try:
opt[list_key] = cfg_val_to_list(opt[list_key])
except KeyError:
pass
try:
opt["package_dir"] = cfg_val_to_dict(opt["package_dir"])
except KeyError:
pass
try:
opt_package_data = dict(cfg.items("options.package_data"))
if not opt_package_data.get("", "").strip():
opt_package_data[""] = opt_package_data["*"]
del opt_package_data["*"]
except (KeyError, NoSectionError):
opt_package_data = {}
try:
opt_extras_require = dict(cfg.items("options.extras_require"))
opt["extras_require"] = {}
for k, v in opt_extras_require.items():
opt["extras_require"][k] = cfg_val_to_list(v)
except NoSectionError:
pass
opt["package_data"] = {}
for k, v in opt_package_data.items():
opt["package_data"][k] = cfg_val_to_list(v)
try:
opt_exclude_package_data = dict(cfg.items("options.exclude_package_data"))
if (
not opt_exclude_package_data.get("", "").strip()
and "*" in opt_exclude_package_data
):
opt_exclude_package_data[""] = opt_exclude_package_data["*"]
del opt_exclude_package_data["*"]
except NoSectionError:
pass
else:
opt["exclude_package_data"] = {}
for k, v in opt_exclude_package_data.items():
opt["exclude_package_data"][k] = cfg_val_to_list(v)
cur_pkgs = opt.get("packages", "").strip()
if "\n" in cur_pkgs:
opt["packages"] = cfg_val_to_list(opt["packages"])
elif cur_pkgs.startswith("find:"):
opt_packages_find = stringify_dict_contents(
dict(cfg.items("options.packages.find"))
)
opt["packages"] = setuptools.find_packages(**opt_packages_find)
return {"metadata": md, "options": opt}
def cut_local_version_on_upload(version):
"""Generate a PEP440 local version if uploading to PyPI."""
import os
import setuptools_scm.version # only present during setup time
IS_PYPI_UPLOAD = os.getenv("PYPI_UPLOAD") == "true" # set in tox.ini
return (
""
if IS_PYPI_UPLOAD
else setuptools_scm.version.get_local_node_and_date(version)
)
if HAS_DIST_INFO_CMD:
class patched_dist_info(setuptools.command.dist_info.dist_info):
def run(self):
self.egg_base = str_if_nested_or_str(self.egg_base)
return setuptools.command.dist_info.dist_info.run(self)
declarative_setup_params = read_configuration("setup.cfg")
"""Declarative metadata and options as read by setuptools."""
setup_params = {}
"""Explicit metadata for passing into setuptools.setup() call."""
setup_params = dict(setup_params, **declarative_setup_params["metadata"])
setup_params = dict(setup_params, **declarative_setup_params["options"])
if HAS_DIST_INFO_CMD:
setup_params["cmdclass"] = {"dist_info": patched_dist_info}
setup_params["use_scm_version"] = {"local_scheme": cut_local_version_on_upload}
# Patch incorrectly decoded package_dir option
# ``egg_info`` demands native strings failing with unicode under Python 2
# Ref https://github.com/pypa/setuptools/issues/1136
setup_params = stringify_dict_contents(setup_params)
if __name__ == "__main__":
setuptools.setup(**setup_params)

79
tox.ini
View File

@ -1,102 +1,29 @@
[tox] [tox]
minversion = 3.12.0 minversion = 3.14.0
envlist = envlist =
lint lint
check py{37}-ansible{28}-unit
py{27,35,36,37}-ansible{27,28,29}
doc
devel
skipdist = True skipdist = True
skip_missing_interpreters = True skip_missing_interpreters = True
isolated_build = True isolated_build = True
[testenv] [testenv]
# Hotfix for https://github.com/pypa/pip/issues/6434
# Resolved by https://github.com/tox-dev/tox/issues/1276
install_command =
python -c 'import subprocess, sys; pip_inst_cmd = sys.executable, "-m", "pip", "install"; subprocess.check_call(pip_inst_cmd + ("pip<19.1", )); subprocess.check_call(pip_inst_cmd + tuple(sys.argv[1:]))' {opts} {packages}
usedevelop = True usedevelop = True
passenv = * passenv = *
setenv = setenv =
ANSIBLE_CALLABLE_WHITELIST={env:ANSIBLE_CALLABLE_WHITELIST:timer,profile_roles} ANSIBLE_CALLABLE_WHITELIST={env:ANSIBLE_CALLABLE_WHITELIST:timer,profile_roles}
PYTHONDONTWRITEBYTECODE=1 PYTHONDONTWRITEBYTECODE=1
PYTEST_ADDOPTS=molecule_hetznercloud/test/unit/ --cov={toxinidir}/molecule/ --no-cov-on-fail {env:PYTEST_ADDOPTS:-n auto}
deps = deps =
ansible27: ansible>=2.7,<2.8
ansible28: ansible>=2.8,<2.9 ansible28: ansible>=2.8,<2.9
ansible29: ansible>=2.9.0b1,<2.10
devel: ansible>=2.9.0b1
devel: docker
extras = extras =
test test
commands = commands =
pip check
devel: pip install -e "git+https://github.com/ansible/molecule.git#egg=molecule"
python -m pytest {posargs} python -m pytest {posargs}
whitelist_externals =
find
sh
[testenv:lint] [testenv:lint]
commands = commands =
# to run a single linter you can do "pre-commit run flake8"
python -m pre_commit run {posargs:--all} python -m pre_commit run {posargs:--all}
deps = pre-commit>=1.18.1 deps = pre-commit>=1.18.1
extras =
skip_install = true skip_install = true
usedevelop = false usedevelop = false
[testenv:check]
envdir = {toxworkdir}/py36-ansible28
deps =
{[testenv]deps}
collective.checkdocs==0.2
twine==1.14.0
usedevelop = False
commands =
python -m pytest --collect-only
python -m setup checkdocs check --metadata --restructuredtext --strict --verbose
python -m twine check .tox/dist/*
[testenv:build-docker]
platform = ^darwin|^linux
usedevelop = False
skip_install = True
deps =
setuptools_scm==3.3.3
packaging # pyup: ignore
commands_pre =
commands =
python ./utils/build-docker.py
whitelist_externals =
sh
[testenv:build-dists-local]
description =
Generate dists which may be not ready
for upload to PyPI because of
containing PEP440 local version part
usedevelop = false
skip_install = true
deps =
pep517 >= 0.5.0
setenv =
commands =
python -m pep517.build \
--source \
--binary \
--out-dir {toxinidir}/dist/ \
{toxinidir}
[testenv:build-dists]
description = Generate dists ready for upload to PyPI
usedevelop = {[testenv:build-dists-local]usedevelop}
skip_install = {[testenv:build-dists-local]skip_install}
deps = {[testenv:build-dists-local]deps}
setenv =
PYPI_UPLOAD = true
commands =
rm -rfv {toxinidir}/dist/
{[testenv:build-dists-local]commands}
whitelist_externals =
rm
{[testenv]whitelist_externals}