tests.py 文件源码

def test_run_migrations_failure(self, migration_name, migration_output):
        """
        Test the first, second, and last migration failing.
        """
        # Using TransactionTestCase sets up the migrations as set up for the test.
        # Reset the release_util migrations to the very beginning - i.e. no tables.
        call_command("migrate", "release_util", "zero", verbosity=0)

        input_yaml = """
        database: 'default',
        migrations:
          - [release_util, 0001_initial]
          - [release_util, 0002_second]
          - [release_util, 0003_third]
        initial_states:
          - [release_util, zero]
        """
        out_file = tempfile.NamedTemporaryFile(suffix='.yml')
        in_file = tempfile.NamedTemporaryFile(suffix='.yml')
        in_file.write(input_yaml.encode('utf-8'))
        in_file.flush()

        # A bogus class for creating a migration object that will raise a CommandError.
        class MigrationFail(object):
            atomic = False
            def state_forwards(self, app_label, state):
                pass
            def database_forwards(self, app_label, schema_editor, from_state, to_state):
                raise CommandError("Yo")

        # Insert the bogus object into the first operation of a migration.
        current_migration_list = release_util.tests.migrations.test_migrations.__dict__[migration_name].__dict__['Migration'].operations
        current_migration_list.insert(0, MigrationFail())

        try:
            # Check the stdout output.
            self._check_command_output(
                cmd="run_migrations",
                cmd_args=(in_file.name,),
                cmd_kwargs={'output_file': out_file.name},
                output=migration_output,
                err_output="Migration error: Migration failed for app 'release_util' - migration '{}'.".format(migration_name),
                exit_value=1
            )
        finally:
            # Whether the test passes or fails, always pop the failure migration of the list.
            current_migration_list.pop(0)

        in_file.close()
        out_file.close()