| # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.cros import dhcp_handling_rule |
| from autotest_lib.client.cros import dhcp_packet |
| from autotest_lib.client.cros import dhcp_test_base |
| from autotest_lib.client.cros import shill_temporary_profile |
| from autotest_lib.client.cros.networking import shill_context |
| |
| class network_DhcpStaticIP(dhcp_test_base.DhcpTestBase): |
| """DHCP test which confirms static IP functionality""" |
| # Length of time the lease from the DHCP server is valid. |
| LEASE_TIME_SECONDS = 60 |
| # We'll fill in the subnet and give this address to the client over DHCP. |
| INTENDED_IP_SUFFIX = '0.0.0.101' |
| # We'll fill in the subnet and supply this as a static IP address. |
| STATIC_IP_SUFFIX = '0.0.0.201' |
| STATIC_IP_NAME_SERVERS = [ '1.1.2.2', '1.1.3.4' ] |
| # Time to wait for the DHCP negotiation protocol to complete. |
| DHCP_NEGOTIATION_TIMEOUT_SECONDS = 10 |
| # Time to wait after DHCP negotiation completes until service is marked |
| # as connected. |
| DHCP_SETUP_TIMEOUT_SECONDS = 3 |
| # Name given to the temporary shill profile we create for this test. |
| TEST_PROFILE_NAME = 'TestStaticIP' |
| # Various parameters that can be set statically. |
| CONFIGURE_STATIC_IP_ADDRESS = 'ip-address' |
| CONFIGURE_STATIC_IP_DNS_SERVERS = 'dns-servers' |
| |
| def create_static_ip_config(self, params): |
| """Creates the configuration dictionary for a StaticIPConfig, given a |
| list of parameters. |
| |
| @param params list of static parameters to configure. |
| @return dict containing the relevant StaticIPConfig parameters. |
| |
| """ |
| self._static_ip_options = {} |
| config = {} |
| if self.CONFIGURE_STATIC_IP_ADDRESS in params: |
| subnet_mask = self.ethernet_pair.interface_subnet_mask |
| static_ip_address = dhcp_test_base.DhcpTestBase.rewrite_ip_suffix( |
| subnet_mask, |
| self.server_ip, |
| self.STATIC_IP_SUFFIX) |
| config['Address'] = static_ip_address |
| |
| prefix_len = self.ethernet_pair.interface_prefix |
| config['Prefixlen'] = prefix_len |
| |
| self._static_ip_options[dhcp_packet.OPTION_REQUESTED_IP] = ( |
| static_ip_address) |
| if self.CONFIGURE_STATIC_IP_DNS_SERVERS in params: |
| config['NameServers'] = self.STATIC_IP_NAME_SERVERS |
| self._static_ip_options[dhcp_packet.OPTION_DNS_SERVERS] = ( |
| self.STATIC_IP_NAME_SERVERS) |
| return config |
| |
| |
| def check_saved_ip(self, service, options): |
| """Check the properties of the Ethernet service to make sure that |
| the address provided by the DHCP server is properly added to the |
| SavedIPConfig. |
| |
| @param service object the Service DBus interface to clear properties. |
| @param options dict parameters that were used to configure the DHCP |
| server. |
| |
| """ |
| intended_ip = options[dhcp_packet.OPTION_REQUESTED_IP] |
| properties = service.GetProperties() |
| saved_address = properties['SavedIPConfig']['Address'] |
| if intended_ip != saved_address: |
| raise error.TestFail('Saved IP address %s is not DHCP address %s' % |
| (saved_address, intended_ip)) |
| |
| |
| def make_lease_negotiation_rules(self, options): |
| """Generate a set of lease negotiation handling rules for a |
| server that will successfully return an IP address to the client. |
| |
| @param options dict of options to be negotiated. In particular, |
| the dhcp_packet.OPTION_REQUESTED_IP element is used to configure |
| the address that will be returned to the client. |
| @return array of DhcpHandlingRule instances which implement the |
| negotiation. |
| |
| """ |
| intended_ip = options[dhcp_packet.OPTION_REQUESTED_IP] |
| rules = [] |
| rules.append(dhcp_handling_rule.DhcpHandlingRule_RespondToDiscovery( |
| intended_ip, |
| self.server_ip, |
| options, |
| {})) |
| rules.append(dhcp_handling_rule.DhcpHandlingRule_RespondToRequest( |
| intended_ip, |
| self.server_ip, |
| options, |
| {})) |
| return rules |
| |
| |
| def test_dhcp_negotiation(self, rules, service): |
| """Perform a DHCP lease negotiation using handler rules from |rules|, |
| and ensure that |service| becomes connected as a result. |
| |
| @param rules array of handling rules that must complete in order for |
| the negotiation to be considered successful. |
| @param service Service DBus object which should become connected as |
| a result of the DHCP negotiation. |
| |
| """ |
| rules[-1].is_final_handler = True |
| self.server.start_test(rules, self.DHCP_NEGOTIATION_TIMEOUT_SECONDS) |
| self.server.wait_for_test_to_finish() |
| if not self.server.last_test_passed: |
| raise error.TestFail('Test server didn\'t get all the messages it ' |
| 'was told to expect during negotiation.') |
| # Wait for the service to enter a "good" state. |
| connect_result = self.shill_proxy.wait_for_property_in( |
| service, |
| self.shill_proxy.SERVICE_PROPERTY_STATE, |
| self.shill_proxy.SERVICE_CONNECTED_STATES, |
| self.DHCP_SETUP_TIMEOUT_SECONDS) |
| (successful, _, association_time) = connect_result |
| if not successful: |
| raise error.TestFail('Ethernet service did not become connected.') |
| |
| |
| def connect_dynamic_ip(self, options, service): |
| """Perform a DHCP negotiation, using |options|. Then check that |
| the IP information configured on client matches the parameters |
| in |options|. |
| |
| @param options dict containing DHCP packet options to be returned |
| to the client during negotiation, and then later checked for |
| consistency. |
| @param service DBus object of the service that should become |
| connected as a result of the negotiation. |
| |
| """ |
| self.test_dhcp_negotiation(self.make_lease_negotiation_rules(options), |
| service) |
| self.check_dhcp_config(options) |
| |
| |
| def connect_static_ip(self, options, service, params): |
| """Perform a DHCP negotiation, using |options|. Then check that |
| the IP information configured on client matches the parameters |
| in |options|, except that the client's IP address should be |
| |static_ip_address|. |
| |
| @param options dict containing DHCP packet options to be returned |
| to the client during negotiation, and then later checked for |
| consistency. |
| @param service DBus object of the service that should become |
| connected as a result of the negotiation. |
| @param params list of static IP parameters we will be verifying. |
| |
| """ |
| rules = self.make_lease_negotiation_rules(options) |
| if self.CONFIGURE_STATIC_IP_ADDRESS in params: |
| # Add a rule that expects the client to release the lease. |
| rules.append(dhcp_handling_rule.DhcpHandlingRule_AcceptRelease( |
| self.server_ip, |
| options, |
| {})) |
| self.test_dhcp_negotiation(rules, service) |
| |
| # Check to make sure that the configured IP address of the client |
| # matches the configured static IP address. |
| static_ip_options = options.copy() |
| static_ip_options.update(self._static_ip_options) |
| self.check_dhcp_config(static_ip_options) |
| self.check_saved_ip(service, options) |
| |
| |
| def test_body(self): |
| """The test main body""" |
| subnet_mask = self.ethernet_pair.interface_subnet_mask |
| intended_ip = dhcp_test_base.DhcpTestBase.rewrite_ip_suffix( |
| subnet_mask, |
| self.server_ip, |
| self.INTENDED_IP_SUFFIX) |
| # Two real name servers, and a bogus one to be unpredictable. |
| dns_servers = ['8.8.8.8', '8.8.4.4', '192.168.87.88'] |
| domain_name = 'corp.google.com' |
| dns_search_list = [ |
| 'corgie.google.com', |
| 'lies.google.com', |
| 'that.is.a.tasty.burger.google.com', |
| ] |
| # This is the pool of information the server will give out to the client |
| # upon request. |
| dhcp_options = { |
| dhcp_packet.OPTION_SERVER_ID : self.server_ip, |
| dhcp_packet.OPTION_SUBNET_MASK : subnet_mask, |
| dhcp_packet.OPTION_IP_LEASE_TIME : self.LEASE_TIME_SECONDS, |
| dhcp_packet.OPTION_REQUESTED_IP : intended_ip, |
| dhcp_packet.OPTION_DNS_SERVERS : dns_servers, |
| dhcp_packet.OPTION_DOMAIN_NAME : domain_name, |
| dhcp_packet.OPTION_DNS_DOMAIN_SEARCH_LIST : dns_search_list, |
| } |
| service = self.find_ethernet_service( |
| self.ethernet_pair.peer_interface_name) |
| |
| manager = self.shill_proxy.manager |
| with shill_temporary_profile.ShillTemporaryProfile( |
| manager, profile_name=self.TEST_PROFILE_NAME): |
| |
| self.connect_dynamic_ip(dhcp_options, service) |
| |
| for params in self._static_param_list: |
| static_ip_config = self.create_static_ip_config(params) |
| with shill_context.StaticIPContext(service, static_ip_config): |
| self.connect_static_ip(dhcp_options, service, params) |
| |
| self.connect_dynamic_ip(dhcp_options, service) |
| |
| |
| def run_once(self, static_param_list): |
| """Setup the static parameter list before calling the DhcpTestBase |
| main loop. |
| |
| @param static_param_list list of iterable properties to configure |
| for each static IP test. |
| |
| """ |
| self._static_param_list = static_param_list |
| super(network_DhcpStaticIP, self).run_once() |
