text:
var should      = require( 'should' );
var sinon       = require( 'sinon' );
var _           = require( 'lodash' );
var uuid        = require( 'node-uuid' );
var proxyquire  = require( 'proxyquire' ).noCallThru();

// Configure proxyquire stubs
var LN_DEPS_MOD_PATH = '../lib/link-dependencies';

// Proxyquire stub for happy-path through the module
var defaultStub = {
    path: {
        resolve: function ( path ) { return path }
    },
    fs: {
        existsSync: function () { return true }
    },
    mkdirp: {
        sync: _.noop
    },

    // Parsing package paths should resolve to an array
    'fake-pkg-path':    {  linkDependencies: [] },
    'package.json':     {  linkDependencies: [] }
};

// Return the stub, but allow for overrides
var getProxyStub = function ( overrides ) {
    return _.assign( {}, defaultStub, overrides );
};

// Spec
describe( 'link-dependencies', function () {

    describe( 'package file validation', function () {

        // Bad package path
        it( 'errors if the package path is invalid', function () {
            var stub = getProxyStub( { fs: { existsSync: function () { return false } } } );
            var lnDeps = proxyquire( LN_DEPS_MOD_PATH, stub );
            lnDeps.bind( null, 'fake-pkg-path' ).should.throw( 'Invalid package path: fake-pkg-path' );
        } );

        // Malformed package JSON
        it( 'errors if it has trouble parsing the package file', function () {
            var UNLIKELY_FILENAME = uuid.v4();
            var ERR_MSG_PATTERN = new RegExp( 'Error parsing package file "' + UNLIKELY_FILENAME + '": .*' );

            var stub = getProxyStub( { path: { resolve: function () { return UNLIKELY_FILENAME } } } );
            var lnDeps = proxyquire( LN_DEPS_MOD_PATH, stub );

            lnDeps.bind().should.throw( ERR_MSG_PATTERN );
        } );
    } );

    describe( 'link dependencies validation', function () {

        // Missing dependency list
        it( 'warns if `linkDependencies` is not found in the package.json file', function () {
            var WARN_MSG = '`linkDependencies` not found in fake-pkg-path. No dependencies will be linked.';

            var stub = getProxyStub( { 'fake-pkg-path': { linkDependencies: null } } );
            var lnDeps = proxyquire( LN_DEPS_MOD_PATH, stub );

            var consoleWarnStub = sinon.stub( global.console, 'warn' );

            lnDeps( 'fake-pkg-path' );

            consoleWarnStub.calledWith( WARN_MSG ).should.be.true;
            consoleWarnStub.calledOnce.should.be.true;
            consoleWarnStub.reset();
        } );

        // Mis-typed dependency list
        it( 'errors if `linkDependencies` is not an array', function () {
            var ERR_MSG = '`linkDependencies` needs to be an array of module paths in fake-pkg-path';
            var stub = getProxyStub( {
                'fake-pkg-path': { linkDependencies: true },
                _: { isArray: function () { return false } }
            } );
            var lnDeps = proxyquire( LN_DEPS_MOD_PATH, stub );
            lnDeps.bind( null, 'fake-pkg-path' ).should.throw( ERR_MSG );
        } );
    } );

    describe( 'output directory handling', function () {

        // node_modules destination directory
        it( 'resolves and creates a node_module directory', function () {
            var pathResolveSpy = sinon.spy( getProxyStub().path.resolve );
            var mkdirpSpy = sinon.spy( getProxyStub().mkdirp.sync );

            var stub = getProxyStub( {
                path: { resolve: pathResolveSpy },
                mkdirp: { sync: mkdirpSpy }
            } );

            var lnDeps = proxyquire( LN_DEPS_MOD_PATH, stub );
            lnDeps();

            pathResolveSpy.calledWith( 'node_modules' ).should.be.true;
            mkdirpSpy.calledAfter( pathResolveSpy );
            mkdirpSpy.calledWith( 'node_modules' ).should.be.true;
            mkdirpSpy.calledOnce.should.be.true;
        } );
    } );

    describe( 'link and unlink', function () {

        // Link deps
        it( 'links the specified dependencies', function () {
            var lnModSpy = sinon.spy();
            var stub = getProxyStub( {
                './link-module': lnModSpy
            } );
            var lnDeps = proxyquire( LN_DEPS_MOD_PATH, stub );
            lnDeps();
            lnModSpy.calledWith( [], 'node_modules', false ).should.be.true;
            lnModSpy.calledOnce.should.be.true;
        } );

        // Unlink deps
        it( 'unlinks the specified dependencies', function () {
            var pathJoinSpy = sinon.spy( function () { return 'path-join-return' } );
            var pathResolveSpy  = sinon.spy( function ( path ) {
                if ( path === 'fake-pkg-path' || path === 'node_modules' ) return path;
                return 'fake-mod-path/package.json'
            } );
            var rmSpy = sinon.spy();

            var stub = getProxyStub( {
                path: {
                    join: pathJoinSpy,
                    resolve: pathResolveSpy
                },
                rimraf: {
                    sync: rmSpy
                },
                'fake-pkg-path': {
                    linkDependencies: [ 'fake-mod-path' ]
                },
                'fake-mod-path/package.json': {
                    name: 'fake-mod-name'
                }
            } );
            var lnDeps = proxyquire( LN_DEPS_MOD_PATH, stub );
            lnDeps( 'fake-pkg-path', true );

            pathJoinSpy.calledWith( 'fake-mod-path', 'package.json' ).should.be.true;
            pathJoinSpy.calledWith( 'node_modules', 'fake-mod-name' ).should.be.true;
            pathJoinSpy.calledTwice.should.be.true;

            pathResolveSpy.calledWith( 'node_modules' ).should.be.true;
            pathResolveSpy.calledWith( 'path-join-return' ).should.be.true;
            pathResolveSpy.calledThrice.should.be.true;

            rmSpy.calledWith( 'path-join-return' ).should.be.true;
            rmSpy.calledOnce.should.be.true;
        } );

        // Unlink deps bad package
        it( '...unless there is an error parsing a dependencie\'s package.json', function () {
            var UNLIKELY_FILENAME = uuid.v4();
            var ERR_MSG_PATTERN = new RegExp( 'Error parsing name from package file ' + UNLIKELY_FILENAME + ': .*' );

            var pathJoinSpy = sinon.spy( function () { return 'path-join-return' } );
            var pathResolveSpy  = sinon.spy( function ( path ) {
                if ( path === 'fake-pkg-path' || path === 'node_modules' ) return path;
                return UNLIKELY_FILENAME;
            } );

            var stub = getProxyStub( {
                path: {
                    join: pathJoinSpy,
                    resolve: pathResolveSpy
                },
                'fake-pkg-path': {
                    linkDependencies: [ 'fake-mod-path' ]
                }
            } );

            var lnDeps = proxyquire( LN_DEPS_MOD_PATH, stub );
            lnDeps.bind( null, 'fake-pkg-path', true ).should.throw();

            pathJoinSpy.calledWith( 'fake-mod-path', 'package.json' ).should.be.true;
            pathJoinSpy.calledOnce.should.be.true;
        } );
    } );
} );


text:
import contextlib
import os

import mock
from oslo_config import cfg

from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
from neutron.common import constants
from neutron.common import exceptions
from neutron.plugins.common import constants as p_const
from neutron.plugins.linuxbridge.agent import linuxbridge_neutron_agent
from neutron.plugins.linuxbridge.common import constants as lconst
from neutron.tests import base

LOCAL_IP = '192.168.0.33'
DEVICE_1 = 'tapabcdef01-12'

class FakeIpLinkCommand(object):
    def set_up(self):
        pass

class FakeIpDevice(object):
    def __init__(self):
        self.link = FakeIpLinkCommand()

class TestLinuxBridge(base.BaseTestCase):

    def setUp(self):
        super(TestLinuxBridge, self).setUp()
        interface_mappings = {'physnet1': 'eth1'}

        with mock.patch.object(linuxbridge_neutron_agent.LinuxBridgeManager,
                               'get_interface_by_ip', return_value=None):
            self.linux_bridge = linuxbridge_neutron_agent.LinuxBridgeManager(
                interface_mappings)

    def test_ensure_physical_in_bridge_invalid(self):
        result = self.linux_bridge.ensure_physical_in_bridge('network_id',
                                                             p_const.TYPE_VLAN,
                                                             'physnetx',
                                                             7)
        self.assertFalse(result)

    def test_ensure_physical_in_bridge_flat(self):
        with mock.patch.object(self.linux_bridge,
                               'ensure_flat_bridge') as flat_bridge_func:
            self.linux_bridge.ensure_physical_in_bridge(
                'network_id', p_const.TYPE_FLAT, 'physnet1', None)
        self.assertTrue(flat_bridge_func.called)

    def test_ensure_physical_in_bridge_vlan(self):
        with mock.patch.object(self.linux_bridge,
                               'ensure_vlan_bridge') as vlan_bridge_func:
            self.linux_bridge.ensure_physical_in_bridge(
                'network_id', p_const.TYPE_VLAN, 'physnet1', 7)
        self.assertTrue(vlan_bridge_func.called)

    def test_ensure_physical_in_bridge_vxlan(self):
        self.linux_bridge.vxlan_mode = lconst.VXLAN_UCAST
        with mock.patch.object(self.linux_bridge,
                               'ensure_vxlan_bridge') as vxlan_bridge_func:
            self.linux_bridge.ensure_physical_in_bridge(
                'network_id', 'vxlan', 'physnet1', 7)
        self.assertTrue(vxlan_bridge_func.called)

class TestLinuxBridgeAgent(base.BaseTestCase):

    def setUp(self):
        super(TestLinuxBridgeAgent, self).setUp()
        # disable setting up periodic state reporting
        cfg.CONF.set_override('report_interval', 0, 'AGENT')
        cfg.CONF.set_default('firewall_driver',
                             'neutron.agent.firewall.NoopFirewallDriver',
                             group='SECURITYGROUP')
        self.get_devices_p = mock.patch.object(ip_lib.IPWrapper, 'get_devices')
        self.get_devices = self.get_devices_p.start()
        self.get_devices.return_value = [ip_lib.IPDevice('eth77')]
        self.get_mac_p = mock.patch('neutron.agent.linux.utils.'
                                    'get_interface_mac')
        self.get_mac = self.get_mac_p.start()
        self.get_mac.return_value = '00:00:00:00:00:01'
        with mock.patch.object(linuxbridge_neutron_agent.LinuxBridgeManager,
                               'get_interface_by_ip', return_value=None):
            self.agent = linuxbridge_neutron_agent.LinuxBridgeNeutronAgentRPC(
                {}, 0)

    def test_treat_devices_removed_with_existed_device(self):
        agent = self.agent
        devices = [DEVICE_1]
        with contextlib.nested(
            mock.patch.object(agent.plugin_rpc, "update_device_down"),
            mock.patch.object(agent.sg_agent, "remove_devices_filter")
        ) as (fn_udd, fn_rdf):
            fn_udd.return_value = {'device': DEVICE_1,
                                   'exists': True}
            with mock.patch.object(linuxbridge_neutron_agent.LOG,
                                   'info') as log:
                resync = agent.treat_devices_removed(devices)
                self.assertEqual(2, log.call_count)
                self.assertFalse(resync)
                self.assertTrue(fn_udd.called)
                self.assertTrue(fn_rdf.called)

    def test_treat_devices_removed_with_not_existed_device(self):
        agent = self.agent
        devices = [DEVICE_1]
        with contextlib.nested(
            mock.patch.object(agent.plugin_rpc, "update_device_down"),
            mock.patch.object(agent.sg_agent, "remove_devices_filter")
        ) as (fn_udd, fn_rdf):
            fn_udd.return_value = {'device': DEVICE_1,
                                   'exists': False}
            with mock.patch.object(linuxbridge_neutron_agent.LOG,
                                   'debug') as log:
                resync = agent.treat_devices_removed(devices)
                self.assertEqual(1, log.call_count)
                self.assertFalse(resync)
                self.assertTrue(fn_udd.called)
                self.assertTrue(fn_rdf.called)

    def test_treat_devices_removed_failed(self):
        agent = self.agent
        devices = [DEVICE_1]
        with contextlib.nested(
            mock.patch.object(agent.plugin_rpc, "update_device_down"),
            mock.patch.object(agent.sg_agent, "remove_devices_filter")
        ) as (fn_udd, fn_rdf):
            fn_udd.side_effect = Exception()
            with mock.patch.object(linuxbridge_neutron_agent.LOG,
                                   'debug') as log:
                resync = agent.treat_devices_removed(devices)
                self.assertEqual(2, log.call_count)
                self.assertTrue(resync)
                self.assertTrue(fn_udd.called)
                self.assertTrue(fn_rdf.called)

    def _test_scan_devices(self, previous, updated,
                           fake_current, expected, sync):
        self.agent.br_mgr = mock.Mock()
        self.agent.br_mgr.get_tap_devices.return_value = fake_current

        self.agent.updated_devices = updated
        results = self.agent.scan_devices(previous, sync)
        self.assertEqual(expected, results)

    def test_scan_devices_no_changes(self):
        previous = {'current': set([1, 2]),
                    'updated': set(),
                    'added': set(),
                    'removed': set()}
        fake_current = set([1, 2])
        updated = set()
        expected = {'current': set([1, 2]),
                    'updated': set(),
                    'added': set(),
                    'removed': set()}

        self._test_scan_devices(previous, updated, fake_current, expected,
                                sync=False)

    def test_scan_devices_added_removed(self):
        previous = {'current': set([1, 2]),
                    'updated': set(),
                    'added': set(),
                    'removed': set()}
        fake_current = set([2, 3])
        updated = set()
        expected = {'current': set([2, 3]),
                    'updated': set(),
                    'added': set([3]),
                    'removed': set([1])}

        self._test_scan_devices(previous, updated, fake_current, expected,
                                sync=False)

    def test_scan_devices_removed_retried_on_sync(self):
        previous = {'current': set([2, 3]),
                    'updated': set(),
                    'added': set(),
                    'removed': set([1])}
        fake_current = set([2, 3])
        updated = set()
        expected = {'current': set([2, 3]),
                    'updated': set(),
                    'added': set([2, 3]),
                    'removed': set([1])}

        self._test_scan_devices(previous, updated, fake_current, expected,
                                sync=True)

    def test_scan_devices_vanished_removed_on_sync(self):
        previous = {'current': set([2, 3]),
                    'updated': set(),
                    'added': set(),
                    'removed': set([1])}
        # Device 2 disappeared.
        fake_current = set([3])
        updated = set()
        # Device 1 should be retried.
        expected = {'current': set([3]),
                    'updated': set(),
                    'added': set([3]),
                    'removed': set([1, 2])}

        self._test_scan_devices(previous, updated, fake_current, expected,
                                sync=True)

    def test_scan_devices_updated(self):
        previous = {'current': set([1, 2]),
                    'updated': set(),
                    'added': set(),
                    'removed': set()}
        fake_current = set([1, 2])
        updated = set([1])
        expected = {'current': set([1, 2]),
                    'updated': set([1]),
                    'added': set(),
                    'removed': set()}

        self._test_scan_devices(previous, updated, fake_current, expected,
                                sync=False)

    def test_scan_devices_updated_non_existing(self):
        previous = {'current': set([1, 2]),
                    'updated': set(),
                    'added': set(),
                    'removed': set()}
        fake_current = set([1, 2])
        updated = set([3])
        expected = {'current': set([1, 2]),
                    'updated': set(),
                    'added': set(),
                    'removed': set()}

        self._test_scan_devices(previous, updated, fake_current, expected,
                                sync=False)

    def test_scan_devices_updated_on_sync(self):
        previous = {'current': set([1, 2]),
                    'updated': set([1]),
                    'added': set(),
                    'removed': set()}
        fake_current = set([1, 2])
        updated = set([2])
        expected = {'current': set([1, 2]),
                    'updated': set([1, 2]),
                    'added': set([1, 2]),
                    'removed': set()}

        self._test_scan_devices(previous, updated, fake_current, expected,
                                sync=True)

    def test_process_network_devices(self):
        agent = self.agent
        device_info = {'current': set(),
                       'added': set(['tap3', 'tap4']),
                       'updated': set(['tap2', 'tap3']),
                       'removed': set(['tap1'])}
        agent.sg_agent.prepare_devices_filter = mock.Mock()
        agent.sg_agent.refresh_firewall = mock.Mock()
        agent.treat_devices_added_updated = mock.Mock(return_value=False)
        agent.treat_devices_removed = mock.Mock(return_value=False)

        agent.process_network_devices(device_info)

        agent.sg_agent.prepare_devices_filter.assert_called_with(
                set(['tap3', 'tap4']))
        self.assertTrue(agent.sg_agent.refresh_firewall.called)
        agent.treat_devices_added_updated.assert_called_with(set(['tap2',
                                                                  'tap3',
                                                                  'tap4']))
        agent.treat_devices_removed.assert_called_with(set(['tap1']))

    def test_treat_devices_added_updated_admin_state_up_true(self):
        agent = self.agent
        mock_details = {'device': 'dev123',
                        'port_id': 'port123',
                        'network_id': 'net123',
                        'admin_state_up': True,
                        'network_type': 'vlan',
                        'segmentation_id': 100,
                        'physical_network': 'physnet1'}
        agent.plugin_rpc = mock.Mock()
        agent.plugin_rpc.get_devices_details_list.return_value = [mock_details]
        agent.br_mgr = mock.Mock()
        agent.br_mgr.add_interface.return_value = True
        resync_needed = agent.treat_devices_added_updated(set(['tap1']))

        self.assertFalse(resync_needed)
        agent.br_mgr.add_interface.assert_called_with('net123', 'vlan',
                                                      'physnet1', 100,
                                                      'port123')
        self.assertTrue(agent.plugin_rpc.update_device_up.called)

    def test_treat_devices_added_updated_admin_state_up_false(self):
        agent = self.agent
        mock_details = {'device': 'dev123',
                        'port_id': 'port123',
                        'network_id': 'net123',
                        'admin_state_up': False,
                        'network_type': 'vlan',
                        'segmentation_id': 100,
                        'physical_network': 'physnet1'}
        agent.plugin_rpc = mock.Mock()
        agent.plugin_rpc.get_devices_details_list.return_value = [mock_details]
        agent.remove_port_binding = mock.Mock()
        resync_needed = agent.treat_devices_added_updated(set(['tap1']))

        self.assertFalse(resync_needed)
        agent.remove_port_binding.assert_called_with('net123', 'port123')
        self.assertFalse(agent.plugin_rpc.update_device_up.called)

class TestLinuxBridgeManager(base.BaseTestCase):
    def setUp(self):
        super(TestLinuxBridgeManager, self).setUp()
        self.interface_mappings = {'physnet1': 'eth1'}

        with mock.patch.object(linuxbridge_neutron_agent.LinuxBridgeManager,
                               'get_interface_by_ip', return_value=None):
            self.lbm = linuxbridge_neutron_agent.LinuxBridgeManager(
                self.interface_mappings)

    def test_interface_exists_on_bridge(self):
        with mock.patch.object(os, 'listdir') as listdir_fn:
            listdir_fn.return_value = ["abc"]
            self.assertTrue(
                self.lbm.interface_exists_on_bridge("br-int", "abc")
            )
            self.assertFalse(
                self.lbm.interface_exists_on_bridge("br-int", "abd")
            )

    def test_get_bridge_name(self):
        nw_id = "123456789101112"
        self.assertEqual(self.lbm.get_bridge_name(nw_id),
                         "brq" + nw_id[0:11])
        nw_id = ""
        self.assertEqual(self.lbm.get_bridge_name(nw_id),
                         "brq")

    def test_get_subinterface_name(self):
        self.assertEqual(self.lbm.get_subinterface_name("eth0", "0"),
                         "eth0.0")
        self.assertEqual(self.lbm.get_subinterface_name("eth0", ""),
                         "eth0.")

    def test_get_tap_device_name(self):
        if_id = "123456789101112"
        self.assertEqual(self.lbm.get_tap_device_name(if_id),
                         constants.TAP_DEVICE_PREFIX + if_id[0:11])
        if_id = ""
        self.assertEqual(self.lbm.get_tap_device_name(if_id),
                         constants.TAP_DEVICE_PREFIX)

    def test_get_vxlan_device_name(self):
        vn_id = constants.MAX_VXLAN_VNI
        self.assertEqual(self.lbm.get_vxlan_device_name(vn_id),
                         "vxlan-" + str(vn_id))
        self.assertIsNone(self.lbm.get_vxlan_device_name(vn_id + 1))

    def test_get_all_neutron_bridges(self):
        br_list = ["br-int", "brq1", "brq2", "br-ex"]
        with mock.patch.object(os, 'listdir') as listdir_fn:
            listdir_fn.return_value = br_list
            self.assertEqual(self.lbm.get_all_neutron_bridges(),
                             br_list[1:3])
            self.assertTrue(listdir_fn.called)

    def test_get_interfaces_on_bridge(self):
        with contextlib.nested(
            mock.patch.object(utils, 'execute'),
            mock.patch.object(os, 'listdir'),
            mock.patch.object(ip_lib, 'device_exists', return_value=True)
        ) as (exec_fn, listdir_fn, dev_exists_fn):
            listdir_fn.return_value = ["qbr1"]
            self.assertEqual(self.lbm.get_interfaces_on_bridge("br0"),
                             ["qbr1"])

    def test_get_interfaces_on_bridge_not_existing(self):
        with mock.patch.object(ip_lib, 'device_exists', return_value=False):
            self.assertEqual([], self.lbm.get_interfaces_on_bridge("br0"))

    def test_get_tap_devices_count(self):
        with mock.patch.object(os, 'listdir') as listdir_fn:
            listdir_fn.return_value = ['tap2101', 'eth0.100', 'vxlan-1000']
            self.assertEqual(self.lbm.get_tap_devices_count('br0'), 1)
            listdir_fn.side_effect = OSError()
            self.assertEqual(self.lbm.get_tap_devices_count('br0'), 0)

    def test_get_interface_by_ip(self):
        with contextlib.nested(
            mock.patch.object(ip_lib.IPWrapper, 'get_devices'),
            mock.patch.object(ip_lib.IpAddrCommand, 'list')
        ) as (get_dev_fn, ip_list_fn):
            device = mock.Mock()
            device.name = 'dev_name'
            get_dev_fn.return_value = [device]
            ip_list_fn.returnvalue = mock.Mock()
            self.assertEqual(self.lbm.get_interface_by_ip(LOCAL_IP),
                             'dev_name')

    def test_get_bridge_for_tap_device(self):
        with contextlib.nested(
            mock.patch.object(self.lbm, "get_all_neutron_bridges"),
            mock.patch.object(self.lbm, "get_interfaces_on_bridge")
        ) as (get_all_qbr_fn, get_if_fn):
            get_all_qbr_fn.return_value = ["br-int", "br-ex"]
            get_if_fn.return_value = ["tap1", "tap2", "tap3"]
            self.assertEqual(self.lbm.get_bridge_for_tap_device("tap1"),
                             "br-int")
            self.assertIsNone(self.lbm.get_bridge_for_tap_device("tap4"))

    def test_is_device_on_bridge(self):
        self.assertTrue(not self.lbm.is_device_on_bridge(""))
        with mock.patch.object(os.path, 'exists') as exists_fn:
            exists_fn.return_value = True
            self.assertTrue(self.lbm.is_device_on_bridge("tap1"))
            exists_fn.assert_called_with(
                "/sys/class/net/tap1/brport"
            )

    def test_get_interface_details(self):
        with contextlib.nested(
            mock.patch.object(ip_lib.IpAddrCommand, 'list'),
            mock.patch.object(ip_lib.IpRouteCommand, 'get_gateway')
        ) as (list_fn, getgw_fn):
            gwdict = dict(gateway='1.1.1.1')
            getgw_fn.return_value = gwdict
            ipdict = dict(cidr='1.1.1.1/24',
                          broadcast='1.1.1.255',
                          scope='global',
                          ip_version=4,
                          dynamic=False)
            list_fn.return_value = ipdict
            ret = self.lbm.get_interface_details("eth0")

            self.assertTrue(list_fn.called)
            self.assertTrue(getgw_fn.called)
            self.assertEqual(ret, (ipdict, gwdict))

    def test_ensure_flat_bridge(self):
        with contextlib.nested(
            mock.patch.object(ip_lib.IpAddrCommand, 'list'),
            mock.patch.object(ip_lib.IpRouteCommand, 'get_gateway')
        ) as (list_fn, getgw_fn):
            gwdict = dict(gateway='1.1.1.1')
            getgw_fn.return_value = gwdict
            ipdict = dict(cidr='1.1.1.1/24',
                          broadcast='1.1.1.255',
                          scope='global',
                          ip_version=4,
                          dynamic=False)
            list_fn.return_value = ipdict
            with mock.patch.object(self.lbm, 'ensure_bridge') as ens:
                self.assertEqual(
                    self.lbm.ensure_flat_bridge("123", "eth0"),
                    "eth0"
                )
                self.assertTrue(list_fn.called)
                self.assertTrue(getgw_fn.called)
                ens.assert_called_once_with("brq123", "eth0",
                                            ipdict, gwdict)

    def test_ensure_vlan_bridge(self):
        with contextlib.nested(
            mock.patch.object(self.lbm, 'ensure_vlan'),
            mock.patch.object(self.lbm, 'ensure_bridge'),
            mock.patch.object(self.lbm, 'get_interface_details'),
        ) as (ens_vl_fn, ens, get_int_det_fn):
            ens_vl_fn.return_value = "eth0.1"
            get_int_det_fn.return_value = (None, None)
            self.assertEqual(self.lbm.ensure_vlan_bridge("123", "eth0", "1"),
                             "eth0.1")
            ens.assert_called_with("brq123", "eth0.1", None, None)

            get_int_det_fn.return_value = ("ips", "gateway")
            self.assertEqual(self.lbm.ensure_vlan_bridge("123", "eth0", "1"),
                             "eth0.1")
            ens.assert_called_with("brq123", "eth0.1", "ips", "gateway")

    def test_ensure_local_bridge(self):
        with mock.patch.object(self.lbm, 'ensure_bridge') as ens_fn:
            self.lbm.ensure_local_bridge("54321")
            ens_fn.assert_called_once_with("brq54321")

    def test_ensure_vlan(self):
        with mock.patch.object(ip_lib, 'device_exists') as de_fn:
            de_fn.return_value = True
            self.assertEqual(self.lbm.ensure_vlan("eth0", "1"), "eth0.1")
            de_fn.return_value = False
            with mock.patch.object(utils, 'execute') as exec_fn:
                exec_fn.return_value = False
                self.assertEqual(self.lbm.ensure_vlan("eth0", "1"), "eth0.1")
                # FIXME(kevinbenton): validate the params to the exec_fn calls
                self.assertEqual(exec_fn.call_count, 2)
                exec_fn.return_value = True
                self.assertIsNone(self.lbm.ensure_vlan("eth0", "1"))
                self.assertEqual(exec_fn.call_count, 3)

    def test_ensure_vxlan(self):
        seg_id = "12345678"
        self.lbm.local_int = 'eth0'
        self.lbm.vxlan_mode = lconst.VXLAN_MCAST
        with mock.patch.object(ip_lib, 'device_exists') as de_fn:
            de_fn.return_value = True
            self.assertEqual(self.lbm.ensure_vxlan(seg_id), "vxlan-" + seg_id)
            de_fn.return_value = False
            with mock.patch.object(self.lbm.ip,
                                   'add_vxlan') as add_vxlan_fn:
                add_vxlan_fn.return_value = FakeIpDevice()
                self.assertEqual(self.lbm.ensure_vxlan(seg_id),
                                 "vxlan-" + seg_id)
                add_vxlan_fn.assert_called_with("vxlan-" + seg_id, seg_id,
                                                group="224.0.0.1",
                                                dev=self.lbm.local_int)
                cfg.CONF.set_override('l2_population', 'True', 'VXLAN')
                self.assertEqual(self.lbm.ensure_vxlan(seg_id),
                                 "vxlan-" + seg_id)
                add_vxlan_fn.assert_called_with("vxlan-" + seg_id, seg_id,
                                                group="224.0.0.1",
                                                dev=self.lbm.local_int,
                                                proxy=True)

    def test_update_interface_ip_details(self):
        gwdict = dict(gateway='1.1.1.1',
                      metric=50)
        ipdict = dict(cidr='1.1.1.1/24',
                      broadcast='1.1.1.255',
                      scope='global',
                      ip_version=4,
                      dynamic=False)
        with contextlib.nested(
            mock.patch.object(ip_lib.IpAddrCommand, 'add'),
            mock.patch.object(ip_lib.IpAddrCommand, 'delete')
        ) as (add_fn, del_fn):
            self.lbm.update_interface_ip_details("br0", "eth0",
                                                 [ipdict], None)
            self.assertTrue(add_fn.called)
            self.assertTrue(del_fn.called)

        with contextlib.nested(
            mock.patch.object(ip_lib.IpRouteCommand, 'add_gateway'),
            mock.patch.object(ip_lib.IpRouteCommand, 'delete_gateway')
        ) as (addgw_fn, delgw_fn):
            self.lbm.update_interface_ip_details("br0", "eth0",
                                                 None, gwdict)
            self.assertTrue(addgw_fn.called)
            self.assertTrue(delgw_fn.called)

    def test_bridge_exists_and_ensure_up(self):
        ip_lib_mock = mock.Mock()
        with mock.patch.object(ip_lib, 'IPDevice', return_value=ip_lib_mock):
            # device exists
            self.assertTrue(self.lbm._bridge_exists_and_ensure_up("br0"))
            self.assertTrue(ip_lib_mock.link.set_up.called)
            # device doesn't exists
            ip_lib_mock.link.set_up.side_effect = RuntimeError
            self.assertFalse(self.lbm._bridge_exists_and_ensure_up("br0"))

    def test_ensure_bridge(self):
        with contextlib.nested(
            mock.patch.object(self.lbm, '_bridge_exists_and_ensure_up'),
            mock.patch.object(utils, 'execute'),
            mock.patch.object(self.lbm, 'update_interface_ip_details'),
            mock.patch.object(self.lbm, 'interface_exists_on_bridge'),
            mock.patch.object(self.lbm, 'is_device_on_bridge'),
            mock.patch.object(self.lbm, 'get_bridge_for_tap_device'),
        ) as (de_fn, exec_fn, upd_fn, ie_fn, if_br_fn, get_if_br_fn):
            de_fn.return_value = False
            exec_fn.return_value = False
            self.assertEqual(self.lbm.ensure_bridge("br0", None), "br0")
            ie_fn.return_Value = False
            self.lbm.ensure_bridge("br0", "eth0")
            upd_fn.assert_called_with("br0", "eth0", None, None)
            ie_fn.assert_called_with("br0", "eth0")

            self.lbm.ensure_bridge("br0", "eth0", "ips", "gateway")
            upd_fn.assert_called_with("br0", "eth0", "ips", "gateway")
            ie_fn.assert_called_with("br0", "eth0")

            exec_fn.side_effect = Exception()
            de_fn.return_value = True
            self.lbm.ensure_bridge("br0", "eth0")
            ie_fn.assert_called_with("br0", "eth0")

            exec_fn.reset_mock()
            exec_fn.side_effect = None
            de_fn.return_value = True
            ie_fn.return_value = False
            get_if_br_fn.return_value = "br1"
            self.lbm.ensure_bridge("br0", "eth0")
            expected = [
                mock.call(['brctl', 'delif', 'br1', 'eth0'],
                          run_as_root=True),
                mock.call(['brctl', 'addif', 'br0', 'eth0'],
                          run_as_root=True),
            ]
            exec_fn.assert_has_calls(expected)

    def test_ensure_physical_in_bridge(self):
        self.assertFalse(
            self.lbm.ensure_physical_in_bridge("123", p_const.TYPE_VLAN,
                                               "phys", "1")
        )
        with mock.patch.object(self.lbm, "ensure_flat_bridge") as flbr_fn:
            self.assertTrue(
                self.lbm.ensure_physical_in_bridge("123", p_const.TYPE_FLAT,
                                                   "physnet1", None)
            )
            self.assertTrue(flbr_fn.called)
        with mock.patch.object(self.lbm, "ensure_vlan_bridge") as vlbr_fn:
            self.assertTrue(
                self.lbm.ensure_physical_in_bridge("123", p_const.TYPE_VLAN,
                                                   "physnet1", "1")
            )
            self.assertTrue(vlbr_fn.called)

        with mock.patch.object(self.lbm, "ensure_vxlan_bridge") as vlbr_fn:
            self.lbm.vxlan_mode = lconst.VXLAN_MCAST
            self.assertTrue(
                self.lbm.ensure_physical_in_bridge("123", p_const.TYPE_VXLAN,
                                                   "physnet1", "1")
            )
            self.assertTrue(vlbr_fn.called)

    def test_add_tap_interface(self):
        with mock.patch.object(ip_lib, "device_exists") as de_fn:
            de_fn.return_value = False
            self.assertFalse(
                self.lbm.add_tap_interface("123", p_const.TYPE_VLAN,
                                           "physnet1", "1", "tap1")
            )

            de_fn.return_value = True
            with contextlib.nested(
                mock.patch.object(self.lbm, "ensure_local_bridge"),
                mock.patch.object(utils, "execute"),
                mock.patch.object(self.lbm, "get_bridge_for_tap_device")
            ) as (en_fn, exec_fn, get_br):
                exec_fn.return_value = False
                get_br.return_value = True
                self.assertTrue(self.lbm.add_tap_interface("123",
                                                           p_const.TYPE_LOCAL,
                                                           "physnet1", None,
                                                           "tap1"))
                en_fn.assert_called_with("123")

                get_br.return_value = False
                exec_fn.return_value = True
                self.assertFalse(self.lbm.add_tap_interface("123",
                                                            p_const.TYPE_LOCAL,
                                                            "physnet1", None,
                                                            "tap1"))

            with contextlib.nested(
                mock.patch.object(self.lbm, "ensure_physical_in_bridge"),
                mock.patch.object(self.lbm, "ensure_tap_mtu"),
                mock.patch.object(self.lbm, "get_bridge_for_tap_device")
            ) as (ens_fn, en_mtu_fn, get_br):
                ens_fn.return_value = False
                self.assertFalse(self.lbm.add_tap_interface("123",
                                                            p_const.TYPE_VLAN,
                                                            "physnet1", "1",
                                                            "tap1"))

                ens_fn.return_value = "eth0.1"
                get_br.return_value = "brq123"
                self.lbm.add_tap_interface("123", p_const.TYPE_VLAN,
                                           "physnet1", "1", "tap1")
                en_mtu_fn.assert_called_once_with("tap1", "eth0.1")

    def test_add_interface(self):
        with mock.patch.object(self.lbm, "add_tap_interface") as add_tap:
            self.lbm.add_interface("123", p_const.TYPE_VLAN, "physnet-1",
                                   "1", "234")
            add_tap.assert_called_with("123", p_const.TYPE_VLAN, "physnet-1",
                                       "1", "tap234")

    def test_delete_vlan_bridge(self):
        with contextlib.nested(
            mock.patch.object(ip_lib, "device_exists"),
            mock.patch.object(self.lbm, "get_interfaces_on_bridge"),
            mock.patch.object(self.lbm, "remove_interface"),
            mock.patch.object(self.lbm, "get_interface_details"),
            mock.patch.object(self.lbm, "update_interface_ip_details"),
            mock.patch.object(self.lbm, "delete_vxlan"),
            mock.patch.object(utils, "execute")
        ) as (de_fn, getif_fn, remif_fn, if_det_fn,
              updif_fn, del_vxlan, exec_fn):
            de_fn.return_value = False
            self.lbm.delete_vlan_bridge("br0")
            self.assertFalse(getif_fn.called)

            de_fn.return_value = True
            getif_fn.return_value = ["eth0", "eth1", "vxlan-1002"]
            if_det_fn.return_value = ("ips", "gateway")
            exec_fn.return_value = False
            self.lbm.delete_vlan_bridge("br0")
            updif_fn.assert_called_with("eth1", "br0", "ips", "gateway")
            del_vxlan.assert_called_with("vxlan-1002")

    def test_delete_vlan_bridge_with_ip(self):
        with contextlib.nested(
            mock.patch.object(ip_lib, "device_exists"),
            mock.patch.object(self.lbm, "get_interfaces_on_bridge"),
            mock.patch.object(self.lbm, "remove_interface"),
            mock.patch.object(self.lbm, "get_interface_details"),
            mock.patch.object(self.lbm, "update_interface_ip_details"),
            mock.patch.object(self.lbm, "delete_vlan"),
            mock.patch.object(utils, "execute")
        ) as (de_fn, getif_fn, remif_fn, if_det_fn,
              updif_fn, del_vlan, exec_fn):
            de_fn.return_value = True
            getif_fn.return_value = ["eth0", "eth1.1"]
            if_det_fn.return_value = ("ips", "gateway")
            exec_fn.return_value = False
            self.lbm.delete_vlan_bridge("br0")
            updif_fn.assert_called_with("eth1.1", "br0", "ips", "gateway")
            self.assertFalse(del_vlan.called)

    def test_delete_vlan_bridge_no_ip(self):
        with contextlib.nested(
            mock.patch.object(ip_lib, "device_exists"),
            mock.patch.object(self.lbm, "get_interfaces_on_bridge"),
            mock.patch.object(self.lbm, "remove_interface"),
            mock.patch.object(self.lbm, "get_interface_details"),
            mock.patch.object(self.lbm, "update_interface_ip_details"),
            mock.patch.object(self.lbm, "delete_vlan"),
            mock.patch.object(utils, "execute")
        ) as (de_fn, getif_fn, remif_fn, if_det_fn,
              updif_fn, del_vlan, exec_fn):
            de_fn.return_value = True
            getif_fn.return_value = ["eth0", "eth1.1"]
            exec_fn.return_value = False
            if_det_fn.return_value = ([], None)
            self.lbm.delete_vlan_bridge("br0")
            del_vlan.assert_called_with("eth1.1")
            self.assertFalse(updif_fn.called)

    def test_delete_vxlan_bridge_no_int_mappings(self):
        interface_mappings = {}
        with mock.patch.object(linuxbridge_neutron_agent.LinuxBridgeManager,
                               'get_interface_by_ip', return_value=None):
            lbm = linuxbridge_neutron_agent.LinuxBridgeManager(
                interface_mappings)

        with contextlib.nested(
            mock.patch.object(ip_lib, "device_exists"),
            mock.patch.object(lbm, "get_interfaces_on_bridge"),
            mock.patch.object(lbm, "remove_interface"),
            mock.patch.object(lbm, "delete_vxlan"),
            mock.patch.object(utils, "execute")
        ) as (de_fn, getif_fn, remif_fn, del_vxlan, exec_fn):
            de_fn.return_value = False
            lbm.delete_vlan_bridge("br0")
            self.assertFalse(getif_fn.called)

            de_fn.return_value = True
            getif_fn.return_value = ["vxlan-1002"]
            exec_fn.return_value = False
            lbm.delete_vlan_bridge("br0")
            del_vxlan.assert_called_with("vxlan-1002")

    def test_remove_empty_bridges(self):
        self.lbm.network_map = {'net1': mock.Mock(), 'net2': mock.Mock()}

        def tap_count_side_effect(*args):
            return 0 if args[0] == 'brqnet1' else 1

        with contextlib.nested(
            mock.patch.object(self.lbm, "delete_vlan_bridge"),
            mock.patch.object(self.lbm, "get_tap_devices_count",
                              side_effect=tap_count_side_effect),
        ) as (del_br_fn, count_tap_fn):
            self.lbm.remove_empty_bridges()
            del_br_fn.assert_called_once_with('brqnet1')

    def test_remove_interface(self):
        with contextlib.nested(
            mock.patch.object(ip_lib, "device_exists"),
            mock.patch.object(self.lbm, "is_device_on_bridge"),
            mock.patch.object(utils, "execute")
        ) as (de_fn, isdev_fn, exec_fn):
            de_fn.return_value = False
            self.assertFalse(self.lbm.remove_interface("br0", "eth0"))
            self.assertFalse(isdev_fn.called)

            de_fn.return_value = True
            isdev_fn.return_value = False
            self.assertTrue(self.lbm.remove_interface("br0", "eth0"))

            isdev_fn.return_value = True
            exec_fn.return_value = True
            self.assertFalse(self.lbm.remove_interface("br0", "eth0"))

            exec_fn.return_value = False
            self.assertTrue(self.lbm.remove_interface("br0", "eth0"))

    def test_delete_vlan(self):
        with contextlib.nested(
            mock.patch.object(ip_lib, "device_exists"),
            mock.patch.object(utils, "execute")
        ) as (de_fn, exec_fn):
            de_fn.return_value = False
            self.lbm.delete_vlan("eth1.1")
            self.assertFalse(exec_fn.called)

            de_fn.return_value = True
            exec_fn.return_value = False
            self.lbm.delete_vlan("eth1.1")
            self.assertTrue(exec_fn.called)

    def _check_vxlan_support(self, expected, vxlan_ucast_supported,
                             vxlan_mcast_supported):
        with contextlib.nested(
            mock.patch.object(self.lbm, 'vxlan_ucast_supported',
                              return_value=vxlan_ucast_supported),
            mock.patch.object(self.lbm, 'vxlan_mcast_supported',
                              return_value=vxlan_mcast_supported)):
            if expected == lconst.VXLAN_NONE:
                self.assertRaises(exceptions.VxlanNetworkUnsupported,
                                  self.lbm.check_vxlan_support)
                self.assertEqual(expected, self.lbm.vxlan_mode)
            else:
                self.lbm.check_vxlan_support()
                self.assertEqual(expected, self.lbm.vxlan_mode)

    def test_check_vxlan_support(self):
        self._check_vxlan_support(expected=lconst.VXLAN_UCAST,
                                  vxlan_ucast_supported=True,
                                  vxlan_mcast_supported=True)
        self._check_vxlan_support(expected=lconst.VXLAN_MCAST,
                                  vxlan_ucast_supported=False,
                                  vxlan_mcast_supported=True)

        self._check_vxlan_support(expected=lconst.VXLAN_NONE,
                                  vxlan_ucast_supported=False,
                                  vxlan_mcast_supported=False)
        self._check_vxlan_support(expected=lconst.VXLAN_NONE,
                                  vxlan_ucast_supported=False,
                                  vxlan_mcast_supported=False)

    def _check_vxlan_ucast_supported(
            self, expected, l2_population, iproute_arg_supported, fdb_append):
        cfg.CONF.set_override('l2_population', l2_population, 'VXLAN')
        with contextlib.nested(
                mock.patch.object(
                    ip_lib, 'device_exists', return_value=False),
                mock.patch.object(self.lbm, 'delete_vxlan', return_value=None),
                mock.patch.object(self.lbm, 'ensure_vxlan', return_value=None),
                mock.patch.object(
                    utils, 'execute',
                    side_effect=None if fdb_append else RuntimeError()),
                mock.patch.object(
                    ip_lib, 'iproute_arg_supported',
                    return_value=iproute_arg_supported)):
                self.assertEqual(expected, self.lbm.vxlan_ucast_supported())

    def test_vxlan_ucast_supported(self):
        self._check_vxlan_ucast_supported(
            expected=False,
            l2_population=False, iproute_arg_supported=True, fdb_append=True)
        self._check_vxlan_ucast_supported(
            expected=False,
            l2_population=True, iproute_arg_supported=False, fdb_append=True)
        self._check_vxlan_ucast_supported(
            expected=False,
            l2_population=True, iproute_arg_supported=True, fdb_append=False)
        self._check_vxlan_ucast_supported(
            expected=True,
            l2_population=True, iproute_arg_supported=True, fdb_append=True)

    def _check_vxlan_mcast_supported(
            self, expected, vxlan_group, iproute_arg_supported):
        cfg.CONF.set_override('vxlan_group', vxlan_group, 'VXLAN')
        with mock.patch.object(
                ip_lib, 'iproute_arg_supported',
                return_value=iproute_arg_supported):
            self.assertEqual(expected, self.lbm.vxlan_mcast_supported())

    def test_vxlan_mcast_supported(self):
        self._check_vxlan_mcast_supported(
            expected=False,
            vxlan_group='',
            iproute_arg_supported=True)
        self._check_vxlan_mcast_supported(
            expected=False,
            vxlan_group='224.0.0.1',
            iproute_arg_supported=False)
        self._check_vxlan_mcast_supported(
            expected=True,
            vxlan_group='224.0.0.1',
            iproute_arg_supported=True)

class TestLinuxBridgeRpcCallbacks(base.BaseTestCase):
    def setUp(self):
        cfg.CONF.set_override('local_ip', LOCAL_IP, 'VXLAN')
        super(TestLinuxBridgeRpcCallbacks, self).setUp()

        class FakeLBAgent(object):
            def __init__(self):
                self.agent_id = 1
                with mock.patch.object(
                        linuxbridge_neutron_agent.LinuxBridgeManager,
                        'get_interface_by_ip', return_value=None):
                    self.br_mgr = (linuxbridge_neutron_agent.
                                   LinuxBridgeManager({'physnet1': 'eth1'}))

                self.br_mgr.vxlan_mode = lconst.VXLAN_UCAST
                segment = mock.Mock()
                segment.network_type = 'vxlan'
                segment.segmentation_id = 1
                self.br_mgr.network_map['net_id'] = segment

        self.lb_rpc = linuxbridge_neutron_agent.LinuxBridgeRpcCallbacks(
            object(),
            FakeLBAgent(),
            object()
        )

    def test_network_delete(self):
        with contextlib.nested(
            mock.patch.object(self.lb_rpc.agent.br_mgr, "get_bridge_name"),
            mock.patch.object(self.lb_rpc.agent.br_mgr, "delete_vlan_bridge")
        ) as (get_br_fn, del_fn):
            get_br_fn.return_value = "br0"
            self.lb_rpc.network_delete("anycontext", network_id="123")
            get_br_fn.assert_called_with("123")
            del_fn.assert_called_with("br0")

    def test_fdb_add(self):
        fdb_entries = {'net_id':
                       {'ports':
                        {'agent_ip': [constants.FLOODING_ENTRY,
                                      ['port_mac', 'port_ip']]},
                        'network_type': 'vxlan',
                        'segment_id': 1}}

        with mock.patch.object(utils, 'execute',
                               return_value='') as execute_fn:
            self.lb_rpc.fdb_add(None, fdb_entries)

            expected = [
                mock.call(['bridge', 'fdb', 'show', 'dev', 'vxlan-1'],
                          run_as_root=True),
                mock.call(['bridge', 'fdb', 'add',
                           constants.FLOODING_ENTRY[0],
                           'dev', 'vxlan-1', 'dst', 'agent_ip'],
                          run_as_root=True,
                          check_exit_code=False),
                mock.call(['ip', 'neigh', 'replace', 'port_ip', 'lladdr',
                           'port_mac', 'dev', 'vxlan-1', 'nud', 'permanent'],
                          run_as_root=True,
                          check_exit_code=False),
                mock.call(['bridge', 'fdb', 'replace', 'port_mac', 'dev',
                           'vxlan-1', 'dst', 'agent_ip'],
                          run_as_root=True,
                          check_exit_code=False),
            ]
            execute_fn.assert_has_calls(expected)

    def test_fdb_ignore(self):
        fdb_entries = {'net_id':
                       {'ports':
                        {LOCAL_IP: [constants.FLOODING_ENTRY,
                                    ['port_mac', 'port_ip']]},
                        'network_type': 'vxlan',
                        'segment_id': 1}}

        with mock.patch.object(utils, 'execute',
                               return_value='') as execute_fn:
            self.lb_rpc.fdb_add(None, fdb_entries)
            self.lb_rpc.fdb_remove(None, fdb_entries)

            self.assertFalse(execute_fn.called)

        fdb_entries = {'other_net_id':
                       {'ports':
                        {'192.168.0.67': [constants.FLOODING_ENTRY,
                                          ['port_mac', 'port_ip']]},
                        'network_type': 'vxlan',
                        'segment_id': 1}}

        with mock.patch.object(utils, 'execute',
                               return_value='') as execute_fn:
            self.lb_rpc.fdb_add(None, fdb_entries)
            self.lb_rpc.fdb_remove(None, fdb_entries)

            self.assertFalse(execute_fn.called)

    def test_fdb_remove(self):
        fdb_entries = {'net_id':
                       {'ports':
                        {'agent_ip': [constants.FLOODING_ENTRY,
                                      ['port_mac', 'port_ip']]},
                        'network_type': 'vxlan',
                        'segment_id': 1}}

        with mock.patch.object(utils, 'execute',
                               return_value='') as execute_fn:
            self.lb_rpc.fdb_remove(None, fdb_entries)

            expected = [
                mock.call(['bridge', 'fdb', 'del',
                           constants.FLOODING_ENTRY[0],
                           'dev', 'vxlan-1', 'dst', 'agent_ip'],
                          run_as_root=True,
                          check_exit_code=False),
                mock.call(['ip', 'neigh', 'del', 'port_ip', 'lladdr',
                           'port_mac', 'dev', 'vxlan-1'],
                          run_as_root=True,
                          check_exit_code=False),
                mock.call(['bridge', 'fdb', 'del', 'port_mac',
                           'dev', 'vxlan-1', 'dst', 'agent_ip'],
                          run_as_root=True,
                          check_exit_code=False),
            ]
            execute_fn.assert_has_calls(expected)

    def test_fdb_update_chg_ip(self):
        fdb_entries = {'chg_ip':
                       {'net_id':
                        {'agent_ip':
                         {'before': [['port_mac', 'port_ip_1']],
                          'after': [['port_mac', 'port_ip_2']]}}}}

        with mock.patch.object(utils, 'execute',
                               return_value='') as execute_fn:
            self.lb_rpc.fdb_update(None, fdb_entries)

            expected = [
                mock.call(['ip', 'neigh', 'replace', 'port_ip_2', 'lladdr',
                           'port_mac', 'dev', 'vxlan-1', 'nud', 'permanent'],
                          run_as_root=True,
                          check_exit_code=False),
                mock.call(['ip', 'neigh', 'del', 'port_ip_1', 'lladdr',
                           'port_mac', 'dev', 'vxlan-1'],
                          run_as_root=True,
                          check_exit_code=False)
            ]
            execute_fn.assert_has_calls(expected)

    def test_fdb_update_chg_ip_empty_lists(self):
        fdb_entries = {'chg_ip': {'net_id': {'agent_ip': {}}}}
        self.lb_rpc.fdb_update(None, fdb_entries)

redpajama_set_name: RedPajamaGithub


text:
package org.hawkular.metrics.scheduler.api;

import static org.hawkular.metrics.datetime.DateTimeService.currentMinute;
import static org.hawkular.metrics.datetime.DateTimeService.getTimeSlice;
import static org.hawkular.metrics.datetime.DateTimeService.now;
import static org.joda.time.Duration.standardMinutes;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

import org.joda.time.Minutes;

/**
 * @author jsanda
 */
public class RepeatingTrigger implements Trigger {

    private Long triggerTime;

    private Long interval;

    private Long delay;

    private Integer repeatCount;

    private Integer executionCount;

    public static class Builder {

        private Long triggerTime;
        private Long interval = TimeUnit.MILLISECONDS.convert(1, TimeUnit.MINUTES);
        private Long delay;
        private Integer repeatCount;

        public Builder withTriggerTime(long time) {
            this.triggerTime = time;
            return this;
        }

        public Builder withInterval(int interval, TimeUnit timeUnit) {
            this.interval = TimeUnit.MILLISECONDS.convert(interval, timeUnit);
            return this;
        }

        public Builder withDelay(int delay, TimeUnit timeUnit) {
            this.delay = TimeUnit.MILLISECONDS.convert(delay, timeUnit);
            return this;
        }

        public Builder withRepeatCount(int count) {
            this.repeatCount = count;
            return this;
        }

        public RepeatingTrigger build() {
            return new RepeatingTrigger(triggerTime, interval, delay, repeatCount);
        }

    }

    private RepeatingTrigger() {
    }

    private RepeatingTrigger(Long triggerTime, Long interval, Long delay, Integer repeatCount) {
        if (triggerTime != null) {
            this.triggerTime = getTimeSlice(triggerTime, standardMinutes(1));
        }
        else if (interval == null && delay == null) {
            this.triggerTime = currentMinute().plusMinutes(1).getMillis();
        }

        this.interval = interval;
        this.delay = delay == null ? Minutes.ONE.toStandardDuration().getMillis() : delay;
        this.repeatCount = repeatCount;
        this.executionCount = 1;

        if (this.triggerTime == null) {
            this.triggerTime = getTimeSlice(now.get().getMillis() + this.delay, standardMinutes(1));
        }
    }

    // TODO reduce visibility?
    // This is for internal use by TaskSchedulerImpl.
    public RepeatingTrigger(long interval, long delay, long triggerTime, int repeatCount, int executionCount) {
        this.interval = interval;
        this.delay = delay;
        this.triggerTime = triggerTime;
        this.executionCount = executionCount;
        this.repeatCount = repeatCount == 0 ? null : repeatCount;
        this.executionCount = executionCount;
    }

    public long getInterval() {
        return interval;
    }

    public long getDelay() {
        return delay;
    }

    @Override
    public long getTriggerTime() {
        return triggerTime;
    }

    public Integer getRepeatCount() {
        return repeatCount;
    }

    public int getExecutionCount() {
        return executionCount;
    }

    @Override
    public Trigger nextTrigger() {
        // TODO what should we do if interval is null?

        if (repeatCount != null && executionCount + 1 > repeatCount) {
            return null;
        }
        RepeatingTrigger next = new RepeatingTrigger();
        next.interval = interval;
        next.delay = delay;
        next.triggerTime = triggerTime + interval;
        next.repeatCount = repeatCount;
        next.executionCount = executionCount + 1;

        return next;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        RepeatingTrigger that = (RepeatingTrigger) o;
        return Objects.equals(triggerTime, that.triggerTime) &&
                Objects.equals(interval, that.interval) &&
                Objects.equals(delay, that.delay);
    }

    @Override
    public int hashCode() {
        return Objects.hash(triggerTime, interval, delay);
    }

    @Override
    public String toString() {
        return "RepeatingTrigger{" +
                "triggerTime=" + triggerTime +
                ", interval=" + interval +
                ", delay=" + delay +
                '}';
    }

}


text:
<html>
<head>
<title>TrySmalltalk</title>
<script src="../../js/amber.js" type="text/javascript"></script>
</head>
<body>
<script type="text/javascript">
	loadAmber({
		files: ['TrySmalltalk.js'],
		prefix: 'examples/trysmalltalk/js',
		ready: function() {
			smalltalk.Browser._openOn_(smalltalk.ProfStef)
			smalltalk.TrySmalltalkWidget._open();
		}});
</script>

</body>
</html>


text:
<!DOCTYPE html>
<html>
  <head>
    <title>Theremin</title>
    <script src="../lib/coffeescript.js"></script>
  </head>
  <body>
    <canvas id="canvas"></canvas>

    <script type="text/coffeescript">
     # Criando o contexto de áudio, oscilador e ganho
     ctx = new (window.AudioConext || window.webkitAudioContext)()
     osc = ctx.createOscillator()
     gain = ctx.createGain()

     # Patching OSC -> GAIN -> OUT
     osc.connect gain
     gain.connect ctx.destination

     # Define parâmetros iniciais do oscilador
     osc.frequency.value = 440.0
     osc.detune.value = 0.0
     osc.type = "sine"

     # Define parâmetro inicias do ganho
     gain.gain.value = 0.01

     # Toca o oscilador daqui a 0 segundos (agora)
     osc.start 0

     # Atualização de parâmetros sonoros pelo mouse
     width = window.innerWidth
     height = window.innerHeight

     maxFrequency = 6000.0
     maxVolume = 0.02

     update = (e) ->
        x = e.clientX
        y = e.clientY

        osc.frequency.value = ((height-y)/height) * maxFrequency
        gain.gain.value = (x/width) * maxVolume

        draw(x, y)

     # Desenha o percurso do mouse
     canvas = document.getElementById('canvas')
     canvas.width = width
     canvas.height = height
     canvasCtx = canvas.getContext('2d')

     draw = (x, y) ->
        canvasCtx.fillStyle = '#000'
        canvasCtx.beginPath()
        canvasCtx.arc x, y, 10, 0, Math.PI*2, false
        canvasCtx.fill()

     document.onmousemove = update
    </script>
  </body>

</html>


text:
const fs = require('fs')
const fse = require('fs-extra')

const nodePackager = require('../helpers/node-packager')
const { spawnSync } = require('../helpers/spawn')
const appPaths = require('../app-paths')
const { log, warn } = require('../helpers/logger')

function ensureWWW (forced) {
  const www = appPaths.resolve.capacitor('www')

  forced === true && fse.removeSync(www)

  if (!fs.existsSync(www)) {
    fse.copySync(
      appPaths.resolve.cli('templates/capacitor/www'),
      appPaths.resolve.capacitor('www')
    )
  }
}

function ensureDeps () {
  if (fs.existsSync(appPaths.resolve.capacitor('node_modules'))) {
    return
  }

  const cmdParam = nodePackager === 'npm'
    ? ['install']
    : []

  log(`Installing Capacitor dependencies...`)
  spawnSync(
    nodePackager,
    cmdParam,
    { cwd: appPaths.capacitorDir, env: { ...process.env, NODE_ENV: 'development' } },
    () => warn(`[FAIL] failed installing dependencies in /src-capacitor`)
  )
}

module.exports = function () {
  ensureWWW()
  ensureDeps()
}

module.exports.ensureWWW = ensureWWW
module.exports.ensureDeps = ensureDeps


text:
package org.teachingkidsprogramming.section08events;

import org.teachingextensions.logo.Tortoise;
import org.teachingextensions.logo.utils.ColorUtils.PenColors;
import org.teachingextensions.logo.utils.EventUtils.MouseLeftClickListener;
import org.teachingextensions.logo.utils.EventUtils.MouseRightClickListener;
import org.teachingextensions.logo.utils.LineAndShapeUtils.Text;
import org.teachingkidsprogramming.recipes.quizzes.graders.SimpleBubbleQuizAdapter;
import org.teachingkidsprogramming.recipes.quizzes.graders.SimpleBubbleQuizGrader;

public class SimpleBubbleQuiz extends SimpleBubbleQuizAdapter
    implements
      MouseLeftClickListener,
      MouseRightClickListener
{
  @Override
  public void question1()
  {
    //code: In the Tortoise background window, have this quiz listen for when the left mouse button is clicked
    Tortoise.getBackgroundWindow().addMouseLeftClickListener(this);
    //action: YOU must left click on the first base to pass this question
  }
  @Override
  public void question2()
  {
    //code: Write "Single!" on the screen in yellow at position 155,135
    new Text("Single!").setTopLeft(155, 135).setPenColor(PenColors.Yellows.Yellow)
        .addTo(Tortoise.getBackgroundWindow());
  }
  @Override
  public void question3()
  {
    //code: In the Tortoise background window, have this quiz listen for when the right mouse button is clicked
    Tortoise.getBackgroundWindow().addMouseRightClickListener(this);
    //action: YOU must right click on the home plate (4th base)  to pass this question
  }
  @Override
  public void question4()
  {
    //code: Write "Home Run!" on the screen in lime green at position 105,235
    new Text("Home Run!").setTopLeft(105, 235).setPenColor(PenColors.Greens.LimeGreen)
        .addTo(Tortoise.getBackgroundWindow());
  }
  public static void main(String[] args)
  {
    new SimpleBubbleQuizGrader().grade(new SimpleBubbleQuiz());
  }
  @Override
  public void onLeftMouseClick(int x, int y)
  {
    if (counter < 3)
    {
      if (counter == 0)
      {
        this.answerQuestion1();
        question2();
        Text baseHit = new Text("Keep Running!");
        baseHit.setPenColor(PenColors.Oranges.Orange);
        baseHit.setTopLeft(105, 35).addTo(Tortoise.getBackgroundWindow());
      }
      drawNextBase();
      counter = counter + 1;
    }
  }
  @Override
  public void onRightMouseClick(int x, int y)
  {
    if (counter != 3) { return; }
    this.answerQuestion3();
    question4();
    drawNextBase();
  }
}


text:
'''
Microsoft Outlook Express (XP)

@author: Peter Parente <parente@cs.unc.edu>
@copyright: Copyright (c) 2008 Peter Parente
@license: BSD License

All rights reserved. This program and the accompanying materials are made
available under the terms of The BSD License which accompanies this
distribution, and is available at
U{http://www.opensource.org/licenses/bsd-license.php}
'''
import re
import UIA, Interface, Config
from UIA import Adapters
from View import Task, Control

# path reused through the script
save_attach_path = '/client[3]/window[5]/client[3]/window[1]/client[3]/window[0]/client[3]/window[1]/tool bar[3]/push button[5]'

class Main(UIA.Macro):
  # the macro that's run at the very start of the program
  def Sequence(self):
    # watch for outlook
    self.WatchForNewWindow(Name='Outlook Express',
                           ClassName='Outlook Express Browser Class')
    # run outlook
    self.RunFile('msimn.exe')
    yield False
    yield True

class DoResendMessage(UIA.Macro):
  def Sequence(self):
    # watch for message window
    self.WatchForNewWindow(ClassName='ATH_Note')
    # open waiting message
    self.SendKeys('^{o}')
    yield False
    # watch for window close
    self.WatchForWindowClose(self.result.Window)
    # send message
    self.SendKeys('%{s}')
    yield False
    yield True

class DoCancelMessageNoConfirm(UIA.Macro):
  def Sequence(self):
    # watch for the are you sure dialog
    self.WatchForNewWindow(Name='Outlook Express', ClassName='#32770',
                           name='confirm')
    # watch for the window closing
    self.WatchForWindowClose(self.model.Window, 'done')
    # close the window
    self.CloseWindow()
    yield False
    # close the confirmation dialog if it appears
    if self.name == 'confirm':
      self.SendKeys('N')
    yield True

class DoSelectSaveAttachments(UIA.Macro):
  def Sequence(self):
    # watch for the attachments dialog
    self.WatchForNewWindow(ClassName='#32770', Name='Save Attachments')
    # give the button the focus
    c = self.model.ChildFromPath(save_attach_path)
    c.SetFocus()
    c.SendKeys('{ENTER}{UP}{ENTER}')
    yield False
    yield True

class DoSelectTaskTab(UIA.Macro):
  tab_list_path = '/dialog[3]/window[14]/page tab list[3]'
  tab_path = '/dialog[3]/window[14]/page tab list[3]/page tab[0]'
  def Sequence(self):
    tab_list = self.model.ChildFromPath(self.tab_list_path)
    tab = self.model.ChildFromPath(self.tab_path)
    tab_list.Select(UIA.Constants.SELFLAG_TAKEFOCUS)
    tab.Select(UIA.Constants.SELFLAG_TAKEFOCUS)
    yield True

class ResendMessage(Task.RunAndReport):
  Name = 'resend message'
  StartMacro = DoResendMessage()

class AttachFiles(Task.FileOpening):
  Name = 'attach files'
  Modal = True
  StartMacro = UIA.StartWindowByKey(key_combo='%{i}a', ClassName='#32770',
                                    Name='Insert Attachment')
  CompleteMacro = UIA.EndWindowByKey(key_combo='{ENTER}')
  CancelMacro = UIA.EndWindowByButton()
  filename_path = '/dialog[3]/window[7]/editable text[3]'

class ChooseFolderStep(Task.FolderBrowsing):
  Name = 'choose destination folder'
  Modal = True
  filename_path = '/dialog[3]/window[6]/editable text[3]'

class ChooseAttachmentStep(Task.FormFill):
  Name = 'choose one attachment to save'
  Modal = True
  attachments_path = '/dialog[3]/window[4]/list[3]'

  def OnInit(self):
    l = Adapters.List(self, self.attachments_path)
    self.AddField(Control.List(self, l, name='', label='file'))

class SaveOneAttachment(Task.Wizard):
  Name = 'save one attachment'
  Modal = True
  StartMacro = DoSelectSaveAttachments()
  CancelMacro = UIA.EndWindowByButton()
  CompleteMacro = UIA.EndWindowByKey(key_combo='%{s}')

  def OnInit(self):
    self.AddStep(ChooseAttachmentStep)
    self.AddStep(ChooseFolderStep)

class SaveAllAttachments(SaveOneAttachment):
  Name = 'save all attachments'

  def OnInit(self):
    self.AddStep(ChooseFolderStep)

class MoveMail(Task.FormFill):
  Name = 'move message to mailbox'
  Modal = True
  folders_path = '/dialog[3]/window[1]/client[3]/window[0]/outline[3]'
  ok_path = '/dialog[3]/window[2]/push button[3]'
  StartMacro = UIA.StartWindowByKey(key_combo='^(+(v))', ClassName='#32770',
                                    Name='move')
  CancelMacro = UIA.EndWindowByButton()
  CompleteMacro = UIA.EndWindowByButton(button_path=ok_path)

  def OnInit(self):
    t = Adapters.Tree(self, self.folders_path)
    self.AddField(Control.Tree(self, t, 'mailboxes', 'mailbox'))

  def IsValid(self):
    ok = Interface.IContext(self).GetObjectAt(self.ok_path)
    if not (ok.State & UIA.Constants.STATE_SYSTEM_UNAVAILABLE):
      return None
    else:
      return 'The message cannot be moved to the selected folder.'

class WriteMail(Task.FormFill):
  Name = 'write mail'
  StartMacro = UIA.StartWindowByKey(ClassName='ATH_Note', key_combo='%f{ENTER 2}')
  CompleteMacro = UIA.EndWindowByKey(key_combo='%{s}')
  CancelMacro = DoCancelMessageNoConfirm()
  to_path = '/client[3]/window[2]/client[3]/window[3]/editable text[3]'
  cc_path = '/client[3]/window[2]/client[3]/window[4]/editable text[3]'
  subject_path = '/client[3]/window[2]/client[3]/window[5]/editable text[3]'
  attachments_path = '/client[3]/window[2]/client[3]/window[2]/list[3]'
  body_path = '/client[3]/window[3]/client[3]/window[0]/client[3]/client[0]'

  def OnInit(self):
    # recepient line
    t = Adapters.EditableTextBox(self, self.to_path, multiline=False)
    self.AddField(Control.TextEntry(self, t, 'recipient', spell=False))
    # carbon copies line
    t = Adapters.EditableTextBox(self, self.cc_path, multiline=False)
    self.AddField(Control.TextEntry(self, t, 'carbon copies',spell=False))
    # subject line
    t = Adapters.EditableTextBox(self, self.subject_path, multiline=False)
    self.AddField(Control.TextEntry(self, t, 'subject'))
    # attachments list
    l = Adapters.EditableList(self, self.attachments_path)
    self.AddField(Control.List(self, l, 'attachments'))
    # body field
    doc = Adapters.EditableDocument(self, self.body_path)
    self.AddField(Control.TextEntry(self, doc, 'message body'))

    # add all secondary task options
    self.AddContextOption(AttachFiles, True)

  def OnLoseFocus(self, message):
    text = self.fields[2].Model.GetAllText()
    if text:
      self.Name = 'continue new message with subject, %s' % text
    else:
      self.Name = 'continue new message'

  def IsValid(self):
    #@todo: not checking email validity
    return None
    regex = '^([^@;,]+@[^@;, ]+(;|,)?\W*)+$'
    # make sure there's at least one valid email address in the to field
    to = self.fields[0].Model.GetAllText()
    if re.match(regex, to) is None:
      return 'Invalid email address as recipient.'
    # make sure all email addresses in the cc field are valid
    cc = self.fields[1].Model.GetAllText()
    if cc != '' and re.match(regex, cc) is None:
      return 'Invalid email address in carbon copies.'
    return None

class ReplyMail(WriteMail):
  Name = 'reply to sender'
  StartMacro = UIA.StartWindowByKey(key_combo='^{r}', ClassName='ATH_Note')

  def OnInit(self):
    # replies have no access to the to field, and start in the message body
    t = Adapters.EditableTextBox(self, self.cc_path, multiline=False)
    self.AddField(Control.TextEntry(self, t, 'carbon copies',spell=False))
    t = Adapters.EditableTextBox(self, self.subject_path, multiline=False)
    self.AddField(Control.TextEntry(self, t, 'subject'))
    l = Adapters.EditableList(self, self.attachments_path)
    self.AddField(Control.List(self, l, 'attachments'))
    doc = Adapters.EditableDocument(self, self.body_path)
    self.AddField(Control.TextEntry(self, doc, 'message body'))
    self.curr = 3
    # add all secondary task options
    self.AddContextOption(AttachFiles, True)

  def IsValid(self):
    # replies are always valid
    return None

  def OnLoseFocus(self, message):
    text = self.fields[1].Model.GetAllText()
    if text:
      self.Name = 'continue reply with subject, %s' % text
    else:
      self.Name = 'continue reply'

class ForwardMail(WriteMail):
  Name = 'forward message'
  StartMacro=UIA.StartWindowByKey(key_combo='^{f}', ClassName='ATH_Note')

  def OnLoseFocus(self, message):
    text = self.fields[2].Model.GetAllText()
    if text:
      self.Name = 'continue forward with subject, %s' % text
    else:
      self.Name = 'continue forward'

class HighlightReading(Control.TextReading):
  start = 0
  def OnRemember(self, message):
    mailbox = self.parent.views[1].Model
    if message.Press:
      Config.log(mailbox.GetSelectedName())
      self.start = self.Model.Chunk.GetCurrentWordBounds()[0]
      if self.start < 0:
        self.start = 0
    else:
      end = self.Model.Chunk.GetCurrentWordBounds()[1]
      Config.log('%d %d' % (self.start, end))
    message.Stop = False

class BrowseMail(Task.LinkedBrowsing):
  Name = 'browse mail'
  Permanence = Task.NO_START|Task.NO_END
  reply_path = '/client[3]/window[1]/client[3]/window[0]/client[3]/window[0]/tool bar[3]/push button[1]'
  forward_path = '/client[3]/window[1]/client[3]/window[0]/client[3]/window[0]/tool bar[3]/push button[2]'
  mailboxes_path = '/client[3]/window[4]/client[3]/window[0]/client[3]/window[1]/client[3]/window[0]/outline[3]'
  messages_path = '/client[3]/window[5]/client[3]/window[0]/client[3]/window[0]/list[3]'
  preview_path = '/client[3]/window[5]/client[3]/window[1]/client[3]/window[0]/client[3]/client[0]/pane[0]'

  def OnInit(self):
    # create models
    cl = Adapters.EditableColumnList(self, self.messages_path, 'Person',
                                     ('Person', 'Subject', 'Received'))
    tr = Adapters.Tree(self, self.mailboxes_path)
    ht = Adapters.SimpleDocument(self, self.preview_path)

    # create and add views in order of influence
    folders = Control.Tree(self, tr, 'mailboxes', label='sibling mailbox')
    messages = Control.List(self, cl, 'messages', label='message',
                            order=['Received', 'Sent'])
    body = HighlightReading(self, ht, 'email body', spell=False)
    self.AddView(folders)
    self.AddView(messages)
    self.AddView(body)

    # define a condition for reply and forward tasks
    def ReplyAndForwardCondition():
      c = Interface.IContext(self).GetObjectAt(self.reply_path)
      try: return not c.IsNotReady()
      except: return False

    # a condition for saving attachments
    def SaveAttachmentsCondition():
      c = Interface.IContext(self).GetObjectAt(save_attach_path)
      try: return not c.IsNotReady()
      except: return False

    # and a condition for resending failed messages
    def ResendMessageCondition():
      name = Interface.IFiniteCollection(self.views[0].Model).GetSelectedName()
      return name == 'Outbox'

    # add secondary tasks to the appropriate controls
    messages.AddContextOption(ReplyMail, True, ReplyAndForwardCondition)
    messages.AddContextOption(ForwardMail, True, ReplyAndForwardCondition)
    messages.AddContextOption(SaveAllAttachments, True, SaveAttachmentsCondition)
    messages.AddContextOption(SaveOneAttachment, True, SaveAttachmentsCondition)
    messages.AddContextOption(ResendMessage, True, ResendMessageCondition)
    messages.AddContextOption(MoveMail, True)
    body.AddContextOption(ReplyMail, True, ReplyAndForwardCondition)
    body.AddContextOption(ForwardMail, True, ReplyAndForwardCondition)
    body.AddContextOption(SaveAllAttachments, True, SaveAttachmentsCondition)
    body.AddContextOption(SaveOneAttachment, True, SaveAttachmentsCondition)
    body.AddContextOption(ResendMessage, True, ResendMessageCondition)
    body.AddContextOption(MoveMail, True)

class AutoConfirmNoSubject(Task.FormFill):
  Modal = True
  Name = 'confirm no subject'
  Permanence = Task.NO_CANCEL
  CompleteMacro = UIA.EndWindowByKey(key_combo='{ENTER}')
  ok_path = '/dialog[3]/window[0]/push button[3]'
  cancel_path = '/dialog[3]/window[1]/push button[3]'
  text_path = '/dialog[3]/window[5]/text[3]'
  check_path = '/dialog[3]/window[3]/check box[3]'

  def OnInit(self):
    bl = Adapters.ButtonList(self, [self.ok_path, self.cancel_path])
    label = Adapters.TextLabel(self, self.text_path)
    self.AddField(Control.List(self, bl, label))
    cb = Adapters.CheckBox(self, self.check_path)
    self.AddField(Control.List(self, cb))

  def Trigger(cls, event, model, container):
    return not (model.State & UIA.Constants.STATE_SYSTEM_SIZEABLE) and \
           model.Name == 'Outlook Express' and model.ClassName == '#32770' and \
           isinstance(container.GetTask(-1), WriteMail)
  Trigger = classmethod(Trigger)

class AutoConfirmDeleteFromTrash(Task.FormFill):
  Modal = True
  Name = 'confirm deletion'
  Permanence = Task.NO_CANCEL
  CompleteMacro = UIA.EndWindowByKey(key_combo='{ENTER}')
  ok_path = '/dialog[3]/window[0]/push button[3]'
  cancel_path = '/dialog[3]/window[1]/push button[3]'
  delete_msg = 'Are you sure you want to permanently delete this message?'

  def OnInit(self):
    bl = Adapters.ButtonList(self, [self.cancel_path, self.ok_path])
    self.AddField(Control.List(self, bl, self.delete_msg))

  def Trigger(cls, event, model, container):
    return not (model.State & UIA.Constants.STATE_SYSTEM_SIZEABLE) and \
           model.Name == 'Outlook Express' and model.ClassName == '#32770' and \
           model.GetChild(3).ChildCount == 4
  Trigger = classmethod(Trigger)

class AutoSendError(Task.FormFill):
  Modal = True
  Name = 'send error'
  Permanence = Task.NO_CANCEL
  CompleteMacro = UIA.HideWindowByKey(key_combo='%h')
  StartMacro = DoSelectTaskTab
  list_path = '/dialog[3]/window[12]/list[3]'

  def OnInit(self):
    cl = Adapters.ColumnList(self, self.list_path, name_key='Action',
                             primary_keys=['Action', 'Status'])
    self.AddField(Control.List(self, cl, 'error list'))

  def Trigger(cls, event, model, container):
    return not (model.State & UIA.Constants.STATE_SYSTEM_SIZEABLE) and \
           model.Name == 'Outlook Express' and model.ClassName == '#32770' and \
           model.GetChild(3).ChildCount == 17
  Trigger = classmethod(Trigger)

Tasks = [BrowseMail, WriteMail]
AutoTasks = [AutoConfirmNoSubject, AutoConfirmDeleteFromTrash, AutoSendError]


text:
echo 'This will take a lot of time...'
docker build -t bo-su/docker-intellij .


text:
// ***********************************************************************
// Assembly         : FenrirFS
// Component        : FenrirDirectory.cs
// Author           : vonderborch
// Created          : 09-22-2016
//
// Version          : 2.0.0
// Last Modified By : vonderborch
// Last Modified On : 05-03-2017
// ***********************************************************************
// <copyright file="FenrirDirectory.cs">
//		Copyright ©  2016
// </copyright>
// <summary>
//      Defines the FenrirDirectory class, an implementation of a the FSDirectory class.
// </summary>
//
// Changelog:
///           - 2.0.0 (05-03-2017) - 2.0.0 release.
///           - 2.0.0 (05-03-2017) - Beta version. Using FS.ParseXXXPath to generate paths. Fixed issue with renaming.
//            - 2.0.0 (01-01-2017) - Beta version. Fixed issue creating directories/files and specifying OpenIfExists.
//            - 2.0.0 (09-24-2016) - Beta version.
// ***********************************************************************
using FenrirFS.Helpers;
using System;
using System.Collections.Generic;
using System.Text.RegularExpressions;
using IO = System.IO;

namespace FenrirFS
{
    /// <summary>
    /// Class FenrirDirectory.
    /// </summary>
    /// <seealso cref="FenrirFS.FSDirectory" />
    public class FenrirDirectory : FSDirectory
    {
        #region Public Constructors

        /// <summary>
        /// Initializes a new instance of the <see cref="FenrirDirectory"/> class.
        /// </summary>
        /// <param name="path">The path.</param>
        public FenrirDirectory(string path) : base(path)
        {
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="FenrirDirectory"/> class.
        /// </summary>
        /// <param name="path">The path.</param>
        /// <param name="name">The name.</param>
        public FenrirDirectory(string path, string name) : base(path, name)
        {
        }

        #endregion Public Constructors

        #region Public Properties

        /// <summary>
        /// Gets a value indicating whether [directory exists].
        /// </summary>
        /// <value><c>true</c> if [directory exists]; otherwise, <c>false</c>.</value>
        public override bool Exists
        {
            get { return IO.Directory.Exists(this); }
        }

        #endregion Public Properties

        #region Public Methods

        /// <summary>
        /// Copies the directory to the specified destination.
        /// </summary>
        /// <param name="destination">The destination.</param>
        /// <param name="collisionOption">The collision option to use if a collision occurs.</param>
        /// <returns><c>true</c> if the copy succeeds, <c>false</c> otherwise.</returns>
        ///  Changelog:
        ///             - 2.0.0 (09-24-2016) - Beta Version.
        public override FSDirectory Copy(string destination, DirectoryCollisionOption collisionOption = DirectoryCollisionOption.FailIfExists)
        {
            if (Exists)
            {
                Validation.NotNullOrWhiteSpaceCheck(destination, nameof(destination));

                switch (collisionOption)
                {
                    case DirectoryCollisionOption.OpenIfExists:
                        return FS.GetDirectory(destination, OpenMode.ThrowIfDoesNotExist);

                    case DirectoryCollisionOption.FailIfExists:
                        return null;

                    case DirectoryCollisionOption.GenerateUniqueName:
                        destination = IOHelper.GenerateUniquePath(destination, false);
                        break;

                    case DirectoryCollisionOption.ReplaceExisting:
                        if (IO.Directory.Exists(destination))
                            IO.Directory.Delete(destination, true);
                        break;

                    case DirectoryCollisionOption.ThrowIfExists:
                        throw new IO.IOException($"A folder with the new path [{destination}] already exists!");
                }

                IO.Directory.CreateDirectory(destination);

                var files = GetFiles();
                for (int i = 0; i < files.Count; i++)
                {
                    try
                    {
                        var newPath = IO.Path.Combine(destination, $"{files[i].Name}{files[i].Extension}");
                        files[i].Copy(newPath, FileCollisionOption.FailIfExists);
                    }
                    catch (Exception ex)
                    {
                        throw new Exception("Exception while copying a sub-directory!", ex);
                    }
                }

                var directories = GetDirectories();
                for (int i = 0; i < directories.Count; i++)
                {
                    try
                    {
                        var newPath = IO.Path.Combine(destination, directories[i].Name);
                        directories[i].Copy(newPath, DirectoryCollisionOption.FailIfExists);
                    }
                    catch (Exception ex)
                    {
                        throw new Exception("Exception while copying a sub-directory!", ex);
                    }
                }

                return FS.GetDirectory(destination, OpenMode.ThrowIfDoesNotExist);
            }

            return null;
        }

        /// <summary>
        /// Creates a new file in the directory.
        /// </summary>
        /// <param name="file">The name (and extension) for the file.</param>
        /// <param name="collisionOption">The collision option to use if a collision occurs.</param>
        /// <returns>A file structure representing the new file.</returns>
        ///  Changelog:
        ///             - 2.0.0 (05-03-2017) - Beta version. Using FS.ParseXXXPath to generate paths.
        ///             - 2.0.0 (01-01-2017) - Beta 3 Version. Fixed a bug that would throw an error if OpenIfExists was specified and the file did not exist.
        ///             - 2.0.0 (09-24-2016) - Beta Version.
        public override FSFile CreateFile(string file, FileCollisionOption collisionOption = FileCollisionOption.FailIfExists)
        {
            Validation.NotNullOrWhiteSpaceCheck(file, nameof(file));

            if (!Exists)
                IO.Directory.CreateDirectory(this);

            string newFullPath = FS.ParseFilePath(this, file);
            switch (collisionOption)
            {
                case FileCollisionOption.OpenIfExists:
                    return FS.GetFile(newFullPath, OpenMode.CreateIfDoesNotExist);

                case FileCollisionOption.FailIfExists:
                    return null;

                case FileCollisionOption.GenerateUniqueName:
                    newFullPath = IOHelper.GenerateUniquePath(newFullPath, true);
                    break;

                case FileCollisionOption.ReplaceExisting:
                    if (IO.File.Exists(newFullPath))
                        IO.File.Delete(newFullPath);
                    break;

                case FileCollisionOption.ThrowIfExists:
                    throw new IO.IOException($"A file with the new path [{newFullPath}] already exists!");
            }

            return FS.GetFile(newFullPath, OpenMode.CreateIfDoesNotExist);
        }

        /// <summary>
        /// Creates a new directory in the directory.
        /// </summary>
        /// <param name="directory">The name for the directory.</param>
        /// <param name="collisionOption">The collision option to use if a collision occurs.</param>
        /// <returns>A directory structure representing the new directory.</returns>
        ///  Changelog:
        ///             - 2.0.0 (05-03-2017) - Beta version. Using FS.ParseXXXPath to generate paths.
        ///             - 2.0.0 (01-01-2017) - Beta 3 Version. Fixed a bug that would throw an error if OpenIfExists was specified and the directory did not exist.
        ///             - 2.0.0 (09-24-2016) - Beta Version.
        public override FSDirectory CreateDirectory(string directory, DirectoryCollisionOption collisionOption = DirectoryCollisionOption.FailIfExists)
        {
            Validation.NotNullOrWhiteSpaceCheck(directory, nameof(directory));

            if (!Exists)
                IO.Directory.CreateDirectory(this);

            string newFullPath = FS.ParseDirectoryPath(this, directory);
            switch (collisionOption)
            {
                case DirectoryCollisionOption.OpenIfExists:
                    return FS.GetDirectory(newFullPath, OpenMode.CreateIfDoesNotExist);

                case DirectoryCollisionOption.FailIfExists:
                    return null;

                case DirectoryCollisionOption.GenerateUniqueName:
                    newFullPath = IOHelper.GenerateUniquePath(newFullPath, false);
                    break;

                case DirectoryCollisionOption.ReplaceExisting:
                    if (IO.Directory.Exists(newFullPath))
                        IO.Directory.Delete(newFullPath);
                    break;

                case DirectoryCollisionOption.ThrowIfExists:
                    throw new IO.IOException($"A folder with the new path [{newFullPath}] already exists!");
            }

            return FS.GetDirectory(newFullPath, OpenMode.CreateIfDoesNotExist);
        }

        /// <summary>
        /// Deletes the directory.
        /// </summary>
        /// <returns><c>true</c> if the directory was deleted, <c>false</c> otherwise.</returns>
        ///  Changelog:
        ///             - 2.0.0 (09-24-2016) - Beta Version.
        public override bool Delete()
        {
            if (Exists)
            {
                IO.Directory.Delete(this);
                return true;
            }
            return false;
        }

        /// <summary>
        /// Deletes a file in the directory.
        /// </summary>
        /// <param name="file">The name of the file to delete.</param>
        /// <returns><c>true</c> if the file was deleted, <c>false</c> otherwise.</returns>
        ///  Changelog:
        ///             - 2.0.0 (09-24-2016) - Beta Version.
        public override bool DeleteFile(string file)
        {
            Validation.NotNullOrWhiteSpaceCheck(file, nameof(file));
            if (!Exists) return false;

            if (IO.File.Exists(file))
            {
                IO.File.Delete(file);
                return true;
            }
            return false;
        }

        /// <summary>
        /// Deletes a directory in the directory.
        /// </summary>
        /// <param name="directory">The name of the directory to delete.</param>
        /// <returns><c>true</c> if the directory was deleted, <c>false</c> otherwise.</returns>
        ///  Changelog:
        ///             - 2.0.0 (09-24-2016) - Beta Version.
        public override bool DeleteDirectory(string directory)
        {
            Validation.NotNullOrWhiteSpaceCheck(directory, nameof(directory));
            if (!Exists) return false;

            if (IO.Directory.Exists(directory))
            {
                IO.Directory.Delete(directory);
                return true;
            }
            return false;
        }

        /// <summary>
        /// Gets the creation time of the directory.
        /// </summary>
        /// <param name="useUtc">if set to <c>true</c> [use UTC].</param>
        /// <returns>The creation time.</returns>
        ///  Changelog:
        ///             - 2.0.0 (09-24-2016) - Beta Version.
        public override DateTime GetCreationTime(bool useUtc = false)
        {
            return !Exists
                ? DateTime.MinValue
                : useUtc
                    ? IO.Directory.GetCreationTimeUtc(this)
                    : IO.Directory.GetCreationTime(this);
        }

        /// <summary>
        /// Gets the last accessed time of the directory.
        /// </summary>
        /// <param name="useUtc">if set to <c>true</c> [use UTC].</param>
        /// <returns>The last accessed time.</returns>
        ///  Changelog:
        ///             - 2.0.0 (09-24-2016) - Beta Version.
        public override DateTime GetLastAccessedTime(bool useUtc = false)
        {
            return !Exists
                ? DateTime.MinValue
                : useUtc
                    ? IO.Directory.GetLastAccessTimeUtc(this)
                    : IO.Directory.GetLastAccessTime(this);
        }

        /// <summary>
        /// Gets the last modified time of the directory.
        /// </summary>
        /// <param name="useUtc">if set to <c>true</c> [use UTC].</param>
        /// <returns>The last modified time.</returns>
        ///  Changelog:
        ///             - 2.0.0 (09-24-2016) - Beta Version.
        public override DateTime GetLastModifiedTime(bool useUtc = false)
        {
            return !Exists
                ? DateTime.MinValue
                : useUtc
                    ? IO.Directory.GetLastWriteTimeUtc(this)
                    : IO.Directory.GetLastWriteTime(this);
        }

        /// <summary>
        /// Moves the directory to the specified specified destination.
        /// </summary>
        /// <param name="destination">The destination.</param>
        /// <param name="collisionOption">The collision option to use if a collision occurs.</param>
        /// <returns><c>true</c> if the move succeeds, <c>false</c> otherwise.</returns>
        ///  Changelog:
        ///             - 2.0.0 (09-24-2016) - Beta Version.
        public override bool Move(string destination, DirectoryCollisionOption collisionOption = DirectoryCollisionOption.FailIfExists)
        {
            if (Exists)
            {
                Validation.NotNullOrWhiteSpaceCheck(destination, nameof(destination));

                switch (collisionOption)
                {
                    case DirectoryCollisionOption.OpenIfExists:
                    case DirectoryCollisionOption.FailIfExists:
                        return false;

                    case DirectoryCollisionOption.GenerateUniqueName:
                        destination = IOHelper.GenerateUniquePath(destination, false);
                        break;

                    case DirectoryCollisionOption.ReplaceExisting:
                        if (IO.Directory.Exists(destination))
                            IO.Directory.Delete(destination);
                        break;

                    case DirectoryCollisionOption.ThrowIfExists:
                        throw new IO.IOException($"A folder with the new path [{destination}] already exists!");
                }

                IO.Directory.Move(this, destination);
                FullPath = destination;
                return true;
            }

            return false;
        }

        /// <summary>
        /// Renames the directory to the specified name.
        /// </summary>
        /// <param name="name">The name.</param>
        /// <param name="collisionOption">The collision option if a file with the new name already exists.</param>
        /// <returns><c>true</c> if the rename was successful, <c>false</c> otherwise.</returns>
        ///  Changelog:
        ///             - 2.0.0 (05-03-2017) - Beta version. Using FS.ParseXXXPath to generate paths. Fixed bug where folder would always move to a subdirectory.
        ///             - 2.0.0 (09-24-2016) - Beta Version.
        public override bool Rename(string name, DirectoryCollisionOption collisionOption = DirectoryCollisionOption.FailIfExists)
        {
            if (Exists)
            {
                Validation.NotNullOrWhiteSpaceCheck(name, nameof(name));

                string newFullPath = FS.ParseDirectoryPath(Path, name);
                switch (collisionOption)
                {
                    case DirectoryCollisionOption.OpenIfExists:
                    case DirectoryCollisionOption.FailIfExists:
                        return false;

                    case DirectoryCollisionOption.GenerateUniqueName:
                        newFullPath = IOHelper.GenerateUniquePath(newFullPath, false);
                        break;

                    case DirectoryCollisionOption.ReplaceExisting:
                        if (IO.Directory.Exists(newFullPath))
                            IO.Directory.Delete(newFullPath);
                        break;

                    case DirectoryCollisionOption.ThrowIfExists:
                        throw new IO.IOException($"A folder with the new path [{newFullPath}] already exists!");
                }

                IO.Directory.Move(this, newFullPath);
                FullPath = newFullPath;
                return true;
            }

            return false;
        }

        #endregion Public Methods

        #region Protected Methods

        /// <summary>
        /// Internal method to get file system entries.
        /// </summary>
        /// <param name="grabFiles">if set to <c>true</c> [grabs files].</param>
        /// <param name="grabDirectories">if set to <c>true</c> [grabs directories].</param>
        /// <param name="searchPattern">The search pattern.</param>
        /// <param name="searchOption">The search option.</param>
        /// <returns>A list of file system entries matching the desired parameters.</returns>
        ///  Changelog:
        ///             - 2.0.0 (09-24-2016) - Beta version.
        protected override List<FSFileSystemEntry> InternalGetFileSystemEntries(bool grabFiles, bool grabDirectories, string searchPattern = "*", SearchOption searchOption = SearchOption.All)
        {
            var filesAndFolders = new List<FSFileSystemEntry>();

            if (Exists)
            {
                // search top directory...
                if (searchOption != SearchOption.SubDirectoriesOnly)
                {
                    if (grabFiles)
                    {
                        foreach (var file in IO.Directory.EnumerateFiles(this))
                        {
                            if (Regex.IsMatch(file, searchPattern))
                                filesAndFolders.Add(new FenrirFile(file));
                        }
                    }
                }

                // return early if we don't need to do anything with any directories...
                if (searchOption == SearchOption.TopDirectoryOnly && !grabDirectories)
                    return filesAndFolders;

                foreach (var folder in IO.Directory.EnumerateDirectories(this))
                {
                    var newFolder = new FenrirDirectory(folder);
                    if (grabDirectories && Regex.IsMatch(folder, searchPattern))
                        filesAndFolders.Add(newFolder);

                    if (searchOption != SearchOption.TopDirectoryOnly)
                    {
                        var result = newFolder.InternalGetFileSystemEntries(grabFiles, grabDirectories, searchPattern, SearchOption.All);

                        for (int i = 0; i < result.Count; i++)
                            filesAndFolders.Add(result[i]);
                    }
                }
            }

            return filesAndFolders;
        }

        #endregion Protected Methods
    }
}