def is_shared_model(model):
    """
    Is the model (or instance of a model) one that should be in the
    public/shared schema?
    """
    if model._is_shared_model:
        return True

    app_model = '{m.app_label}.{m.model_name}'.format(m=model._meta).lower()

    # These should be case insensitive!
    if app_model in REQUIRED_SHARED_MODELS:
        return True

    if app_model in [x.lower() for x in settings.SHARED_MODELS]:
        return True

    # Sometimes, we want a join table to be private.
    if app_model in [x.lower() for x in settings.PRIVATE_MODELS]:
        return False

    # if all fields are auto or fk, then we are a join model,
    # and if all related objects are shared, then we must
    # also be shared, unless we were explicitly marked as private
    # above.
    if _is_join_model(model):
        return all(
            is_shared_model(field.remote_field.model)
            for field in model._meta.fields
            if field.remote_field
        )

    return False def is_shared_table(table, apps=apps):
    """
    Is the model from the provided database table name shared?

    We may need to look and see if we can work out which models
    this table joins.
    """
    if table in REQUIRED_SHARED_TABLES:
        return True

    # Get a mapping of all table names to models.
    models = _get_models(apps, inspect.stack())

    table_map = {
        x._meta.db_table: x for x in models
        if not x._meta.proxy
    }

    # If we have a match, see if that one is shared.
    if table in table_map:
        return is_shared_model(table_map[table])

    # It may be a join table.
    through = _get_join_model(table, table_map)
    if through:
        return is_shared_model(through)

    # Not a join table: just assume that it's not shared.
    return False def activate_template_schema():
    """
    Activate the template schema.

    You probably don't want to do this. Sometimes you do (like for instance
    to apply migrations).
    """
    from .signals import schema_pre_activate, schema_post_activate

    _thread_locals.schema = None
    schema_name = settings.TEMPLATE_SCHEMA
    schema_pre_activate.send(sender=None, schema_name=schema_name)
    _set_search_path(schema_name)
    if _get_search_path() != schema_name:
        raise SchemaNotFound('Template schema was not activated. It seems "{0}" is active.'.format(_get_search_path()))
    schema_post_activate.send(sender=None, schema_name=schema_name) def deactivate_schema(schema=None):
    """
    Deactivate the provided (or current) schema.
    """
    from .signals import schema_pre_activate, schema_post_activate

    cursor = connection.cursor()
    schema_pre_activate.send(sender=None, schema_name=None)
    cursor.execute('SET search_path TO "$user",{0}'.format(settings.PUBLIC_SCHEMA))
    schema_post_activate.send(sender=None, schema_name=None)
    _thread_locals.schema = None
    cursor.close() class AwareModel(models.Model):
    name = models.CharField(max_length=10, unique=True)
    status = models.BooleanField(default=False)
    factor = models.PositiveSmallIntegerField(default=7)

    objects = MultiSchemaManager() class CoReferentialModelA(models.Model):
    name = models.CharField(max_length=10, unique=True)
    other = models.ForeignKey('tests.CoReferentialModelB',
                              on_delete=models.CASCADE,
                              related_name='model_a',
                              null=True, blank=True) class CoReferentialModelB(models.Model):
    name = models.CharField(max_length=10, unique=True)
    other = models.ForeignKey('tests.CoReferentialModelA',
                              on_delete=models.CASCADE,
                              related_name='model_b',
                              null=True, blank=True) class ModelA(models.Model):
    pass class ModelB(models.Model):
    pass class ModelBPrefix(models.Model):
    model_a = models.ManyToManyField(ModelA) class NaiveModel(SharedSchemaModel):
    name = models.CharField(max_length=10, unique=True)
    status = models.BooleanField(default=False) class SelfReferentialModel(models.Model):
    name = models.CharField(max_length=10, unique=True)
    parent = models.ForeignKey('tests.SelfReferentialModel',
                               on_delete=models.CASCADE,
                               related_name='children',
                               null=True, blank=True)
    from unittest.mock import patch
    from mock import patch
from django.test import TestCase, override_settings
from django.db import connection
from boardinghouse.schema import (
    is_shared_model, is_shared_table,
    activate_template_schema, deactivate_schema,
)
from ..models import (
    AwareModel,
    CoReferentialModelA,
    CoReferentialModelB,
    ModelA,
    ModelB,
    ModelBPrefix,
    NaiveModel,
    SelfReferentialModel,
)
        from django.contrib.admin.models import LogEntry
        from django.contrib.auth.models import User, Group, Permission
try:
except ImportError:





class TestIsSharedModel(TestCase):
    def test_aware_model(self):
        self.assertFalse(is_shared_model(AwareModel()))

    def test_naive_model(self):
        self.assertTrue(is_shared_model(NaiveModel()))

    def test_self_referential_model(self):
        self.assertFalse(is_shared_model(SelfReferentialModel()))

    def test_co_referential_models(self):
        self.assertFalse(is_shared_model(CoReferentialModelA()))
        self.assertFalse(is_shared_model(CoReferentialModelB()))

    def test_contrib_models(self):

        self.assertTrue(is_shared_model(User()))
        self.assertTrue(is_shared_model(Permission()))
        self.assertTrue(is_shared_model(Group()))
        self.assertTrue(is_shared_model(LogEntry()))

    def test_prefix(self):
        self.assertFalse(is_shared_model(ModelA))
        self.assertFalse(is_shared_model(ModelB))
        self.assertFalse(is_shared_model(ModelBPrefix))


class TestIsSharedTable(TestCase):
    def test_schema_table(self):
        self.assertTrue(is_shared_table('boardinghouse_schema'))
        self.assertTrue(is_shared_table('boardinghouse_schema_users'))

    def test_aware_model_table(self):
        self.assertFalse(is_shared_table(AwareModel._meta.db_table))

    def test_naive_model_table(self):
        self.assertTrue(is_shared_table(NaiveModel._meta.db_table))

    def test_join_tables(self):
        self.assertTrue(is_shared_table('auth_group_permissions'))
        self.assertFalse(is_shared_table('auth_user_groups'))

    def test_prefix_clash(self):
        self.assertFalse(is_shared_table('tests_modelb'))


class TestTemplateSchemaActivation(TestCase):
    @override_settings(TEMPLATE_SCHEMA='__template_schema__')
    def test_exception_when_no_template_schema_found(self):
        with self.assertRaises(Exception):

            activate_template_schema()