263 lines
9.7 KiB
Python
263 lines
9.7 KiB
Python
# Copyright 2018 AT&T Intellectual Property. All other rights reserved.
|
|
#
|
|
# 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.
|
|
"""Tests for the default node_lookup provided with the deployment group
|
|
functionality.
|
|
"""
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from shipyard_airflow.common.deployment_group.deployment_group import (
|
|
GroupNodeSelector
|
|
)
|
|
from shipyard_airflow.common.deployment_group.errors import (
|
|
InvalidDeploymentGroupNodeLookupError
|
|
)
|
|
from shipyard_airflow.common.deployment_group.node_lookup import (
|
|
NodeLookup, _generate_node_filter, _validate_selectors
|
|
)
|
|
from drydock_provisioner import error as errors
|
|
|
|
|
|
class TestNodeLookup:
|
|
def test_validate_selectors(self):
|
|
"""Tests the _validate_selectors function"""
|
|
try:
|
|
_validate_selectors([GroupNodeSelector({})])
|
|
_validate_selectors([])
|
|
except:
|
|
# No exceptions expected.
|
|
assert False
|
|
|
|
with pytest.raises(InvalidDeploymentGroupNodeLookupError) as idgnle:
|
|
_validate_selectors(None)
|
|
assert "iterable of GroupNodeSelectors" in str(idgnle.value)
|
|
|
|
with pytest.raises(InvalidDeploymentGroupNodeLookupError) as idgnle:
|
|
_validate_selectors(["bad!"])
|
|
assert "all input elements in the selectors" in str(idgnle.value)
|
|
|
|
with pytest.raises(InvalidDeploymentGroupNodeLookupError) as idgnle:
|
|
_validate_selectors(["bad!", "also bad!"])
|
|
assert "all input elements in the selectors" in str(idgnle.value)
|
|
|
|
with pytest.raises(InvalidDeploymentGroupNodeLookupError) as idgnle:
|
|
_validate_selectors([GroupNodeSelector({}), "bad!"])
|
|
assert "all input elements in the selectors" in str(idgnle.value)
|
|
|
|
def test_generate_node_filter(self):
|
|
"""Tests the _generate_node_filter function"""
|
|
sel = GroupNodeSelector({
|
|
'node_names': [],
|
|
'node_labels': ['label1:label1'],
|
|
'node_tags': ['tag1', 'tag2'],
|
|
'rack_names': ['rack3', 'rack1'],
|
|
})
|
|
nf = _generate_node_filter([sel])
|
|
assert nf == {
|
|
'filter_set': [{
|
|
'filter_type': 'intersection',
|
|
'node_names': [],
|
|
'node_tags': ['tag1', 'tag2'],
|
|
'rack_names': ['rack3', 'rack1'],
|
|
'node_labels': {'label1': 'label1'}}
|
|
],
|
|
'filter_set_type': 'union'
|
|
}
|
|
|
|
sel2 = GroupNodeSelector({
|
|
'node_names': ['node1', 'node2', 'node3', 'node4', 'node5'],
|
|
'node_labels': ['label1:label1', 'label2:label2'],
|
|
'node_tags': ['tag1', 'tag2'],
|
|
'rack_names': ['rack3', 'rack1'],
|
|
})
|
|
nf = _generate_node_filter([sel, sel2])
|
|
assert nf == {
|
|
'filter_set': [
|
|
{
|
|
'filter_type': 'intersection',
|
|
'node_names': [],
|
|
'node_tags': ['tag1', 'tag2'],
|
|
'rack_names': ['rack3', 'rack1'],
|
|
'node_labels': {'label1': 'label1'}
|
|
},
|
|
{
|
|
'filter_type': 'intersection',
|
|
'node_names': ['node1', 'node2', 'node3', 'node4',
|
|
'node5'],
|
|
'node_tags': ['tag1', 'tag2'],
|
|
'rack_names': ['rack3', 'rack1'],
|
|
'node_labels': {'label1': 'label1', 'label2': 'label2'}
|
|
}
|
|
],
|
|
'filter_set_type': 'union'
|
|
}
|
|
|
|
sel3 = GroupNodeSelector({})
|
|
sel4 = GroupNodeSelector({
|
|
'node_names': [],
|
|
'node_labels': [],
|
|
'node_tags': [],
|
|
'rack_names': [],
|
|
})
|
|
nf = _generate_node_filter([sel, sel3, sel4])
|
|
assert nf == {
|
|
'filter_set': [{
|
|
'filter_type': 'intersection',
|
|
'node_names': [],
|
|
'node_tags': ['tag1', 'tag2'],
|
|
'rack_names': ['rack3', 'rack1'],
|
|
'node_labels': {'label1': 'label1'}}
|
|
],
|
|
'filter_set_type': 'union'
|
|
}
|
|
|
|
nf = _generate_node_filter([sel3, sel4])
|
|
assert nf is None
|
|
|
|
def test_generate_node_filter_more_labels(self):
|
|
"""Tests the _generate_node_filter function"""
|
|
sel = GroupNodeSelector({
|
|
'node_names': [],
|
|
'node_labels': ['label1:label1',
|
|
'label2:enabled',
|
|
'control-plane: bicycle'],
|
|
'node_tags': [],
|
|
'rack_names': [],
|
|
})
|
|
nf = _generate_node_filter([sel])
|
|
assert nf == {
|
|
'filter_set': [{
|
|
'filter_type': 'intersection',
|
|
'node_names': [],
|
|
'node_tags': [],
|
|
'rack_names': [],
|
|
'node_labels': {'label1': 'label1',
|
|
'label2': 'enabled',
|
|
'control-plane': 'bicycle'}
|
|
}],
|
|
'filter_set_type': 'union'
|
|
}
|
|
|
|
def test_generate_node_filter_only_rack(self):
|
|
"""Tests the _generate_node_filter function"""
|
|
sel = GroupNodeSelector({
|
|
'node_names': [],
|
|
'node_labels': [],
|
|
'node_tags': [],
|
|
'rack_names': ['RACK1', 'RACK2'],
|
|
})
|
|
nf = _generate_node_filter([sel])
|
|
assert nf == {
|
|
'filter_set': [{
|
|
'filter_type': 'intersection',
|
|
'node_names': [],
|
|
'node_tags': [],
|
|
'rack_names': ['RACK1', 'RACK2'],
|
|
'node_labels': {}
|
|
}],
|
|
'filter_set_type': 'union'
|
|
}
|
|
|
|
|
|
|
|
@mock.patch('shipyard_airflow.common.deployment_group.node_lookup'
|
|
'._get_nodes_for_filter', return_value=['node1', 'node2'])
|
|
def test_NodeLookup_lookup(self, *args):
|
|
"""Test the functionality of the setup and lookup functions"""
|
|
nl = NodeLookup(mock.MagicMock(), {"design": "ref"})
|
|
|
|
assert nl.design_ref == {"design": "ref"}
|
|
assert nl.drydock_client
|
|
|
|
sel = GroupNodeSelector({
|
|
'node_names': [],
|
|
'node_labels': ['label1:label1'],
|
|
'node_tags': ['tag1', 'tag2'],
|
|
'rack_names': ['rack3', 'rack1'],
|
|
})
|
|
|
|
resp = nl.lookup([sel])
|
|
assert resp == ['node1', 'node2']
|
|
|
|
@mock.patch('shipyard_airflow.common.deployment_group.node_lookup'
|
|
'._get_nodes_for_filter',
|
|
side_effect=errors.ClientError("nope"))
|
|
def test_NodeLookup_lookup_retry(self, get_nodes):
|
|
"""Test the functionality of the setup and lookup functions"""
|
|
nl = NodeLookup(mock.MagicMock(), {"design": "ref"}, retry_delay=0.1)
|
|
sel = GroupNodeSelector({
|
|
'node_names': [],
|
|
'node_labels': [],
|
|
'node_tags': [],
|
|
'rack_names': [],
|
|
})
|
|
with pytest.raises(errors.ClientError) as ex:
|
|
resp = nl.lookup([sel])
|
|
assert get_nodes.call_count == 3
|
|
|
|
@mock.patch('shipyard_airflow.common.deployment_group.node_lookup'
|
|
'._get_nodes_for_filter',
|
|
side_effect=Exception("nope"))
|
|
def test_NodeLookup_lookup_retry_exception(self, get_nodes):
|
|
"""Test the functionality of the setup and lookup functions"""
|
|
nl = NodeLookup(mock.MagicMock(), {"design": "ref"}, retry_delay=0.1)
|
|
sel = GroupNodeSelector({
|
|
'node_names': [],
|
|
'node_labels': [],
|
|
'node_tags': [],
|
|
'rack_names': [],
|
|
})
|
|
with pytest.raises(Exception) as ex:
|
|
resp = nl.lookup([sel])
|
|
assert get_nodes.call_count == 3
|
|
|
|
@mock.patch('shipyard_airflow.common.deployment_group.node_lookup'
|
|
'._get_nodes_for_filter',
|
|
side_effect=errors.ClientUnauthorizedError("nope"))
|
|
def test_NodeLookup_lookup_client_unauthorized(self, get_nodes):
|
|
"""Test the functionality of the setup and lookup functions"""
|
|
nl = NodeLookup(mock.MagicMock(), {"design": "ref"}, retry_delay=0.1)
|
|
sel = GroupNodeSelector({
|
|
'node_names': [],
|
|
'node_labels': [],
|
|
'node_tags': [],
|
|
'rack_names': [],
|
|
})
|
|
with pytest.raises(errors.ClientUnauthorizedError) as ex:
|
|
resp = nl.lookup([sel])
|
|
assert get_nodes.call_count == 1
|
|
|
|
@mock.patch('shipyard_airflow.common.deployment_group.node_lookup'
|
|
'._get_nodes_for_filter',
|
|
side_effect=errors.ClientForbiddenError("nope"))
|
|
def test_NodeLookup_lookup_client_forbidden(self, get_nodes):
|
|
"""Test the functionality of the setup and lookup functions"""
|
|
nl = NodeLookup(mock.MagicMock(), {"design": "ref"}, retry_delay=0.1)
|
|
sel = GroupNodeSelector({
|
|
'node_names': [],
|
|
'node_labels': [],
|
|
'node_tags': [],
|
|
'rack_names': [],
|
|
})
|
|
with pytest.raises(errors.ClientForbiddenError) as ex:
|
|
resp = nl.lookup([sel])
|
|
assert get_nodes.call_count == 1
|
|
|
|
def test_NodeLookup_lookup_missing_design_ref(self):
|
|
"""Test the functionality of the setup and lookup functions"""
|
|
with pytest.raises(InvalidDeploymentGroupNodeLookupError) as idgnle:
|
|
NodeLookup(mock.MagicMock(), {})
|
|
assert 'An incomplete design ref' in str(idgnle.value)
|