| # Copyright 2015 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. |
| |
| """ |
| Base class for DHCPv6 tests. This class just sets up a little bit of plumbing, |
| like a virtual ethernet device with one end that looks like a real ethernet |
| device to shill and a DHCPv6 test server on the end that doesn't look like a |
| real ethernet interface to shill. Child classes should override test_body() |
| with the logic of their test. |
| """ |
| |
| import logging |
| import time |
| import traceback |
| |
| from autotest_lib.client.bin import test |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.common_lib.cros import virtual_ethernet_pair |
| from autotest_lib.client.cros import dhcpv6_test_server |
| from autotest_lib.client.cros.networking import shill_proxy |
| |
| # These are keys that may be used with the DBus dictionary returned from |
| # Dhcpv6TestBase.get_interface_ipconfig(). |
| DHCPV6_KEY_ADDRESS = 'Address' |
| DHCPV6_KEY_DELEGATED_PREFIX = 'DelegatedPrefix' |
| DHCPV6_KEY_DELEGATED_PREFIX_LENGTH = 'DelegatedPrefixLength' |
| DHCPV6_KEY_NAMESERVERS = 'NameServers' |
| DHCPV6_KEY_SEARCH_DOMAIN_LIST = 'SearchDomains' |
| |
| # After DHCPv6 completes, an ipconfig should appear shortly after |
| IPCONFIG_POLL_COUNT = 5 |
| IPCONFIG_POLL_PERIOD_SECONDS = 1 |
| |
| class Dhcpv6TestBase(test.test): |
| """Parent class for tests that work verify DHCPv6 behavior.""" |
| version = 1 |
| |
| def get_device(self, interface_name): |
| """Finds the corresponding Device object for an interface with |
| the name |interface_name|. |
| |
| @param interface_name string The name of the interface to check. |
| |
| @return DBus interface object representing the associated device. |
| |
| """ |
| return self.shill_proxy.find_object('Device', |
| {'Name': interface_name}) |
| |
| |
| def find_ethernet_service(self, interface_name): |
| """Finds the corresponding service object for an Ethernet interface. |
| |
| @param interface_name string The name of the associated interface |
| |
| @return Service object representing the associated service. |
| |
| """ |
| device = self.get_device(interface_name) |
| device_path = shill_proxy.ShillProxy.dbus2primitive(device.object_path) |
| return self.shill_proxy.find_object('Service', {'Device': device_path}) |
| |
| |
| def get_interface_ipconfig_objects(self, interface_name): |
| """ |
| Returns a list of dbus object proxies for |interface_name|. |
| Returns an empty list if no such interface exists. |
| |
| @param interface_name string name of the device to query (e.g., "eth0"). |
| |
| @return list of objects representing DBus IPConfig RPC endpoints. |
| |
| """ |
| device = self.get_device(interface_name) |
| if device is None: |
| return [] |
| |
| device_properties = device.GetProperties(utf8_strings=True) |
| proxy = self.shill_proxy |
| |
| ipconfig_object = proxy.DBUS_TYPE_IPCONFIG |
| return filter(bool, |
| [ proxy.get_dbus_object(ipconfig_object, property_path) |
| for property_path in device_properties['IPConfigs'] ]) |
| |
| |
| def get_interface_ipconfig(self, interface_name): |
| """ |
| Returns a dictionary containing settings for an |interface_name| set |
| via DHCPv6. Returns None if no such interface or setting bundle on |
| that interface can be found in shill. |
| |
| @param interface_name string name of the device to query (e.g., "eth0"). |
| |
| @return dict containing the the properties of the IPConfig stripped |
| of DBus meta-data or None. |
| |
| """ |
| dhcp_properties = None |
| for ipconfig in self.get_interface_ipconfig_objects(interface_name): |
| logging.info('Looking at ipconfig %r', ipconfig) |
| ipconfig_properties = ipconfig.GetProperties(utf8_strings=True) |
| if 'Method' not in ipconfig_properties: |
| logging.info('Found ipconfig object with no method field') |
| continue |
| if ipconfig_properties['Method'] != 'dhcp6': |
| logging.info('Found ipconfig object with method != dhcp6') |
| continue |
| if dhcp_properties != None: |
| raise error.TestFail('Found multiple ipconfig objects ' |
| 'with method == dhcp6') |
| dhcp_properties = ipconfig_properties |
| if dhcp_properties is None: |
| logging.info('Did not find IPConfig object with method == dhcp6') |
| return None |
| logging.info('Got raw dhcp config dbus object: %s.', dhcp_properties) |
| return shill_proxy.ShillProxy.dbus2primitive(dhcp_properties) |
| |
| |
| def run_once(self): |
| self._server = None |
| self._server_ip = None |
| self._ethernet_pair = None |
| self._shill_proxy = shill_proxy.ShillProxy() |
| try: |
| # TODO(zqiu): enable DHCPv6 for peer interface, either by restarting |
| # shill with appropriate command line options or via a new DBUS |
| # command. |
| self._ethernet_pair = virtual_ethernet_pair.VirtualEthernetPair( |
| interface_ip=None, |
| peer_interface_name='pseudoethernet0', |
| peer_interface_ip=None, |
| interface_ipv6=dhcpv6_test_server.DHCPV6_SERVER_ADDRESS) |
| self._ethernet_pair.setup() |
| if not self._ethernet_pair.is_healthy: |
| raise error.TestFail('Could not create virtual ethernet pair.') |
| self._server_ip = self._ethernet_pair.interface_ip |
| self._server = dhcpv6_test_server.Dhcpv6TestServer( |
| self._ethernet_pair.interface_name) |
| self._server.start() |
| self.test_body() |
| except (error.TestFail, error.TestNAError): |
| # Pass these through without modification. |
| raise |
| except Exception as e: |
| logging.error('Caught exception: %s.', str(e)) |
| logging.error('Trace: %s', traceback.format_exc()) |
| raise error.TestFail('Caught exception: %s.' % str(e)) |
| finally: |
| if self._server is not None: |
| self._server.stop() |
| if self._ethernet_pair is not None: |
| self._ethernet_pair.teardown() |
| |
| def test_body(self): |
| """ |
| Override this method with the body of your test. You may safely assume |
| that the the properties exposed by DhcpTestBase correctly return |
| references to the test apparatus. |
| """ |
| raise error.TestFail('No test body implemented') |
| |
| @property |
| def server_ip(self): |
| """ |
| Return the IP address of the side of the interface that the DHCPv6 test |
| server is bound to. The server itself is bound the the broadcast |
| address on the interface. |
| """ |
| return self._server_ip |
| |
| @property |
| def server(self): |
| """ |
| Returns a reference to the DHCP test server. Use this to add handlers |
| and run tests. |
| """ |
| return self._server |
| |
| @property |
| def ethernet_pair(self): |
| """ |
| Returns a reference to the virtual ethernet pair created to run DHCP |
| tests on. |
| """ |
| return self._ethernet_pair |
| |
| @property |
| def shill_proxy(self): |
| """ |
| Returns a the shill proxy instance. |
| """ |
| return self._shill_proxy |
| |
| |
| def check_dhcpv6_config(self): |
| """ |
| Compare the DHCPv6 ipconfig with DHCP lease parameters to ensure |
| that the DUT attained the correct values. |
| |
| """ |
| # Retrieve DHCPv6 configuration. |
| for attempt in range(IPCONFIG_POLL_COUNT): |
| dhcpv6_config = self.get_interface_ipconfig( |
| self.ethernet_pair.peer_interface_name) |
| # Wait until both IP address and delegated prefix are obtained. |
| if (dhcpv6_config is not None and |
| dhcpv6_config.get(DHCPV6_KEY_ADDRESS) and |
| dhcpv6_config.get(DHCPV6_KEY_DELEGATED_PREFIX)): |
| break; |
| time.sleep(IPCONFIG_POLL_PERIOD_SECONDS) |
| else: |
| raise error.TestFail('Failed to retrieve DHCPv6 ipconfig object ' |
| 'from shill.') |
| |
| # Verify Non-temporary Address prefix. |
| address = dhcpv6_config.get(DHCPV6_KEY_ADDRESS) |
| actual_prefix = address[:address.index('::')] |
| expected_prefix = dhcpv6_test_server.DHCPV6_SERVER_SUBNET_PREFIX[: |
| dhcpv6_test_server.DHCPV6_SERVER_SUBNET_PREFIX.index('::')] |
| if actual_prefix != expected_prefix: |
| raise error.TestFail('Address prefix mismatch: ' |
| 'actual %s expected %s.' % |
| (actual_prefix, expected_prefix)) |
| # Verify Non-temporary Address suffix. |
| actual_suffix = int(address[address.index('::')+2:], 16) |
| if (actual_suffix < dhcpv6_test_server.DHCPV6_ADDRESS_RANGE_LOW or |
| actual_suffix > dhcpv6_test_server.DHCPV6_ADDRESS_RANGE_HIGH): |
| raise error.TestFail('Invalid address suffix: ' |
| 'actual %x expected (%x-%x)' % |
| (actual_suffix, |
| dhcpv6_test_server.DHCPV6_ADDRESS_RANGE_LOW, |
| dhcpv6_test_server.DHCPV6_ADDRESS_RANGE_HIGH)) |
| |
| # Verify delegated prefix. |
| delegated_prefix = dhcpv6_config.get(DHCPV6_KEY_DELEGATED_PREFIX) |
| for x in range( |
| dhcpv6_test_server.DHCPV6_PREFIX_DELEGATION_INDEX_LOW, |
| dhcpv6_test_server.DHCPV6_PREFIX_DELEGATION_INDEX_HIGH+1): |
| valid_prefix = \ |
| dhcpv6_test_server.DHCPV6_PREFIX_DELEGATION_RANGE_FORMAT % x |
| if delegated_prefix == valid_prefix: |
| break; |
| else: |
| raise error.TestFail('Invalid delegated prefix: %s' % |
| (delegated_prefix)) |
| # Verify delegated prefix length. |
| delegated_prefix_length = \ |
| int(dhcpv6_config.get(DHCPV6_KEY_DELEGATED_PREFIX_LENGTH)) |
| expected_delegated_prefix_length = \ |
| dhcpv6_test_server.DHCPV6_PREFIX_DELEGATION_PREFIX_LENGTH |
| if delegated_prefix_length != expected_delegated_prefix_length: |
| raise error.TestFail('Delegated prefix length mismatch: ' |
| 'actual %d expected %d' % |
| (delegated_prefix_length, |
| expected_delegated_prefix_length)) |
| |
| # Verify name servers. |
| actual_name_servers = dhcpv6_config.get(DHCPV6_KEY_NAMESERVERS) |
| expected_name_servers = \ |
| dhcpv6_test_server.DHCPV6_NAME_SERVERS.split(',') |
| if actual_name_servers != expected_name_servers: |
| raise error.TestFail('Name servers mismatch: actual %r expected %r' |
| % (actual_name_servers, expected_name_servers)) |
| # Verify domain search. |
| actual_domain_search = dhcpv6_config.get(DHCPV6_KEY_SEARCH_DOMAIN_LIST) |
| expected_domain_search = \ |
| dhcpv6_test_server.DHCPV6_DOMAIN_SEARCH.split(',') |
| if actual_domain_search != expected_domain_search: |
| raise error.TestFail('Domain search list mismatch: ' |
| 'actual %r expected %r' % |
| (actual_domain_search, expected_domain_search)) |
