diff --git a/lib/main/models/__init__.py b/lib/main/models/__init__.py index cfc2a15dc4..de08b3881d 100644 --- a/lib/main/models/__init__.py +++ b/lib/main/models/__init__.py @@ -920,24 +920,6 @@ class JobTemplate(CommonModel): job.save() return job - # project has one default playbook but really should have a list of playbooks and flags ... - # ssh-agent bash - # ssh-add ... < key entry - # - # playbook in source control is already on the disk - - # we'll extend ansible core to have callback context like - # self.context.playbook - # self.context.runner - # and the callback will read the environment for ACOM_CELERY_JOB_ID or similar - # and log tons into the database - - # the ansible commander setup instructions will include installing the database logging callback - # inventory script is going to need some way to load Django models - # it is documented on ansible.cc under API docs and takes two parameters - # --list - # -- host - def get_absolute_url(self): import lib.urls return reverse(lib.urls.views_JobTemplateDetail, args=(self.pk,)) diff --git a/lib/main/tasks.py b/lib/main/tasks.py index fdcc4287b4..1e2e0d95cb 100644 --- a/lib/main/tasks.py +++ b/lib/main/tasks.py @@ -19,6 +19,7 @@ import logging import os import select import subprocess +import tempfile import time import traceback from celery import Task @@ -61,6 +62,7 @@ class RunJob(Task): ''' creds = job.credential if creds and creds.ssh_key_data: + # FIXME: File permissions? handle, path = tempfile.mkstemp() f = os.fdopen(handle, 'w') f.write(creds.ssh_key_data) @@ -140,11 +142,10 @@ class RunJob(Task): args.append(job.playbook) # relative path to project.local_path ssh_key_path = kwargs.get('ssh_key_path', '') if ssh_key_path: - cmd = '; '.join([subprocess.list2cmdline(['ssh-add', ssh_key_path]), - subprocess.list2cmdline(args)]) - return ['ssh-agent', 'sh', '-c', cmd] - else: - return args + cmd = ' '.join([subprocess.list2cmdline(['ssh-add', ssh_key_path]), + '&&', subprocess.list2cmdline(args)]) + args = ['ssh-agent', 'sh', '-c', cmd] + return args def capture_subprocess_output(self, proc, timeout=1.0): ''' @@ -223,7 +224,6 @@ class RunJob(Task): status, stdout, stderr = 'error', '', '' logfile = cStringIO.StringIO() logfile_pos = logfile.tell() - print 'ARGS:', repr(args) child = pexpect.spawn(args[0], args[1:], cwd=cwd, env=env) child.logfile_read = logfile job_canceled = False diff --git a/lib/main/tests/commands.py b/lib/main/tests/commands.py index a67d8d4d05..12cdf1c1b0 100644 --- a/lib/main/tests/commands.py +++ b/lib/main/tests/commands.py @@ -19,13 +19,15 @@ import os import StringIO import sys import tempfile +from django.conf import settings from django.core.management import call_command from django.core.management.base import CommandError from django.utils.timezone import now from lib.main.models import * from lib.main.tests.base import BaseTest -__all__ = ['AcomInventoryTest', 'AcomCallbackEventTest'] +__all__ = ['RunCommandAsScriptTest', 'AcomInventoryTest', + 'AcomCallbackEventTest'] class BaseCommandTest(BaseTest): ''' @@ -34,11 +36,13 @@ class BaseCommandTest(BaseTest): def setUp(self): super(BaseCommandTest, self).setUp() + self._sys_path = [x for x in sys.path] self._environ = dict(os.environ.items()) self._temp_files = [] def tearDown(self): super(BaseCommandTest, self).tearDown() + sys.path = self._sys_path for k,v in self._environ.items(): if os.environ.get(k, None) != v: os.environ[k] = v @@ -54,6 +58,7 @@ class BaseCommandTest(BaseTest): Run a management command and capture its stdout/stderr along with any exceptions. ''' + command_runner = options.pop('command_runner', call_command) stdin_fileobj = options.pop('stdin_fileobj', None) options.setdefault('verbosity', 1) options.setdefault('interactive', False) @@ -66,7 +71,7 @@ class BaseCommandTest(BaseTest): sys.stderr = StringIO.StringIO() result = None try: - result = call_command(name, *args, **options) + result = command_runner(name, *args, **options) except Exception, e: result = e except SystemExit, e: @@ -79,6 +84,23 @@ class BaseCommandTest(BaseTest): sys.stderr = original_stderr return result, captured_stdout, captured_stderr +class RunCommandAsScriptTest(BaseCommandTest): + ''' + Test helper to run management command as standalone script. + ''' + + def test_run_command_as_script(self): + from lib.main.management.commands import run_command_as_script + os.environ['ACOM_TEST_DATABASE_NAME'] = settings.DATABASES['default']['NAME'] + # FIXME: Not sure how to test ImportError for settings module. + def run_cmd(name, *args, **kwargs): + return run_command_as_script(name) + result, stdout, stderr = self.run_command('version', + command_runner=run_cmd) + self.assertEqual(result, None) + self.assertTrue(stdout) + self.assertFalse(stderr) + class AcomInventoryTest(BaseCommandTest): ''' Test cases for acom_inventory management command. diff --git a/lib/main/tests/tasks.py b/lib/main/tests/tasks.py index 55c5c4fdd4..e79d50b371 100644 --- a/lib/main/tests/tasks.py +++ b/lib/main/tests/tasks.py @@ -22,6 +22,7 @@ from django.conf import settings from django.test.utils import override_settings from lib.main.models import * from lib.main.tests.base import BaseTransactionTest +from lib.main.tasks import RunJob TEST_PLAYBOOK = '''- hosts: test-group gather_facts: False @@ -39,6 +40,68 @@ TEST_PLAYBOOK2 = '''- hosts: test-group command: test 1 = 0 ''' +TEST_SSH_KEY_DATA = '''-----BEGIN RSA PRIVATE KEY----- +MIIEpQIBAAKCAQEAyQ8F5bbgjHvk4SZJsKI9OmJKMFxZqRhvx4LaqjLTKbBwRBsY +1/C00NPiZn70dKbeyV7RNVZxuzM6yd3D3lwTdbDu/eJ0x72t3ch+TdLt/aenyy10 +IvZyhSlxCLDkDaVVPFYJOQzVS8TkdOi6ZHc+R0c0A+4ZE8OQ8C0zIKtUTHqRk4/v +gYK5guhNS0DdgWkBj6K+r/9D4bqdPTJPt4S7H75vb1tBgseiqftEkLYOhTK2gsCi +5uJgpG4zPQY4Kk/97dbW7pwcvPkr1rKkAwEJ27Bfo+DBv3oEx3SinpXQtOrH1aEO +RHSXldBaymdBtVLUhjxDlnnQ7Ps+fNX04R7N4QIDAQABAoIBAQClEDxbNyRqsVxa +q8BbzxZNVFxsD6Vceb9rIDa8/DT4SO4iO8zNm8QWnZ2FYDz5d/X3hGxlSa7dbVWa +XQJtD1K6kKPks4IEaejP58Ypxj20vWu4Fnz+Jy4lvLwb0n2n5lBv1IKF389NATw9 +7sL3sB3lDsPZZiQYYbogNDuBWqc+kP0zD84bONsM/B2HMRm9BRv2UsZf+zKU4pTA +UqHffyjmw7LqHmbtVjwVcUsC+xcE4kCuWLvabFnTWOSnWECyIw2+trxKdwCXbfzG +s5rn4Dj+aEKimzFaRpTSVx6w4yw9xw/EjsSaZ88jKSpTP8ocCut6zv+P/JwlukEX +4A4FxqyxAoGBAOp3G9EIAAWijcIgO5OdiZNEqVyqd3yyPzT6d/q7bf4dpVCZiLNA +bRmge83aMc4g2Dpkn/++It3bDmnXXGg+BZSX5KT9JLklXchaw9phv9J0diZEUvYS +mSQafbUGIqYnYzns3TU0cbgITs1iVIEstHYjGr3J88nDG+HFCHboxa93AoGBANuG +cDFgyvm79+haK2fHhUCZgaFFYBpkpuz+zjDjzIytOzymWa2gD9jIa7mvdvoH2ge3 +AVG0vy+n9cJaqJMuLkhdI01wVlqY9wvDHFyZCXyIvKVPMljKeTvCNGCupsG4R171 +gSKT5ryOx58MGbE7knAZC+QWpwxFpdpbfej6g7NnAoGBAMz6ipAJbXN/tG0FnvAj +pxXfzizcPw/+CTI40tGaMMQbiN5ZC+CiL39bBUFnQ2mQ31jVheegg3zvuL8hb4EW +z+wjitoPEZ7nowC5EUaHdJr6BBzaWKkWg1nD6yhqj7ow7xfCE3YjPlQEt1fpYjV4 +LuClOgi4WPCIKYUMq6TBRaprAoGAVrEjs0xPPApQH5EkXQp9BALbH23/Qs0G4sbJ +dKMxT0jGAPCMr7VrLKgRarXxXVImdy99NOAVNGO2+PbGZcEyA9/MJjO71nFb9mgp +1iOVjHmPThUVg90JvWC3QIsYTZ5RiR2Yzqfr0gDsslGb/9LPxLcPbBbKB12l3rKM +6amswvcCgYEAvgcSlTfAkI3ac8rB70HuDmSdqKblIiQjtPtT/ixXaFkZOmHRr4AE +KepMRDnaO/ldPDPEWCGqPzEM0t/0jS8/hCu3zLHHpZ+0LnHq+EXkOI0/GB4P+z5l +Vz3kouC0BTav0rCEnDop/cWMTiAp/XhKXfrTTTOra/F8l2xD8n/mnzY= +-----END RSA PRIVATE KEY-----''' + +TEST_SSH_KEY_DATA_LOCKED = '''-----BEGIN RSA PRIVATE KEY----- +Proc-Type: 4,ENCRYPTED +DEK-Info: AES-128-CBC,6B4E92AF4C29DE26FD8535D81825BDE6 + +pg8YplxPpfzgEGUiko34DGaYklyGyYKXjOrGFGyLoquNAVNFyewT34dDrZi0IAaE +79wMVcdlHbrJfZz8ML8I/ft6zM6BdlwZExH4y9DRAaktY3yIXxSvowBQ6ljh3wUy +M6m0afOfVjT22V8hLFgX0yTQ6P9zTG1cmj6+JQWTsMJ5EP3rnFK5CyrJXP48B3GI +GgE66rkXDvcKlVeIrbrpcTyfmEpafPgVRJYCDFXxeO/BfKgUFVxFq1PgFbvGQMmD +wA6EsyRrN+aoub1sqzj8tM8e4nwEi0EifdRShkFeqH4GUOKypanTXfCqwFBgYi5a +i3YwSnniZZPwCniGR5cl8oetrc5dubq/IR0txsGi2lO6zJEWdSer/EadS0QAll4S +yXrSc/lFaez1VmVe/8aoBKDOHhe7jV3YXAuqCeB4o/SThB/9Gad44MTbqFH3d7cD +k+F0Cjup7LZqZpXeB7ZHRG/Yt9MtBzwDVmEWaxA1WIN5a8xyZEVzRswSi4lZX69z +Va7eTKcrCbHOQmIbLZGRiZbAbfgriwwxQCJWELv80h+A754Bhi23n3WzcT094fRi +cqK//HcHHXxYGmrfUbHYcj+GCQ07Uk2ZR3qglmPISUCgfZwM9k0LpXudWE8vmF2S +pAnbgxgrfUMtpu5EAO+d8Sn5wQLVD7YzPBUhM4PYfYUbJnRoZQryuR4lqCzcg0te +BM8x1LzSXyBEbQaonuMzSz1hCQ9hZpUwUEqDWAT3cPNmgyWkXQ1P8ehJhTmryGJw +/GHxNzMZDGj+bBKo7ic3r1g3ZmmlSU1EVxMLvRBKhdc1XicBVqepDma6/LEpj+5X +oplR+3Q0QSQ8CchcSxYtOpI3UBCatpyu09GtfzS+7bI5I7FVYUccR83+oQlKpPHC +5O2irB8JeXqAY679fx2N4i0E6l5Xr5AjUtOBCNil0Y70eOf9ER6i7kGakR7bUtk5 +fQn8Em9pLsYYalnekn4sxyHpGq59KgNPjQiJRByYidSJ/oyNbmtPlxfXLwpuicd2 +8HLm1e0UeGidfF/bSlySwDzy1ZlSr/Apdcn9ou5hfhaGuQvjr9SvJwxQFNRMPdHj +ukBSDGuxyyU+qBrWJhFsymiZAWDofY/4GzgMu4hh0PwN5arzoTxnLHmc/VFttyMx +nP7bTaa9Sr54TlMr7NuKTzz5biXKjqJ9AZKIUF2+ERebjV0hMpJ5NPsLwPUnA9kx +R3tl1JL2Ia82ovS81Ghff/cBZsx/+LQYa+ac4eDTyXxyg4ei5tPwOlzz7pDKJAr9 +XEh2X6rywCNghEMZPaOQLiEDLJ2is6P4OarSa/yoU4OMetpFfwZ0oJSCmGlEa+CF +zeJ80yXhU1Ru2eqiUjCAUg25BFPwoiMJDc6jWWow7OrXCQsw7Ddo2ncy1p9QeWjM +2R4ojPHWuXKYxvwVSc8NZHASlycBCaxHLDAEyH4avOSDPWOB1H5t+RrNmo0qgush +0aRo6F7BjzB2rA4E+xu2u11TBfF8iB3PC919/vxnkXF97NqezsaCz6VbRlsU0A+B +wwoi+P4JlJF6ZuhuDv6mhmBCSdXdc1bvimvdpOljhThr+cG5mM08iqWGKdA665cw +-----END RSA PRIVATE KEY----- +''' + +TEST_SSH_KEY_DATA_UNLOCK = 'unlockme' + @override_settings(CELERY_ALWAYS_EAGER=True, CELERY_EAGER_PROPAGATES_EXCEPTIONS=True) class BaseCeleryTest(BaseTransactionTest): @@ -49,7 +112,7 @@ class BaseCeleryTest(BaseTransactionTest): @override_settings(ANSIBLE_TRANSPORT='local') class RunJobTest(BaseCeleryTest): ''' - Test cases for run_job celery task. + Test cases for RunJob celery task. ''' def setUp(self): @@ -65,14 +128,25 @@ class RunJobTest(BaseCeleryTest): self.group = self.inventory.groups.create(name='test-group', inventory=self.inventory) self.group.hosts.add(self.host) + self.project = None + self.credential = None # Pass test database name in environment for use by the inventory script. os.environ['ACOM_TEST_DATABASE_NAME'] = settings.DATABASES['default']['NAME'] + # Monkeypatch RunJob to capture list of command line arguments. + self.original_build_args = RunJob.build_args + self.run_job_args = None + def new_build_args(_self, job, **kw): + args = self.original_build_args(_self, job, **kw) + self.run_job_args = args + return args + RunJob.build_args = new_build_args def tearDown(self): super(RunJobTest, self).tearDown() os.environ.pop('ACOM_TEST_DATABASE_NAME', None) if self.test_project_path: shutil.rmtree(self.test_project_path, True) + RunJob.build_args = self.original_build_args def create_test_credential(self, **kwargs): opts = { @@ -93,28 +167,54 @@ class RunJobTest(BaseCeleryTest): self.project = self.make_projects(self.normal_django_user, 1, playbook_content)[0] self.organization.projects.add(self.project) - def create_test_job(self, **kwargs): + def create_test_job_template(self, **kwargs): opts = { 'name': 'test-job-template', 'inventory': self.inventory, 'project': self.project, - 'playbook': self.project.available_playbooks[0], + 'credential': self.credential, } + try: + opts['playbook'] = self.project.available_playbooks[0] + except (AttributeError, IndexError): + pass opts.update(kwargs) self.job_template = JobTemplate.objects.create(**opts) - return self.job_template.create_job() + return self.job_template + + def create_test_job(self, **kwargs): + job_template = kwargs.pop('job_template', None) + if job_template: + self.job = job_template.create_job(**kwargs) + else: + opts = { + 'name': 'test-job', + 'inventory': self.inventory, + 'project': self.project, + 'credential': self.credential, + } + try: + opts['playbook'] = self.project.available_playbooks[0] + except (AttributeError, IndexError): + pass + opts.update(kwargs) + self.job = Job.objects.create(**opts) + return self.job def test_run_job(self): self.create_test_project(TEST_PLAYBOOK) - job = self.create_test_job() + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template) self.assertEqual(job.status, 'new') - job.start() + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) self.assertEqual(job.status, 'pending') job = Job.objects.get(pk=job.pk) #print 'stdout:', job.result_stdout #print 'stderr:', job.result_stderr #print job.status #print settings.DATABASES + #print self.run_job_args self.assertEqual(job.status, 'successful') self.assertTrue(job.result_stdout) job_events = job.job_events.all() @@ -134,9 +234,11 @@ class RunJobTest(BaseCeleryTest): def test_check_job(self): self.create_test_project(TEST_PLAYBOOK) - job = self.create_test_job(job_type='check') + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template, job_type='check') self.assertEqual(job.status, 'new') - job.start() + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) self.assertEqual(job.status, 'pending') job = Job.objects.get(pk=job.pk) self.assertEqual(job.status, 'successful') @@ -158,9 +260,11 @@ class RunJobTest(BaseCeleryTest): def test_run_job_that_fails(self): self.create_test_project(TEST_PLAYBOOK2) - job = self.create_test_job() + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template) self.assertEqual(job.status, 'new') - job.start() + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) self.assertEqual(job.status, 'pending') job = Job.objects.get(pk=job.pk) self.assertEqual(job.status, 'failed') @@ -181,9 +285,11 @@ class RunJobTest(BaseCeleryTest): def test_check_job_where_task_would_fail(self): self.create_test_project(TEST_PLAYBOOK2) - job = self.create_test_job(job_type='check') + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template, job_type='check') self.assertEqual(job.status, 'new') - job.start() + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) self.assertEqual(job.status, 'pending') job = Job.objects.get(pk=job.pk) # Since we don't actually run the task, the --check should indicate @@ -203,3 +309,175 @@ class RunJobTest(BaseCeleryTest): self.assertEqual(job.unreachable_hosts.count(), 0) self.assertEqual(job.skipped_hosts.count(), 1) self.assertEqual(job.processed_hosts.count(), 1) + + def test_cancel_job(self): + self.create_test_project(TEST_PLAYBOOK) + job_template = self.create_test_job_template() + # The cancel_flag isn't checked until after the job is started, so + # setting it here will allow the job to start, then interrupt it. + job = self.create_test_job(job_template=job_template, cancel_flag=True) + self.assertEqual(job.status, 'new') + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) + self.assertEqual(job.status, 'pending') + job = Job.objects.get(pk=job.pk) + self.assertEqual(job.status, 'canceled') + + def test_extra_job_options(self): + self.create_test_project(TEST_PLAYBOOK) + job_template = self.create_test_job_template(use_sudo=True, forks=3, + verbosity=2, + extra_vars={'foo': 1}) + job = self.create_test_job(job_template=job_template) + self.assertEqual(job.status, 'new') + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) + self.assertEqual(job.status, 'pending') + job = Job.objects.get(pk=job.pk) + # Job may fail if current user doesn't have password-less sudo + # privileges, but we're mainly checking the command line arguments. + self.assertTrue(job.status in ('successful', 'failed')) + self.assertTrue(job.result_stdout) + self.assertTrue('--sudo' in self.run_job_args) + self.assertTrue('--forks=3' in self.run_job_args) + self.assertTrue('-vv' in self.run_job_args) + self.assertTrue('--extra-vars=foo=1' in self.run_job_args) + + def test_limit_option(self): + self.create_test_project(TEST_PLAYBOOK) + job_template = self.create_test_job_template(limit='bad.example.com') + job = self.create_test_job(job_template=job_template) + self.assertEqual(job.status, 'new') + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) + self.assertEqual(job.status, 'pending') + job = Job.objects.get(pk=job.pk) + self.assertEqual(job.status, 'failed') + self.assertTrue('--limit=bad.example.com' in self.run_job_args) + + def test_ssh_username_and_password(self): + self.create_test_credential(ssh_username='sshuser', + ssh_password='sshpass') + self.create_test_project(TEST_PLAYBOOK) + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template) + self.assertEqual(job.status, 'new') + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) + self.assertEqual(job.status, 'pending') + job = Job.objects.get(pk=job.pk) + self.assertEqual(job.status, 'successful') + self.assertTrue('--user=sshuser' in self.run_job_args) + self.assertTrue('--ask-pass' in self.run_job_args) + + def test_ssh_ask_password(self): + self.create_test_credential(ssh_password='ASK') + self.create_test_project(TEST_PLAYBOOK) + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template) + self.assertEqual(job.status, 'new') + self.assertTrue(job.get_passwords_needed_to_start()) + self.assertTrue('ssh_password' in job.get_passwords_needed_to_start()) + self.assertFalse(job.start()) + self.assertEqual(job.status, 'new') + self.assertTrue(job.start(ssh_password='sshpass')) + self.assertEqual(job.status, 'pending') + job = Job.objects.get(pk=job.pk) + self.assertEqual(job.status, 'successful') + self.assertTrue('--ask-pass' in self.run_job_args) + + def test_sudo_username_and_password(self): + self.create_test_credential(sudo_username='sudouser', + sudo_password='sudopass') + self.create_test_project(TEST_PLAYBOOK) + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template) + self.assertEqual(job.status, 'new') + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) + self.assertEqual(job.status, 'pending') + job = Job.objects.get(pk=job.pk) + # Job may fail if current user doesn't have password-less sudo + # privileges, but we're mainly checking the command line arguments. + self.assertTrue(job.status in ('successful', 'failed')) + self.assertTrue('--sudo-user=sudouser' in self.run_job_args) + self.assertTrue('--ask-sudo-pass' in self.run_job_args) + + def test_sudo_ask_password(self): + self.create_test_credential(sudo_password='ASK') + self.create_test_project(TEST_PLAYBOOK) + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template) + self.assertEqual(job.status, 'new') + self.assertTrue(job.get_passwords_needed_to_start()) + self.assertTrue('sudo_password' in job.get_passwords_needed_to_start()) + self.assertFalse(job.start()) + self.assertEqual(job.status, 'new') + self.assertTrue(job.start(sudo_password='sudopass')) + self.assertEqual(job.status, 'pending') + job = Job.objects.get(pk=job.pk) + # Job may fail if current user doesn't have password-less sudo + # privileges, but we're mainly checking the command line arguments. + self.assertTrue(job.status in ('successful', 'failed')) + self.assertTrue('--ask-sudo-pass' in self.run_job_args) + + def test_unlocked_ssh_key(self): + self.create_test_credential(ssh_key_data=TEST_SSH_KEY_DATA) + self.create_test_project(TEST_PLAYBOOK) + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template) + self.assertEqual(job.status, 'new') + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) + self.assertEqual(job.status, 'pending') + job = Job.objects.get(pk=job.pk) + self.assertEqual(job.status, 'successful') + self.assertTrue('ssh-agent' in self.run_job_args) + + def test_locked_ssh_key_with_password(self): + self.create_test_credential(ssh_key_data=TEST_SSH_KEY_DATA_LOCKED, + ssh_key_unlock=TEST_SSH_KEY_DATA_UNLOCK) + self.create_test_project(TEST_PLAYBOOK) + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template) + self.assertEqual(job.status, 'new') + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) + self.assertEqual(job.status, 'pending') + job = Job.objects.get(pk=job.pk) + self.assertEqual(job.status, 'successful') + self.assertTrue('ssh-agent' in self.run_job_args) + self.assertTrue('Bad passphrase' not in job.result_stdout) + + def test_locked_ssh_key_with_bad_password(self): + self.create_test_credential(ssh_key_data=TEST_SSH_KEY_DATA_LOCKED, + ssh_key_unlock='not the passphrase') + self.create_test_project(TEST_PLAYBOOK) + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template) + self.assertEqual(job.status, 'new') + self.assertFalse(job.get_passwords_needed_to_start()) + self.assertTrue(job.start()) + self.assertEqual(job.status, 'pending') + job = Job.objects.get(pk=job.pk) + self.assertEqual(job.status, 'failed') + self.assertTrue('ssh-agent' in self.run_job_args) + self.assertTrue('Bad passphrase' in job.result_stdout) + + def test_locked_ssh_key_ask_password(self): + self.create_test_credential(ssh_key_data=TEST_SSH_KEY_DATA_LOCKED, + ssh_key_unlock='ASK') + self.create_test_project(TEST_PLAYBOOK) + job_template = self.create_test_job_template() + job = self.create_test_job(job_template=job_template) + self.assertEqual(job.status, 'new') + self.assertTrue(job.get_passwords_needed_to_start()) + self.assertTrue('ssh_key_unlock' in job.get_passwords_needed_to_start()) + self.assertFalse(job.start()) + self.assertEqual(job.status, 'new') + self.assertTrue(job.start(ssh_key_unlock=TEST_SSH_KEY_DATA_UNLOCK)) + self.assertEqual(job.status, 'pending') + job = Job.objects.get(pk=job.pk) + self.assertEqual(job.status, 'successful') + self.assertTrue('ssh-agent' in self.run_job_args) + self.assertTrue('Bad passphrase' not in job.result_stdout)