mirror of
https://github.com/ansible/awx.git
synced 2026-01-14 11:20:39 -03:30
Merge pull request #1878 from wwitzel3/test-refactors
Refactoring /old tests to use py.test
This commit is contained in:
commit
ba1e4cb19a
75
awx/main/tests/functional/commands/test_commands.py
Normal file
75
awx/main/tests/functional/commands/test_commands.py
Normal file
@ -0,0 +1,75 @@
|
||||
import sys
|
||||
import pytest
|
||||
|
||||
try:
|
||||
from cStringIO import StringIO
|
||||
except ImportError:
|
||||
from StringIO import StringIO
|
||||
|
||||
from django.core.management import call_command
|
||||
|
||||
from awx.main.models import Instance
|
||||
|
||||
from awx.main.management.commands.update_password import UpdatePassword
|
||||
from awx.main.management.commands.remove_instance import Command as RemoveInstance
|
||||
|
||||
def run_command(name, *args, **options):
|
||||
command_runner = options.pop('command_runner', call_command)
|
||||
stdin_fileobj = options.pop('stdin_fileobj', None)
|
||||
options.setdefault('verbosity', 1)
|
||||
options.setdefault('interactive', False)
|
||||
original_stdin = sys.stdin
|
||||
original_stdout = sys.stdout
|
||||
original_stderr = sys.stderr
|
||||
if stdin_fileobj:
|
||||
sys.stdin = stdin_fileobj
|
||||
sys.stdout = StringIO()
|
||||
sys.stderr = StringIO()
|
||||
result = None
|
||||
try:
|
||||
result = command_runner(name, *args, **options)
|
||||
except Exception as e:
|
||||
result = e
|
||||
finally:
|
||||
captured_stdout = sys.stdout.getvalue()
|
||||
captured_stderr = sys.stderr.getvalue()
|
||||
sys.stdin = original_stdin
|
||||
sys.stdout = original_stdout
|
||||
sys.stderr = original_stderr
|
||||
return result, captured_stdout, captured_stderr
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"username,password,expected,changed", [
|
||||
('admin', 'dingleberry', 'Password updated\n', True),
|
||||
('admin', 'admin', 'Password not updated\n', False),
|
||||
(None, 'foo', 'username required', False),
|
||||
('admin', None, 'password required', False),
|
||||
]
|
||||
)
|
||||
def test_update_password_command(mocker, username, password, expected, changed):
|
||||
with mocker.patch.object(UpdatePassword, 'update_password', return_value=changed):
|
||||
result, stdout, stderr = run_command('update_password', username=username, password=password)
|
||||
if result is None:
|
||||
assert stdout == expected
|
||||
else:
|
||||
assert str(result) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"primary,hostname,startswith,exception", [
|
||||
(True, "127.0.0.1", "Cannot remove primary", None),
|
||||
(False, "127.0.0.2", "Successfully removed", None),
|
||||
(False, "127.0.0.3", "No matching instance", Instance.DoesNotExist),
|
||||
]
|
||||
)
|
||||
def test_remove_instance_command(mocker, primary, hostname, startswith, exception):
|
||||
mock_instance = mocker.MagicMock(primary=primary, enforce_unique_find=True)
|
||||
with mocker.patch.object(Instance.objects, 'get', return_value=mock_instance, side_effect=exception):
|
||||
with mocker.patch.object(RemoveInstance, 'include_option_hostname_uuid_find'):
|
||||
with mocker.patch.object(RemoveInstance, 'get_unique_fields', return_value={'hostname':hostname, 'uuid':1}):
|
||||
result, stdout, stderr = run_command("remove_instance", hostname=hostname)
|
||||
if result is None:
|
||||
assert stdout.startswith(startswith)
|
||||
else:
|
||||
assert str(result).startswith(startswith)
|
||||
@ -1,39 +0,0 @@
|
||||
# Copyright (c) 2015 Ansible, Inc.
|
||||
# All Rights Reserved
|
||||
|
||||
# Python
|
||||
import uuid
|
||||
|
||||
# AWX
|
||||
from awx.main.tests.base import BaseTest
|
||||
from command_base import BaseCommandMixin
|
||||
from awx.main.models import * # noqa
|
||||
|
||||
__all__ = ['RemoveInstanceCommandFunctionalTest']
|
||||
|
||||
class RemoveInstanceCommandFunctionalTest(BaseCommandMixin, BaseTest):
|
||||
uuids = []
|
||||
instances = []
|
||||
|
||||
def setup_instances(self):
|
||||
self.uuids = [uuid.uuid4().hex for x in range(0, 3)]
|
||||
self.instances.append(Instance(uuid=settings.SYSTEM_UUID, primary=True, hostname='127.0.0.1'))
|
||||
self.instances.append(Instance(uuid=self.uuids[0], primary=False, hostname='127.0.0.2'))
|
||||
self.instances.append(Instance(uuid=self.uuids[1], primary=False, hostname='127.0.0.3'))
|
||||
self.instances.append(Instance(uuid=self.uuids[2], primary=False, hostname='127.0.0.4'))
|
||||
for x in self.instances:
|
||||
x.save()
|
||||
|
||||
def setUp(self):
|
||||
super(RemoveInstanceCommandFunctionalTest, self).setUp()
|
||||
self.create_test_license_file()
|
||||
self.setup_instances()
|
||||
self.setup_users()
|
||||
|
||||
def test_default(self):
|
||||
self.assertEqual(Instance.objects.filter(hostname="127.0.0.2").count(), 1)
|
||||
result, stdout, stderr = self.run_command('remove_instance', hostname='127.0.0.2')
|
||||
self.assertIsNone(result)
|
||||
self.assertEqual(stdout, 'Successfully removed instance (uuid="%s",hostname="127.0.0.2",role="secondary").\n' % (self.uuids[0]))
|
||||
self.assertEqual(Instance.objects.filter(hostname="127.0.0.2").count(), 0)
|
||||
|
||||
@ -1,37 +0,0 @@
|
||||
# Copyright (c) 2015 Ansible, Inc.
|
||||
# All Rights Reserved
|
||||
|
||||
# AWX
|
||||
from awx.main.tests.base import BaseTest
|
||||
from command_base import BaseCommandMixin
|
||||
|
||||
# Django
|
||||
from django.core.management.base import CommandError
|
||||
|
||||
__all__ = ['UpdatePasswordCommandFunctionalTest']
|
||||
|
||||
class UpdatePasswordCommandFunctionalTest(BaseCommandMixin, BaseTest):
|
||||
def setUp(self):
|
||||
super(UpdatePasswordCommandFunctionalTest, self).setUp()
|
||||
self.create_test_license_file()
|
||||
self.setup_instances()
|
||||
self.setup_users()
|
||||
|
||||
def test_updated_ok(self):
|
||||
result, stdout, stderr = self.run_command('update_password', username='admin', password='dingleberry')
|
||||
self.assertEqual(stdout, 'Password updated\n')
|
||||
|
||||
def test_same_password(self):
|
||||
result, stdout, stderr = self.run_command('update_password', username='admin', password='admin')
|
||||
self.assertEqual(stdout, 'Password not updated\n')
|
||||
|
||||
def test_error_username_required(self):
|
||||
result, stdout, stderr = self.run_command('update_password', password='foo')
|
||||
self.assertIsInstance(result, CommandError)
|
||||
self.assertEqual(str(result), 'username required')
|
||||
|
||||
def test_error_password_required(self):
|
||||
result, stdout, stderr = self.run_command('update_password', username='admin')
|
||||
self.assertIsInstance(result, CommandError)
|
||||
self.assertEqual(str(result), 'password required')
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user