79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
import unittest
|
|
|
|
import mock
|
|
|
|
from os_faults.api.node_collection import Host, NodeCollection
|
|
from os_faults.drivers.services.process import ServiceAsProcess
|
|
|
|
|
|
class TestServiceAsProcess(unittest.TestCase):
|
|
|
|
def test_unplug(self):
|
|
config = {
|
|
'grep': 'test-service',
|
|
'port': ['tcp', 8000],
|
|
}
|
|
cloud_management = mock.Mock()
|
|
|
|
service = ServiceAsProcess('test-service', config,
|
|
mock.Mock(), cloud_management)
|
|
|
|
hosts = [Host('10.1.1.10')]
|
|
node_collection = NodeCollection(cloud_management=cloud_management,
|
|
hosts=hosts)
|
|
service.get_nodes = mock.Mock(return_value=node_collection)
|
|
|
|
# run the command
|
|
service.unplug()
|
|
|
|
# verify
|
|
expected_task = {
|
|
'iptables': {
|
|
'chain': 'INPUT',
|
|
'protocol': 'tcp',
|
|
'jump': 'DROP',
|
|
'destination_port': '8000',
|
|
'action': 'insert',
|
|
'state': 'present',
|
|
'comment': 'Added by os-faults',
|
|
},
|
|
'become': 'yes',
|
|
}
|
|
cloud_management.execute_on_cloud.assert_called_once_with(
|
|
hosts, expected_task)
|
|
|
|
def test_unplug_with_other_port(self):
|
|
config = {
|
|
'grep': 'test-service',
|
|
'port': ['tcp', 8000],
|
|
}
|
|
cloud_management = mock.Mock()
|
|
|
|
service = ServiceAsProcess('test-service', config,
|
|
mock.Mock(), cloud_management)
|
|
|
|
hosts = [Host('10.1.1.10')]
|
|
node_collection = NodeCollection(cloud_management=cloud_management,
|
|
hosts=hosts)
|
|
service.get_nodes = mock.Mock(return_value=node_collection)
|
|
|
|
# run the command
|
|
service.unplug(direction='egress', other_port=['udp', 10000])
|
|
|
|
# verify
|
|
expected_task = {
|
|
'iptables': {
|
|
'chain': 'OUTPUT',
|
|
'protocol': 'udp',
|
|
'jump': 'DROP',
|
|
'destination_port': '10000',
|
|
'action': 'insert',
|
|
'state': 'present',
|
|
'comment': 'Added by os-faults',
|
|
},
|
|
'become': 'yes',
|
|
}
|
|
cloud_management.execute_on_cloud.assert_called_once_with(
|
|
hosts, expected_task)
|
|
|