Python lambdas and variable scope

The issue

Recently, a colleague posted the following code snippet and asked for an explanation of why it didn’t work as they expected:

x = []
for i in range(1, 10):
    x.append(lambda z: "%d" % i + z
print(list(x[i]("banzaii") for i in range(0, 9))

which results in:

['9banzaii', '9banzaii', '9banzaii', '9banzaii', '9banzaii', '9banzaii', '9banzaii', '9banzaii', '9banzaii']

instead of the expected:

['1banzaii', '2banzaii', '3banzaii', '4banzaii', '5banzaii', '6banzaii', '7banzaii', '8banzaii', '9banzaii']

The explanation

The key is that the variable i is outside of the scope of the created lambda. Only the variable z is in the lambda scope. Essentially it would be like writing:

x=[]
def fn(z):
    return "%d" % i + z

for i in range(1, 10):
    x.append(fn)

which makes it a lot more obvious what is going on. The global variable i is used each time the function is actually called—which is after the for loop is executed and i is now 9.

The fix

The fix is simple: make sure that i is in the scope of the function that is appended to array x.

Solution 1: using a closure

def makefunc(i):
    def inner(z):
        return "%d" % i + z
    return inner

x=[]
for i in range(1, 10):
    x.append(makefunc(i))

In this example, a closure around the function inner passes i into the returned function’s scope.

Solution 2: using lambda to create an anonymous closure

x=[]
for i in range(1, 10):
    x.append((lambda n: lambda z: "%d" % n + z)(i))

Here, we are appending the result of calling a lambda function with the argument i which returns a lambda that uses that value. This is exactly equivalent to the previous example, but for people who like one-liners. :)

Solution 3: using a lambda with a default argument

x=[]
for i in range(1, 10):
    x.append(lambda z, n=i: "%d" % n + z)

This solution ensures that we are using a variable n that is in the lambda’s scope and is initialized by default to i. A possible issue with this solution is if the caller of the function in x[i] passes an additional argument which overrides the default value of n. With the previous solutions, this bug would raise a TypeError exception instead of running successfully with (most likely) unintended results.

Neutron external sanity checks

Intro

Recently, a new helper script was added to neutron to facilitate testing whether or not a running system supports all of the Neutron configuration options that have been enabled. Before this script, ad hoc code kept being added to Neutron to check component version numbers at runtime. Due to the increasing number of features added and the brittleness inherent in testing version numbers for feature availability, this functionality was moved into an external check that could be run once at deployment time and specifically test for system suitability by trying to use the required features. For example, if VXLAN tunnels are enabled, the script can be run to verify that the installed version of Open vSwitch is capable of creating VXLAN ports by actually trying to create the port and removing it.

Usage

Checks can be run either by passing CLI arguments to test a system for specific feature support, or by passing in config files from the system and automatically testing each of the enabled features covered by the test script.

For example, to test specifically for Open vSwitch VXLAN support from the CLI:

neutron-sanity-check —ovs-vxlan

To run all tests that map to enabled configuration options, pass in the config-file options for each of the configs to be tested. For example:

neutron-sanity-check —config-file /etc/neutron/neutron.conf —config-file /etc/neutron/plugins/ml2/ml2_conf.ini

Adding new tests

To add a new test, first define the option in neutron/cmd/sanity_check.py in the OPTS variable using the existing options as a guide. The BoolOptCallback is just like a standard BoolOpt, but also allows you to pass a callback function which actually defines the test that should be run by the script.

Example:

OPTS = [
    BoolOptCallback('ovs_vxlan', check_ovs_vxlan, default=False,
                    help=_('Check for vxlan support')),
]

Next, add a check to enable_tests_from_config() to override the CLI option for the test defined in OPTS to True based on applicable config values.

For example:

if 'vxlan' in cfg.CONF.AGENT.tunnel_types:
    cfg.CONF.set_override('ovs_vxlan', True)

Next, add the new test to neutron/cmd/sanity/checks.py. Tests should test the actual feature required and not rely on version numbers. Tests should also return the system state to the way it was before the test was run. Return value should be True for passing, False for failing.

For example:

def vxlan_supported(root_helper, from_ip='192.0.2.1', to_ip='192.0.2.2'):
     name = "vxlantest-" + utils.get_random_string(6)
     with ovs_lib.OVSBridge(name, root_helper) as br:
         port = br.add_tunnel_port(from_ip, to_ip, const.TYPE_VXLAN)
         return port != ovs_const.INVALID_OFPORT

Next, go back to neutron/cmd/sanity_check.py and add the callback function that was defined in OPTS. This function should take no arguments and should log a helpful error message on failure. If testing via config options is supported, it should use the cfg.CONF option that the test depends on. The return value should be True for passing, False for failing. The reason this function is separate from the neutron/cmd/sanity/checks.py version is that the tests in sanity_check.py must all have the same arguments and generally will use values from cfg.CONF, whereas for testing purposes other argument values may be necessary.

Example:

def check_ovs_vxlan():
    result = checks.vxlan_supported(root_helper=cfg.CONF.AGENT.root_helper)
    if not result:
        LOG.error(_('Check for Open vSwitch VXLAN support failed. '
                    'Please ensure that the version of openvswitch '
                    'being used has VXLAN support.'))
    return result

Finally, define a functional test in neutron/tests/functional/sanity/test_ovs_sanity.py which ensures that they defined check function runs through to completion. This will alert developers if an API change is made that breaks the sanity check functions. For example:

def test_ovs_vxlan_support_runs(self):
    """This test just ensures that the test in neutron-sanity-check
        can run through without error, without mocking anything out
    """
    self.check_sudo_enabled()
    checks.vxlan_supported(self.root_helper)