← Back to branch summary

~ubuntu-branches/ubuntu/vivid/landscape-client/vivid

« back to all changes in this revision

Viewing changes to landscape/tests/test_configuration.py

  • Committer: Package Import Robot
  • Author(s): Andreas Hasenack
  • Date: 2014-11-20 21:03:56 UTC
  • mfrom: (1.1.34)
  • Revision ID: package-import@ubuntu.com-20141120210356-bdl520tpnxddxtm0
Tags: 14.11-0ubuntu1
https://launchpad.net/bugs/1389686
https://launchpad.net/bugs/1374501
https://launchpad.net/bugs/1369635
https://launchpad.net/bugs/1362506
https://launchpad.net/bugs/1362506
https://launchpad.net/bugs/1342646
https://launchpad.net/bugs/1326940
https://launchpad.net/bugs/1320236
* New upstream version:
  - Fix regression occurring when performing Landscape-driven release
    upgrades (LP: #1389686)
  - Fix regression occurring when switching the client between different
    Landscape servers (LP #376134)
  - Support reporting QEMU virtualization (LP: #1374501)
  - Bump Juju integration message format (LP: #1369635, LP: #1362506)
  - Drop provisioning registration message (LP: #1362506)
  - Drop cloud registration message (LP: #1342646)
  - Fix handling broken packages (LP: #1326940)
  - Add new Swift usage message type (LP: #1320236)

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/tests/test_configuration.py
15
15
    register, setup, main, setup_init_script_and_start_client,
16
16
    stop_client_and_disable_init_script, ConfigurationError,
17
17
    ImportOptionError, store_public_key_data,
18
 
    fetch_base64_ssl_public_certificate, bootstrap_tree)
 
18
    bootstrap_tree)
19
19
from landscape.broker.registration import InvalidCredentialsError
20
20
from landscape.sysvconfig import SysVConfig, ProcessError
21
21
from landscape.tests.helpers import (
629
629
 
630
630
    def test_access_group_defined_on_command_line(self):
631
631
        """
632
 
        An access group defined on the command line can be verified by the
633
 
        user.
 
632
        When an access group is provided on the command line, do not prompt
 
633
        the user for it.
634
634
        """
635
635
        raw_input_mock = self.mocker.replace(raw_input, passthrough=False)
636
636
        self.expect(raw_input_mock(ANY)).count(0)
820
820
                "--ping-interval", "30",
821
821
                "--http-proxy", "",
822
822
                "--https-proxy", "",
823
 
                "--otp", "",
824
 
                "--tags", "",
825
 
                "--provisioning-otp", ""]
 
823
                "--tags", ""]
826
824
        config = self.get_config(args)
827
825
        setup(config)
828
826
        self.assertConfigEqual(
837
835
            "https_proxy = \n"
838
836
            "url = https://landscape.canonical.com/message-system\n"
839
837
            "exchange_interval = 900\n"
840
 
            "otp = \n"
841
838
            "ping_interval = 30\n"
842
839
            "ping_url = http://landscape.canonical.com/ping\n"
843
 
            "provisioning_otp = \n"
844
 
            "urgent_exchange_interval = 60\n"
845
 
             % config.data_path)
 
840
            "urgent_exchange_interval = 60\n" % config.data_path)
846
841
 
847
842
    def test_silent_setup_without_computer_title(self):
848
843
        """A computer title is required."""
862
857
        config = self.get_config(["--silent", "-t", "rex"])
863
858
        self.assertRaises(ConfigurationError, setup, config)
864
859
 
865
 
    def test_silent_setup_with_otp(self):
866
 
        """
867
 
        If the OTP is specified, there is no need to pass the account name and
868
 
        the computer title.
869
 
        """
870
 
        sysvconfig_mock = self.mocker.patch(SysVConfig)
871
 
        sysvconfig_mock.set_start_on_boot(True)
872
 
        self.mocker.replay()
873
 
 
874
 
        config = self.get_config(["--silent", "--otp", "otp1"])
875
 
        setup(config)
876
 
 
877
 
        self.assertEqual("otp1", config.otp)
878
 
 
879
 
    def test_silent_setup_with_provisioning_otp(self):
880
 
        """
881
 
        If the provisioning OTP is specified, there is no need to pass the
882
 
        account name and the computer title.
883
 
        """
884
 
        sysvconfig_mock = self.mocker.patch(SysVConfig)
885
 
        sysvconfig_mock.set_start_on_boot(True)
886
 
        sysvconfig_mock.restart_landscape()
887
 
        self.mocker.replay()
888
 
 
889
 
        config = self.get_config(["--silent", "--provisioning-otp", "otp1"])
890
 
        setup(config)
891
 
 
892
 
        self.assertEqual("otp1", config.provisioning_otp)
893
 
 
894
860
    def test_silent_script_users_imply_script_execution_plugin(self):
895
861
        """
896
862
        If C{--script-users} is specified, without C{ScriptExecution} in the
1719
1685
        service = self.broker_service
1720
1686
 
1721
1687
        registration_mock = self.mocker.replace(service.registration)
1722
 
        config_mock = self.mocker.replace(service.config)
1723
1688
        print_text_mock = self.mocker.replace(print_text)
1724
1689
        reactor_mock = self.mocker.patch(FakeReactor)
1725
1690
 
1733
1698
        time_mock.sleep(ANY)
1734
1699
        self.mocker.count(1)
1735
1700
 
1736
 
        # After a nice dance the configuration is reloaded.
1737
 
        config_mock.reload()
1738
 
 
1739
1701
        # The register() method is called.  We fire the "registration-done"
1740
1702
        # event after it's done, so that it cascades into a deferred callback.
1741
1703
 
1773
1735
 
1774
1736
        self.log_helper.ignore_errors(InvalidCredentialsError)
1775
1737
        registration_mock = self.mocker.replace(service.registration)
1776
 
        config_mock = self.mocker.replace(service.config)
1777
1738
        print_text_mock = self.mocker.replace(print_text)
1778
1739
        reactor_mock = self.mocker.patch(FakeReactor)
1779
1740
 
1787
1748
        time_mock.sleep(ANY)
1788
1749
        self.mocker.count(1)
1789
1750
 
1790
 
        # After a nice dance the configuration is reloaded.
1791
 
        config_mock.reload()
1792
 
 
1793
1751
        # The register() method is called.  We fire the "registration-failed"
1794
1752
        # event after it's done, so that it cascades into a deferred errback.
1795
1753
 
1828
1786
        service = self.broker_service
1829
1787
 
1830
1788
        registration_mock = self.mocker.replace(service.registration)
1831
 
        config_mock = self.mocker.replace(service.config)
1832
1789
        print_text_mock = self.mocker.replace(print_text)
1833
1790
        reactor_mock = self.mocker.patch(FakeReactor)
1834
1791
 
1842
1799
        time_mock.sleep(ANY)
1843
1800
        self.mocker.count(1)
1844
1801
 
1845
 
        # After a nice dance the configuration is reloaded.
1846
 
        config_mock.reload()
1847
 
 
1848
1802
        def register_done():
1849
1803
            service.reactor.fire("exchange-failed")
1850
1804
        registration_mock.register()
1878
1832
        service = self.broker_service
1879
1833
 
1880
1834
        registration_mock = self.mocker.replace(service.registration)
1881
 
        config_mock = self.mocker.replace(service.config)
1882
1835
        print_text_mock = self.mocker.replace(print_text)
1883
1836
        reactor_mock = self.mocker.patch(FakeReactor)
1884
1837
        remote_mock = self.mocker.patch(RemoteBroker)
1898
1851
        time_mock.sleep(ANY)
1899
1852
        self.mocker.count(1)
1900
1853
 
1901
 
        # After a nice dance the configuration is reloaded.
1902
 
        config_mock.reload()
1903
 
 
1904
1854
        remote_mock.call_on_event(ANY)
1905
1855
        self.mocker.result(succeed(None))
1906
1856
 
2109
2059
        connector = connector_factory(ANY, configuration)
2110
2060
        connector.connect(max_retries=0, quiet=True)
2111
2061
        self.mocker.result(succeed(remote_broker))
2112
 
        remote_broker.reload_configuration()
2113
 
        self.mocker.result(succeed(None))
2114
2062
        remote_broker.call_on_event(ANY)
2115
2063
        self.mocker.result(succeed(None))
2116
2064
 
2160
2108
        self.assertEqual(key_filename,
2161
2109
                         store_public_key_data(config, "123456789"))
2162
2110
        self.assertEqual("123456789", open(key_filename, "r").read())
2163
 
 
2164
 
    def test_fetch_base64_ssl(self):
2165
 
        """
2166
 
        L{fetch_base64_ssl_public_certificate} should pull a JSON object from
2167
 
        http://providedhostname/get-ca-cert. And return the custom_ca_cert data
2168
 
        if it exists.
2169
 
        """
2170
 
        base64_cert = "base64:  MTIzNDU2Nzg5MA=="  # encoded woo hoo
2171
 
        fetch_mock = self.mocker.replace("landscape.lib.fetch.fetch")
2172
 
        fetch_mock("http://fakehost/get-ca-cert", insecure=True)
2173
 
        self.mocker.result(
2174
 
            "{\"custom_ca_cert\": \"%s\"}" % base64_cert)
2175
 
        self.mocker.replay()
2176
 
 
2177
 
        def check_info(info):
2178
 
            self.assertEqual(
2179
 
                "Fetching CA certificate from fakehost if available...",
2180
 
                str(info))
2181
 
 
2182
 
        content = fetch_base64_ssl_public_certificate("fakehost",
2183
 
                                                      on_info=check_info)
2184
 
        self.assertEqual(base64_cert, content)
2185
 
 
2186
 
    def test_fetch_base64_ssl_no_custom_ca(self):
2187
 
        """
2188
 
        L{fetch_base64_ssl_public_certificate} should pull a JSON object from
2189
 
        http://providedhostname/get-ca-cert. And return the custom_ca_cert data
2190
 
        if it exists, otherwise it should return an empty string.""
2191
 
        """
2192
 
        fetch_mock = self.mocker.replace("landscape.lib.fetch.fetch")
2193
 
        fetch_mock("http://fakehost/get-ca-cert", insecure=True)
2194
 
        self.mocker.result("{}")
2195
 
 
2196
 
        print_text_mock = self.mocker.replace(print_text)
2197
 
        print_text_mock(
2198
 
            "Fetching CA certificate from fakehost if available...")
2199
 
        print_text_mock("No custom CA certificate available for fakehost.")
2200
 
        self.mocker.replay()
2201
 
 
2202
 
        content = fetch_base64_ssl_public_certificate("fakehost",
2203
 
                                                      on_info=print_text)
2204
 
        self.assertEqual("", content)
2205
 
 
2206
 
    def test_fetch_base64_ssl_with_http_code_fetch_error(self):
2207
 
        fetch_mock = self.mocker.replace("landscape.lib.fetch.fetch")
2208
 
        fetch_mock("http://fakehost/get-ca-cert", insecure=True)
2209
 
        self.mocker.throw(HTTPCodeError(404, ""))
2210
 
        print_text_mock = self.mocker.replace(print_text)
2211
 
        print_text_mock(
2212
 
            "Fetching CA certificate from fakehost if available...")
2213
 
        self.mocker.replay()
2214
 
 
2215
 
        def check_error(error):
2216
 
            self.assertEqual("Unable to fetch CA certificate from discovered "
2217
 
                             "server fakehost: Server does not support client "
2218
 
                             "auto-registation.",
2219
 
                             str(error))
2220
 
 
2221
 
        content = fetch_base64_ssl_public_certificate("fakehost",
2222
 
                                                      on_info=print_text,
2223
 
                                                      on_error=check_error)
2224
 
        self.assertEquals("", content)
2225
 
 
2226
 
    def test_fetch_base64_ssl_with_pycurl_error(self):
2227
 
        fetch_mock = self.mocker.replace("landscape.lib.fetch.fetch")
2228
 
        fetch_mock("http://fakehost/get-ca-cert", insecure=True)
2229
 
        self.mocker.throw(PyCurlError(60, "pycurl message"))
2230
 
        print_text_mock = self.mocker.replace(print_text)
2231
 
        print_text_mock(
2232
 
            "Fetching CA certificate from fakehost if available...")
2233
 
        self.mocker.replay()
2234
 
 
2235
 
        def check_error(error):
2236
 
            self.assertEqual("Unable to fetch CA certificate from fakehost: "
2237
 
                             "Error 60: pycurl message",
2238
 
                             str(error))
2239
 
 
2240
 
        content = fetch_base64_ssl_public_certificate("fakehost",
2241
 
                                                      on_info=print_text,
2242
 
                                                      on_error=check_error)
2243
 
        self.assertEquals("", content)