# -*- coding: utf-8 -*- from __future__ import absolute_import, unicode_literals, print_function # Import python libraries import logging import os.path import shutil import sys import tempfile import textwrap # Import Salt Libraries import salt.utils.yaml as yaml # Import pytest libraries import pytest from pytestskipmarkers.utils import ports from saltfactories.utils import random_string, running_username # Import Pepper libraries import pepper import pepper.script log = logging.getLogger(__name__) @pytest.fixture(scope='session') def sshd_config_dir(salt_factories): config_dir = salt_factories.get_root_dir_for_daemon("sshd") yield config_dir shutil.rmtree(str(config_dir), ignore_errors=True) @pytest.fixture(scope='session') def session_sshd_server(salt_factories, sshd_config_dir, session_master): sshd_config_dict = { "Protocol": "2", # Turn strict modes off so that we can operate in /tmp "StrictModes": "no", # Logging "SyslogFacility": "AUTH", "LogLevel": "INFO", # Authentication: "LoginGraceTime": "120", "PermitRootLogin": "without-password", "PubkeyAuthentication": "yes", # Don't read the user's ~/.rhosts and ~/.shosts files "IgnoreRhosts": "yes", "HostbasedAuthentication": "no", # To enable empty passwords, change to yes (NOT RECOMMENDED) "PermitEmptyPasswords": "no", # Change to yes to enable challenge-response passwords (beware issues with # some PAM modules and threads) "ChallengeResponseAuthentication": "no", # Change to no to disable tunnelled clear text passwords "PasswordAuthentication": "no", "X11Forwarding": "no", "X11DisplayOffset": "10", "PrintMotd": "no", "PrintLastLog": "yes", "TCPKeepAlive": "yes", "AcceptEnv": "LANG LC_*", "UsePAM": "yes", } factory = salt_factories.get_sshd_daemon( sshd_config_dict=sshd_config_dict, config_dir=sshd_config_dir, ) with factory.started(): yield factory @pytest.fixture(scope='session') def session_ssh_roster_config(session_sshd_server, session_master): roster_contents = """ localhost: host: 127.0.0.1 port: {} user: {} priv: {} mine_functions: test.arg: ['itworked'] """.format( session_sshd_server.listen_port, running_username(), session_sshd_server.client_key ) with pytest.helpers.temp_file( "roster", roster_contents, session_master.config_dir ) as roster_file: yield roster_file @pytest.fixture(scope='session') def salt_api_port(): ''' Returns an unused localhost port for the api port ''' return ports.get_unused_localhost_port() @pytest.fixture(scope='session') def pepperconfig(salt_api_port): config = textwrap.dedent(''' [main] SALTAPI_URL=http://localhost:{0}/ SALTAPI_USER=pepper SALTAPI_PASS=pepper SALTAPI_EAUTH=sharedsecret [pepper] SALTAPI_URL=http://localhost:{0}/ SALTAPI_USER=pepper SALTAPI_PASS=pepper SALTAPI_EAUTH=sharedsecret [baduser] SALTAPI_URL=http://localhost:{0}/ SALTAPI_USER=saltdev SALTAPI_PASS=saltdev SALTAPI_EAUTH=pam [badapi] SALTAPI_URL=git://localhost:{0}/ [noapi] SALTAPI_USER=pepper SALTAPI_PASS=pepper SALTAPI_EAUTH=sharedsecret [noopts] SALTAPI_URL=http://localhost:{0}/ '''.format(salt_api_port)) with open('tests/.pepperrc', 'w') as pepper_file: print(config, file=pepper_file) yield os.remove('tests/.pepperrc') @pytest.fixture def pepper_client(session_salt_api, salt_api_port): client = pepper.Pepper('http://localhost:{0}'.format(salt_api_port)) client.login('pepper', 'pepper', 'sharedsecret') return client @pytest.fixture def tokfile(): tokdir = tempfile.mkdtemp() yield os.path.join(tokdir, 'peppertok.json') shutil.rmtree(tokdir) @pytest.fixture def output_file(): ''' Returns the path to the salt master configuration file ''' out_dir = tempfile.mkdtemp() yield os.path.join(out_dir, 'output') shutil.rmtree(out_dir) @pytest.fixture(params=['/run', '/login']) def pepper_cli(request, session_salt_api, salt_api_port, output_file, session_sshd_server): ''' Wrapper to invoke Pepper with common params and inside an empty env ''' if request.config.getoption('--salt-api-backend') == 'rest_tornado' and request.param == '/run': pytest.xfail("rest_tornado does not support /run endpoint until next release") def_args = [ '--out=json', '--output-file={0}'.format(output_file), '-c', 'tests/.pepperrc', ] if request.param == '/run': def_args = ['--run-uri'] + def_args def _run_pepper_cli(*args, **kwargs): sys.argv = ['pepper', '-p', kwargs.pop('profile', 'main')] + def_args + list(args) exitcode = pepper.script.Pepper()() try: with open(output_file, 'r') as result: try: return yaml.load(result) except yaml.parser.ParserError: result.seek(0) return [yaml.load('{0}}}'.format(ret).strip('"')) for ret in result.read().split('}"\n') if ret] except Exception as exc: log.error('ExitCode %s: %s', exitcode, exc) return exitcode return _run_pepper_cli @pytest.fixture(scope='session') def session_master_factory(request, salt_factories, session_master_config_overrides): return salt_factories.salt_master_daemon( random_string("master-"), overrides=session_master_config_overrides ) @pytest.fixture(scope='session') def session_master(session_master_factory): with session_master_factory.started(): yield session_master_factory @pytest.fixture(scope='session') def session_master_config_overrides(request, salt_api_port, salt_api_backend): return { salt_api_backend: { 'port': salt_api_port, 'disable_ssl': True, }, 'external_auth': { 'sharedsecret': { 'pepper': [ '.*', '@jobs', '@wheel', '@runner', ], }, }, 'sharedsecret': 'pepper', 'token_expire': 94670856, 'ignore_host_keys': True, 'ssh_wipe': True, 'netapi_enable_clients': [ 'local', 'local_async', 'local_subset', 'ssh', 'runner', 'runner_async', 'wheel', 'wheel_async', 'run' ] } @pytest.fixture(scope='session') def session_minion_factory(session_master_factory): """Return a factory for a randomly named minion connected to master.""" minion_factory = session_master_factory.salt_minion_daemon(random_string("minion-")) minion_factory.after_terminate( pytest.helpers.remove_stale_minion_key, session_master_factory, minion_factory.id ) return minion_factory @pytest.fixture(scope='session') def session_minion(session_master, session_minion_factory): # noqa assert session_master.is_running() with session_minion_factory.started(): yield session_minion_factory @pytest.fixture(scope='session') def session_minion_id(session_minion): return session_minion.id @pytest.fixture(scope='session') def salt_api_backend(request): ''' Return the salt-api backend (cherrypy or tornado) ''' backend = request.config.getoption('--salt-api-backend') if backend is not None: return backend backend = request.config.getini('salt_api_backend') if backend is not None: return backend return 'rest_cherrypy' @pytest.fixture(scope='session') def session_salt_api_factory(session_master_factory): return session_master_factory.salt_api_daemon() @pytest.fixture(scope='session') def session_salt_api(session_master, session_salt_api_factory): assert session_master.is_running() with session_salt_api_factory.started(): yield session_salt_api_factory def pytest_addoption(parser): parser.addoption( '--salt-api-backend', action='store', default='rest_cherrypy', help='which backend to use for salt-api, must be one of rest_cherrypy or rest_tornado', )