Skip to content

Commit

Permalink
Merge pull request #126 from peopledoc/fix-0.7
Browse files Browse the repository at this point in the history
  • Loading branch information
Joachim Jablon authored Dec 20, 2021
2 parents 206f54c + 9414135 commit 9d8899b
Show file tree
Hide file tree
Showing 3 changed files with 89 additions and 101 deletions.
7 changes: 2 additions & 5 deletions septentrion/utils.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,9 +44,6 @@ def since(iterable: Iterable[T], value: T) -> Iterable[T]:
"""
Returns the values from iterable starting after the element is found
>>> list(until(range(300), 297))
[298, 299]
[297, 298, 299]
"""
it = itertools.dropwhile((lambda x: x != value), iterable)
# Drop the first element
next(it)
yield from it
yield from itertools.dropwhile((lambda x: x != value), iterable)
175 changes: 80 additions & 95 deletions tests/unit/test_core.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,13 @@

import pytest

from septentrion import configuration, core, exceptions, versions
from septentrion import configuration, core, exceptions
from septentrion.versions import Version


@pytest.fixture
def known_versions(mocker):
versions_ = [versions.Version.from_string(v) for v in ("0", "1.1", "1.2", "1.3")]
versions_ = [Version.from_string(v) for v in ("0", "1.1", "1.2", "1.3")]
mocker.patch("septentrion.core.files.get_known_versions", return_value=versions_)

return versions_
Expand All @@ -18,26 +18,24 @@ def test_get_applied_versions(mocker, known_versions):
mocker.patch(
"septentrion.core.db.get_applied_versions",
return_value=[
versions.Version.from_string("1.0"),
versions.Version.from_string("1.1"),
Version.from_string("1.0"),
Version.from_string("1.1"),
],
)
settings = configuration.Settings()
versions_ = core.get_applied_versions(settings=settings)

assert versions_ == [versions.Version.from_string("1.1")]
assert versions_ == [Version.from_string("1.1")]


def test_get_closest_version_unknown_target_version(known_versions):
settings = configuration.Settings(
target_version=versions.Version.from_string("1.5")
)
settings = configuration.Settings(target_version=Version.from_string("1.5"))

# target_version is not a known version
with pytest.raises(ValueError):
core.get_closest_version(
settings=settings,
target_version=versions.Version.from_string("1.5"),
target_version=Version.from_string("1.5"),
sql_tpl="schema_{}.sql",
existing_files=[],
)
Expand All @@ -48,20 +46,20 @@ def test_get_closest_version_ok(known_versions):

version = core.get_closest_version(
settings=settings,
target_version=versions.Version.from_string("1.1"),
target_version=Version.from_string("1.1"),
sql_tpl="schema_{}.sql",
existing_files=["schema_1.0.sql", "schema_1.1.sql"],
)

assert version == versions.Version.from_string("1.1")
assert version == Version.from_string("1.1")


def test_get_closest_version_schema_doesnt_exist(known_versions):
settings = configuration.Settings()

version = core.get_closest_version(
settings=settings,
target_version=versions.Version.from_string("1.1"),
target_version=Version.from_string("1.1"),
sql_tpl="schema_{}.sql",
existing_files=["schema_1.0.sql", "schema_1.2.sql"],
)
Expand All @@ -75,10 +73,10 @@ def test_get_closest_version_earlier_schema(known_versions):

version = core.get_closest_version(
settings=settings,
target_version=versions.Version.from_string("1.3"),
target_version=Version.from_string("1.3"),
sql_tpl="schema_{}.sql",
existing_files=["schema_1.0.sql", "schema_1.1.sql"],
force_version=versions.Version.from_string("1.2"),
force_version=Version.from_string("1.2"),
)

assert version is None
Expand All @@ -93,10 +91,10 @@ def test_get_closest_version_schema_force_ko(known_versions):
with pytest.raises(ValueError):
core.get_closest_version(
settings=settings,
target_version=versions.Version.from_string("1.1"),
target_version=Version.from_string("1.1"),
sql_tpl="schema_{}.sql",
existing_files=["schema_1.0.sql", "schema_1.1.sql"],
force_version=versions.Version.from_string("1.4"),
force_version=Version.from_string("1.4"),
)


Expand All @@ -105,24 +103,24 @@ def test_get_closest_version_schema_force_ok(known_versions):

version = core.get_closest_version(
settings=settings,
target_version=versions.Version.from_string("1.3"),
target_version=Version.from_string("1.3"),
sql_tpl="schema_{}.sql",
existing_files=["schema_1.2.sql", "schema_1.3.sql"],
force_version=versions.Version.from_string("1.2"),
force_version=Version.from_string("1.2"),
)

assert version == versions.Version.from_string("1.2")
assert version == Version.from_string("1.2")


def test_get_closest_version_schema_force_dont_exist(known_versions):
settings = configuration.Settings()

version = core.get_closest_version(
settings=settings,
target_version=versions.Version.from_string("1.3"),
target_version=Version.from_string("1.3"),
sql_tpl="schema_{}.sql",
existing_files=["schema_1.1.sql", "schema_1.3.sql"],
force_version=versions.Version.from_string("1.2"),
force_version=Version.from_string("1.2"),
)

# schema_1.2.sql doesn't exist
Expand All @@ -134,138 +132,125 @@ def test_get_best_schema_version_ok(mocker, known_versions):
"septentrion.core.files.get_special_files",
return_value=["schema_1.1.sql", "schema_1.2.sql"],
)
settings = configuration.Settings(
target_version=versions.Version.from_string("1.2")
)
settings = configuration.Settings(target_version=Version.from_string("1.2"))

version = core.get_best_schema_version(settings=settings)

assert version == versions.Version.from_string("1.2")
assert version == Version.from_string("1.2")


def test_get_best_schema_version_ko(mocker, known_versions):
mocker.patch(
"septentrion.core.files.get_special_files",
return_value=["schema_1.0.sql", "schema_1.3.sql"],
)
settings = configuration.Settings(
target_version=versions.Version.from_string("1.2")
)
settings = configuration.Settings(target_version=Version.from_string("1.2"))

with pytest.raises(exceptions.SeptentrionException):
core.get_best_schema_version(settings=settings)


def test_build_migration_plan_unknown_version(known_versions):
settings = configuration.Settings(
target_version=versions.Version.from_string("1.5")
)
from_version = versions.Version.from_string("0")
settings = configuration.Settings(target_version=Version.from_string("1.5"))
from_version = Version.from_string("0")

with pytest.raises(ValueError):
list(core.build_migration_plan(settings, from_version=from_version))


def test_build_migration_plan_ok(mocker, known_versions):
mocker.patch("septentrion.core.db.get_applied_migrations", return_value=[])
def test_build_migration_plan_db(mocker, known_versions):
# What a mock hell ><

# So first, we mock db.get_applied_migrations to tell the following story:
# - on 1.1, only migration "a" was previously applied.
# - on 1.2, no migration was previously applied.
mocker.patch(
"septentrion.core.files.get_migrations_files_mapping",
return_value={
"file.ddl.sql": pathlib.Path(
"tests/test_data/sql/17.1/manual/file.ddl.sql"
),
"file.dml.sql": pathlib.Path(
"tests/test_data/sql/17.1/manual/file.dml.sql"
),
},
"septentrion.db.get_applied_migrations",
side_effect=lambda settings, version: {
Version.from_string("1.1"): ["a"],
Version.from_string("1.2"): [],
}[version],
)
# Then, regarding the migration files that exist on the disk:
# - There are 2 files for 1.1 (so one already applied and one new)
# - 1 file for 1.2
# - 1 file for 1.3
mocker.patch(
"septentrion.files.get_migrations_files_mapping",
side_effect=lambda settings, version: {
Version.from_string("1.1"): {
"a": pathlib.Path("a"),
"b": pathlib.Path("b"),
},
Version.from_string("1.2"): {
"c": pathlib.Path("c"),
},
Version.from_string("1.3"): {"d": pathlib.Path("d")},
}[version],
)
mocker.patch("septentrion.core.files.is_manual_migration", return_value=True)
# The contents of each migration is ignored
mocker.patch("septentrion.files.file_lines_generator", return_value="")
# Migration "c" is a manual migration
mocker.patch(
"septentrion.files.is_manual_migration",
side_effect=(
lambda migration_path, migration_contents: str(migration_path) == "c"
),
)
# We'll apply migrations up until 1.2 included
settings = configuration.Settings(
target_version=versions.Version.from_string("1.2")
target_version=Version.from_string("1.2"),
)
from_version = versions.Version.from_string("0")
# And we'll start at version 1.1 included
from_version = Version.from_string("1.1")
plan = core.build_migration_plan(settings=settings, from_version=from_version)

expected = [
{
"version": Version.from_string("1.1"),
"plan": [
(
"file.ddl.sql",
False,
pathlib.Path("tests/test_data/sql/17.1/manual/file.ddl.sql"),
True,
),
(
"file.dml.sql",
False,
pathlib.Path("tests/test_data/sql/17.1/manual/file.dml.sql"),
True,
),
# On 1.1, migration a is already applied. It's not manual
("a", True, pathlib.Path("a"), False),
# migration b, though needs to be applied. It's not manual
("b", False, pathlib.Path("b"), False),
],
"version": versions.Version.from_string("1.1"),
},
{
"version": Version.from_string("1.2"),
"plan": [
(
"file.ddl.sql",
False,
pathlib.Path("tests/test_data/sql/17.1/manual/file.ddl.sql"),
True,
),
(
"file.dml.sql",
False,
pathlib.Path("tests/test_data/sql/17.1/manual/file.dml.sql"),
True,
),
# migration c also needs to be applied. It's manual (the last True)
("c", False, pathlib.Path("c"), True),
],
"version": versions.Version.from_string("1.2"),
},
]
assert list(plan) == expected


def test_build_migration_plan_db_uptodate(mocker, known_versions):
mocker.patch(
"septentrion.core.db.get_applied_migrations",
return_value=[Version.from_string("1.1"), Version.from_string("1.2")],
)
settings = configuration.Settings(
target_version=versions.Version.from_string("1.2"),
)

from_version = versions.Version.from_string("0")
plan = core.build_migration_plan(settings=settings, from_version=from_version)

expected = [
{"plan": [], "version": versions.Version.from_string("1.1")},
{"plan": [], "version": versions.Version.from_string("1.2")},
]
assert list(plan) == expected


def test_build_migration_plan_with_schema(mocker, known_versions):
mocker.patch("septentrion.core.db.get_applied_migrations", return_value=[])
settings = configuration.Settings(target_version="1.2")
from_version = versions.Version.from_string("1.1")
from_version = Version.from_string("1.1")

plan = list(core.build_migration_plan(settings=settings, from_version=from_version))

expected = [
{"plan": [], "version": versions.Version.from_string("1.2")},
{"plan": [], "version": Version.from_string("1.1")},
{"plan": [], "version": Version.from_string("1.2")},
]
assert list(plan) == expected


def test_build_migration_plan_with_no_target_version(mocker, known_versions):
mocker.patch("septentrion.core.db.get_applied_migrations", return_value=[])
settings = configuration.Settings(target_version=None)
from_version = versions.Version.from_string("1.1")
from_version = Version.from_string("1.1")

plan = list(core.build_migration_plan(settings=settings, from_version=from_version))

expected = [
{"plan": [], "version": versions.Version.from_string("1.2")},
{"plan": [], "version": versions.Version.from_string("1.3")},
{"plan": [], "version": Version.from_string("1.1")},
{"plan": [], "version": Version.from_string("1.2")},
{"plan": [], "version": Version.from_string("1.3")},
]
assert list(plan) == expected
8 changes: 7 additions & 1 deletion tests/unit/test_utils.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import pytest

from septentrion.utils import is_version, until
from septentrion.utils import is_version, since, until


@pytest.mark.parametrize("value,expected", [("1.2", True), ("bananas", False)])
Expand All @@ -17,3 +17,9 @@ def test_until():
def test_until_error():
with pytest.raises(ValueError):
list(until(range(5), 10))


def test_since():
values = list(since(range(300), 297))

assert values == [297, 298, 299]

0 comments on commit 9d8899b

Please sign in to comment.