mirror of
https://github.com/carrotquest/django-clickhouse.git
synced 2024-11-11 03:46:46 +03:00
48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
from django.test import TestCase, override_settings
|
|
from infi.clickhouse_orm.migrations import MigrationHistory
|
|
|
|
from django_clickhouse.database import connections
|
|
from django_clickhouse.migrations import migrate_app
|
|
from django_clickhouse.routers import DefaultRouter
|
|
from tests.clickhouse_models import ClickHouseTestModel
|
|
|
|
|
|
class NoMigrateRouter(DefaultRouter):
|
|
def allow_migrate(self, db_alias, app_label, model=None, **hints):
|
|
return False
|
|
|
|
|
|
def table_exists(db, model_class):
|
|
res = db.select(
|
|
"SELECT * FROM system.tables WHERE `database`='%s' AND `name`='%s'"
|
|
% (db.db_name, model_class.table_name())
|
|
)
|
|
res = list(res)
|
|
return bool(res)
|
|
|
|
|
|
@override_settings(CLICKHOUSE_MIGRATE_WITH_DEFAULT_DB=False)
|
|
class MigrateAppTest(TestCase):
|
|
def setUp(self):
|
|
self.db = connections['default']
|
|
|
|
# Clean all database data
|
|
self.db.drop_database()
|
|
self.db.db_exists = False
|
|
self.db.create_database()
|
|
|
|
def test_migrate_app(self):
|
|
migrate_app('tests', 'default')
|
|
self.assertTrue(table_exists(self.db, ClickHouseTestModel))
|
|
|
|
self.assertEqual(1, self.db.count(MigrationHistory))
|
|
|
|
# Migrations are already applied no actions should be done
|
|
migrate_app('tests', 'default')
|
|
self.assertEqual(1, self.db.count(MigrationHistory))
|
|
|
|
@override_settings(CLICKHOUSE_DATABASE_ROUTER=NoMigrateRouter)
|
|
def test_router_not_allowed(self):
|
|
migrate_app('tests', 'default')
|
|
self.assertFalse(table_exists(self.db, ClickHouseTestModel))
|