@register('settings')
def check_template_prefix_stats_with_underscore(app_configs=None, **kwargs):
    """Ensure that the prefix for schema template internal names starts with underscore.

    This is required because a leading underscore is the trigger that the indicated
    schema is not a "regular" schema, and should not be activated according to the
    normal rules.
    """
    from django.conf import settings

    if not settings.BOARDINGHOUSE_TEMPLATE_PREFIX.startswith('_'):
        return [Error('BOARDINGHOUSE_TEMPLATE_PREFIX must start with an underscore',
                      id='boardinghouse.contrib.template.E001')]

    return [] class SchemaTemplate(SharedSchemaMixin, models.Model):
    """
    A ``boardinghouse.contrib.template.models.SchemaTemplate`` can be used
    for creating a new schema complete with some initial data.
    """
    template_schema_id = models.AutoField(primary_key=True)
    name = models.CharField(max_length=128, unique=True)
    is_active = models.BooleanField(default=True)
    description = models.TextField(null=True, blank=True)

    objects = SchemaTemplateQuerySet.as_manager()

    class Meta:
        default_permissions = ('add', 'change', 'delete', 'view', 'activate', 'clone')
        verbose_name = lazy(verbose_name, six.text_type)()
        verbose_name_plural = lazy(verbose_name_plural, six.text_type)()

    def __str__(self):
        return self.name

    @property
    def schema(self):
        return '{0}{1}'.format(settings.BOARDINGHOUSE_TEMPLATE_PREFIX, self.pk)

    def activate(self):
        activate_schema(self.schema)

    def deactivate(self):
        deactivate_schema() def verbose_name():
    if 'makemigrations' in sys.argv:
        return u'template schema'
    return u'template {0}'.format(get_schema_model()._meta.verbose_name) def verbose_name_plural():
    if 'makemigrations' in sys.argv:
        return u'template schemata'
    return u'template {0}'.format(get_schema_model()._meta.verbose_name_plural) class Schema(AbstractSchema):
    """
    The default schema model.

    Unless you set `settings.BOARDINGHOUSE_SCHEMA_MODEL`, this model will
    be used for storing the schema objects.
    """
    users = models.ManyToManyField(
        settings.AUTH_USER_MODEL,
        blank=True, related_name='schemata',
        help_text=_(u'Which users may access data from this schema.')
    )

    class Meta:
        app_label = 'boardinghouse'
        verbose_name_plural = 'schemata'
        swappable = 'BOARDINGHOUSE_SCHEMA_MODEL' def _schema_exists(schema_name, cursor=None):
    if cursor:
        cursor.execute('''SELECT schema_name
                            FROM information_schema.schemata
                           WHERE schema_name = %s''',
                       [schema_name])
        return bool(cursor.fetchone())

    cursor = connection.cursor()
    try:
        return _schema_exists(schema_name, cursor)
    finally:
        cursor.close() def _get_schema(schema_name):
    """
    Get the matching active schema object for the given name,
    if it exists.
    """
    for handler, response in find_schema.send(sender=None, schema=schema_name):
        if response:
            return response def get_active_schema_name():
    """
    Get the currently active schema.

    This requires a database query to ask it what the current `search_path` is.
    """
    active_schema = getattr(_thread_locals, 'schema', None)

    if not active_schema:
        reported_schema = _get_search_path()[0]

        if _get_schema(reported_schema):
            active_schema = reported_schema
        else:
            active_schema = None

        _thread_locals.schema = active_schema

    return active_schema class AwareModel(models.Model):
    name = models.CharField(max_length=10, unique=True)
    status = models.BooleanField(default=False)
    factor = models.PositiveSmallIntegerField(default=7)

    objects = MultiSchemaManager() def get_table_list():
    with connection.cursor() as cursor:
        table_list = connection.introspection.get_table_list(cursor)
    if table_list and not isinstance(table_list[0], six.string_types):
        table_list = [table.name for table in table_list]
    return table_list
import sys
import uuid
from django.test import TestCase, override_settings
from django.contrib.auth.models import User
from django.db import connection, migrations, models
from django.db.migrations.state import ProjectState
from django.utils import six
from boardinghouse.contrib.template.apps import check_template_prefix_stats_with_underscore
from boardinghouse.contrib.template.models import SchemaTemplate
from boardinghouse.contrib.template.models import verbose_name, verbose_name_plural
from boardinghouse.models import Schema
from boardinghouse.schema import (
    _schema_exists, _get_schema,
    get_active_schema_name,
)
from ..models import AwareModel
from .utils import get_table_list





CREDENTIALS = {
    'username': 'username',
    'password': 'password'
}


class TestContribTemplate(TestCase):
    def test_find_schema_finds_templates(self):
        template = SchemaTemplate.objects.create(name='Foo')
        self.assertEqual(template, _get_schema(template.schema))

    def test_template_str(self):
        template = SchemaTemplate(name='Foo')
        self.assertEqual(u'Foo', six.text_type(template))

    def test_templates_can_be_created_and_destroyed(self):
        template = SchemaTemplate.objects.create(name='Foo')
        self.assertTrue(_schema_exists(template.schema))
        template.activate()
        self.assertEqual(get_active_schema_name(), template.schema)
        template.deactivate()
        template.delete()
        self.assertFalse(_schema_exists(template.schema))

    def test_templates_cannot_be_activated_normally(self):
        template = SchemaTemplate.objects.create(name='Foo')
        User.objects.create_user(**CREDENTIALS)
        self.client.login(**CREDENTIALS)
        response = self.client.get('/__change_schema__/{0}/'.format(template.schema))
        self.assertEqual(403, response.status_code)

    def test_templates_can_be_activated_by_superuser(self):
        template = SchemaTemplate.objects.create(name='Foo')
        User.objects.create_superuser(email='superuser@example.com', **CREDENTIALS)
        self.client.login(**CREDENTIALS)
        response = self.client.get('/__change_schema__/{0}/'.format(template.schema))
        self.assertEqual(200, response.status_code)

    def test_invalid_template_raises_forbidden(self):
        template = SchemaTemplate.objects.create(name='Foo')
        User.objects.create_superuser(email='superuser@example.com', **CREDENTIALS)
        self.client.login(**CREDENTIALS)
        response = self.client.get('/__change_schema__/{0}1/'.format(template.schema))
        self.assertEqual(403, response.status_code)

    def test_cloning_templates_clones_data(self):
        template = SchemaTemplate.objects.create(name='Foo')
        template.activate()

        aware = AwareModel.objects.create(name=uuid.uuid4().hex[:10])

        schema = Schema(name='cloned', schema='cloned')
        schema._clone = template.schema
        schema.save()

        schema.activate()
        AwareModel.objects.get(name=aware.name)

    def test_editing_template_does_not_change_template_data(self):
        template = SchemaTemplate.objects.create(name='Foo')
        template.activate()

        original = AwareModel.objects.create(name=uuid.uuid4().hex[:10])

        schema = Schema(name='cloned', schema='cloned')
        schema._clone = template.schema
        schema.save()

        schema.activate()
        cloned = AwareModel.objects.get(name=original.name)
        cloned.status = True
        cloned.save()

        template.activate()
        self.assertEqual(1, AwareModel.objects.filter(status=False).count())
        self.assertEqual(0, AwareModel.objects.filter(status=True).count())

        schema.activate()
        self.assertEqual(0, AwareModel.objects.filter(status=False).count())
        self.assertEqual(1, AwareModel.objects.filter(status=True).count())

    def test_migrations_apply_to_templates(self):
        template = SchemaTemplate.objects.create(name='a')
        operation = migrations.CreateModel("Pony", [
            ('pony_id', models.AutoField(primary_key=True)),
            ('pink', models.IntegerField(default=1)),
        ])
        project_state = ProjectState()
        new_state = project_state.clone()
        operation.state_forwards('tests', new_state)

        template.activate()
        self.assertFalse('tests_pony' in get_table_list())

        with connection.schema_editor() as editor:
            operation.database_forwards('tests', editor, project_state, new_state)
        template.activate()
        self.assertTrue('tests_pony' in get_table_list())

        with connection.schema_editor() as editor:
            operation.database_backwards('tests', editor, new_state, project_state)
        template.activate()
        self.assertFalse('tests_pony' in get_table_list())

    def test_verbose_name_in_makemigrations(self):
        sys.argv.append('makemigrations')

        self.assertEqual('template schema', verbose_name())