<commit_before>#    Copyright 2013 Mirantis, Inc.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.


import logging
import unittest
from devops.helpers.helpers import wait
from fuelweb_test.helpers import Ebtables
from fuelweb_test.integration.base_node_test_case import BaseNodeTestCase
from fuelweb_test.integration.decorators import snapshot_errors, \
    debug, fetch_logs
from fuelweb_test.settings import EMPTY_SNAPSHOT

logging.basicConfig(
    format=':%(lineno)d: %(asctime)s %(message)s',
    level=logging.DEBUG
)

logger = logging.getLogger(__name__)
logwrap = debug(logger)


class TestNode(BaseNodeTestCase):
    @logwrap
    @fetch_logs
    def test_release_upload(self):
        self.prepare_environment()
        self._upload_sample_release()

    @logwrap
    @fetch_logs
    def test_http_returns_no_error(self):
        self.prepare_environment()
        self.client.get_root()

    @logwrap
    @fetch_logs
    def test_create_empty_cluster(self):
        self.prepare_environment()
        self.create_cluster(name='empty')

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_node_deploy(self):
        self.prepare_environment()
        self.bootstrap_nodes(self.nodes().slaves[:1])

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_updating_nodes_in_cluster(self):
        self.prepare_environment()
        cluster_id = self.create_cluster(name='empty')
        nodes = {'slave-01': ['controller']}
        self.bootstrap_nodes(self.nodes().slaves[:1])
        self.update_nodes(cluster_id, nodes)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_one_node_provisioning(self):
        self.prepare_environment()
        cluster_id = self.create_cluster(name="provision")
        self._basic_provisioning(
            cluster_id=cluster_id,
            nodes_dict={'controller': ['slave-01']}
        )
        self.run_OSTF(cluster_id=cluster_id, should_fail=12, should_pass=12)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_simple_cluster_flat(self):
        cluster_id = self.prepare_environment(settings={
            'nodes': {
                'slave-01': ['controller'],
                'slave-02': ['compute']
            }
        })
        self.assertClusterReady(
            'slave-01', smiles_count=6, networks_count=1, timeout=300)
        self.get_ebtables(cluster_id, self.nodes().slaves[:2]).restore_vlans()
        task = self._run_network_verify(cluster_id)
        self.assertTaskSuccess(task, 60 * 2)
        self.run_OSTF(cluster_id=cluster_id, should_fail=6, should_pass=18)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_simple_cluster_vlan(self):
        self.prepare_environment()
        cluster_name = 'simple_vlan'
        nodes = {'slave-01': ['controller'], 'slave-02': ['compute']}
        cluster_id = self.create_cluster(name=cluster_name)
        self.update_vlan_network_fixed(cluster_id, amount=8, network_size=32)
        self._basic_provisioning(cluster_id, nodes)
        self.assertClusterReady(
            'slave-01', smiles_count=6, networks_count=8, timeout=300)
        self.get_ebtables(cluster_id, self.nodes().slaves[:2]).restore_vlans()
        task = self._run_network_verify(cluster_id)
        self.assertTaskSuccess(task, 60 * 2)
        self.run_OSTF(cluster_id=cluster_id, should_fail=6, should_pass=18)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_network_config(self):
        cluster_id = self.prepare_environment(settings={
            'nodes': {
                'slave-01': ['controller'],
                'slave-02': ['compute']
            }
        })
        slave = self.nodes().slaves[0]
        node = self.get_node_by_devops_node(slave)
        self.assertNetworkConfiguration(node)
        self.run_OSTF(cluster_id=cluster_id, should_fail=6, should_pass=18)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_node_deletion(self):
        cluster_id = self.prepare_environment(settings={
            'nodes': {
                'slave-01': ['controller'],
                'slave-02': ['compute']
            }
        })
        nailgun_nodes = self.update_nodes(
            cluster_id, {'slave-01': ['controller']}, False, True)
        task = self.deploy_cluster(cluster_id)
        self.assertTaskSuccess(task)
        nodes = filter(lambda x: x["pending_deletion"] is True, nailgun_nodes)
        self.assertTrue(
            len(nodes) == 1,
            "Verify 1 node has pending deletion status"
        )
        wait(lambda: self.is_node_discovered(
            nodes[0]),
            timeout=3 * 60
        )
        self.run_OSTF(cluster_id=cluster_id, should_fail=24, should_pass=0)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_network_verify_with_blocked_vlan(self):
        self.prepare_environment()
        cluster_name = 'net_verify'
        nodes_dict = {'slave-01': ['controller'], 'slave-02': ['compute']}

        cluster_id = self.create_cluster(name=cluster_name)
        devops_nodes = self.nodes().slaves[:1]
        self.bootstrap_nodes(devops_nodes)

        ebtables = self.get_ebtables(cluster_id, devops_nodes)
        ebtables.restore_vlans()
        self.update_nodes(cluster_id, nodes_dict)
        try:
            ebtables.block_first_vlan()
            task = self._run_network_verify(cluster_id)
            self.assertTaskFailed(task, 60 * 2)
        finally:
            ebtables.restore_first_vlan()

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_multinic_bootstrap_booting(self):
        self.prepare_environment()
        slave = self.nodes().slaves[0]
        mac_addresses = [interface.mac_address for interface in
                         slave.interfaces.filter(network__name='internal')]
        try:
            for mac in mac_addresses:
                Ebtables.block_mac(mac)
            for mac in mac_addresses:
                Ebtables.restore_mac(mac)
                slave.destroy(verbose=False)
                self.nodes().admins[0].revert(EMPTY_SNAPSHOT)
                nailgun_slave = self.bootstrap_nodes([slave])[0]
                self.assertEqual(mac.upper(), nailgun_slave['mac'].upper())
                Ebtables.block_mac(mac)<commit_msg>Fix bugs<commit_after>#    Copyright 2013 Mirantis, Inc.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.


import logging
import unittest
from devops.helpers.helpers import wait
from fuelweb_test.helpers import Ebtables
from fuelweb_test.integration.base_node_test_case import BaseNodeTestCase
from fuelweb_test.integration.decorators import snapshot_errors, \
    debug, fetch_logs
from fuelweb_test.settings import EMPTY_SNAPSHOT

logging.basicConfig(
    format=':%(lineno)d: %(asctime)s %(message)s',
    level=logging.DEBUG
)

logger = logging.getLogger(__name__)
logwrap = debug(logger)


class TestNode(BaseNodeTestCase):
    @logwrap
    @fetch_logs
    def test_release_upload(self):
        self.prepare_environment()
        self._upload_sample_release()

    @logwrap
    @fetch_logs
    def test_http_returns_no_error(self):
        self.prepare_environment()
        self.client.get_root()

    @logwrap
    @fetch_logs
    def test_create_empty_cluster(self):
        self.prepare_environment()
        self.create_cluster(name='empty')

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_node_deploy(self):
        self.prepare_environment()
        self.bootstrap_nodes(self.nodes().slaves[:1])

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_updating_nodes_in_cluster(self):
        self.prepare_environment()
        cluster_id = self.create_cluster(name='empty')
        nodes = {'slave-01': ['controller']}
        self.bootstrap_nodes(self.nodes().slaves[:1])
        self.update_nodes(cluster_id, nodes)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_one_node_provisioning(self):
        self.prepare_environment()
        cluster_id = self.create_cluster(name="provision")
        self._basic_provisioning(
            cluster_id=cluster_id,
            nodes_dict={'slave-01': ['controller']}
        )
        self.run_OSTF(cluster_id=cluster_id, should_fail=12, should_pass=12)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_simple_cluster_flat(self):
        cluster_id = self.prepare_environment(settings={
            'nodes': {
                'slave-01': ['controller'],
                'slave-02': ['compute']
            }
        })
        self.assertClusterReady(
            'slave-01', smiles_count=6, networks_count=1, timeout=300)
        self.get_ebtables(cluster_id, self.nodes().slaves[:2]).restore_vlans()
        task = self._run_network_verify(cluster_id)
        self.assertTaskSuccess(task, 60 * 2)
        self.run_OSTF(cluster_id=cluster_id, should_fail=6, should_pass=18)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_simple_cluster_vlan(self):
        self.prepare_environment()
        cluster_name = 'simple_vlan'
        nodes = {'slave-01': ['controller'], 'slave-02': ['compute']}
        cluster_id = self.create_cluster(name=cluster_name)
        self.update_vlan_network_fixed(cluster_id, amount=8, network_size=32)
        self._basic_provisioning(cluster_id, nodes)
        self.assertClusterReady(
            'slave-01', smiles_count=6, networks_count=8, timeout=300)
        self.get_ebtables(cluster_id, self.nodes().slaves[:2]).restore_vlans()
        task = self._run_network_verify(cluster_id)
        self.assertTaskSuccess(task, 60 * 2)
        self.run_OSTF(cluster_id=cluster_id, should_fail=6, should_pass=18)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_network_config(self):
        cluster_id = self.prepare_environment(settings={
            'nodes': {
                'slave-01': ['controller'],
                'slave-02': ['compute']
            }
        })
        slave = self.nodes().slaves[0]
        node = self.get_node_by_devops_node(slave)
        self.assertNetworkConfiguration(node)
        self.run_OSTF(cluster_id=cluster_id, should_fail=6, should_pass=18)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_node_deletion(self):
        cluster_id = self.prepare_environment(settings={
            'nodes': {
                'slave-01': ['controller'],
                'slave-02': ['compute']
            }
        })
        nailgun_nodes = self.update_nodes(
            cluster_id, {'slave-01': ['controller']}, False, True)
        task = self.deploy_cluster(cluster_id)
        self.assertTaskSuccess(task)
        nodes = filter(lambda x: x["pending_deletion"] is True, nailgun_nodes)
        self.assertTrue(
            len(nodes) == 1,
            "Verify 1 node has pending deletion status"
        )
        wait(lambda: self.is_node_discovered(
            nodes[0]),
            timeout=3 * 60
        )
        self.run_OSTF(cluster_id=cluster_id, should_fail=24, should_pass=0)

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_network_verify_with_blocked_vlan(self):
        self.prepare_environment()
        cluster_name = 'net_verify'
        nodes_dict = {'slave-01': ['controller'], 'slave-02': ['compute']}

        cluster_id = self.create_cluster(name=cluster_name)
        devops_nodes = self.nodes().slaves[:2]
        self.bootstrap_nodes(devops_nodes)

        ebtables = self.get_ebtables(cluster_id, devops_nodes)
        ebtables.restore_vlans()
        self.update_nodes(cluster_id, nodes_dict)
        try:
            ebtables.block_first_vlan()
            task = self._run_network_verify(cluster_id)
            self.assertTaskFailed(task, 60 * 2)
        finally:
            ebtables.restore_first_vlan()

    @snapshot_errors
    @logwrap
    @fetch_logs
    def test_multinic_bootstrap_booting(self):
        self.prepare_environment()
        slave = self.nodes().slaves[0]
        mac_addresses = [interface.mac_address for interface in
                         slave.interfaces.filter(network__name='internal')]
        try:
            for mac in mac_addresses:
                Ebtables.block_mac(mac)
            for mac in mac_addresses:
                Ebtables.restore_mac(mac)
                slave.destroy(verbose=False)
                self.nodes().admins[0].revert(EMPTY_SNAPSHOT)
                nailgun_slave = self.bootstrap_nodes([slave])[0]
                self.assertEqual(mac.upper(), nailgun_slave['mac'].upper())
                Ebtables.block_mac(mac)
{'fraction_non_alphanumeric': 0.10260959562356768, 'fraction_numerical': 0.013454572336807866, 'mean_word_length': 2.706301369863014, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4025349', 'n_tokens_mistral': 4178, 'n_tokens_neox': 4060, 'n_words': 736}