1
"""Tests for assess_model_migration module."""
5
from mock import call, Mock, patch
9
import assess_model_migration as amm
10
from deploy_stack import BootstrapManager
21
class TestParseArgs(TestCase):
23
def test_default_args(self):
24
args = amm.parse_args(
25
["an-env", "/bin/juju", "/tmp/logs", "an-env-mod"])
32
temp_env_name='an-env-mod',
45
fake_stdout = StringIO.StringIO()
46
with parse_error(self) as fake_stderr:
47
with patch("sys.stdout", fake_stdout):
48
amm.parse_args(["--help"])
49
self.assertEqual("", fake_stderr.getvalue())
51
"Test model migration feature", fake_stdout.getvalue())
54
class TestGetBootstrapManagers(TestCase):
55
def test_returns_two_bs_managers(self):
56
ret_bs = [Mock(), Mock()]
57
with temp_dir() as log_dir:
58
args = argparse.Namespace(logs=log_dir)
60
BootstrapManager, 'from_args', side_effect=ret_bs):
61
bs1, bs2 = amm.get_bootstrap_managers(args)
62
self.assertEqual(bs1, ret_bs[0])
63
self.assertEqual(bs2, ret_bs[1])
65
def test_gives_second_manager_unique_env(self):
67
mock_bs1.temp_env_name = 'testing-env-name'
68
ret_bs = [mock_bs1, Mock()]
69
with temp_dir() as log_dir:
70
args = argparse.Namespace(logs=log_dir)
71
with patch.object(BootstrapManager, 'from_args',
73
bs1, bs2 = amm.get_bootstrap_managers(args)
74
self.assertEqual(bs2.temp_env_name, 'testing-env-name-b')
76
def test_creates_unique_log_dirs(self):
77
ret_bs = [Mock(), Mock()]
78
args = argparse.Namespace(logs='/some/path')
79
with patch.object(BootstrapManager, 'from_args', side_effect=ret_bs):
80
with patch.object(amm, '_new_log_dir') as log_dir:
81
bs1, bs2 = amm.get_bootstrap_managers(args)
82
self.assertEqual(2, log_dir.call_count)
85
[call(args.logs, 'a'), call(args.logs, 'b')])
88
class TestNewLogDir(TestCase):
90
def test_returns_created_log_path(self):
91
with temp_dir() as log_dir_path:
93
expected_path = '{}/env-testing'.format(log_dir_path)
94
log_dir = amm._new_log_dir(log_dir_path, post_fix)
95
self.assertEqual(log_dir, expected_path)
97
def test_creates_new_log_dir(self):
98
with temp_dir() as log_dir_path:
100
expected_path = '{}/env-testing'.format(log_dir_path)
101
amm._new_log_dir(log_dir_path, post_fix)
102
self.assertTrue(os.path.exists(expected_path))
105
class TestWaitForModel(TestCase):
106
# Check that it returns an error if the model never comes up.
107
# Pass in a timeout for the model check
108
def test_raises_exception_when_timeout_occurs(self):
109
with patch.object(until_timeout, 'next', side_effect=StopIteration()):
110
with self.assertRaises(AssertionError):
111
amm.wait_for_model(Mock(), 'TestModelName')
113
def test_returns_when_model_found(self):
115
mock_client.get_models.return_value = dict(
117
dict(name='TestModelName')])
118
amm.wait_for_model(mock_client, 'TestModelName')
120
def test_pauses_between_failed_matches(self):
122
mock_client.get_models.side_effect = [
123
dict(models=[]), # Failed check
124
dict(models=[dict(name='TestModelName')]), # Successful check
126
with patch.object(amm, 'sleep') as mock_sleep:
127
amm.wait_for_model(mock_client, 'TestModelName')
128
mock_sleep.assert_called_once_with(1)
131
class TestMain(TestCase):
134
argv = ["an-env", "/bin/juju", "/tmp/logs", "an-env-mod", "--verbose"]
137
with patch.object(amm, "configure_logging", autospec=True) as mock_cl:
138
with patch.object(amm, "assess_model_migration",
139
autospec=True) as mock_assess:
140
with patch.object(amm, "get_bootstrap_managers",
141
return_value=[bs_1, bs_2]):
143
mock_cl.assert_called_once_with(logging.DEBUG)
144
mock_assess.assert_called_once_with(bs_1, bs_2, False)
added
removed
tests/test_assess_model_migration.py
