def test_run_migrations_success(self):
"""
Test the migration success path.
"""
# 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]
"""
output = [
{
'database': 'default',
'duration': None,
'failed_migration': None,
'migration': 'all',
'output': None,
'succeeded_migrations': [
['release_util', '0001_initial'],
['release_util', '0002_second'],
['release_util', '0003_third'],
],
'traceback': None,
'succeeded': True,
},
]
out_file = tempfile.NamedTemporaryFile(suffix='.yml')
in_file = tempfile.NamedTemporaryFile(suffix='.yml')
in_file.write(input_yaml.encode('utf-8'))
in_file.flush()
# Check the stdout output against the expected output.
self._check_command_output(
cmd='run_migrations',
cmd_args=(in_file.name,),
cmd_kwargs={'output_file': out_file.name},
output=output,
)
in_file.close()
# Check the contents of the output file against the expected output.
with open(out_file.name, 'r') as f:
output_yaml = f.read()
parsed_yaml = yaml.safe_load(output_yaml)
self.assertTrue(isinstance(parsed_yaml, list))
parsed_yaml = self._null_certain_fields(parsed_yaml)
self.assertEqual(yaml.dump(output), yaml.dump(parsed_yaml))
out_file.close()
评论列表
文章目录