163 lines
6.8 KiB
Python
163 lines
6.8 KiB
Python
# hobo - portal to configure and deploy applications
|
|
# Copyright (C) 2019 Entr'ouvert
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify it
|
|
# under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from django.contrib.auth.models import User
|
|
from django.core.management import call_command
|
|
from django.core.management.base import CommandError
|
|
from django.db import connection
|
|
from tenant_schemas.utils import tenant_context
|
|
|
|
from hobo.multitenant.middleware import TenantMiddleware
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def configuration(settings, tmpdir):
|
|
settings.TENANT_BASE = str(tmpdir.mkdir('tenants'))
|
|
|
|
|
|
def schema_exists(schema_name):
|
|
with connection.cursor() as cursor:
|
|
cursor.execute('select schema_name from information_schema.schemata')
|
|
return schema_name in [row[0] for row in cursor.fetchall()]
|
|
|
|
|
|
def test_create_tenant(db):
|
|
assert not schema_exists('www_example_com')
|
|
call_command('create_tenant', 'www.example.com')
|
|
assert schema_exists('www_example_com')
|
|
tenants = list(TenantMiddleware.get_tenants())
|
|
assert len(tenants) == 1
|
|
tenant = tenants[0]
|
|
assert tenant.domain_url == 'www.example.com'
|
|
assert tenant.schema_name == 'www_example_com'
|
|
with tenant_context(tenant):
|
|
User.objects.create(username='admin')
|
|
|
|
|
|
def test_create_tenant_failure(db, caplog):
|
|
with mock.patch(
|
|
'hobo.multitenant.management.commands.migrate_schemas.MigrateSchemasCommand.handle'
|
|
) as handle:
|
|
handle.side_effect = CommandError('unable to migrate')
|
|
assert not schema_exists('www_example_com')
|
|
with pytest.raises(CommandError) as exc_info:
|
|
call_command('create_tenant', 'www.example.com')
|
|
assert str(exc_info.value) == 'tenant creation failed (unable to migrate)'
|
|
assert not schema_exists('www_example_com')
|
|
|
|
with connection.cursor() as cursor:
|
|
cursor.execute('select schema_name from information_schema.schemata')
|
|
assert 'www_example_com' not in [row[0] for row in cursor.fetchall()]
|
|
|
|
|
|
def test_migrate_schemas_skip_applied(db, capsys):
|
|
assert not schema_exists('www_example_com')
|
|
call_command('create_tenant', 'www.example.com')
|
|
captured = capsys.readouterr()
|
|
assert 'Running migrate for schema www_example_com' in captured.out
|
|
call_command('migrate_schemas', verbosity=1)
|
|
captured = capsys.readouterr()
|
|
assert 'Skipping migrations of tenant www.example.com' in captured.out
|
|
call_command('migrate_schemas', 'common', '0001_initial', verbosity=1)
|
|
captured = capsys.readouterr()
|
|
assert 'Running migrate for tenant www.example.com' in captured.out
|
|
assert 'Unapplying common.0002' in captured.out
|
|
call_command('migrate_schemas', verbosity=1)
|
|
captured = capsys.readouterr()
|
|
assert 'Running migrate for tenant www.example.com' in captured.out
|
|
assert 'Applying common.0002' in captured.out
|
|
call_command('migrate_schemas', verbosity=1)
|
|
captured = capsys.readouterr()
|
|
assert 'Skipping migrations of tenant www.example.com' in captured.out
|
|
|
|
|
|
def test_create_tenant_from_existing_tenant(db):
|
|
assert not schema_exists('www_example_com')
|
|
call_command('create_tenant', 'www.example.com')
|
|
assert schema_exists('www_example_com')
|
|
tenants = list(TenantMiddleware.get_tenants())
|
|
assert len(tenants) == 1
|
|
tenant = tenants[0]
|
|
assert tenant.domain_url == 'www.example.com'
|
|
assert tenant.schema_name == 'www_example_com'
|
|
with tenant_context(tenant):
|
|
User.objects.create(username='admin')
|
|
|
|
assert not schema_exists('www_newexample_com')
|
|
call_command('create_tenant', '--legacy-hostname', 'www.example.com', 'www.newexample.com')
|
|
assert not schema_exists('www_example_com')
|
|
assert schema_exists('www_newexample_com')
|
|
tenants = list(TenantMiddleware.get_tenants())
|
|
assert len(tenants) == 1
|
|
tenant = tenants[0]
|
|
assert tenant.domain_url == 'www.newexample.com'
|
|
assert tenant.schema_name == 'www_newexample_com'
|
|
|
|
|
|
def test_create_tenant_from_existing_tenant_error_too_many_hosts(db):
|
|
assert not schema_exists('host1_com')
|
|
assert not schema_exists('host2_com')
|
|
assert not schema_exists('host3_com')
|
|
with pytest.raises(CommandError) as exc_info:
|
|
call_command('create_tenant', '--legacy-hostname', 'host1.com', 'host2.com', 'host3.com')
|
|
assert str(exc_info.value) == 'You must specify only hostname when using --legacy-hostname'
|
|
assert not schema_exists('host1_com')
|
|
assert not schema_exists('host2_com')
|
|
assert not schema_exists('host3_com')
|
|
|
|
|
|
def test_create_tenant_from_existing_tenant_error_does_not_exists(db):
|
|
assert not schema_exists('host1_com')
|
|
assert not schema_exists('host2_com')
|
|
with pytest.raises(CommandError) as exc_info:
|
|
call_command('create_tenant', '--legacy-hostname', 'host1.com', 'host2.com')
|
|
assert str(exc_info.value) == 'legacy tenant does not exists'
|
|
assert not schema_exists('host1_com')
|
|
assert not schema_exists('host2_com')
|
|
|
|
|
|
def test_create_tenant_from_existing_tenant_but_already_exists(db):
|
|
assert not schema_exists('host1_com')
|
|
assert not schema_exists('host2_com')
|
|
call_command('create_tenant', 'host1.com')
|
|
call_command('create_tenant', 'host2.com')
|
|
assert schema_exists('host1_com')
|
|
assert schema_exists('host2_com')
|
|
with pytest.raises(CommandError) as exc_info:
|
|
call_command('create_tenant', '--legacy-hostname', 'host1.com', 'host2.com')
|
|
assert str(exc_info.value) == 'tenant already exists'
|
|
|
|
|
|
def test_migrate_schemas_eta(db, capsys):
|
|
call_command('create_tenant', 'host1.com')
|
|
call_command('create_tenant', 'host2.com')
|
|
# all skipped, no ETA is displayed
|
|
call_command('migrate_schemas', verbosity=1)
|
|
captured = capsys.readouterr()
|
|
assert 'Skipping migrations of tenant host1.com (1/2)' in captured.out
|
|
assert 'Skipping migrations of tenant host2.com (2/2)' in captured.out
|
|
assert 'migrate_schemas ETA: 2' not in captured.out
|
|
# force re-migration and re-migrate, ETA is displayed
|
|
call_command('migrate_schemas', 'common', '0001_initial', verbosity=1)
|
|
call_command('migrate_schemas', verbosity=1)
|
|
captured = capsys.readouterr()
|
|
assert 'Running migrate for tenant host1.com (1/2)' in captured.out
|
|
assert 'Running migrate for tenant host2.com (2/2)' in captured.out
|
|
assert 'migrate_schemas ETA: 2' in captured.out
|