Projet

Général

Profil

0001-multitenant-wrap-schema-creation-in-atomic-23119.patch

Benjamin Dauvergne, 08 mars 2019 00:00

Télécharger (4,33 ko)

Voir les différences: 

Subject: [PATCH] multitenant: wrap schema creation in atomic() (#23119)


 .../management/commands/create_tenant.py      |  5 +-
 tests_multitenant/test_create_tenant.py       | 65 +++++++++++++++++++
 2 files changed, 68 insertions(+), 2 deletions(-)
 create mode 100644 tests_multitenant/test_create_tenant.py
hobo/multitenant/management/commands/create_tenant.py
1 1 
import os
2 2 
import sys
3 3 

  		




  4
  
  
    
from django.db import connection

  		




  
  4
  
    
from django.db import connection, transaction

  		




  5
  5
  
    
from django.core.management.base import CommandError, BaseCommand

  		




  6
  6
  
    

  		




  7
  7
  
    
from hobo.multitenant.middleware import TenantMiddleware, get_tenant_model

  		




  ......




  47
  47
  
    
                    print

  		




  48
  48
  
    
                    print self.style.NOTICE("=== Creating schema ") \

  		




  49
  49
  
    
                        + self.style.SQL_TABLE(tenant.schema_name)

  		




  50
  
  
    
                tenant.create_schema(check_if_exists=True)

  		




  
  50
  
    
                with transaction.atomic():

  		




  
  51
  
    
                    tenant.create_schema(check_if_exists=True)

  		




  51
  52
  
    
            except Exception as e:

  		




  52
  53
  
    
                os.rmdir(tenant_dir_tmp)

  		




  53
  54
  
    
                raise CommandError('tenant creation failed (%s)' % str(e))

  		












  

    
      tests_multitenant/test_create_tenant.py
    
  





  
  1
  
    
# hobo - portal to configure and deploy applications

  		




  
  2
  
    
# Copyright (C) 2019  Entr'ouvert

  		




  
  3
  
    
#

  		




  
  4
  
    
# This program is free software: you can redistribute it and/or modify it

  		




  
  5
  
    
# under the terms of the GNU Affero General Public License as published

  		




  
  6
  
    
# by the Free Software Foundation, either version 3 of the License, or

  		




  
  7
  
    
# (at your option) any later version.

  		




  
  8
  
    
#

  		




  
  9
  
    
# This program is distributed in the hope that it will be useful,

  		




  
  10
  
    
# but WITHOUT ANY WARRANTY; without even the implied warranty of

  		




  
  11
  
    
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

  		




  
  12
  
    
# GNU Affero General Public License for more details.

  		




  
  13
  
    
#

  		




  
  14
  
    
# You should have received a copy of the GNU Affero General Public License

  		




  
  15
  
    
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

  		




  
  16
  
    

  		




  
  17
  
    
from django.db import connection

  		




  
  18
  
    
from django.core.management import call_command

  		




  
  19
  
    
from django.core.management.base import CommandError

  		




  
  20
  
    
from django.contrib.auth.models import User

  		




  
  21
  
    

  		




  
  22
  
    
from tenant_schemas.utils import tenant_context

  		




  
  23
  
    

  		




  
  24
  
    
from hobo.multitenant.middleware import TenantMiddleware

  		




  
  25
  
    

  		




  
  26
  
    
import mock

  		




  
  27
  
    
import pytest

  		




  
  28
  
    

  		




  
  29
  
    

  		




  
  30
  
    
@pytest.fixture(autouse=True)

  		




  
  31
  
    
def configuration(settings, tmpdir):

  		




  
  32
  
    
    settings.TENANT_BASE = str(tmpdir.mkdir('tenants'))

  		




  
  33
  
    

  		




  
  34
  
    

  		




  
  35
  
    
def schema_exists(schema_name):

  		




  
  36
  
    
    with connection.cursor() as cursor:

  		




  
  37
  
    
        cursor.execute('select schema_name from information_schema.schemata')

  		




  
  38
  
    
        return 'www_example_com' in [row[0] for row in cursor.fetchall()]

  		




  
  39
  
    

  		




  
  40
  
    

  		




  
  41
  
    
def test_create_tenant(db):

  		




  
  42
  
    
    assert not schema_exists('www_example_com')

  		




  
  43
  
    
    call_command('create_tenant', 'www.example.com')

  		




  
  44
  
    
    assert schema_exists('www_example_com')

  		




  
  45
  
    
    tenants = list(TenantMiddleware.get_tenants())

  		




  
  46
  
    
    assert len(tenants) == 1

  		




  
  47
  
    
    tenant = tenants[0]

  		




  
  48
  
    
    assert tenant.domain_url == 'www.example.com'

  		




  
  49
  
    
    assert tenant.schema_name == 'www_example_com'

  		




  
  50
  
    
    with tenant_context(tenant):

  		




  
  51
  
    
        User.objects.create(username='admin')

  		




  
  52
  
    

  		




  
  53
  
    

  		




  
  54
  
    
def test_create_tenant_failure(db, caplog):

  		




  
  55
  
    
    with mock.patch('hobo.multitenant.management.commands.migrate_schemas.MigrateSchemasCommand.handle') as handle:

  		




  
  56
  
    
        handle.side_effect = CommandError('unable to migrate')

  		




  
  57
  
    
        assert not schema_exists('www_example_com')

  		




  
  58
  
    
        with pytest.raises(CommandError) as exc_info:

  		




  
  59
  
    
            call_command('create_tenant', 'www.example.com')

  		




  
  60
  
    
        assert str(exc_info.value) == 'tenant creation failed (unable to migrate)'

  		




  
  61
  
    
        assert not schema_exists('www_example_com')

  		




  
  62
  
    

  		




  
  63
  
    
    with connection.cursor() as cursor:

  		




  
  64
  
    
        cursor.execute('select schema_name from information_schema.schemata')

  		




  
  65
  
    
        assert 'www_example_com' not in [row[0] for row in cursor.fetchall()]

  		




  0
  
  
    
-