<commit_before># -*- coding: utf-8 -*-

import os
import pytest
import subprocess

from alembic.config import Config as AlembicConfig
from alembic.script import ScriptDirectory
from os import path

MIGRATION_PATH = path.join(path.dirname(__file__), '..', 'alembic', 'versions')

ALL_MIGRATIONS = [x.split('.')[0].split('_')[0]
                  for x in os.listdir(MIGRATION_PATH)
                  if x.endswith('.py')]


def list_migrations(cfg_path, head):
    cfg = AlembicConfig(cfg_path)
    script = ScriptDirectory.from_config(cfg)
    migrations = [x.revision
                  for x in script.walk_revisions(base='base', head=head)]
    migrations.reverse()
    return migrations


def upgrade(alembic_config, migration):
    subprocess.check_call(['alembic', 'upgrade', migration],
                          cwd=path.dirname(alembic_config))


def downgrade(alembic_config, migration):
    subprocess.check_call(['alembic', 'downgrade', migration],
                          cwd=path.dirname(alembic_config))


@pytest.mark.parametrize('migration', ALL_MIGRATIONS)
def test_alembic_migration_upgrade(alembic_config, config, migration):
    # run migrations in sequence from base -> head
    for mig in list_migrations(alembic_config, migration):
        upgrade(alembic_config, mig)


@pytest.mark.parametrize('migration', ALL_MIGRATIONS)
def test_alembic_migration_downgrade(alembic_config, config, migration):
    # upgrade to the parameterized test case ("head")
    upgrade(alembic_config, migration)

    # run migrations in sequence from "head" -> base
    migrations = list_migrations(alembic_config, migration)
    migrations.reverse()

    for mig in migrations:
        downgrade(alembic_config, mig)<commit_msg>added test up/downgrade schema idempotence<commit_after># -*- coding: utf-8 -*-

import os
import pytest
import subprocess

from alembic.config import Config as AlembicConfig
from alembic.script import ScriptDirectory
from os import path
from sqlalchemy import text

from db import db
from journalist_app import create_app

MIGRATION_PATH = path.join(path.dirname(__file__), '..', 'alembic', 'versions')

ALL_MIGRATIONS = [x.split('.')[0].split('_')[0]
                  for x in os.listdir(MIGRATION_PATH)
                  if x.endswith('.py')]


def list_migrations(cfg_path, head):
    cfg = AlembicConfig(cfg_path)
    script = ScriptDirectory.from_config(cfg)
    migrations = [x.revision
                  for x in script.walk_revisions(base='base', head=head)]
    migrations.reverse()
    return migrations


def upgrade(alembic_config, migration):
    subprocess.check_call(['alembic', 'upgrade', migration],
                          cwd=path.dirname(alembic_config))


def downgrade(alembic_config, migration):
    subprocess.check_call(['alembic', 'downgrade', migration],
                          cwd=path.dirname(alembic_config))


def get_schema(app):
    with app.app_context():
        return list(db.engine.execute(text('''
            SELECT type, name, tbl_name, sql
            FROM sqlite_master
            ORDER BY type, name, tbl_name
            ''')))


@pytest.mark.parametrize('migration', ALL_MIGRATIONS)
def test_alembic_migration_upgrade(alembic_config, config, migration):
    # run migrations in sequence from base -> head
    for mig in list_migrations(alembic_config, migration):
        upgrade(alembic_config, mig)


@pytest.mark.parametrize('migration', ALL_MIGRATIONS)
def test_alembic_migration_downgrade(alembic_config, config, migration):
    # upgrade to the parameterized test case ("head")
    upgrade(alembic_config, migration)

    # run migrations in sequence from "head" -> base
    migrations = list_migrations(alembic_config, migration)
    migrations.reverse()

    for mig in migrations:
        downgrade(alembic_config, mig)


@pytest.mark.parametrize('migration', ALL_MIGRATIONS)
def test_schema_unchanged_after_up_then_downgrade(alembic_config,
                                                  config,
                                                  migration):
    # Create the app here. Using a fixture will init the database.
    app = create_app(config)

    migrations = list_migrations(alembic_config, migration)

    if len(migrations) > 1:
        target = migrations[-2]
        upgrade(alembic_config, target)
    else:
        # The first migration is the degenerate case where we don't need to
        # get the database to some base state.
        pass

    original_schema = get_schema(app)

    upgrade(alembic_config, '+1')
    downgrade(alembic_config, '-1')

    reverted_schema = get_schema(app)

    # The initial migration is a degenerate case because it creates the table
    # 'alembic_version', but rolling back the migration doesn't clear it.
    if len(migrations) == 1:
        reverted_schema = list(filter(lambda x: x[2] != 'alembic_version',
                                      reverted_schema))

    assert reverted_schema == original_schema
{'fraction_non_alphanumeric': 0.09678068410462777, 'fraction_numerical': 0.002414486921529175, 'mean_word_length': 3.1913996627318717, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3288881', 'n_tokens_mistral': 1524, 'n_tokens_neox': 1507, 'n_words': 360}