← Back to branch summary

~ubuntu-branches/ubuntu/lucid/landscape-client/lucid-updates

« back to all changes in this revision

Viewing changes to landscape/tests/test_watchdog.py

  • Committer: Package Import Robot
  • Author(s): Andreas Hasenack
  • Date: 2012-04-10 14:28:48 UTC
  • mfrom: (1.1.27)
  • mto: This revision was merged to the branch mainline in revision 35.
  • Revision ID: package-import@ubuntu.com-20120410142848-7xsy4g2xii7y7ntc
Import upstream version 12.04.3

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/tests/test_watchdog.py
18
18
    WatchDogConfiguration, bootstrap_list,
19
19
    MAXIMUM_CONSECUTIVE_RESTARTS, RESTART_BURST_DELAY, run,
20
20
    Broker, Monitor, Manager)
 
21
from landscape.lib.dns import discover_server
 
22
from landscape.configuration import (
 
23
    fetch_base64_ssl_public_certificate, print_text)
21
24
from landscape.amp import ComponentProtocolFactory, RemoteComponentConnector
22
25
from landscape.broker.amp import RemoteBrokerConnector
 
26
from landscape.deployment import Configuration
23
27
from landscape.reactor import TwistedReactor
24
28
 
25
29
import landscape.watchdog
962
966
        self.configuration = WatchDogConfiguration()
963
967
        self.data_path = self.makeDir()
964
968
        self.log_dir = self.makeDir()
965
 
        self.configuration.load(["--data-path", self.data_path,
 
969
        self.config_filename = self.makeFile("[client]\n")
 
970
        self.configuration.load(["--config", self.config_filename,
 
971
                                 "--data-path", self.data_path,
966
972
                                 "--log-dir", self.log_dir])
967
973
 
968
974
    def test_daemonize(self):
1166
1172
        self.assertEqual(service.exit_code, 2)
1167
1173
        return result
1168
1174
 
 
1175
    def test_autodiscover_config_write_with_pubkey(self):
 
1176
        """
 
1177
        When server_autodiscover is set True, and the config.ssl_public_key
 
1178
        already exists, ensure we update and write the config file with the
 
1179
        discovered server urls.
 
1180
        """
 
1181
        self.configuration.server_autodiscover = True
 
1182
        self.configuration.ssl_public_key = "/tmp/fakepubkey.ssl"
 
1183
 
 
1184
        service = WatchDogService(self.configuration)
 
1185
 
 
1186
        # Validate appropriate initial config options
 
1187
        self.assertEquals("https://landscape.canonical.com/message-system",
 
1188
                          service._config.url)
 
1189
        self.assertEquals("/tmp/fakepubkey.ssl",
 
1190
                          service._config.ssl_public_key)
 
1191
        self.assertTrue(service._config.server_autodiscover)
 
1192
 
 
1193
        bootstrap_list_mock = self.mocker.patch(bootstrap_list)
 
1194
        bootstrap_list_mock.bootstrap(data_path=self.data_path,
 
1195
                                      log_dir=self.log_dir)
 
1196
 
 
1197
        self.mocker.order()
 
1198
 
 
1199
        discover_mock = self.mocker.replace(discover_server, passthrough=False)
 
1200
        discover_mock(self.configuration.autodiscover_srv_query_string,
 
1201
                      self.configuration.autodiscover_a_query_string)
 
1202
        self.mocker.result(succeed("fakehostname"))
 
1203
 
 
1204
        watchdog_mock = self.mocker.replace(service.watchdog)
 
1205
        watchdog_mock.check_running()
 
1206
        self.mocker.result(succeed([]))
 
1207
        watchdog_mock.start()
 
1208
        self.mocker.result(succeed(None))
 
1209
        self.mocker.replay()
 
1210
 
 
1211
        # trigger something to ensure autodiscover() is called
 
1212
        service.startService()
 
1213
 
 
1214
        # Reload config to validate config.write() was called with changes
 
1215
        config = Configuration()
 
1216
        config.load(["--config", self.config_filename])
 
1217
        self.assertFalse(config.server_autodiscover)
 
1218
        self.assertEquals("https://fakehostname/message-system",
 
1219
                         config.url)
 
1220
        self.assertEquals("http://fakehostname/ping",
 
1221
                         config.ping_url)
 
1222
        self.assertEquals("/tmp/fakepubkey.ssl", config.ssl_public_key)
 
1223
 
 
1224
    def test_autodiscover_config_write_without_pubkey(self):
 
1225
        """
 
1226
        WatchDogService should attempt to fetch the custom CA cert from the
 
1227
        discovered server if server_autodiscover=True and ssl_public_key is
 
1228
        undefined. If the discovered server has a custom signed CA cert, that
 
1229
        should be saved and its file path should be written to to configuration
 
1230
        file.
 
1231
        """
 
1232
        base64_cert = "base64:  MTIzNDU2Nzg5MA=="  # encoded from 1234567890
 
1233
 
 
1234
        key_filename = os.path.join(self.data_path,
 
1235
            os.path.basename(self.config_filename + ".ssl_public_key"))
 
1236
 
 
1237
        self.configuration.server_autodiscover = True
 
1238
 
 
1239
        service = WatchDogService(self.configuration)
 
1240
 
 
1241
        # Validate appropriate initial config options
 
1242
        self.assertEquals(None, self.configuration.ssl_public_key)
 
1243
        self.assertTrue(self.configuration.server_autodiscover)
 
1244
 
 
1245
        discover_mock = self.mocker.replace(discover_server, passthrough=False)
 
1246
        discover_mock(self.configuration.autodiscover_srv_query_string,
 
1247
                      self.configuration.autodiscover_a_query_string)
 
1248
        self.mocker.result(succeed("fakehostname"))
 
1249
 
 
1250
        fetch_ca_mock = self.mocker.replace(
 
1251
            fetch_base64_ssl_public_certificate, passthrough=False)
 
1252
 
 
1253
        fetch_ca_mock("fakehostname", on_info=ANY, on_error=ANY)
 
1254
        self.mocker.result(base64_cert)
 
1255
 
 
1256
        print_text_mock = self.mocker.replace(print_text)
 
1257
        print_text_mock("Writing SSL CA certificate to %s..." % key_filename)
 
1258
 
 
1259
        watchdog_mock = self.mocker.replace(service.watchdog)
 
1260
        watchdog_mock.check_running()
 
1261
        self.mocker.result(succeed([]))
 
1262
        watchdog_mock.start()
 
1263
        self.mocker.result(succeed(None))
 
1264
        self.mocker.replay()
 
1265
 
 
1266
        service.startService()
 
1267
 
 
1268
        # Reload config file to validate config.write() was called with changes
 
1269
        config = Configuration()
 
1270
        config.load(["--config", self.config_filename])
 
1271
        self.assertFalse(config.server_autodiscover)
 
1272
        self.assertEquals("https://fakehostname/message-system",
 
1273
                         config.url)
 
1274
        self.assertEquals("http://fakehostname/ping",
 
1275
                         config.ping_url)
 
1276
        self.assertEquals(key_filename, config.ssl_public_key)
 
1277
        self.assertEqual("1234567890", open(key_filename, "r").read())
 
1278
 
 
1279
    def test_autodiscover_config_write_without_pubkey_no_custom_ca(self):
 
1280
        """
 
1281
        When server_autodiscover is set True, and the config does not have an
 
1282
        ssl_public_key defined WatchDogService should attempt to fetch the
 
1283
        custom CA cert from the discovered server.
 
1284
        """
 
1285
        self.configuration.server_autodiscover = True
 
1286
 
 
1287
        service = WatchDogService(self.configuration)
 
1288
 
 
1289
        # Validate appropriate initial config options
 
1290
        self.assertEquals(None, self.configuration.ssl_public_key)
 
1291
        self.assertTrue(self.configuration.server_autodiscover)
 
1292
 
 
1293
        discover_mock = self.mocker.replace(discover_server, passthrough=False)
 
1294
        discover_mock(self.configuration.autodiscover_srv_query_string,
 
1295
                      self.configuration.autodiscover_a_query_string)
 
1296
        self.mocker.result(succeed("fakehostname"))
 
1297
 
 
1298
        fetch_ca_mock = self.mocker.replace(
 
1299
            fetch_base64_ssl_public_certificate, passthrough=False)
 
1300
 
 
1301
        fetch_ca_mock("fakehostname", on_info=ANY, on_error=ANY)
 
1302
        self.mocker.result("")  # No Custom CA cert found
 
1303
 
 
1304
        watchdog_mock = self.mocker.replace(service.watchdog)
 
1305
        watchdog_mock.check_running()
 
1306
        self.mocker.result(succeed([]))
 
1307
        watchdog_mock.start()
 
1308
        self.mocker.result(succeed(None))
 
1309
        self.mocker.replay()
 
1310
 
 
1311
        service.startService()
 
1312
 
 
1313
        # Reload config file to validate config.write() was called with changes
 
1314
        config = Configuration()
 
1315
        config.load(["--config", self.config_filename])
 
1316
        self.assertFalse(config.server_autodiscover)
 
1317
        self.assertEquals("https://fakehostname/message-system",
 
1318
                         config.url)
 
1319
        self.assertEquals("http://fakehostname/ping",
 
1320
                         config.ping_url)
 
1321
        self.assertEquals(None, config.ssl_public_key)
 
1322
 
1169
1323
    def test_bootstrap(self):
1170
1324
 
1171
1325
        data_path = self.makeDir()