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

import os
import pytest
import re
import subprocess

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

from . import conftest

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')]

WHITESPACE_REGEX = re.compile(r'\s+')


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():
        result = list(db.engine.execute(text('''
            SELECT type, name, tbl_name, sql
            FROM sqlite_master
            ORDER BY type, name, tbl_name
            ''')))

    return {(x[0], x[1], x[2]): x[3] for x in result}


def assert_schemas_equal(left, right):
    for (k, v) in list(left.items()):
        if k not in right:
            raise AssertionError(
                'Left contained {} but right did not'.format(k))
        if not ddl_equal(v, right[k]):
            raise AssertionError(
                'Schema for {} did not match:\nLeft:\n{}\nRight:\n{}'
                .format(k, v, right[k]))
        right.pop(k)

    if right:
        raise AssertionError(
            'Right had additional tables: {}'.format(list(right.keys())))


def ddl_equal(left, right):
    '''Check the "tokenized" DDL is equivalent because, because sometimes
        Alembic schemas append columns on the same line to the DDL comes out
        like:

        column1 TEXT NOT NULL, column2 TEXT NOT NULL

        and SQLAlchemy comes out:

        column1 TEXT NOT NULL,
        column2 TEXT NOT NULL
    '''
    # ignore the autoindex cases
    if left is None and right is None:
        return True

    left = [x for x in WHITESPACE_REGEX.split(left) if x]
    right = [x for x in WHITESPACE_REGEX.split(right) if x]

    # Strip commas and quotes
    left = [x.replace("\"", "").replace(",", "") for x in left]
    right = [x.replace("\"", "").replace(",", "") for x in right]

    return sorted(left) == sorted(right)


def test_alembic_head_matches_db_models(journalist_app,
                                        alembic_config,
                                        config):
    '''This test is to make sure that our database models in `models.py` are
       always in sync with the schema generated by `alembic upgrade head`.
    '''
    models_schema = get_schema(journalist_app)

    config.DATABASE_FILE = config.DATABASE_FILE + '.new'
    # Use the fixture to rewrite the config with the new URI
    conftest.alembic_config(config)

    # Create database file
    subprocess.check_call(['sqlite3', config.DATABASE_FILE, '.databases'])
    upgrade(alembic_config, 'head')

    # Recreate the app to get a new SQLALCHEMY_DATABASE_URI
    app = create_app(config)
    alembic_schema = get_schema(app)

    # The initial migration creates the table 'alembic_version', but this is
    # not present in the schema created by `db.create_all()`.
    alembic_schema = {k: v for k, v in list(alembic_schema.items())
                      if k[2] != 'alembic_version'}

    assert_schemas_equal(alembic_schema, models_schema)


@pytest.mark.parametrize('migration', ALL_MIGRATIONS)<commit_msg>Fix pytest complaint about test_alembic<commit_after># -*- coding: utf-8 -*-

import os
import pytest
import re
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')]

WHITESPACE_REGEX = re.compile(r'\s+')


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():
        result = list(db.engine.execute(text('''
            SELECT type, name, tbl_name, sql
            FROM sqlite_master
            ORDER BY type, name, tbl_name
            ''')))

    return {(x[0], x[1], x[2]): x[3] for x in result}


def assert_schemas_equal(left, right):
    for (k, v) in list(left.items()):
        if k not in right:
            raise AssertionError(
                'Left contained {} but right did not'.format(k))
        if not ddl_equal(v, right[k]):
            raise AssertionError(
                'Schema for {} did not match:\nLeft:\n{}\nRight:\n{}'
                .format(k, v, right[k]))
        right.pop(k)

    if right:
        raise AssertionError(
            'Right had additional tables: {}'.format(list(right.keys())))


def ddl_equal(left, right):
    '''Check the "tokenized" DDL is equivalent because, because sometimes
        Alembic schemas append columns on the same line to the DDL comes out
        like:

        column1 TEXT NOT NULL, column2 TEXT NOT NULL

        and SQLAlchemy comes out:

        column1 TEXT NOT NULL,
        column2 TEXT NOT NULL
    '''
    # ignore the autoindex cases
    if left is None and right is None:
        return True

    left = [x for x in WHITESPACE_REGEX.split(left) if x]
    right = [x for x in WHITESPACE_REGEX.split(right) if x]

    # Strip commas and quotes
    left = [x.replace("\"", "").replace(",", "") for x in left]
    right = [x.replace("\"", "").replace(",", "") for x in right]

    return sorted(left) == sorted(right)


def test_alembic_head_matches_db_models(journalist_app,
                                        alembic_config,
                                        config):
    '''This test is to make sure that our database models in `models.py` are
       always in sync with the schema generated by `alembic upgrade head`.
    '''
    models_schema = get_schema(journalist_app)

    os.remove(config.DATABASE_FILE)

    # Create database file
    subprocess.check_call(['sqlite3', config.DATABASE_FILE, '.databases'])
    upgrade(alembic_config, 'head')

    # Recreate the app to get a new SQLALCHEMY_DATABASE_URI
    app = create_app(config)
    alembic_schema = get_schema(app)

    # The initial migration creates the table 'alembic_version', but this is
    # not present in the schema created by `db.create_all()`.
    alembic_schema = {k: v for k, v in list(alembic_schema.items())
                      if k[2] != 'alembic_version'}

    assert_schemas_equal(alembic_schema, models_schema)


@pytest.mark.parametrize('migration', ALL_MIGRATIONS)
{'fraction_non_alphanumeric': 0.10839340885684862, 'fraction_numerical': 0.0033470648815653965, 'mean_word_length': 2.650845864661654, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '797410', 'n_tokens_mistral': 2442, 'n_tokens_neox': 2358, 'n_words': 690}