From dc722ec17af87501e9ef0f52a5f9daab85a958e1 Mon Sep 17 00:00:00 2001 From: Santiago Lamora Date: Wed, 31 Mar 2021 10:39:47 +0200 Subject: [PATCH] Set env variable to skip REST_API tests. --- .../databases/tests/functional_tests/tests.py | 113 +++++++++--------- .../lists/tests/functional_tests/tests.py | 79 ++++++------ .../mailboxes/tests/functional_tests/tests.py | 4 + .../tests/functional_tests/tests.py | 3 + .../webapps/tests/functional_tests/tests.py | 10 +- 5 files changed, 113 insertions(+), 96 deletions(-) diff --git a/orchestra/contrib/databases/tests/functional_tests/tests.py b/orchestra/contrib/databases/tests/functional_tests/tests.py index c13ad946..55bef281 100644 --- a/orchestra/contrib/databases/tests/functional_tests/tests.py +++ b/orchestra/contrib/databases/tests/functional_tests/tests.py @@ -1,22 +1,24 @@ -import MySQLdb import os import socket import time +import unittest +import MySQLdb from django.conf import settings as djsettings from django.core.management.base import CommandError from django.core.urlresolvers import reverse -from selenium.webdriver.support.select import Select - from orchestra.admin.utils import change_url -from orchestra.contrib.orchestration.models import Server, Route +from orchestra.contrib.orchestration.models import Route, Server from orchestra.utils.sys import sshrun -from orchestra.utils.tests import (BaseLiveServerTestCase, random_ascii, save_response_on_error, - snapshot_on_error) +from orchestra.utils.tests import (BaseLiveServerTestCase, random_ascii, + save_response_on_error, snapshot_on_error) +from selenium.webdriver.support.select import Select from ... import backends, settings from ...models import Database, DatabaseUser +TEST_REST_API = int(os.getenv('TEST_REST_API', '0')) + class DatabaseTestMixin(object): MASTER_SERVER = os.environ.get('ORCHESTRA_SECOND_SERVER', 'localhost') @@ -24,40 +26,40 @@ class DatabaseTestMixin(object): 'orchestra.contrib.orchestration', 'orcgestra.apps.databases', ) - + def setUp(self): super(DatabaseTestMixin, self).setUp() self.add_route() djsettings.DEBUG = True - + def add_route(self): raise NotImplementedError - + 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 test_add(self): dbname = '%s_database' % random_ascii(5) username = '%s_dbuser' % random_ascii(5) password = '@!?%spppP001' % random_ascii(5) self.add(dbname, username, password) self.validate_create_table(dbname, username, password) - + def test_delete(self): dbname = '%s_database' % random_ascii(5) username = '%s_dbuser' % random_ascii(5) @@ -68,7 +70,7 @@ class DatabaseTestMixin(object): self.delete_user(username) self.validate_delete(dbname, username, password) self.validate_delete_user(dbname, username) - + def test_change_password(self): dbname = '%s_database' % random_ascii(5) username = '%s_dbuser' % random_ascii(5) @@ -81,7 +83,7 @@ class DatabaseTestMixin(object): self.change_password(username, new_password) self.validate_login_error(dbname, username, password) self.validate_create_table(dbname, username, new_password) - + def test_add_user(self): dbname = '%s_database' % random_ascii(5) username = '%s_dbuser' % random_ascii(5) @@ -98,7 +100,7 @@ class DatabaseTestMixin(object): self.add_user_to_db(username2, dbname) self.validate_create_table(dbname, username, password) self.validate_create_table(dbname, username2, password2) - + def test_delete_user(self): dbname = '%s_database' % random_ascii(5) username = '%s_dbuser' % random_ascii(5) @@ -117,7 +119,7 @@ class DatabaseTestMixin(object): self.delete_user(username2) self.validate_login_error(dbname, username2, password2) self.validate_delete_user(username2, password2) - + def test_swap_user(self): dbname = '%s_database' % random_ascii(5) username = '%s_dbuser' % random_ascii(5) @@ -137,7 +139,7 @@ class DatabaseTestMixin(object): class MySQLControllerMixin(object): db_type = 'mysql' - + def setUp(self): super(MySQLControllerMixin, self).setUp() # Get local ip address used to reach self.MASTER_SERVER @@ -145,7 +147,7 @@ class MySQLControllerMixin(object): s.connect((self.MASTER_SERVER, 22)) settings.DATABASES_DEFAULT_HOST = s.getsockname()[0] s.close() - + def add_route(self): server = Server.objects.create(name=self.MASTER_SERVER) backend = backends.MySQLController.get_name() @@ -154,22 +156,22 @@ class MySQLControllerMixin(object): match = "databaseuser.type == '%s'" % self.db_type backend = backends.MySQLUserController.get_name() Route.objects.create(backend=backend, match=match, host=server) - + def validate_create_table(self, name, username, password): db = MySQLdb.connect(host=self.MASTER_SERVER, port=3306, user=username, passwd=password, db=name) cur = db.cursor() cur.execute('CREATE TABLE table_%s ( id INT ) ;' % random_ascii(10)) - + def validate_login_error(self, dbname, username, password): self.assertRaises(MySQLdb.OperationalError, self.validate_create_table, dbname, username, password ) - + def validate_delete(self, dbname, username, password): self.validate_login_error(dbname, username, password) self.assertRaises(CommandError, sshrun, self.MASTER_SERVER, 'mysql %s' % dbname, display=False) - + def validate_delete_user(self, name, username): context = { 'name': name, @@ -181,11 +183,12 @@ class MySQLControllerMixin(object): """mysql mysql -e 'SELECT * FROM user WHERE user="%(username)s";'""" % context, display=False).stdout) +@unittest.skipUnless(TEST_REST_API, "REST API tests") class RESTDatabaseMixin(DatabaseTestMixin): def setUp(self): super(RESTDatabaseMixin, self).setUp() self.rest_login() - + @save_response_on_error def add(self, dbname, username, password): user = self.rest.databaseusers.create(username=username, password=password, type=self.db_type) @@ -193,31 +196,31 @@ class RESTDatabaseMixin(DatabaseTestMixin): 'username': user.username }] self.rest.databases.create(name=dbname, users=users, type=self.db_type) - + @save_response_on_error def delete(self, dbname): self.rest.databases.retrieve(name=dbname).delete() - + @save_response_on_error def change_password(self, username, password): user = self.rest.databaseusers.retrieve(username=username).get() user.set_password(password) - + @save_response_on_error def add_user(self, username, password): self.rest.databaseusers.create(username=username, password=password, type=self.db_type) - + @save_response_on_error def add_user_to_db(self, username, dbname): user = self.rest.databaseusers.retrieve(username=username).get() db = self.rest.databases.retrieve(name=dbname).get() db.users.append(user) db.save() - + @save_response_on_error def delete_user(self, username): self.rest.databaseusers.retrieve(username=username).delete() - + @save_response_on_error def swap_user(self, username, username2, dbname): user = self.rest.databaseusers.retrieve(username=username2).get() @@ -231,84 +234,84 @@ class AdminDatabaseMixin(DatabaseTestMixin): def setUp(self): super(AdminDatabaseMixin, self).setUp() self.admin_login() - + @snapshot_on_error def add(self, dbname, username, password): url = self.live_server_url + reverse('admin:databases_database_add') self.selenium.get(url) - + type_input = self.selenium.find_element_by_id('id_type') type_select = Select(type_input) type_select.select_by_value(self.db_type) - + name_field = self.selenium.find_element_by_id('id_name') name_field.send_keys(dbname) - + 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) - + name_field.submit() self.assertNotEqual(url, self.selenium.current_url) - + @snapshot_on_error def delete(self, dbname): db = Database.objects.get(name=dbname) self.admin_delete(db) - + @snapshot_on_error def change_password(self, username, password): user = DatabaseUser.objects.get(username=username) self.admin_change_password(user, password) - + @snapshot_on_error def add_user(self, username, password): url = self.live_server_url + reverse('admin:databases_databaseuser_add') self.selenium.get(url) - + type_input = self.selenium.find_element_by_id('id_type') type_select = Select(type_input) type_select.select_by_value(self.db_type) - + 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) - + username_field.submit() self.assertNotEqual(url, self.selenium.current_url) - + @snapshot_on_error def add_user_to_db(self, username, dbname): database = Database.objects.get(name=dbname, type=self.db_type) url = self.live_server_url + change_url(database) self.selenium.get(url) - + user = DatabaseUser.objects.get(username=username, type=self.db_type) users_from = self.selenium.find_element_by_id('id_users_from') users_select = Select(users_from) users_select.select_by_value(str(user.pk)) - + add_user = self.selenium.find_element_by_id('id_users_add_link') add_user.click() - + save = self.selenium.find_element_by_name('_save') save.submit() self.assertNotEqual(url, self.selenium.current_url) - + @snapshot_on_error def swap_user(self, username, username2, dbname): database = Database.objects.get(name=dbname, type=self.db_type) url = self.live_server_url + change_url(database) self.selenium.get(url) - + # remove user "username" user = DatabaseUser.objects.get(username=username, type=self.db_type) users_to = self.selenium.find_element_by_id('id_users_to') @@ -317,7 +320,7 @@ class AdminDatabaseMixin(DatabaseTestMixin): remove_user = self.selenium.find_element_by_id('id_users_remove_link') remove_user.click() time.sleep(0.2) - + # add user "username2" user = DatabaseUser.objects.get(username=username2, type=self.db_type) users_from = self.selenium.find_element_by_id('id_users_from') @@ -326,11 +329,11 @@ class AdminDatabaseMixin(DatabaseTestMixin): add_user = self.selenium.find_element_by_id('id_users_add_link') add_user.click() time.sleep(0.2) - + save = self.selenium.find_element_by_name('_save') save.submit() self.assertNotEqual(url, self.selenium.current_url) - + @snapshot_on_error def delete_user(self, username): user = DatabaseUser.objects.get(username=username) diff --git a/orchestra/contrib/lists/tests/functional_tests/tests.py b/orchestra/contrib/lists/tests/functional_tests/tests.py index 5ccc6acd..8ba93abc 100644 --- a/orchestra/contrib/lists/tests/functional_tests/tests.py +++ b/orchestra/contrib/lists/tests/functional_tests/tests.py @@ -1,24 +1,26 @@ import os import smtplib import time -import requests +import unittest from email.mime.text import MIMEText +import requests from django.conf import settings as djsettings from django.core.management.base import CommandError from django.core.urlresolvers import reverse -from selenium.webdriver.support.select import Select - from orchestra.admin.utils import change_url from orchestra.contrib.domains.models import Domain -from orchestra.contrib.orchestration.models import Server, Route +from orchestra.contrib.orchestration.models import Route, Server from orchestra.utils.sys import sshrun -from orchestra.utils.tests import (BaseLiveServerTestCase, random_ascii, snapshot_on_error, - save_response_on_error) +from orchestra.utils.tests import (BaseLiveServerTestCase, random_ascii, + save_response_on_error, snapshot_on_error) +from selenium.webdriver.support.select import Select from ... import backends, settings from ...models import List +TEST_REST_API = int(os.getenv('TEST_REST_API', '0')) + class ListMixin(object): MASTER_SERVER = os.environ.get('ORCHESTRA_SLAVE_SERVER', 'localhost') @@ -27,12 +29,12 @@ class ListMixin(object): 'orchestra.contrib.domains', 'orchestra.contrib.lists', ) - + def setUp(self): super(ListMixin, self).setUp() self.add_route() djsettings.DEBUG = True - + def validate_add(self, name, address=None): sshrun(self.MASTER_SERVER, 'list_members %s' % name, display=False) if not address: @@ -44,11 +46,11 @@ class ListMixin(object): sshrun(self.MASTER_SERVER, 'grep -v ":\|^\s\|^$\|-\|\.\|\s" /var/spool/mail/nobody | base64 -d | grep "%s"' % request_address, display=False) - + def validate_login(self, name, password): url = 'http://%s/cgi-bin/mailman/admin/%s' % (settings.LISTS_DEFAULT_DOMAIN, name) self.assertEqual(200, requests.post(url, data={'adminpw': password}).status_code) - + def validate_delete(self, name): context = { 'name': name, @@ -62,7 +64,7 @@ class ListMixin(object): 'grep "^\s*$(domain)s\s*$" %(virtual_domain)s' % context, display=False) self.assertRaises(CommandError, sshrun, self.MASTER_SERVER, 'list_lists | grep -i "^\s*%(name)s\s"' % context, display=False) - + def subscribe(self, subscribe_address): msg = MIMEText('') msg['To'] = subscribe_address @@ -76,12 +78,12 @@ class ListMixin(object): server.sendmail(msg['From'], msg['To'], msg.as_string()) finally: server.quit() - + def add_route(self): server = Server.objects.create(name=self.MASTER_SERVER) backend = backends.MailmanController.get_name() Route.objects.create(backend=backend, match=True, host=server) - + def test_add(self): name = '%s_list' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) @@ -90,7 +92,7 @@ class ListMixin(object): self.validate_add(name) self.validate_login(name, password) self.addCleanup(self.delete, name) - + def test_add_with_address(self): name = '%s_list' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) @@ -102,7 +104,7 @@ class ListMixin(object): self.addCleanup(self.delete, name) # Mailman doesn't support changing the address, only the domain self.validate_add(name, address="%s@%s" % (address_name, address_domain)) - + def test_change_password(self): name = '%s_list' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) @@ -113,7 +115,7 @@ class ListMixin(object): new_password = '@!?%spppP001' % random_ascii(5) self.change_password(name, new_password) self.validate_login(name, new_password) - + def test_change_domain(self): name = '%s_list' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) @@ -128,7 +130,7 @@ class ListMixin(object): address_domain = Domain.objects.create(name=domain_name, account=self.account) self.update_domain(name, domain_name) self.validate_add(name, address="%s@%s" % (address_name, address_domain)) - + def test_change_address_name(self): name = '%s_list' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) @@ -142,7 +144,7 @@ class ListMixin(object): address_name = '%s_name' % random_ascii(10) self.update_address_name(name, address_name) self.validate_add(name, address="%s@%s" % (address_name, address_domain)) - + def test_delete(self): name = '%s_list' % random_ascii(10) password = '@!?%spppP001' % random_ascii(5) @@ -158,11 +160,12 @@ class ListMixin(object): self.validate_delete(name) +@unittest.skipUnless(TEST_REST_API, "REST API tests") class RESTListMixin(ListMixin): def setUp(self): super(RESTListMixin, self).setUp() self.rest_login() - + @save_response_on_error def add(self, name, password, admin_email, address_name=None, address_domain=None): extra = {} @@ -172,22 +175,22 @@ class RESTListMixin(ListMixin): 'address_domain': self.rest.domains.retrieve(name=address_domain.name).get(), }) self.rest.lists.create(name=name, password=password, admin_email=admin_email, **extra) - + @save_response_on_error def delete(self, name): self.rest.lists.retrieve(name=name).delete() - + @save_response_on_error def change_password(self, name, password): mail_list = self.rest.lists.retrieve(name=name).get() mail_list.set_password(password) - + @save_response_on_error def update_domain(self, name, domain_name): mail_list = self.rest.lists.retrieve(name=name).get() domain = self.rest.domains.retrieve(name=domain_name).get() mail_list.update(address_domain=domain) - + @save_response_on_error def update_address_name(self, name, address_name): mail_list = self.rest.lists.retrieve(name=name).get() @@ -198,70 +201,70 @@ class AdminListMixin(ListMixin): def setUp(self): super(AdminListMixin, self).setUp() self.admin_login() - + @snapshot_on_error def add(self, name, password, admin_email, address_name=None, address_domain=None): url = self.live_server_url + reverse('admin:lists_list_add') self.selenium.get(url) - + name_field = self.selenium.find_element_by_id('id_name') name_field.send_keys(name) - + 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) - + admin_email_field = self.selenium.find_element_by_id('id_admin_email') admin_email_field.send_keys(admin_email) - + if address_name: address_name_field = self.selenium.find_element_by_id('id_address_name') address_name_field.send_keys(address_name) - + domain = Domain.objects.get(name=address_domain) domain_input = self.selenium.find_element_by_id('id_address_domain') domain_select = Select(domain_input) domain_select.select_by_value(str(domain.pk)) - + name_field.submit() self.assertNotEqual(url, self.selenium.current_url) - + @snapshot_on_error def delete(self, name): mail_list = List.objects.get(name=name) self.admin_delete(mail_list) - + @snapshot_on_error def change_password(self, name, password): mail_list = List.objects.get(name=name) self.admin_change_password(mail_list, password) - + @snapshot_on_error def update_domain(self, name, domain_name): mail_list = List.objects.get(name=name) url = self.live_server_url + change_url(mail_list) self.selenium.get(url) - + domain = Domain.objects.get(name=domain_name) domain_input = self.selenium.find_element_by_id('id_address_domain') domain_select = Select(domain_input) domain_select.select_by_value(str(domain.pk)) - + save = self.selenium.find_element_by_name('_save') save.submit() self.assertNotEqual(url, self.selenium.current_url) - + @snapshot_on_error def update_address_name(self, name, address_name): mail_list = List.objects.get(name=name) url = self.live_server_url + change_url(mail_list) self.selenium.get(url) - + address_name_field = self.selenium.find_element_by_id('id_address_name') address_name_field.clear() address_name_field.send_keys(address_name) - + save = self.selenium.find_element_by_name('_save') save.submit() self.assertNotEqual(url, self.selenium.current_url) diff --git a/orchestra/contrib/mailboxes/tests/functional_tests/tests.py b/orchestra/contrib/mailboxes/tests/functional_tests/tests.py index ccfb1c22..66459968 100644 --- a/orchestra/contrib/mailboxes/tests/functional_tests/tests.py +++ b/orchestra/contrib/mailboxes/tests/functional_tests/tests.py @@ -4,6 +4,7 @@ import poplib import smtplib import time import textwrap +import unittest from email.mime.text import MIMEText from django.apps import apps @@ -21,6 +22,8 @@ from orchestra.utils.tests import BaseLiveServerTestCase, random_ascii, snapshot from ... import backends, settings from ...models import Mailbox +TEST_REST_API = int(os.getenv('TEST_REST_API', '0')) + class MailboxMixin(object): MASTER_SERVER = os.environ.get('ORCHESTRA_SLAVE_SERVER', 'localhost') @@ -235,6 +238,7 @@ class MailboxMixin(object): # TODO test autoreply +@unittest.skipUnless(TEST_REST_API, "REST API tests") class RESTMailboxMixin(MailboxMixin): def setUp(self): super(RESTMailboxMixin, self).setUp() diff --git a/orchestra/contrib/systemusers/tests/functional_tests/tests.py b/orchestra/contrib/systemusers/tests/functional_tests/tests.py index 4f7e68e0..96c38c37 100644 --- a/orchestra/contrib/systemusers/tests/functional_tests/tests.py +++ b/orchestra/contrib/systemusers/tests/functional_tests/tests.py @@ -2,6 +2,7 @@ import ftplib import os import re import time +import unittest from functools import partial import paramiko @@ -21,6 +22,7 @@ from ... import backends from ...models import SystemUser +TEST_REST_API = int(os.getenv('TEST_REST_API', '0')) r = partial(run, silent=True, display=False) sshr = partial(sshrun, silent=True, display=False) @@ -185,6 +187,7 @@ class SystemUserMixin(object): # TODO test resources +@unittest.skipUnless(TEST_REST_API, "REST API tests") class RESTSystemUserMixin(SystemUserMixin): def setUp(self): super(RESTSystemUserMixin, self).setUp() diff --git a/orchestra/contrib/webapps/tests/functional_tests/tests.py b/orchestra/contrib/webapps/tests/functional_tests/tests.py index b3cf8dbb..72536f15 100644 --- a/orchestra/contrib/webapps/tests/functional_tests/tests.py +++ b/orchestra/contrib/webapps/tests/functional_tests/tests.py @@ -1,16 +1,19 @@ import ftplib import os +import unittest from io import StringIO from django.conf import settings as djsettings - -from orchestra.contrib.orchestration.models import Server, Route +from orchestra.contrib.orchestration.models import Route, Server from orchestra.contrib.systemusers.backends import UNIXUserController -from orchestra.utils.tests import BaseLiveServerTestCase, random_ascii, snapshot_on_error, save_response_on_error +from orchestra.utils.tests import BaseLiveServerTestCase, random_ascii, save_response_on_error, snapshot_on_error from ... import backends +TEST_REST_API = int(os.getenv('TEST_REST_API', '0')) + + class WebAppMixin(object): MASTER_SERVER = os.environ.get('ORCHESTRA_MASTER_SERVER', 'localhost') DEPENDENCIES = ( @@ -74,6 +77,7 @@ class PHPFPMWebAppMixin(StaticWebAppMixin): ) +@unittest.skipUnless(TEST_REST_API, "REST API tests") class RESTWebAppMixin(object): def setUp(self): super(RESTWebAppMixin, self).setUp()