From c505f9a3c6e5510b6ce3bf5aa1b0a068521964fc Mon Sep 17 00:00:00 2001 From: Santiago Lamora Date: Tue, 30 Mar 2021 13:11:41 +0200 Subject: [PATCH] Replace SystemUserBackend with UNIXUserController --- .../tests/functional_tests/tests.py | 104 +++++++++--------- .../webapps/tests/functional_tests/tests.py | 22 ++-- 2 files changed, 63 insertions(+), 63 deletions(-) diff --git a/orchestra/contrib/systemusers/tests/functional_tests/tests.py b/orchestra/contrib/systemusers/tests/functional_tests/tests.py index 681326bf..4f7e68e0 100644 --- a/orchestra/contrib/systemusers/tests/functional_tests/tests.py +++ b/orchestra/contrib/systemusers/tests/functional_tests/tests.py @@ -31,35 +31,35 @@ class SystemUserMixin(object): 'orchestra.contrib.orchestration', 'orcgestra.apps.systemusers', ) - + def setUp(self): super(SystemUserMixin, self).setUp() self.add_route() djsettings.DEBUG = True - + def add_route(self): master = Server.objects.create(name=self.MASTER_SERVER) - backend = backends.SystemUserBackend.get_name() + backend = backends.UNIXUserController.get_name() Route.objects.create(backend=backend, match=True, host=master) - + def save(self): raise NotImplementedError - + def add(self): raise NotImplementedError - + def delete(self): raise NotImplementedError - + def update(self): raise NotImplementedError - + def disable(self): raise NotImplementedError - + def add_group(self, username, groupname): raise NotImplementedError - + def validate_user(self, username): idcmd = sshr(self.MASTER_SERVER, "id %s" % username) self.assertEqual(0, idcmd.exit_code) @@ -69,7 +69,7 @@ class SystemUserMixin(object): idgroups = idcmd.stdout.strip().split(' ')[2] idgroups = re.findall(r'\d+\((\w+)\)', idgroups) self.assertEqual(set(groups), set(idgroups)) - + def validate_delete(self, username): self.assertRaises(SystemUser.DoesNotExist, SystemUser.objects.get, username=username) self.assertRaises(CommandError, @@ -81,19 +81,19 @@ class SystemUserMixin(object): self.assertRaises(CommandError, sshrun, self.MASTER_SERVER, 'grep "^%s:" /etc/shadow' % username, display=False) # Home will be deleted on account delete, see test_delete_account - + def validate_ftp(self, username, password): ftp = ftplib.FTP(self.MASTER_SERVER) ftp.login(user=username, passwd=password) ftp.close() - + def validate_sftp(self, username, password): transport = paramiko.Transport((self.MASTER_SERVER, 22)) transport.connect(username=username, password=password) sftp = paramiko.SFTPClient.from_transport(transport) sftp.listdir() sftp.close() - + def validate_ssh(self, username, password): ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) @@ -103,14 +103,14 @@ class SystemUserMixin(object): channel.exec_command('ls') self.assertEqual(0, channel.recv_exit_status()) channel.close() - + def test_add(self): username = '%s_systemuser' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) self.add(username, password) self.addCleanup(self.delete, username) self.validate_user(username) - + def test_ftp(self): username = '%s_systemuser' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) @@ -120,7 +120,7 @@ class SystemUserMixin(object): self.validate_sftp, username, password) self.assertRaises(paramiko.AuthenticationException, self.validate_ssh, username, password) - + def test_sftp(self): username = '%s_systemuser' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) @@ -128,14 +128,14 @@ class SystemUserMixin(object): self.addCleanup(self.delete, username) self.validate_sftp(username, password) self.assertRaises(AssertionError, self.validate_ssh, username, password) - + def test_ssh(self): username = '%s_systemuser' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) self.add(username, password, shell='/bin/bash') self.addCleanup(self.delete, username) self.validate_ssh(username, password) - + def test_delete(self): username = '%s_systemuser' % random_ascii(10) password = '@!?%sppppP001' % random_ascii(5) @@ -144,7 +144,7 @@ class SystemUserMixin(object): self.delete(username) self.validate_delete(username) self.assertRaises(Exception, self.delete, self.account.username) - + def test_add_group(self): username = '%s_systemuser' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) @@ -161,7 +161,7 @@ class SystemUserMixin(object): groups = list(user.groups.values_list('username', flat=True)) self.assertIn(username2, groups) self.validate_user(username) - + def test_disable(self): username = '%s_systemuser' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) @@ -171,7 +171,7 @@ class SystemUserMixin(object): self.disable(username) self.validate_user(username) self.assertRaises(ftplib.error_perm, self.validate_ftp, username, password) - + def test_change_password(self): username = '%s_systemuser' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) @@ -192,38 +192,38 @@ class RESTSystemUserMixin(SystemUserMixin): # create main user self.save(self.account.username) self.addCleanup(self.delete_account, self.account.username) - + @save_response_on_error def add(self, username, password, shell='/dev/null'): self.rest.systemusers.create(username=username, password=password, shell=shell) - + @save_response_on_error def delete(self, username): user = self.rest.systemusers.retrieve(username=username).get() user.delete() - + @save_response_on_error def add_group(self, username, groupname): user = self.rest.systemusers.retrieve(username=username).get() user.groups.append({'username': groupname}) user.save() - + @save_response_on_error def disable(self, username): user = self.rest.systemusers.retrieve(username=username).get() user.is_active = False user.save() - + @save_response_on_error def save(self, username): user = self.rest.systemusers.retrieve(username=username).get() user.save() - + @save_response_on_error def change_password(self, username, password): user = self.rest.systemusers.retrieve(username=username).get() user.set_password(password) - + def delete_account(self, username): self.rest.account.delete() @@ -235,42 +235,42 @@ class AdminSystemUserMixin(SystemUserMixin): # create main user self.save(self.account.username) self.addCleanup(self.delete_account, self.account.username) - + @snapshot_on_error def add(self, username, password, shell='/dev/null'): url = self.live_server_url + reverse('admin:systemusers_systemuser_add') self.selenium.get(url) - + username_field = self.selenium.find_element_by_id('id_username') username_field.send_keys(username) - + password_field = self.selenium.find_element_by_id('id_password1') password_field.send_keys(password) password_field = self.selenium.find_element_by_id('id_password2') password_field.send_keys(password) - + shell_input = self.selenium.find_element_by_id('id_shell') shell_select = Select(shell_input) shell_select.select_by_value(shell) - + username_field.submit() self.assertNotEqual(url, self.selenium.current_url) - + @snapshot_on_error def delete(self, username): user = SystemUser.objects.get(username=username) self.admin_delete(user) - + @snapshot_on_error def delete_account(self, username): account = Account.objects.get(username=username) self.admin_delete(account) - + @snapshot_on_error def disable(self, username): user = SystemUser.objects.get(username=username) self.admin_disable(user) - + @snapshot_on_error def add_group(self, username, groupname): user = SystemUser.objects.get(username=username) @@ -282,7 +282,7 @@ class AdminSystemUserMixin(SystemUserMixin): save = self.selenium.find_element_by_name('_save') save.submit() self.assertNotEqual(url, self.selenium.current_url) - + @snapshot_on_error def save(self, username): user = SystemUser.objects.get(username=username) @@ -291,7 +291,7 @@ class AdminSystemUserMixin(SystemUserMixin): save = self.selenium.find_element_by_name('_save') save.submit() self.assertNotEqual(url, self.selenium.current_url) - + @snapshot_on_error def change_password(self, username, password): user = SystemUser.objects.get(username=username) @@ -307,37 +307,37 @@ class AdminSystemUserTest(AdminSystemUserMixin, BaseLiveServerTestCase): def test_create_account(self): url = self.live_server_url + reverse('admin:accounts_account_add') self.selenium.get(url) - + account_username = '%s_account' % random_ascii(10) username = self.selenium.find_element_by_id('id_username') username.send_keys(account_username) - + account_password = '@!?%spppP001' % random_ascii(5) password = self.selenium.find_element_by_id('id_password1') password.send_keys(account_password) password = self.selenium.find_element_by_id('id_password2') password.send_keys(account_password) - + full_name = random_ascii(10) full_name_field = self.selenium.find_element_by_id('id_full_name') full_name_field.send_keys(full_name) - + account_email = 'orchestra@orchestra.lan' email = self.selenium.find_element_by_id('id_email') email.send_keys(account_email) - + contact_short_name = random_ascii(10) short_name = self.selenium.find_element_by_id('id_contacts-0-short_name') short_name.send_keys(contact_short_name) - + email = self.selenium.find_element_by_id('id_contacts-0-email') email.send_keys(account_email) email.submit() self.assertNotEqual(url, self.selenium.current_url) - + self.addCleanup(self.delete_account, account_username) self.assertEqual(0, sshr(self.MASTER_SERVER, "id %s" % account_username).exit_code) - + @snapshot_on_error def test_delete_account(self): home = self.account.main_systemuser.get_home() @@ -347,7 +347,7 @@ class AdminSystemUserTest(AdminSystemUserMixin, BaseLiveServerTestCase): self.account = self.create_account(username=self.account.username, superuser=True) self.selenium.delete_all_cookies() self.admin_login() - + @snapshot_on_error def test_disable_account(self): username = '%s_systemuser' % random_ascii(10) @@ -357,18 +357,18 @@ class AdminSystemUserTest(AdminSystemUserMixin, BaseLiveServerTestCase): self.validate_ftp(username, password) self.disable(username) self.validate_user(username) - + disable = reverse('admin:accounts_account_disable', args=(self.account.pk,)) url = self.live_server_url + disable self.selenium.get(url) confirmation = self.selenium.find_element_by_name('post') confirmation.submit() self.assertNotEqual(url, self.selenium.current_url) - + self.assertRaises(ftplib.error_perm, self.validate_ftp, username, password) self.selenium.get(url) self.assertNotEqual(url, self.selenium.current_url) - + # Reenable for test cleanup self.account.is_active = True self.account.save() diff --git a/orchestra/contrib/webapps/tests/functional_tests/tests.py b/orchestra/contrib/webapps/tests/functional_tests/tests.py index 414a2316..322557ab 100644 --- a/orchestra/contrib/webapps/tests/functional_tests/tests.py +++ b/orchestra/contrib/webapps/tests/functional_tests/tests.py @@ -5,7 +5,7 @@ from io import StringIO from django.conf import settings as djsettings from orchestra.contrib.orchestration.models import Server, Route -from orchestra.contrib.systemusers.backends import SystemUserBackend +from orchestra.contrib.systemusers.backends import UNIXUserController from orchestra.utils.tests import BaseLiveServerTestCase, random_ascii, snapshot_on_error, save_response_on_error from ... import backends @@ -18,20 +18,20 @@ class WebAppMixin(object): 'orchestra.contrib.systemusers', 'orchestra.contrib.webapps', ) - + def setUp(self): super(WebAppMixin, self).setUp() self.add_route() djsettings.DEBUG = True - + def add_route(self): server, __ = Server.objects.get_or_create(name=self.MASTER_SERVER) - backend = SystemUserBackend.get_name() + backend = UNIXUserController.get_name() Route.objects.get_or_create(backend=backend, match=True, host=server) backend = self.backend.get_name() match = 'webapp.type == "%s"' % self.type_value Route.objects.create(backend=backend, match=match, host=server) - + def upload_webapp(self, name): try: ftp = ftplib.FTP(self.MASTER_SERVER) @@ -44,7 +44,7 @@ class WebAppMixin(object): index.close() finally: ftp.close() - + def test_add(self): name = '%s_%s_webapp' % (random_ascii(10), self.type_value) self.add_webapp(name) @@ -85,16 +85,16 @@ class RESTWebAppMixin(object): self.rest_login() # create main user self.save_systemuser() - + @save_response_on_error def save_systemuser(self): systemuser = self.rest.systemusers.retrieve().get() systemuser.update(is_active=True) - + @save_response_on_error def add_webapp(self, name, options=[]): self.rest.webapps.create(name=name, type=self.type_value, options=options) - + @save_response_on_error def delete_webapp(self, name): self.rest.webapps.retrieve(name=name).delete() @@ -106,11 +106,11 @@ class AdminWebAppMixin(WebAppMixin): self.admin_login() # create main user self.save_systemuser() - + @snapshot_on_error def save_systemuser(self): url = '' - + @snapshot_on_error def add(self, name, password, admin_email): pass