Update Tests:

* don't use the project fixtures anymore but the one Django project
* update and refactor cache tests to be less brittle
* explicitly test mysql, postgresql and sqlite3 databases
* remove 'multiple_call' tests since it is not possible anymore
(with the Django singleton)
* update Travis and tox (dropping Python 3.4)
This commit is contained in:
David Wobrock committed 2019-05-12 15:32:57 +02:00
1 parent f35c9177f4
commit 5390fd507f
9 files changed
+303 -668

No files matched your search

+3 -5
View File
@@ -1,17 +1,15 @@
sudo: false sudo: false
language: python language: python
dist: xenial dist: xenial
services:
- mysql
- postgresql
matrix: matrix:
include: include:
- python: 2.7 - python: 2.7
env: TOXENV=py27-django111 env: TOXENV=py27-django111
- python: 3.4
env: TOXENV=py34-django111
- python: 3.4
env: TOXENV=py34-django20
- python: 3.5 - python: 3.5
env: TOXENV=py35-django111 env: TOXENV=py35-django111
- python: 3.5 - python: 3.5
+11 -63
View File
@@ -13,70 +13,18 @@
# limitations under the License. # limitations under the License.
import os import os
import shutil
_BASE_DIR = os.path.join( _PROJ_TEST_DIR = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
'tests/') "tests/",
_FIXTURES_FOLDER = os.path.join(_BASE_DIR, 'test_project_fixtures/') "test_project",
ADD_NOT_NULL_COLUMN_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_project_add_not_null_column/')
CREATE_TABLE_WITH_NOT_NULL_COLUMN_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_project_create_table_with_not_null_column/')
DROP_COLUMN_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_project_drop_column/')
RENAME_COLUMN_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_project_rename_column/')
ALTER_COLUMN_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_project_alter_column')
RENAME_TABLE_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_project_rename_table/')
ADD_NOT_NULL_COLUMN_FOLLOWED_BY_DEFAULT_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_project_add_not_null_column_followed_by_default/')
MULTI_COMMIT_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_project_multi_commit/')
CORRECT_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_correct_project/')
DELETED_MIGRATION_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_deleted_migration_project/')
IGNORE_MIGRATION_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_project_ignore_migration/')
NOT_DJANGO_GIT_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_git_project/')
NOT_GIT_DJANGO_PROJECT = os.path.join(
_FIXTURES_FOLDER, 'test_django_without_git_project/')
NON_GIT_ROOT_GIT_FOLDER = os.path.join(
_FIXTURES_FOLDER, 'test_non_root_git_project/'
)
NON_GIT_ROOT_DJANGO_PROJECT = os.path.join(
NON_GIT_ROOT_GIT_FOLDER, 'django_project/'
) )
CREATE_TABLE_WITH_NOT_NULL_COLUMN = "app_create_table_with_not_null_column"
ALL_GIT_PROJECTS = ( ADD_NOT_NULL_COLUMN = "app_add_not_null_column"
NOT_DJANGO_GIT_PROJECT, DROP_COLUMN = "app_drop_column"
MULTI_COMMIT_PROJECT, RENAME_COLUMN = "app_rename_column"
DELETED_MIGRATION_PROJECT, RENAME_TABLE = "app_rename_table"
NON_GIT_ROOT_GIT_FOLDER, IGNORE_MIGRATION = "app_ignore_migration"
) ADD_NOT_NULL_COLUMN_FOLLOWED_BY_DEFAULT = "app_add_not_null_column_followed_by_default"
ALTER_COLUMN = "app_alter_column"
def prepare_git_project(path):
"""Copy the git/ folder to .git/ so it is
actually versioned"""
src_git = os.path.join(path, 'git/')
dest_git = os.path.join(path, '.git/')
try:
shutil.copytree(src_git, dest_git)
except OSError:
pass # .git/ might already exist
def clear_all_git_projects():
"""Delete all .git/ folders in test projects"""
for project_path in ALL_GIT_PROJECTS:
git_path = os.path.join(project_path, '.git/')
shutil.rmtree(git_path, ignore_errors=True)
+179 -98
View File
@@ -12,13 +12,17 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import os import os
import shutil
import unittest
import sys import sys
import unittest
from tests import fixtures from django.conf import settings
from django.db.migrations import Migration
from django_migration_linter import MigrationLinter, Cache, DEFAULT_CACHE_PATH from django_migration_linter import (
MigrationLinter,
analyse_sql_statements,
get_migration_abspath,
)
if sys.version_info >= (3, 3): if sys.version_info >= (3, 3):
import unittest.mock as mock import unittest.mock as mock
@@ -26,131 +30,208 @@ else:
import mock import mock
class CacheTest(unittest.TestCase): class CacheTestCase(unittest.TestCase):
MIGRATION_FILE = os.path.join(fixtures.ALTER_COLUMN_PROJECT, 'test_app', 'migrations', '0001_initial.py') def setUp(self):
self.test_project_path = os.path.dirname(settings.BASE_DIR)
def test_cache_normal(self): @mock.patch(
cache_file = os.path.join(DEFAULT_CACHE_PATH, 'test_project_add_not_null_column.pickle') "django_migration_linter.MigrationLinter._gather_all_migrations",
if os.path.exists(cache_file): return_value=[
os.remove(cache_file) Migration("0001_create_table", "app_add_not_null_column"),
linter = MigrationLinter(fixtures.ADD_NOT_NULL_COLUMN_PROJECT) Migration("0002_add_new_not_null_field", "app_add_not_null_column"),
],
)
def test_cache_normal(self, *args):
linter = MigrationLinter(self.test_project_path)
linter.old_cache.clear()
linter.old_cache.save()
with mock.patch.object(MigrationLinter, 'get_sql', wraps=linter.get_sql)as sql_mock: with mock.patch(
"django_migration_linter.migration_linter.analyse_sql_statements",
wraps=analyse_sql_statements,
) as analyse_sql_statements_mock:
linter.lint_all_migrations() linter.lint_all_migrations()
self.assertEqual(sql_mock.call_count, 2) self.assertEqual(2, analyse_sql_statements_mock.call_count)
cache = Cache( cache = linter.new_cache
fixtures.ADD_NOT_NULL_COLUMN_PROJECT,
DEFAULT_CACHE_PATH
)
cache.load() cache.load()
self.assertEqual(cache['3ef74e7f3e53e273e2fc95379248d58d']['result'], 'OK') self.assertEqual("OK", cache["4a3770a405738d457e2d23e17fb1f3aa"]["result"])
self.assertEqual(cache['e1e312b6d08ecbe017c25c58fc2be257']['result'], 'ERR') self.assertEqual("ERR", cache["19fd3ea688fc05e2cc2a6e67c0b7aa17"]["result"])
self.assertListEqual( self.assertListEqual(
cache['e1e312b6d08ecbe017c25c58fc2be257']['errors'], cache["19fd3ea688fc05e2cc2a6e67c0b7aa17"]["errors"],
[{'err_msg': 'RENAMING tables', 'code': 'RENAME_TABLE', 'table': None, 'column': None}] [
{
"err_msg": "RENAMING tables",
"code": "RENAME_TABLE",
"table": None,
"column": None,
}
],
) )
# Start the Linter again -> should use cache now. # Start the Linter again -> should use cache now.
linter = MigrationLinter(fixtures.ADD_NOT_NULL_COLUMN_PROJECT) linter = MigrationLinter(self.test_project_path)
with mock.patch.object(MigrationLinter, 'get_sql', wraps=linter.get_sql) as sql_mock: with mock.patch(
"django_migration_linter.migration_linter.analyse_sql_statements",
wraps=analyse_sql_statements,
) as analyse_sql_statements_mock:
linter.lint_all_migrations() linter.lint_all_migrations()
self.assertEqual(sql_mock.call_count, 0) analyse_sql_statements_mock.assert_not_called()
self.assertTrue(linter.has_errors) self.assertTrue(linter.has_errors)
def test_cache_ignored(self): @mock.patch(
cache_file = os.path.join(DEFAULT_CACHE_PATH, 'test_project_ignore_migration.pickle') "django_migration_linter.MigrationLinter._gather_all_migrations",
if os.path.exists(cache_file): return_value=[
os.remove(cache_file) Migration("0001_create_table", "app_add_not_null_column"),
linter = MigrationLinter(fixtures.IGNORE_MIGRATION_PROJECT) Migration("0002_add_new_not_null_field", "app_add_not_null_column"),
],
)
def test_cache_different_databases(self, *args):
linter = MigrationLinter(self.test_project_path, database="mysql")
linter.old_cache.clear()
linter.old_cache.save()
with mock.patch.object(MigrationLinter, 'get_sql', wraps=linter.get_sql) as sql_mock: linter = MigrationLinter(self.test_project_path, database="sqlite")
linter.old_cache.clear()
linter.old_cache.save()
with mock.patch(
"django_migration_linter.migration_linter.analyse_sql_statements",
wraps=analyse_sql_statements,
) as analyse_sql_statements_mock:
linter.lint_all_migrations() linter.lint_all_migrations()
self.assertEqual(sql_mock.call_count, 2) self.assertEqual(2, analyse_sql_statements_mock.call_count)
cache = Cache( cache = linter.new_cache
fixtures.IGNORE_MIGRATION_PROJECT,
DEFAULT_CACHE_PATH
)
cache.load() cache.load()
self.assertEqual(cache['63230606af0eccaef7f1f78c537c624c']['result'], 'OK') self.assertEqual("OK", cache["4a3770a405738d457e2d23e17fb1f3aa"]["result"])
self.assertEqual(cache['5c5ca1780a9f28439c1defc1f32af894']['result'], 'IGNORE') self.assertEqual("ERR", cache["19fd3ea688fc05e2cc2a6e67c0b7aa17"]["result"])
self.assertListEqual(
cache["19fd3ea688fc05e2cc2a6e67c0b7aa17"]["errors"],
[
{
"err_msg": "RENAMING tables",
"code": "RENAME_TABLE",
"table": None,
"column": None,
}
],
)
# Start the Linter again but with different database, should not be the same cache
linter = MigrationLinter(self.test_project_path, database="mysql")
with mock.patch(
"django_migration_linter.migration_linter.analyse_sql_statements",
wraps=analyse_sql_statements,
) as analyse_sql_statements_mock:
linter.lint_all_migrations()
self.assertEqual(2, analyse_sql_statements_mock.call_count)
cache = linter.new_cache
cache.load()
self.assertEqual("OK", cache["4a3770a405738d457e2d23e17fb1f3aa"]["result"])
self.assertEqual("ERR", cache["19fd3ea688fc05e2cc2a6e67c0b7aa17"]["result"])
self.assertListEqual(
cache["19fd3ea688fc05e2cc2a6e67c0b7aa17"]["errors"],
[
{
"err_msg": "NOT NULL constraint on columns",
"code": "NOT_NULL",
"table": "app_add_not_null_column_a",
"column": "new_not_null_field",
}
],
)
self.assertTrue(linter.has_errors)
@mock.patch(
"django_migration_linter.MigrationLinter._gather_all_migrations",
return_value=[
Migration("0001_initial", "app_ignore_migration"),
Migration("0002_ignore_migration", "app_ignore_migration"),
],
)
def test_cache_ignored(self, *args):
linter = MigrationLinter(self.test_project_path, ignore_name_contains="0001")
linter.old_cache.clear()
linter.old_cache.save()
with mock.patch(
"django_migration_linter.migration_linter.analyse_sql_statements",
wraps=analyse_sql_statements,
) as analyse_sql_statements_mock:
linter.lint_all_migrations()
self.assertEqual(1, analyse_sql_statements_mock.call_count)
cache = linter.new_cache
cache.load()
self.assertEqual("IGNORE", cache["0fab48322ba76570da1a3c193abb77b5"]["result"])
# Start the Linter again -> should use cache now. # Start the Linter again -> should use cache now.
linter = MigrationLinter(fixtures.IGNORE_MIGRATION_PROJECT) linter = MigrationLinter(self.test_project_path)
with mock.patch.object(MigrationLinter, 'get_sql', wraps=linter.get_sql) as sql_mock: with mock.patch(
"django_migration_linter.migration_linter.analyse_sql_statements",
wraps=analyse_sql_statements,
) as analyse_sql_statements_mock:
linter.lint_all_migrations() linter.lint_all_migrations()
self.assertEqual(sql_mock.call_count, 0) self.assertEqual(1, analyse_sql_statements_mock.call_count)
def test_cache_ignored_command_line(self): @mock.patch(
cache_file = os.path.join(DEFAULT_CACHE_PATH, 'test_project_ignore_migration.pickle') "django_migration_linter.MigrationLinter._gather_all_migrations",
if os.path.exists(cache_file): return_value=[
os.remove(cache_file) Migration("0002_add_new_not_null_field", "app_add_not_null_column")
linter = MigrationLinter(fixtures.IGNORE_MIGRATION_PROJECT, ],
ignore_name_contains='0001') )
def test_cache_modified(self, *args):
linter = MigrationLinter(self.test_project_path)
linter.old_cache.clear()
linter.old_cache.save()
with mock.patch.object(MigrationLinter, 'get_sql', wraps=linter.get_sql) as sql_mock: with mock.patch(
"django_migration_linter.migration_linter.analyse_sql_statements",
wraps=analyse_sql_statements,
) as analyse_sql_statements_mock:
linter.lint_all_migrations() linter.lint_all_migrations()
self.assertEqual(sql_mock.call_count, 1) self.assertEqual(1, analyse_sql_statements_mock.call_count)
cache = Cache( cache = linter.new_cache
fixtures.IGNORE_MIGRATION_PROJECT,
DEFAULT_CACHE_PATH
)
cache.load() cache.load()
self.assertNotIn('63230606af0eccaef7f1f78c537c624c', cache) self.assertEqual("ERR", cache["19fd3ea688fc05e2cc2a6e67c0b7aa17"]["result"])
self.assertEqual(cache['5c5ca1780a9f28439c1defc1f32af894']['result'], 'IGNORE')
def test_cache_modified(self): # Get the content of the migration file and mock the open call to append
cache_file = os.path.join(DEFAULT_CACHE_PATH, 'test_project_alter_column.pickle') # some content to change the hash
if os.path.exists(cache_file): migration_path = get_migration_abspath(
os.remove(cache_file) "app_add_not_null_column", "0002_add_new_not_null_field"
linter = MigrationLinter(fixtures.ALTER_COLUMN_PROJECT)
linter.lint_all_migrations()
cache = Cache(
fixtures.ALTER_COLUMN_PROJECT,
DEFAULT_CACHE_PATH
) )
with open(migration_path, "rb") as f:
file_content = f.read()
file_content += b"# test comment"
linter = MigrationLinter(self.test_project_path)
with mock.patch(
"django_migration_linter.migration_linter.open",
mock.mock_open(read_data=file_content),
):
with mock.patch(
"django_migration_linter.migration_linter.analyse_sql_statements",
wraps=analyse_sql_statements,
) as analyse_sql_statements_mock:
linter.lint_all_migrations()
self.assertEqual(1, analyse_sql_statements_mock.call_count)
cache = linter.new_cache
cache.load() cache.load()
self.assertEqual(cache['8589aa107b6da296c4b49cd2681d2230']['result'], 'OK', self.assertNotIn("19fd3ea688fc05e2cc2a6e67c0b7aa17", cache)
'If this fails, tearDown might have failed to remove ' self.assertEqual(1, len(cache))
'the modification from tests/test_project_fixtures/' self.assertEqual("ERR", cache["a25768641a0ad526fad199f97c303784"]["result"])
'test_project_alter_column/test_app/migrations/0001_initial.py'
)
self.assertEqual(cache['8f54c4a434cfaa9838e8ca12eb988255']['result'], 'ERR')
self.assertListEqual(
cache['8f54c4a434cfaa9838e8ca12eb988255']['errors'],
[{u'err_msg': u'ALTERING columns (Could be backward compatible. '
u'You may ignore this migration.)',
u'code': u'ALTER_COLUMN',
u'table': u'test_app_a',
u'column': None}
]
)
# Modify migration
backup_migration_file = self.MIGRATION_FILE + "_backup"
shutil.copy2(self.MIGRATION_FILE, backup_migration_file)
with open(self.MIGRATION_FILE, "a") as f:
f.write("# modification at the end of the file")
# Start the Linter again -> Cache should look different now
linter = MigrationLinter(fixtures.ALTER_COLUMN_PROJECT)
linter.lint_all_migrations()
cache = Cache(
fixtures.ALTER_COLUMN_PROJECT,
DEFAULT_CACHE_PATH
)
cache.load()
shutil.copy2(backup_migration_file, self.MIGRATION_FILE)
os.remove(backup_migration_file)
self.assertNotIn('8589aa107b6da296c4b49cd2681d2230', cache)
self.assertEqual(cache['fbee628b1ab4bd1c14f8a4b41123e7cf']['result'], 'OK')
-298
View File
@@ -1,298 +0,0 @@
# Copyright 2019 3YOURMIND GmbH
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import shutil
import unittest
from subprocess import Popen, PIPE
from django_migration_linter import utils, DEFAULT_CACHE_PATH, constants
from tests import fixtures
import sys
class CallLinterFromCommandLineTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
super(CallLinterFromCommandLineTest, cls).setUpClass()
linter_name = 'django-migration-linter'
cls.linter_exec = '{0}/bin/{1}'.format(sys.prefix, linter_name) if hasattr(sys, 'real_prefix') else linter_name
def test_call_linter_cmd_line_working(self):
cmd = '{0} --no-cache {1}'.format(self.linter_exec, fixtures.CORRECT_PROJECT)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
lines = list(map(utils.clean_bytes_to_str, process.stdout.readlines()))
self.assertEqual(len(lines), 6)
self.assertEqual(
sorted(lines[:4]),
sorted([
"(test_app1, 0001_initial)... OK",
"(test_app1, 0002_a_new_null_field)... OK",
"(test_app2, 0001_foo)... OK",
"(test_app3, 0001_initial)... OK",
])
)
def test_call_linter_cmd_line_errors(self):
cmd = '{0} --no-cache {1}'.format(
self.linter_exec,
fixtures.ADD_NOT_NULL_COLUMN_PROJECT)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertNotEqual(process.returncode, 0)
lines = list(map(utils.clean_bytes_to_str, process.stdout.readlines()))
self.assertEqual(len(lines), 5)
self.assertTrue(lines[0].endswith('OK'))
self.assertTrue(lines[1].endswith('ERR'))
self.assertTrue('RENAMING tables' in lines[2])
def test_call_linter_cmd_line_exclude_apps(self):
cmd = '{0} --no-cache {1} --exclude-apps test_app2'.format(
self.linter_exec,
fixtures.CORRECT_PROJECT)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
lines = list(map(utils.clean_bytes_to_str, process.stdout.readlines()))
self.assertEqual(len(lines), 6)
self.assertEqual(
sorted(lines[:4]),
sorted([
"(test_app1, 0001_initial)... OK",
"(test_app1, 0002_a_new_null_field)... OK",
"(test_app2, 0001_foo)... IGNORE",
"(test_app3, 0001_initial)... OK",
])
)
def test_call_linter_cmd_line_include_apps(self):
cmd = '{0} --no-cache {1} --include-apps test_app2'.format(
self.linter_exec,
fixtures.CORRECT_PROJECT)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
lines = list(map(utils.clean_bytes_to_str, process.stdout.readlines()))
self.assertEqual(len(lines), 6)
self.assertEqual(
sorted(lines[:4]),
sorted([
"(test_app1, 0001_initial)... IGNORE",
"(test_app1, 0002_a_new_null_field)... IGNORE",
"(test_app2, 0001_foo)... OK",
"(test_app3, 0001_initial)... IGNORE",
])
)
def test_call_linter_cmd_line_ignore_name(self):
cmd = '{0} --no-cache {1} --ignore-name 0001_initial'.format(
self.linter_exec,
fixtures.CORRECT_PROJECT)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
lines = list(map(utils.clean_bytes_to_str, process.stdout.readlines()))
self.assertEqual(len(lines), 6)
self.assertEqual(
sorted(lines[:4]),
sorted([
"(test_app1, 0001_initial)... IGNORE",
"(test_app1, 0002_a_new_null_field)... OK",
"(test_app2, 0001_foo)... OK",
"(test_app3, 0001_initial)... IGNORE",
])
)
def test_call_linter_cmd_line_ignore_name_contains(self):
cmd = '{0} --no-cache {1} --ignore-name-contains 0001'.format(
self.linter_exec,
fixtures.CORRECT_PROJECT)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
lines = list(map(utils.clean_bytes_to_str, process.stdout.readlines()))
self.assertEqual(len(lines), 6)
self.assertEqual(
sorted(lines[:4]),
sorted([
"(test_app1, 0001_initial)... IGNORE",
"(test_app1, 0002_a_new_null_field)... OK",
"(test_app2, 0001_foo)... IGNORE",
"(test_app3, 0001_initial)... IGNORE",
])
)
def test_call_linter_cmd_line_cache(self):
cache_file = os.path.join(DEFAULT_CACHE_PATH, 'test_correct_project.pickle')
if os.path.exists(cache_file):
os.remove(cache_file)
cmd = '{0} {1}'.format(
self.linter_exec,
fixtures.CORRECT_PROJECT
)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertTrue(os.path.exists(DEFAULT_CACHE_PATH))
def test_call_linter_cmd_line_cache_path(self):
if os.path.exists('/tmp/migration-linter-cache-tests/'):
shutil.rmtree('/tmp/migration-linter-cache-tests/')
cmd = '{0} {1} --cache-path={2}'.format(
self.linter_exec,
fixtures.CORRECT_PROJECT,
'/tmp/migration-linter-cache-tests/'
)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
self.assertTrue(os.path.exists('/tmp/migration-linter-cache-tests'))
def test_call_linter_cmd_line_no_cache(self):
cache_file = os.path.join(DEFAULT_CACHE_PATH, 'test_correct_project.pickle')
if os.path.exists(cache_file):
os.remove(cache_file)
cmd = '{0} --no-cache {1}'.format(
self.linter_exec,
fixtures.CORRECT_PROJECT
)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
self.assertFalse(os.path.exists(cache_file))
def test_call_linter_cmd_line_git_id(self):
cmd = '{0} --no-cache {1} d7125d5f4f0cc9623f670a66c54f131acc50032d'.format(
self.linter_exec,
fixtures.MULTI_COMMIT_PROJECT)
fixtures.prepare_git_project(fixtures.MULTI_COMMIT_PROJECT)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
lines = list(map(utils.clean_bytes_to_str, process.stdout.readlines()))
self.assertEqual(len(lines), 3)
self.assertTrue(lines[0].endswith('OK'))
self.assertTrue(lines[1].startswith('*** Summary'))
def test_call_linter_with_deleted_migrations(self):
cmd = '{0} --no-cache {1} 154ecf6119325cc3b1f3f5a4e709bfbd61a4a4ba'.format(
self.linter_exec,
fixtures.DELETED_MIGRATION_PROJECT)
fixtures.prepare_git_project(fixtures.DELETED_MIGRATION_PROJECT)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
lines = list(map(utils.clean_bytes_to_str, process.stdout.readlines()))
self.assertEqual(len(lines), 3)
self.assertTrue(lines[0].endswith('OK'))
self.assertTrue(lines[1].startswith('*** Summary'))
def test_call_from_within_project(self):
cmd = 'cd {0} && {1} --no-cache .'.format(
fixtures.CORRECT_PROJECT,
self.linter_exec)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
lines = list(map(utils.clean_bytes_to_str, process.stdout.readlines()))
self.assertEqual(len(lines), 6)
self.assertEqual(
sorted(lines[:4]),
sorted([
"(test_app1, 0001_initial)... OK",
"(test_app1, 0002_a_new_null_field)... OK",
"(test_app2, 0001_foo)... OK",
"(test_app3, 0001_initial)... OK",
])
)
def test_call_project_non_git_root(self):
# could use tag: version_ok (but doesn't work in Travis)
cmd = '{0} --no-cache {1} 2c46f438a89972f9ce2d5151a825a5bfb7f7db4b'.format(
self.linter_exec,
fixtures.NON_GIT_ROOT_DJANGO_PROJECT)
fixtures.prepare_git_project(fixtures.NON_GIT_ROOT_GIT_FOLDER)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
lines = list(map(utils.clean_bytes_to_str, process.stdout.readlines()))
self.assertEqual(len(lines), 3)
self.assertTrue(lines[0].endswith('OK'))
self.assertTrue(lines[1].startswith('*** Summary'))
def test_call_project_non_git_root_ko(self):
# could use tag: version_ko (but doesn't work in Travis)
cmd = '{0} --no-cache {1} 0140e142724fc58944797f9ddc2ebf964146339a'.format(
self.linter_exec,
fixtures.NON_GIT_ROOT_DJANGO_PROJECT)
fixtures.prepare_git_project(fixtures.NON_GIT_ROOT_GIT_FOLDER)
process = Popen(
cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertNotEqual(process.returncode, 0)
lines = list(map(utils.clean_bytes_to_str, process.stdout.readlines()))
self.assertEqual(len(lines), 5)
self.assertTrue(lines[0].endswith('ERR'))
self.assertTrue(lines[2].endswith('OK'))
self.assertTrue(lines[3].startswith('*** Summary'))
class VersionOptionLinterFromCommandLineTest(CallLinterFromCommandLineTest):
def test_call_with_version_option(self):
cmd = "{} --version".format(self.linter_exec)
process = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
process_read_stream = process.stderr if sys.version_info.major == 2 else process.stdout
lines = list(map(utils.clean_bytes_to_str, process_read_stream.readlines()))
self.assertEqual(len(lines), 1)
self.assertEqual(lines[0], "django-migration-linter {}".format(constants.__version__))
def test_call_with_short_version_option(self):
cmd = "{} -V".format(self.linter_exec)
process = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)
process.wait()
self.assertEqual(process.returncode, 0)
process_read_stream = process.stderr if sys.version_info.major == 2 else process.stdout
lines = list(map(utils.clean_bytes_to_str, process_read_stream.readlines()))
self.assertEqual(len(lines), 1)
self.assertEqual(lines[0], "django-migration-linter {}".format(constants.__version__))
+59 -62
View File
@@ -11,100 +11,97 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import os
import unittest import unittest
from django.conf import settings
from django_migration_linter import MigrationLinter from django_migration_linter import MigrationLinter
from tests import fixtures from tests import fixtures
class BaseBackwardCompatibilityDetection(object): class BaseBackwardCompatibilityDetection(object):
def tearDown(self, *args, **kwargs): def setUp(self, *args, **kwargs):
fixtures.clear_all_git_projects() self.test_project_path = os.path.dirname(settings.BASE_DIR)
super( return super(BaseBackwardCompatibilityDetection, self).setUp(*args, **kwargs)
BaseBackwardCompatibilityDetection,
self).tearDown(*args, **kwargs)
def _test_linter_finds_errors(self, path, commit_id=None): def _test_linter_finds_errors(self, app=None, commit_id=None):
linter = MigrationLinter(path, database=self.DATABASE, no_cache=True) linter = self._launch_linter(app, commit_id)
linter.lint_all_migrations(git_commit_id=commit_id)
self.assertTrue(linter.has_errors) self.assertTrue(linter.has_errors)
self.assertNotEqual(linter.nb_valid + linter.nb_erroneous, 0)
def _test_linter_finds_no_errors(self, path, commit_id=None): def _test_linter_finds_no_errors(self, app=None, commit_id=None):
linter = MigrationLinter(path, database=self.DATABASE, no_cache=True) linter = self._launch_linter(app, commit_id)
linter.lint_all_migrations(git_commit_id=commit_id)
self.assertFalse(linter.has_errors) self.assertFalse(linter.has_errors)
self.assertNotEqual(linter.nb_valid + linter.nb_erroneous, 0)
def _launch_linter(self, app=None, commit_id=None):
if app is not None:
app = [app]
linter = MigrationLinter(
self.test_project_path,
include_apps=app,
database=next(iter(self.databases)),
no_cache=True,
)
linter.lint_all_migrations(git_commit_id=commit_id)
return linter
# *** Tests *** # *** Tests ***
def test_create_table_with_not_null_column(self): def test_create_table_with_not_null_column(self):
test_project_path = fixtures.CREATE_TABLE_WITH_NOT_NULL_COLUMN_PROJECT app = fixtures.CREATE_TABLE_WITH_NOT_NULL_COLUMN
self._test_linter_finds_no_errors(test_project_path) self._test_linter_finds_no_errors(app)
def test_detect_adding_not_null_column(self): def test_detect_adding_not_null_column(self):
test_project_path = fixtures.ADD_NOT_NULL_COLUMN_PROJECT app = fixtures.ADD_NOT_NULL_COLUMN
self._test_linter_finds_errors(test_project_path) self._test_linter_finds_errors(app)
def test_detect_drop_column(self): def test_detect_drop_column(self):
test_project_path = fixtures.DROP_COLUMN_PROJECT app = fixtures.DROP_COLUMN
self._test_linter_finds_errors(test_project_path) self._test_linter_finds_errors(app)
def test_detect_rename_column(self): def test_detect_rename_column(self):
test_project_path = fixtures.RENAME_COLUMN_PROJECT app = fixtures.RENAME_COLUMN
self._test_linter_finds_errors(test_project_path) self._test_linter_finds_errors(app)
def test_detect_rename_table(self): def test_detect_rename_table(self):
test_project_path = fixtures.RENAME_TABLE_PROJECT app = fixtures.RENAME_TABLE
self._test_linter_finds_errors(test_project_path) self._test_linter_finds_errors(app)
def test_ignore_migration(self): def test_ignore_migration(self):
test_project_path = fixtures.IGNORE_MIGRATION_PROJECT app = fixtures.IGNORE_MIGRATION
self._test_linter_finds_no_errors(test_project_path) self._test_linter_finds_no_errors(app)
def test_accept_not_null_column_followed_by_adding_default(self): def test_accept_not_null_column_followed_by_adding_default(self):
test_project_path = \ app = fixtures.ADD_NOT_NULL_COLUMN_FOLLOWED_BY_DEFAULT
fixtures.ADD_NOT_NULL_COLUMN_FOLLOWED_BY_DEFAULT_PROJECT self._test_linter_finds_no_errors(app)
self._test_linter_finds_no_errors(test_project_path)
def test_detect_alter_column(self): def test_detect_alter_column(self):
test_project_path = fixtures.ALTER_COLUMN_PROJECT app = fixtures.ALTER_COLUMN
self._test_linter_finds_errors(test_project_path) self._test_linter_finds_errors(app)
def test_no_specify_git_hash(self): def test_with_git_ref(self):
test_project_path = fixtures.MULTI_COMMIT_PROJECT self._test_linter_finds_errors(commit_id="v0.1.4")
fixtures.prepare_git_project(test_project_path)
self._test_linter_finds_errors(test_project_path, commit_id=None)
def test_specify_git_hash_by_commit_hash(self):
test_project_path = fixtures.MULTI_COMMIT_PROJECT
fixtures.prepare_git_project(test_project_path)
self._test_linter_finds_no_errors(
test_project_path,
commit_id='d7125d5f4f0cc9623f670a66c54f131acc50032d')
#def test_specify_git_hash_by_tag(self):
# test_project_path = fixtures.MULTI_COMMIT_PROJECT
# fixtures.prepare_git_project(test_project_path)
# self._test_linter_finds_no_errors(test_project_path, commit_id='tag1')
class MySqlBackwardCompatibilityDetectionTest(unittest.TestCase, BaseBackwardCompatibilityDetection): class SqliteBackwardCompatibilityDetectionTestCase(
DATABASE = "mysql" BaseBackwardCompatibilityDetection, unittest.TestCase
):
class SqliteBackwardCompatibilityDetectionTest(unittest.TestCase, BaseBackwardCompatibilityDetection): databases = ["sqlite"]
DATABASE = "sqlite"
def test_accept_not_null_column_followed_by_adding_default(self): def test_accept_not_null_column_followed_by_adding_default(self):
test_project_path = \ app = fixtures.ADD_NOT_NULL_COLUMN_FOLLOWED_BY_DEFAULT
fixtures.ADD_NOT_NULL_COLUMN_FOLLOWED_BY_DEFAULT_PROJECT self._test_linter_finds_errors(app)
self._test_linter_finds_errors(test_project_path)
def test_specify_git_hash_by_commit_hash(self):
test_project_path = fixtures.MULTI_COMMIT_PROJECT
fixtures.prepare_git_project(test_project_path)
self._test_linter_finds_errors(
test_project_path,
commit_id='d7125d5f4f0cc9623f670a66c54f131acc50032d')
class PostgresqlBackwardCompatibilityDetectionTest(unittest.TestCase, BaseBackwardCompatibilityDetection): class MySqlBackwardCompatibilityDetectionTestCase(
DATABASE = "postgresql" BaseBackwardCompatibilityDetection, unittest.TestCase
):
databases = ["mysql"]
class PostgresqlBackwardCompatibilityDetectionTestCase(
BaseBackwardCompatibilityDetection, unittest.TestCase
):
databases = ["postgresql"]
-40
View File
@@ -1,40 +0,0 @@
# Copyright 2019 3YOURMIND GmbH
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import unittest
import os
from django_migration_linter import MigrationLinter
from django_migration_linter.migration import Migration
from tests import fixtures
class TestMultipleLinters(unittest.TestCase):
def test_multiple_linters(self):
l1 = MigrationLinter(fixtures.ADD_NOT_NULL_COLUMN_PROJECT)
l2 = MigrationLinter(fixtures.RENAME_COLUMN_PROJECT)
l3 = MigrationLinter(fixtures.CORRECT_PROJECT)
l1.lint_migration(Migration(os.path.join(fixtures.ADD_NOT_NULL_COLUMN_PROJECT, 'test_app/migrations/0002_add_new_not_null_field.py')))
l2.lint_migration(Migration(os.path.join(fixtures.RENAME_COLUMN_PROJECT, 'test_app/migrations/0002_rename_column.py')))
l3.lint_migration(Migration(os.path.join(fixtures.CORRECT_PROJECT, 'test_app1/migrations/0001_initial.py')))
l3.lint_migration(Migration(os.path.join(fixtures.CORRECT_PROJECT, 'test_app1/migrations/0002_a_new_null_field.py')))
self.assertTrue(l1.has_errors)
self.assertTrue(l2.has_errors)
self.assertFalse(l3.has_errors)
self.assertEqual(l1.nb_total, 1)
self.assertEqual(l2.nb_total, 1)
self.assertEqual(l3.nb_total, 2)
+31 -53
View File
@@ -13,85 +13,63 @@
# limitations under the License. # limitations under the License.
import unittest import unittest
import os
from django.db.migrations import Migration
from django_migration_linter import MigrationLinter from django_migration_linter import MigrationLinter
from django_migration_linter.migration import Migration
from tests import fixtures
class TestLinterFunctions(unittest.TestCase): class LinterFunctionsTestCase(unittest.TestCase):
def test_get_sql(self): def test_get_sql(self):
project_path = fixtures.ADD_NOT_NULL_COLUMN_PROJECT linter = MigrationLinter()
linter = MigrationLinter(project_path) sql_statements = linter.get_sql("app_add_not_null_column", "0001")
sql_statements = linter.get_sql('test_app', '0001')
self.assertEqual(len(sql_statements), 6) self.assertEqual(len(sql_statements), 6)
self.assertEqual(sql_statements[0], 'BEGIN;') self.assertEqual(sql_statements[0], "BEGIN;")
self.assertEqual(sql_statements[-1], 'COMMIT;') self.assertEqual(sql_statements[-1], "COMMIT;")
def test_has_errors(self): def test_has_errors(self):
project_path = fixtures.MULTI_COMMIT_PROJECT linter = MigrationLinter()
linter = MigrationLinter(project_path)
self.assertFalse(linter.has_errors) self.assertFalse(linter.has_errors)
m = Migration(os.path.join(project_path, 'test_app/migrations/0001_create_table.py')) m = Migration("0001_create_table", "app_add_not_null_column")
linter.lint_migration(m) linter.lint_migration(m)
self.assertFalse(linter.has_errors) self.assertFalse(linter.has_errors)
m = Migration(os.path.join(project_path, 'test_app/migrations/0002_add_new_not_null_field.py')) m = Migration("0002_add_new_not_null_field", "app_add_not_null_column")
linter.lint_migration(m) linter.lint_migration(m)
self.assertTrue(linter.has_errors) self.assertTrue(linter.has_errors)
m = Migration(os.path.join(project_path, 'test_app/migrations/0001_create_table.py')) m = Migration("0001_create_table", "app_add_not_null_column")
linter.lint_migration(m) linter.lint_migration(m)
self.assertTrue(linter.has_errors) self.assertTrue(linter.has_errors)
def test_linter_creation(self):
with self.assertRaises(ValueError):
MigrationLinter('foo/')
with self.assertRaises(ValueError):
MigrationLinter('/dev/null')
with self.assertRaises(ValueError):
MigrationLinter(fixtures.NOT_DJANGO_GIT_PROJECT)
def test_ignore_migration_include_apps(self): def test_ignore_migration_include_apps(self):
linter = MigrationLinter( linter = MigrationLinter(include_apps=("app_add_not_null_column",))
fixtures.ADD_NOT_NULL_COLUMN_PROJECT, self.assertTrue(linter.should_ignore_migration("app_correct", "0001"))
include_apps=('test_app2',)) self.assertTrue(linter.should_ignore_migration("app_correct", "0002"))
self.assertTrue(linter.should_ignore_migration('test_app1', '0001')) self.assertFalse(
self.assertFalse(linter.should_ignore_migration('test_app2', '0001')) linter.should_ignore_migration("app_add_not_null_column", "0001")
)
def test_ignore_migration_exclude_apps(self): def test_ignore_migration_exclude_apps(self):
linter = MigrationLinter( linter = MigrationLinter(exclude_apps=("app_add_not_null_column",))
fixtures.ADD_NOT_NULL_COLUMN_PROJECT, self.assertFalse(linter.should_ignore_migration("app_correct", "0001"))
exclude_apps=('test_app1',)) self.assertFalse(linter.should_ignore_migration("app_correct", "0002"))
self.assertTrue(linter.should_ignore_migration('test_app1', '0001')) self.assertTrue(
self.assertFalse(linter.should_ignore_migration('test_app2', '0001')) linter.should_ignore_migration("app_add_not_null_column", "0001")
)
def test_ignore_migration_name_contains(self): def test_ignore_migration_name_contains(self):
linter = MigrationLinter( linter = MigrationLinter(ignore_name_contains="foo")
fixtures.ADD_NOT_NULL_COLUMN_PROJECT, self.assertFalse(linter.should_ignore_migration("app_correct", "0001_initial"))
ignore_name_contains='foo') self.assertTrue(linter.should_ignore_migration("app_correct", "0002_foo"))
self.assertTrue(linter.should_ignore_migration('test_app', '0001_foo'))
self.assertFalse(linter.should_ignore_migration('test_app', '0002_bar'))
def test_ignore_migration_full_name(self): def test_ignore_migration_full_name(self):
linter = MigrationLinter( linter = MigrationLinter(ignore_name=("0002_foo",))
fixtures.ADD_NOT_NULL_COLUMN_PROJECT, self.assertFalse(linter.should_ignore_migration("app_correct", "0001_initial"))
ignore_name=('0001_foo',)) self.assertTrue(linter.should_ignore_migration("app_correct", "0002_foo"))
self.assertTrue(linter.should_ignore_migration('test_app', '0001_foo'))
self.assertFalse(linter.should_ignore_migration('test_app', '0002_bar'))
def test_gather_all_migrations(self): def test_gather_all_migrations(self):
linter = MigrationLinter(fixtures.CORRECT_PROJECT) linter = MigrationLinter()
migrations = linter._gather_all_migrations() migrations = linter._gather_all_migrations()
self.assertEqual(len(migrations), 4) self.assertGreater(len(list(migrations)), 1)
self.assertEqual(
sorted([(m.app_name, m.name) for m in migrations]),
sorted([
("test_app1", "0001_initial"),
("test_app1", "0002_a_new_null_field"),
("test_app2", "0001_foo"),
("test_app3", "0001_initial"),
])
)
+16 -44
View File
@@ -12,74 +12,46 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import os
import unittest import unittest
from tests import fixtures from django_migration_linter.utils import split_path, split_migration_path
from django_migration_linter.utils import is_django_project, is_directory, find_project_settings_module, \
split_path, split_migration_path
class UtilityFunctionTest(unittest.TestCase): class SplitPathTestCase(unittest.TestCase):
def test_detect_django_project(self):
self.assertTrue(is_django_project(fixtures.ADD_NOT_NULL_COLUMN_PROJECT))
def test_detect_not_django_project(self):
self.assertFalse(is_django_project(fixtures._BASE_DIR))
def test_detect_directory(self):
self.assertTrue(is_directory(fixtures.ADD_NOT_NULL_COLUMN_PROJECT))
def test_detect_not_directory(self):
self.assertFalse(is_directory(
os.path.join(fixtures.ADD_NOT_NULL_COLUMN_PROJECT, 'manage.py')))
def test_find_project_settings_module_same_project_name(self):
expected = 'test_project_add_not_null_column.settings'
actual = find_project_settings_module(fixtures.ADD_NOT_NULL_COLUMN_PROJECT)
self.assertEqual(actual, expected)
def test_find_project_settings_module_different_project_name(self):
expected = 'test_project.settings'
actual = find_project_settings_module(fixtures.CORRECT_PROJECT)
self.assertEqual(actual, expected)
class SplitPathTest(unittest.TestCase):
def test_split_path(self): def test_split_path(self):
split = split_path('foo/bar/fuz.py') split = split_path("foo/bar/fuz.py")
self.assertEqual(split, ["foo", "bar", "fuz.py"]) self.assertEqual(split, ["foo", "bar", "fuz.py"])
def test_split_full_path(self): def test_split_full_path(self):
split = split_path('/foo/bar/fuz.py') split = split_path("/foo/bar/fuz.py")
self.assertEqual(split, ["/", "foo", "bar", "fuz.py"]) self.assertEqual(split, ["/", "foo", "bar", "fuz.py"])
def test_split_folder_path(self): def test_split_folder_path(self):
split = split_path('/foo/bar') split = split_path("/foo/bar")
self.assertEqual(split, ["/", "foo", "bar"]) self.assertEqual(split, ["/", "foo", "bar"])
def test_split_folder_path_trailing_slash(self): def test_split_folder_path_trailing_slash(self):
split = split_path('/foo/bar/') split = split_path("/foo/bar/")
self.assertEqual(split, ["/", "foo", "bar"]) self.assertEqual(split, ["/", "foo", "bar"])
def test_split_folder_path_trailing_slashes(self): def test_split_folder_path_trailing_slashes(self):
split = split_path('/foo/bar///') split = split_path("/foo/bar///")
self.assertEqual(split, ["/", "foo", "bar"]) self.assertEqual(split, ["/", "foo", "bar"])
def test_split_migration_long_path(self): def test_split_migration_long_path(self):
input_path = 'apps/the_app/migrations/0001_stuff.py' input_path = "apps/the_app/migrations/0001_stuff.py"
app, mig = split_migration_path(input_path) app, mig = split_migration_path(input_path)
self.assertEqual(app, 'the_app') self.assertEqual(app, "the_app")
self.assertEqual(mig, '0001_stuff') self.assertEqual(mig, "0001_stuff")
def test_split_migration_path(self): def test_split_migration_path(self):
input_path = 'the_app/migrations/0001_stuff.py' input_path = "the_app/migrations/0001_stuff.py"
app, mig = split_migration_path(input_path) app, mig = split_migration_path(input_path)
self.assertEqual(app, 'the_app') self.assertEqual(app, "the_app")
self.assertEqual(mig, '0001_stuff') self.assertEqual(mig, "0001_stuff")
def test_split_migration_full_path(self): def test_split_migration_full_path(self):
input_path = '/home/user/djangostuff/apps/the_app/migrations/0001_stuff.py' input_path = "/home/user/djangostuff/apps/the_app/migrations/0001_stuff.py"
app, mig = split_migration_path(input_path) app, mig = split_migration_path(input_path)
self.assertEqual(app, 'the_app') self.assertEqual(app, "the_app")
self.assertEqual(mig, '0001_stuff') self.assertEqual(mig, "0001_stuff")
+4 -5
View File
@@ -1,12 +1,11 @@
[tox] [tox]
envlist = envlist =
py{27,34,35,36,37}-django111 py27-django111
py{34,35,36,37}-django20 py{35,36,37}-django{111,20,21}
py{35,36,37}-django21
lint lint
[testenv] [testenv]
commands = py.test {posargs} commands = ./manage.py test --no-input {posargs}
extras = test extras = test
pip_pre = true pip_pre = true
deps = deps =
@@ -23,4 +22,4 @@ deps =
black black
commands = commands =
flake8 --max-line-length=88 django_migration_linter flake8 --max-line-length=88 django_migration_linter
black --check django_migration_linter black --check django_migration_linter/ tests/ manage.py setup.py