729
731
config.computer_title = None
730
732
self.broker_service.identity.secure_id = None
731
733
self.assertFalse(handler.should_register())
736
class IsCloudManagedTests(LandscapeTest):
739
super(IsCloudManagedTests, self).setUp()
743
def fake_fetch(self, url, connect_timeout=None):
744
self.urls.append((url, connect_timeout))
745
return self.responses.pop(0)
747
def mock_socket(self):
749
Mock out socket usage by is_cloud_managed to wait for the network.
751
# Mock the socket.connect call that it also does
752
socket_class = self.mocker.replace("socket.socket", passthrough=False)
753
socket = socket_class()
754
socket.connect((EC2_HOST, 80))
758
def test_is_managed(self):
760
L{is_cloud_managed} returns True if the EC2 user-data contains Landscape
761
instance information. It fetches the EC2 data with low timeouts.
763
user_data = {"otps": ["otp1"], "exchange-url": "http://exchange",
764
"ping-url": "http://ping"}
765
self.responses = [dumps(user_data), "0"]
770
self.assertTrue(is_cloud_managed(self.fake_fetch))
773
[(EC2_API + "/user-data", 5),
774
(EC2_API + "/meta-data/ami-launch-index", 5)])
776
def test_is_managed_index(self):
777
user_data = {"otps": ["otp1", "otp2"],
778
"exchange-url": "http://exchange",
779
"ping-url": "http://ping"}
780
self.responses = [dumps(user_data), "1"]
783
self.assertTrue(is_cloud_managed(self.fake_fetch))
785
def test_is_managed_wrong_index(self):
786
user_data = {"otps": ["otp1"], "exchange-url": "http://exchange",
787
"ping-url": "http://ping"}
788
self.responses = [dumps(user_data), "1"]
791
self.assertFalse(is_cloud_managed(self.fake_fetch))
793
def test_is_managed_exchange_url(self):
794
user_data = {"otps": ["otp1"], "ping-url": "http://ping"}
795
self.responses = [dumps(user_data), "0"]
798
self.assertFalse(is_cloud_managed(self.fake_fetch))
800
def test_is_managed_ping_url(self):
801
user_data = {"otps": ["otp1"], "exchange-url": "http://exchange"}
802
self.responses = [dumps(user_data), "0"]
805
self.assertFalse(is_cloud_managed(self.fake_fetch))
807
def test_is_managed_bpickle(self):
808
self.responses = ["some other user data", "0"]
811
self.assertFalse(is_cloud_managed(self.fake_fetch))
813
def test_is_managed_no_data(self):
814
self.responses = ["", "0"]
817
self.assertFalse(is_cloud_managed(self.fake_fetch))
819
def test_is_managed_fetch_not_found(self):
820
def fake_fetch(url, connect_timeout=None):
821
raise HTTPCodeError(404, "ohnoes")
824
self.assertFalse(is_cloud_managed(fake_fetch))
826
def test_is_managed_fetch_error(self):
827
def fake_fetch(url, connect_timeout=None):
828
raise FetchError(7, "couldn't connect to host")
831
self.assertFalse(is_cloud_managed(fake_fetch))
833
def test_waits_for_network(self):
835
is_cloud_managed will wait until the network before trying to fetch
838
user_data = {"otps": ["otp1"], "exchange-url": "http://exchange",
839
"ping-url": "http://ping"}
840
self.responses = [dumps(user_data), "0"]
843
time_sleep = self.mocker.replace("time.sleep", passthrough=False)
844
socket_class = self.mocker.replace("socket.socket", passthrough=False)
845
socket_obj = socket_class()
846
socket_obj.connect((EC2_HOST, 80))
847
self.mocker.throw(socket.error("woops"))
849
socket_obj = socket_class()
850
socket_obj.connect((EC2_HOST, 80))
851
self.mocker.result(None)
854
self.assertTrue(is_cloud_managed(self.fake_fetch))
856
def test_waiting_times_out(self):
858
We'll only wait five minutes for the network to come up.
860
def fake_fetch(url, connect_timeout=None):
861
raise FetchError(7, "couldn't connect to host")
864
time_sleep = self.mocker.replace("time.sleep", passthrough=False)
865
time_time = self.mocker.replace("time.time", passthrough=False)
867
self.mocker.result(100)
868
socket_class = self.mocker.replace("socket.socket", passthrough=False)
869
socket_obj = socket_class()
870
socket_obj.connect((EC2_HOST, 80))
871
self.mocker.throw(socket.error("woops"))
874
self.mocker.result(401)
876
# Mocking time.time is dangerous, because the test harness calls it. So
877
# we explicitly reset mocker before returning from the test.
879
self.assertFalse(is_cloud_managed(fake_fetch))