<commit_before># -*- coding: utf-8 -*-
# Copyright (c) 2015 The Pycroft Authors. See the AUTHORS file.
# This file is part of the Pycroft project and licensed under the terms of
# the Apache License, Version 2.0. See the LICENSE file for details.
from contextlib import contextmanager
from functools import partial
import os
import random
import string
import unittest
from flask import url_for, _request_ctx_stack
from flask.ext import testing
from fixture.style import NamedDataStyle
from fixture import SQLAlchemyFixture, DataTestCase
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import SingletonThreadPool
import sys
from werkzeug.routing import IntegerConverter, UnicodeConverter
from pycroft._compat import iteritems
from pycroft.model import session
from pycroft.model import _all, drop_db_model, create_db_model


engine = None
connection = None
_setup_stack = 0


def setup():
    global engine, connection, _setup_stack
    _setup_stack += 1
    if _setup_stack > 1:
        return
    try:
        uri = os.environ['PYCROFT_DB_URI']
    except KeyError:
        raise RuntimeError("Environment variable PYCROFT_DB_URI must be "
                           "set to an SQLalchemy connection string.")
    engine = create_engine(uri, poolclass=SingletonThreadPool)
    connection = engine.connect()
    drop_db_model(connection)
    create_db_model(connection)


def teardown():
    global engine, connection, _setup_stack
    _setup_stack -= 1
    if _setup_stack > 0:
        return
    drop_db_model(connection)
    connection.close()
    engine = None
    connection = None


class FixtureDataTestBase(DataTestCase, unittest.TestCase):
    """A TestCase baseclass that handles database fixtures.

    You only need to define a `datasets` class member with a list of
    the fixture DataSets. The type of the fixture element will be taken
    from the name of the DataSet class. It needs "Data" as suffic. So if
    you want to provide fixtures for the User model the name of the DataSet
    has to be "UserData". See also test_property.py for an example.

    If you overwrite the `tearDown` or `setUpClass` methods don't forget
    to call the ones in the superclass.
    """
    @classmethod
    def setUpClass(cls):
        setup()
        cls.fixture = SQLAlchemyFixture(
            env=_all, style=NamedDataStyle(),
            engine=connection
        )

    @classmethod
    def tearDownClass(cls):
        teardown()

    def setUp(self):
        super(FixtureDataTestBase, self).setUp()
        self.transaction = connection.begin_nested()
        s = scoped_session(sessionmaker(bind=connection))
        session.set_scoped_session(s)

    def tearDown(self):
        # Rollback the session
        session.session.rollback()
        session.Session.remove()
        # Rollback the outer transaction to the savepoint
        self.transaction.rollback()
        super(FixtureDataTestBase, self).tearDown()

    @contextmanager
    def _rollback_with_context(self, context):
        with context:
            try:
                yield
            except:
                # Check if the session is in “partial rollback” state
                if not session.session.is_active:
                    session.session.rollback()
                raise

    def assertRaises(self, excClass, callableObj=None, *args, **kwargs):<commit_msg>Cleanup fixtures if setUp methods fail<commit_after># -*- coding: utf-8 -*-
# Copyright (c) 2015 The Pycroft Authors. See the AUTHORS file.
# This file is part of the Pycroft project and licensed under the terms of
# the Apache License, Version 2.0. See the LICENSE file for details.
from contextlib import contextmanager
from functools import partial
import os
import random
import string
import unittest
from flask import url_for, _request_ctx_stack
from flask.ext import testing
from fixture.style import NamedDataStyle
from fixture import SQLAlchemyFixture, DataTestCase
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import SingletonThreadPool
import sys
from werkzeug.routing import IntegerConverter, UnicodeConverter
from pycroft._compat import iteritems
from pycroft.model import session
from pycroft.model import _all, drop_db_model, create_db_model


engine = None
connection = None
_setup_stack = 0


def setup():
    global engine, connection, _setup_stack
    _setup_stack += 1
    if _setup_stack > 1:
        return
    try:
        uri = os.environ['PYCROFT_DB_URI']
    except KeyError:
        raise RuntimeError("Environment variable PYCROFT_DB_URI must be "
                           "set to an SQLalchemy connection string.")
    engine = create_engine(uri, poolclass=SingletonThreadPool)
    connection = engine.connect()
    drop_db_model(connection)
    create_db_model(connection)


def teardown():
    global engine, connection, _setup_stack
    _setup_stack -= 1
    if _setup_stack > 0:
        return
    drop_db_model(connection)
    connection.close()
    engine = None
    connection = None


class FixtureDataTestBase(DataTestCase, unittest.TestCase):
    """A TestCase baseclass that handles database fixtures.

    You only need to define a `datasets` class member with a list of
    the fixture DataSets. The type of the fixture element will be taken
    from the name of the DataSet class. It needs "Data" as suffic. So if
    you want to provide fixtures for the User model the name of the DataSet
    has to be "UserData". See also test_property.py for an example.

    If you overwrite the `tearDown` or `setUpClass` methods don't forget
    to call the ones in the superclass.
    """
    @classmethod
    def setUpClass(cls):
        setup()
        cls.fixture = SQLAlchemyFixture(
            env=_all, style=NamedDataStyle(),
            engine=connection
        )

    @classmethod
    def tearDownClass(cls):
        teardown()

    def setUp(self):
        super(FixtureDataTestBase, self).setUp()
        self.transaction = connection.begin_nested()
        s = scoped_session(sessionmaker(bind=connection))
        session.set_scoped_session(s)
        self.addCleanup(self.cleanup)

    def _rollback(self):
        # Rollback the session
        session.session.rollback()
        session.Session.remove()
        # Rollback the outer transaction to the savepoint
        self.transaction.rollback()
        self.transaction = None

    def tearDown(self):
        self._rollback()
        super(FixtureDataTestBase, self).tearDown()

    def cleanup(self):
        if self.transaction is None:
            return
        self._rollback()
        self.data.teardown()

    @contextmanager
    def _rollback_with_context(self, context):
        with context:
            try:
                yield
            except:
                # Check if the session is in “partial rollback” state
                if not session.session.is_active:
                    session.session.rollback()
                raise

    def assertRaises(self, excClass, callableObj=None, *args, **kwargs):
{'fraction_non_alphanumeric': 0.06544943820224719, 'fraction_numerical': 0.0033707865168539327, 'mean_word_length': 2.916941694169417, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3052388', 'n_tokens_mistral': 1998, 'n_tokens_neox': 1910, 'n_words': 667}