1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
# Copyright 2009-2010 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
# pylint: disable-msg=E0702
"""Test handling of EC2 machine images."""
__metaclass__ = type
from unittest import TestCase
from devscripts.ec2test.instance import EC2Instance
from lp.testing.fakemethod import FakeMethod
class FakeAccount:
"""Helper for setting up an `EC2Instance` without EC2."""
acquire_private_key = FakeMethod()
acquire_security_group = FakeMethod()
class FakeOutput:
"""Pretend stdout/stderr output from EC2 instance."""
output = "Fake output."
class FakeBotoInstance:
"""Helper for setting up an `EC2Instance` without EC2."""
id = 0
state = 'running'
public_dns_name = 'fake-instance'
update = FakeMethod()
stop = FakeMethod()
get_console_output = FakeOutput
class FakeReservation:
"""Helper for setting up an `EC2Instance` without EC2."""
def __init__(self):
self.instances = [FakeBotoInstance()]
class FakeImage:
"""Helper for setting up an `EC2Instance` without EC2."""
run = FakeMethod(result=FakeReservation())
class FakeFailure(Exception):
"""A pretend failure from the test runner."""
class TestEC2Instance(TestCase):
"""Test running of an `EC2Instance` without EC2."""
def _makeInstance(self):
"""Set up an `EC2Instance`, with stubbing where needed.
`EC2Instance.shutdown` is replaced with a `FakeMethod`, so check
its call_count to see whether it's been invoked.
"""
session_name = None
image = FakeImage()
instance_type = 'c1.xlarge'
demo_networks = None
account = FakeAccount()
from_scratch = None
user_key = None
login = None
region = None
instance = EC2Instance(
session_name, image, instance_type, demo_networks, account,
from_scratch, user_key, login,
region)
instance.shutdown = FakeMethod()
instance._report_traceback = FakeMethod()
instance.log = FakeMethod()
return instance
def _runInstance(self, instance, runnee=None, headless=False):
"""Set up and run an `EC2Instance` (but without EC2)."""
if runnee is None:
runnee = FakeMethod()
instance.set_up_and_run(False, not headless, runnee)
def test_EC2Instance_test_baseline(self):
# The EC2 instances we set up have neither started nor been shut
# down. After running, they have started.
# Not a very useful test, except it establishes the basic
# assumptions for the other tests.
instance = self._makeInstance()
runnee = FakeMethod()
self.assertEqual(0, runnee.call_count)
self.assertEqual(0, instance.shutdown.call_count)
self._runInstance(instance, runnee=runnee)
self.assertEqual(1, runnee.call_count)
def test_set_up_and_run_headful(self):
# A non-headless run executes all tests in the instance, then
# shuts down.
instance = self._makeInstance()
self._runInstance(instance, headless=False)
self.assertEqual(1, instance.shutdown.call_count)
def test_set_up_and_run_headless(self):
# An asynchronous, headless run kicks off the tests on the
# instance but does not shut it down.
instance = self._makeInstance()
self._runInstance(instance, headless=True)
self.assertEqual(0, instance.shutdown.call_count)
def test_set_up_and_run_headful_failure(self):
# If the test runner barfs, the instance swallows the exception
# and shuts down.
instance = self._makeInstance()
runnee = FakeMethod(failure=FakeFailure("Headful barfage."))
self._runInstance(instance, runnee=runnee, headless=False)
self.assertEqual(1, instance.shutdown.call_count)
def test_set_up_and_run_headless_failure(self):
# If the instance's test runner fails to set up for a headless
# run, the instance swallows the exception and shuts down.
instance = self._makeInstance()
runnee = FakeMethod(failure=FakeFailure("Headless boom."))
self._runInstance(instance, runnee=runnee, headless=True)
self.assertEqual(1, instance.shutdown.call_count)
|