django-orchestra/orchestra/apps/databases/tests/functional_tests/tests.py

145 lines
4.7 KiB
Python
Raw Normal View History

2014-10-07 13:08:59 +00:00
import MySQLdb
import os
2014-10-09 17:04:12 +00:00
import time
2014-10-03 14:02:11 +00:00
from functools import partial
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.apps.accounts.models import Account
from orchestra.apps.orchestration.models import Server, Route
from orchestra.utils.system import run
2014-10-07 13:08:59 +00:00
from orchestra.utils.tests import (BaseLiveServerTestCase, random_ascii, save_response_on_error,
snapshot_on_error)
2014-10-03 14:02:11 +00:00
from ... import backends, settings
2014-10-07 13:50:59 +00:00
from ...models import Database, DatabaseUser
2014-10-03 14:02:11 +00:00
class DatabaseTestMixin(object):
2014-10-07 13:08:59 +00:00
MASTER_SERVER = os.environ.get('ORCHESTRA_SECOND_SERVER', 'localhost')
2014-10-03 14:02:11 +00:00
DEPENDENCIES = (
'orchestra.apps.orchestration',
'orcgestra.apps.databases',
)
def setUp(self):
2014-10-07 13:08:59 +00:00
super(DatabaseTestMixin, self).setUp()
2014-10-03 14:02:11 +00:00
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):
2014-10-07 13:08:59 +00:00
dbname = '%s_database' % random_ascii(5)
2014-10-09 17:04:12 +00:00
username = '%s_dbuser' % random_ascii(5)
2014-10-07 13:08:59 +00:00
password = '@!?%spppP001' % random_ascii(5)
self.add(dbname, username, password)
self.validate_create_table(dbname, username, password)
2014-10-03 14:02:11 +00:00
2014-10-07 13:08:59 +00:00
class MySQLBackendMixin(object):
db_type = 'mysql'
2014-10-09 17:04:12 +00:00
def setUp(self):
super(MySQLBackendMixin, self).setUp()
settings.DATABASES_DEFAULT_HOST = '10.228.207.207'
2014-10-03 14:02:11 +00:00
def add_route(self):
2014-10-07 13:08:59 +00:00
server = Server.objects.create(name=self.MASTER_SERVER)
backend = backends.MySQLBackend.get_name()
match = "database.type == '%s'" % self.db_type
Route.objects.create(backend=backend, match=match, host=server)
match = "databaseuser.type == '%s'" % self.db_type
backend = backends.MySQLUserBackend.get_name()
Route.objects.create(backend=backend, match=match, host=server)
2014-10-03 14:02:11 +00:00
def validate_create_table(self, name, username, password):
2014-10-07 13:08:59 +00:00
db = MySQLdb.connect(host=self.MASTER_SERVER, port=3306, user=username, passwd=password, db=name)
2014-10-03 14:02:11 +00:00
cur = db.cursor()
2014-10-09 17:04:12 +00:00
cur.execute('CREATE TABLE test ( id INT ) ;')
2014-10-03 14:02:11 +00:00
def validate_delete(self, name, username, password):
self.asseRaises(MySQLdb.ConnectionError,
2014-10-07 13:08:59 +00:00
self.validate_create_table, name, username, password)
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)
2014-10-09 17:04:12 +00:00
# TODO fucking nested objects
self.rest.databases.create(name=dbname, roles=[{'user': user.url}], type=self.db_type)
2014-10-07 13:08:59 +00:00
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)
2014-10-07 13:50:59 +00:00
@snapshot_on_error
def delete(self, dbname):
db = Database.objects.get(name=dbname)
self.admin_delete(db)
@snapshot_on_error
def delete_user(self, username):
user = DatabaseUser.objects.get(username=username)
self.admin_delete(user)
2014-10-07 13:08:59 +00:00
2014-10-03 14:02:11 +00:00
2014-10-07 13:08:59 +00:00
class RESTMysqlDatabaseTest(MySQLBackendMixin, RESTDatabaseMixin, BaseLiveServerTestCase):
pass
2014-10-03 14:02:11 +00:00
2014-10-07 13:08:59 +00:00
class AdminMysqlDatabaseTest(MySQLBackendMixin, AdminDatabaseMixin, BaseLiveServerTestCase):
pass