diff options
Diffstat (limited to 'laas/tests')
28 files changed, 1591 insertions, 0 deletions
diff --git a/laas/tests/test_action_add_management_vlan.py b/laas/tests/test_action_add_management_vlan.py new file mode 100644 index 0000000..608e3d8 --- /dev/null +++ b/laas/tests/test_action_add_management_vlan.py @@ -0,0 +1,87 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import add_management_vlan +import json +import mock + + +class ManagementVlanTest(BaseActionTestCase): + action_cls = add_management_vlan.ManagementVlanAction + + def setUp(self): + super(ManagementVlanTest, self).setUp() + self.action = self.get_action_instance() + host_info = { + "interfaces": { + "mac1": { + "mac": "mac1", + "bus": "bus1", + "switch": "switch1", + "port": "Ethernet1/1", + "name": "ifname1" + }, + "mac2": { + "mac": "mac2", + "bus": "bus2", + "switch": "switch1", + "port": "Ethernet1/2", + "name": "ifname2" + } + } + } + self.action.action_service.set_value("host1", json.dumps(host_info), local=False) + switch_info = {"user": "user", "password": "password"} + self.action.action_service.set_value("switch_switch1", json.dumps(switch_info), local=False) + + def hasConsecutiveCalls(self, args, mock_obj): + """ + args is a list of arguments as tuples. This method asserts that + mock was called with those arguments in that order + """ + if len(args) < 1: + return True + for call_index in range(len(mock_obj.call_args_list)): + arg_index = 0 + while mock_obj.call_args_list[call_index] == (args[arg_index],): + call_index += 1 + arg_index += 1 + if arg_index == len(args): + return True + return False + + def test_vlans(self): + with mock.patch('actions.actions.add_management_vlan.NXCommand') as Mocked: + self.action.run(["host1"]) + self.assertTrue(Mocked.called) + + # assert that the correct commands are run in order for each interface + # but we dont care about the order of the interfaces + mocked = Mocked.return_value + self.assertTrue(mocked.add_command.called) + expected_calls = [ + ("interface Ethernet1/1",), + ("switchport mode trunk",), + ("switchport trunk allowed vlan 98,99",), + ("switchport trunk native vlan 98",), + ] + self.assertTrue(self.hasConsecutiveCalls(expected_calls, mocked.add_command)) + + expected_calls[0] = ("interface Ethernet1/2",) + self.assertTrue(self.hasConsecutiveCalls(expected_calls, mocked.add_command)) + + self.assertEqual(mocked.add_command.call_count, 8) + self.assertEqual(Mocked.call_count, mocked.execute.call_count) diff --git a/laas/tests/test_action_detectHostsToBoot.py b/laas/tests/test_action_detectHostsToBoot.py new file mode 100644 index 0000000..d7a990e --- /dev/null +++ b/laas/tests/test_action_detectHostsToBoot.py @@ -0,0 +1,74 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import detectHostsToBoot +import json +import responses + + +class DetectHostsTest(BaseActionTestCase): + action_cls = detectHostsToBoot.DetectHostsAction + + def setUp(self): + super(DetectHostsTest, self).setUp() + # this is not documented + self.action_service.set_value("lab_auth_token", "my_auth_token", local=False) + self.action = self.get_action_instance(config={ + "dashboard": { + "address": "http://my.dashboard.com", + "lab_name": "my_lab" + } + }) + self.hosts_url = "http://my.dashboard.com/api/labs/my_lab/hosts/" + self.action.action_service.set_value("hosts_to_boot", "[]", local=False) + + def hostDetected(self, host): + detected_hosts = json.loads(self.action.action_service.get_value("hosts_to_boot", local=False)) + return host in detected_hosts + + @responses.activate + def test_single_host_detected(self): + self.action.action_service.set_value("hosts", '["host1"]', local=False) + responses.add(responses.GET, self.hosts_url + "host1", json={"booked": True}) + self.action.run() + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], "my_auth_token") + self.assertTrue(self.hostDetected("host1")) + + @responses.activate + def test_single_host_not_detected(self): + self.action.action_service.set_value("hosts", '["host1"]', local=False) + responses.add(responses.GET, self.hosts_url + "host1", json={"booked": False}) + self.action.run() + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], "my_auth_token") + self.assertEqual(self.action.action_service.get_value("hosts_to_boot", local=False), "[]") + + @responses.activate + def test_multiple_hosts(self): + self.action.action_service.set_value("hosts", '["host1", "host2", "host3"]', local=False) + responses.add(responses.GET, self.hosts_url + "host1", json={"booked": True}) + responses.add(responses.GET, self.hosts_url + "host2", json={"booked": True}) + responses.add(responses.GET, self.hosts_url + "host3", json={"booked": False}) + + self.action.run() + self.assertEqual(len(responses.calls), 3) + self.assertEqual(responses.calls[0].request.headers['auth-token'], "my_auth_token") + self.assertEqual(responses.calls[1].request.headers['auth-token'], "my_auth_token") + self.assertEqual(responses.calls[2].request.headers['auth-token'], "my_auth_token") + self.assertTrue(self.hostDetected("host1")) + self.assertTrue(self.hostDetected("host2")) + self.assertFalse(self.hostDetected("host3")) diff --git a/laas/tests/test_action_detect_hardware_tasks.py b/laas/tests/test_action_detect_hardware_tasks.py new file mode 100644 index 0000000..d27ac1d --- /dev/null +++ b/laas/tests/test_action_detect_hardware_tasks.py @@ -0,0 +1,42 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import detect_hardware_tasks +import json + + +class DetectHWTestCase(BaseActionTestCase): + action_cls = detect_hardware_tasks.DetectHardwareTasksAction + + def setUp(self): + super(DetectHWTestCase, self).setUp() + self.action = self.get_action_instance() + job_data = { + "hardware": { + "task1": { + "power": "on", + "image": 1, + "hostname": "host1" + } + } + } + self.action.action_service.set_value("job_1", json.dumps(job_data), local=False) + + def test_detect_hardware_tasks(self): + result = self.action.run(job_id=1, task_id="task1") + self.assertTrue(result['power']) + self.assertTrue(result['hostname']) + self.assertTrue(result['image']) diff --git a/laas/tests/test_action_error_task.py b/laas/tests/test_action_error_task.py new file mode 100644 index 0000000..c1f9fe5 --- /dev/null +++ b/laas/tests/test_action_error_task.py @@ -0,0 +1,40 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import error_task +import responses + + +class ErrorTaskTestCase(BaseActionTestCase): + action_cls = error_task.ErrorTaskAction + + def setUp(self): + super(ErrorTaskTestCase, self).setUp() + self.action_service.set_value("lab_auth_token", "my_auth_token", local=False) + self.action = self.get_action_instance(config={ + "dashboard": { + "address": "http://my.dashboard.com", + "lab_name": "my_lab" + } + }) + + @responses.activate + def test_error_task(self): + responses.add(responses.POST, "http://my.dashboard.com/api/labs/my_lab/jobs/1/task1") + self.action.run(job_id=1, task_id="task1") + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], "my_auth_token") + self.assertEqual(responses.calls[0].request.body, "status=300") diff --git a/laas/tests/test_action_finish_task.py b/laas/tests/test_action_finish_task.py new file mode 100644 index 0000000..85a842e --- /dev/null +++ b/laas/tests/test_action_finish_task.py @@ -0,0 +1,40 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import finish_task +import responses + + +class FinishTaskTestCase(BaseActionTestCase): + action_cls = finish_task.FinishTaskAction + + def setUp(self): + super(FinishTaskTestCase, self).setUp() + self.action_service.set_value("lab_auth_token", "my_auth_token", local=False) + self.action = self.get_action_instance(config={ + "dashboard": { + "address": "http://my.dashboard.com", + "lab_name": "my_lab" + } + }) + + @responses.activate + def test_error_task(self): + responses.add(responses.POST, "http://my.dashboard.com/api/labs/my_lab/jobs/1/task1") + self.action.run(job_id=1, task_id="task1") + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], "my_auth_token") + self.assertEqual(responses.calls[0].request.body, "status=200") diff --git a/laas/tests/test_action_fog.py b/laas/tests/test_action_fog.py new file mode 100644 index 0000000..fcbc0c0 --- /dev/null +++ b/laas/tests/test_action_fog.py @@ -0,0 +1,84 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions.lib import fog +import responses +import json + + +class FogTestCase(BaseActionTestCase): + action_cls = fog.FogAction + + def setUp(self): + super(FogTestCase, self).setUp() + self.action = self.get_action_instance(config={ + "fog": { + "address": "http://my.fog.com/fog/", + "api_key": "my_api_key", + "user_key": "my_user_key", + } + }) + + def assertGoodHeader(self, request): + self.assertEqual(request.headers['fog-api-token'], "my_api_key") + self.assertEqual(request.headers['fog-user-token'], "my_user_key") + # TODO: content type? only required when I send a body + + @responses.activate + def test_fog_create_image(self): + responses.add(responses.POST, "http://my.fog.com/fog/image") + payload = {"key1": "v1", "key2": "v2"} + self.action.createImage(payload) + self.assertEqual(len(responses.calls), 1) + self.assertGoodHeader(responses.calls[0].request) + self.assertEqual(payload, json.loads(responses.calls[0].request.body)) + + @responses.activate + def test_fog_get_image(self): + payload = { + "id": 42, + "key": "value", + "name": "fakeImage" + } + responses.add(responses.GET, "http://my.fog.com/fog/image/42", json=payload) + result = self.action.getImage(img=42) + self.assertEqual(len(responses.calls), 1) + self.assertGoodHeader(responses.calls[0].request) + self.assertEqual(result, payload) + + @responses.activate + def test_fog_delete_task(self): + responses.add(responses.DELETE, "http://my.fog.com/fog/fog/host/42/cancel") + self.action.delTask(42) + self.assertEqual(len(responses.calls), 1) + self.assertGoodHeader(responses.calls[0].request) + + @responses.activate + def test_get_host_number(self): + payload = { + "hosts": [ + {"name": "host1", "id": 1}, + {"name": "host2", "id": 2}, + {"name": "host3", "id": 3}, + {"name": "host4", "id": 4}, + {"name": "host5", "id": 5}, + ] + } + responses.add(responses.GET, "http://my.fog.com/fog/host", json=payload) + result = self.action.getHostNumber("host4") + self.assertEqual(len(responses.calls), 1) + self.assertGoodHeader(responses.calls[0].request) + self.assertEqual(result, 4) diff --git a/laas/tests/test_action_fog_capture_host.py b/laas/tests/test_action_fog_capture_host.py new file mode 100644 index 0000000..03cf44b --- /dev/null +++ b/laas/tests/test_action_fog_capture_host.py @@ -0,0 +1,53 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import fog_captureHost +import responses +import json + + +class FogCaptureHostTestCase(BaseActionTestCase): + action_cls = fog_captureHost.StartCaptureAction + + def setUp(self): + super(FogCaptureHostTestCase, self).setUp() + self.action = self.get_action_instance(config={ + "fog": { + "address": "http://my.fog.com/fog/", + "api_key": "my_api_key", + "user_key": "my_user_key", + } + }) + + def assertGoodHeader(self, request): + self.assertEqual(request.headers['fog-api-token'], "my_api_key") + self.assertEqual(request.headers['fog-user-token'], "my_user_key") + # TODO: content type? only required when I send a body + + @responses.activate + def test_fog_capture_host(self): + responses.add(responses.POST, "http://my.fog.com/fog/host/42/task") + self.action.action_service.set_value("host1", json.dumps({"fog_name": "fog_host1"}), local=False) + responses.add(responses.GET, "http://my.fog.com/fog/host", json={ + "hosts": [ + {"name": "fog_host1", "id": 42}, + ] + }) + self.action.run(host="host1") + self.assertEqual(len(responses.calls), 2) + self.assertGoodHeader(responses.calls[0].request) + self.assertGoodHeader(responses.calls[1].request) + self.assertEqual(json.loads(responses.calls[1].request.body), {"taskTypeID": 2}) diff --git a/laas/tests/test_action_fog_change_image.py b/laas/tests/test_action_fog_change_image.py new file mode 100644 index 0000000..fd11a46 --- /dev/null +++ b/laas/tests/test_action_fog_change_image.py @@ -0,0 +1,59 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import fog_changeImage +import responses +import json + + +class FogChangeImageTestCase(BaseActionTestCase): + action_cls = fog_changeImage.ChangeImageAction + + def setUp(self): + super(FogChangeImageTestCase, self).setUp() + self.action = self.get_action_instance(config={ + "fog": { + "address": "http://my.fog.com/fog/", + "api_key": "my_api_key", + "user_key": "my_user_key", + } + }) + + def assertGoodHeader(self, request): + self.assertEqual(request.headers['fog-api-token'], "my_api_key") + self.assertEqual(request.headers['fog-user-token'], "my_user_key") + # TODO: content type? only required when I send a body + + @responses.activate + def test_fog_change_image(self): + responses.add(responses.POST, "http://my.fog.com/fog/host/42/edit") + self.action.action_service.set_value("host1", json.dumps({"fog_name": "fog_host1"}), local=False) + responses.add(responses.GET, "http://my.fog.com/fog/host", json={ + "hosts": [ + {"name": "fog_host1", "id": 42}, + ] + }) + responses.add(responses.GET, "http://my.fog.com/fog/host/42", json={ + "name": "fog_host1", + "imageID": 99, + "description": "test host" + }) + self.action.run(host="host1", image=3) + self.assertEqual(len(responses.calls), 3) + self.assertGoodHeader(responses.calls[0].request) + self.assertGoodHeader(responses.calls[1].request) + self.assertGoodHeader(responses.calls[2].request) + self.assertEqual(int(json.loads(responses.calls[2].request.body)['imageID']), 3) diff --git a/laas/tests/test_action_fog_create_snapshot.py b/laas/tests/test_action_fog_create_snapshot.py new file mode 100644 index 0000000..8cb52a0 --- /dev/null +++ b/laas/tests/test_action_fog_create_snapshot.py @@ -0,0 +1,76 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import fog_createSnapshot +import responses +import json + + +class FogCreateSnapshotTestCase(BaseActionTestCase): + action_cls = fog_createSnapshot.FogCreateSnapshotAction + + def setUp(self): + super(FogCreateSnapshotTestCase, self).setUp() + self.action = self.get_action_instance(config={ + "fog": { + "address": "http://my.fog.com/fog/", + "api_key": "my_api_key", + "user_key": "my_user_key", + } + }) + + def assertGoodHeader(self, request): + self.assertEqual(request.headers['fog-api-token'], "my_api_key") + self.assertEqual(request.headers['fog-user-token'], "my_user_key") + # TODO: content type? only required when I send a body + + @responses.activate + def test_fog_create_snapshot(self): + responses.add(responses.GET, "http://my.fog.com/fog/host", json={ + "hosts": [ + {"name": "fog_host1", "id": 42}, + ] + }) + responses.add(responses.GET, "http://my.fog.com/fog/host/42", json={ + "name": "fog_host1", + "imagename": "orig_img", + "description": "test host" + }) + responses.add(responses.GET, "http://my.fog.com/fog/image", json={ + "images": [ + {"name": "orig_img", "id": 10}, + ] + }) + responses.add(responses.GET, "http://my.fog.com/fog/image/10", json={ + "imagePartitionTypeID": 1, + "toReplicate": False, + "isEnabled": True, + "compress": 6, + "storagegroups": 1, + "osID": 50, + "imageTypeID": 4 + }) + responses.add(responses.POST, "http://my.fog.com/fog/image") + self.action.action_service.set_value("host1", json.dumps({"fog_name": "fog_host1"}), local=False) + + self.action.run(host="host1", name="my_snapshot") + self.assertEqual(len(responses.calls), 5) + self.assertGoodHeader(responses.calls[0].request) + self.assertGoodHeader(responses.calls[1].request) + self.assertGoodHeader(responses.calls[2].request) + self.assertGoodHeader(responses.calls[3].request) + self.assertGoodHeader(responses.calls[4].request) + self.assertEqual(json.loads(responses.calls[-1].request.body)['name'], "my_snapshot") diff --git a/laas/tests/test_action_fog_start_imaging.py b/laas/tests/test_action_fog_start_imaging.py new file mode 100644 index 0000000..97601e9 --- /dev/null +++ b/laas/tests/test_action_fog_start_imaging.py @@ -0,0 +1,54 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import fog_startImaging +import responses +import json + + +class FogStartImageTestCase(BaseActionTestCase): + action_cls = fog_startImaging.StartImagingAction + + def setUp(self): + super(FogStartImageTestCase, self).setUp() + self.action = self.get_action_instance(config={ + "fog": { + "address": "http://my.fog.com/fog/", + "api_key": "my_api_key", + "user_key": "my_user_key", + } + }) + + def assertGoodHeader(self, request): + self.assertEqual(request.headers['fog-api-token'], "my_api_key") + self.assertEqual(request.headers['fog-user-token'], "my_user_key") + # TODO: content type? only required when I send a body + + @responses.activate + def test_fog_start_imaging(self): + responses.add(responses.GET, "http://my.fog.com/fog/host", json={ + "hosts": [ + {"name": "fog_host1", "id": 42}, + ] + }) + responses.add(responses.POST, "http://my.fog.com/fog/host/42/task") + self.action.action_service.set_value("host1", json.dumps({"fog_name": "fog_host1"}), local=False) + + self.action.run(host="host1") + self.assertEqual(len(responses.calls), 2) + self.assertGoodHeader(responses.calls[0].request) + self.assertGoodHeader(responses.calls[1].request) + self.assertEqual(int(json.loads(responses.calls[-1].request.body)['taskTypeID']), 1) diff --git a/laas/tests/test_action_get_all_macs.py b/laas/tests/test_action_get_all_macs.py new file mode 100644 index 0000000..015b826 --- /dev/null +++ b/laas/tests/test_action_get_all_macs.py @@ -0,0 +1,62 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import get_all_macs +import json + + +class GetMacsTestCase(BaseActionTestCase): + action_cls = get_all_macs.MacAction + + def setUp(self): + super(GetMacsTestCase, self).setUp() + self.action = self.get_action_instance() + + def test_single_mac(self): + self.action.action_service.set_value("host1", json.dumps({ + "interfaces": { + "mac1": { + "mac": "mac1", + "speed": 42 + } + } + }), local=False) + result = self.action.run(host="host1") + self.assertEqual(result, "mac1") + + def test_multiple_macs(self): + self.action.action_service.set_value("host1", json.dumps({ + "interfaces": { + "mac1": { + "mac": "mac1", + "speed": 42 + }, + "mac2": { + "mac": "mac2", + "speed": 42 + }, + "mac3": { + "mac": "mac3", + "speed": 42 + } + } + }), local=False) + result = self.action.run(host="host1") + parsed_results = set(result.split("|")) + expected_results = set(["mac1", "mac2", "mac3"]) + self.assertTrue( + parsed_results.issubset(expected_results) and expected_results.issubset(parsed_results) + ) diff --git a/laas/tests/test_action_get_host_type.py b/laas/tests/test_action_get_host_type.py new file mode 100644 index 0000000..5858dcb --- /dev/null +++ b/laas/tests/test_action_get_host_type.py @@ -0,0 +1,37 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import get_host_type + + +class FinishTaskTestCase(BaseActionTestCase): + action_cls = get_host_type.HostTypeAction + + def setUp(self): + super(FinishTaskTestCase, self).setUp() + self.action = self.get_action_instance() + + def test_hpe_host_type(self): + result = self.action.run(host="hpe5") + self.assertEqual(result, "hpe") + + def test_arm_host_type(self): + result = self.action.run(host="arm50") + self.assertEqual(result, "arm") + + def test_bad_host_type(self): + result = self.action.run(host="unknown") + self.assertFalse(result) diff --git a/laas/tests/test_action_get_ipmi_hostname.py b/laas/tests/test_action_get_ipmi_hostname.py new file mode 100644 index 0000000..418241a --- /dev/null +++ b/laas/tests/test_action_get_ipmi_hostname.py @@ -0,0 +1,46 @@ +############################################################################## +# Copyright 2018 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase + +from actions.actions import get_ipmi_hostname +import socket + +hosts = {"host1": "ipmi_host1", "host2": "ipmi_host2"} + + +class IpmiHostnameActionTestCase(BaseActionTestCase): + action_cls = get_ipmi_hostname.ipmi_infoAction + + def setUp(self): + super(IpmiHostnameActionTestCase, self).setUp() + self.skipTest("not read") + + def test_goodHostname_givesRightResult(self): + action = self.get_action_instance() + for key in hosts.keys(): + self.assertEquals(hosts[key], action.run(host=key)) + + def test_results_resolvable(self): + action = self.get_action_instance() + for key in hosts.keys(): + # socket will return ip as a string if it can, which is truthy + self.asserTrue(socket.gethostbyname(action.run(host=key))) + + def test_badHostname_throwsError(self): + bad_host = "abc_I_dont_know_thee" + action = self.get_action_instance() + with self.assertRaises(IndexError): + action.run(host=bad_host) diff --git a/laas/tests/test_action_get_ipmi_password.py b/laas/tests/test_action_get_ipmi_password.py new file mode 100644 index 0000000..b9bc8bc --- /dev/null +++ b/laas/tests/test_action_get_ipmi_password.py @@ -0,0 +1,38 @@ +############################################################################## +# Copyright 2018 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import get_ipmi_password + +hosts = ["ipmi_hostname1", "ipmi_hostname2"] + + +class IpmiPasswordActionTestCase(BaseActionTestCase): + action_cls = get_ipmi_password.ipmi_passwdAction + + def setUp(self): + super(IpmiPasswordActionTestCase, self).setUp() + self.skipTest("not read") + + def test_goodHost_returnsValue(self): + action = self.get_action_instance() + for host in hosts: + self.assertTrue(action.run(host=host)) + + def test_badHost_throwsError(self): + bad_host = "abc_IDontKnowThee" + action = self.get_action_instance() + with self.assertRaises(IndexError): + action.run(host=bad_host) diff --git a/laas/tests/test_action_get_ipmi_username.py b/laas/tests/test_action_get_ipmi_username.py new file mode 100644 index 0000000..1b71683 --- /dev/null +++ b/laas/tests/test_action_get_ipmi_username.py @@ -0,0 +1,38 @@ +############################################################################## +# Copyright 2018 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import get_ipmi_username + +hosts = ["ipmi_host1", "ipmi_host2"] + + +class IpmiUserActionTestCase(BaseActionTestCase): + action_cls = get_ipmi_username.ipmi_userAction + + def setUp(self): + super(IpmiUserActionTestCase, self).setUp() + self.skipTest("not ready") + + def test_goodHost_returnsValue(self): + action = self.get_action_instance() + for host in hosts: + self.assertTrue(action.run(host=host)) + + def test_badHost_throwsError(self): + bad_host = "abc_IDontKnowThee" + action = self.get_action_instance() + with self.assertRaises(IndexError): + action.run(host=bad_host) diff --git a/laas/tests/test_action_get_jumphost.py b/laas/tests/test_action_get_jumphost.py new file mode 100644 index 0000000..a2eb9a9 --- /dev/null +++ b/laas/tests/test_action_get_jumphost.py @@ -0,0 +1,33 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import get_jumphost + + +class FinishTaskTestCase(BaseActionTestCase): + action_cls = get_jumphost.GetJumphostAction + + def setUp(self): + super(FinishTaskTestCase, self).setUp() + self.action = self.get_action_instance() + + def test_get_jumphost(self): + result = self.action.run(hosts=[{"host1": "Compute"}, {"host2": "Jumphost"}]) + self.assertEqual(result, "host2") + + def test_with_no_jumphost(self): + result = self.action.run(hosts=[{"host1": "Compute"}, {"host2": "Controller"}]) + self.assertFalse(result) diff --git a/laas/tests/test_action_get_task.py b/laas/tests/test_action_get_task.py new file mode 100644 index 0000000..f9a37af --- /dev/null +++ b/laas/tests/test_action_get_task.py @@ -0,0 +1,44 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import get_task +import json + + +class GetTaskTestCase(BaseActionTestCase): + action_cls = get_task.GetTaskAction + + def setUp(self): + super(GetTaskTestCase, self).setUp() + self.action = self.get_action_instance() + + def test_get_task_multiple_tasks(self): + self.action.action_service.set_value("job_1", json.dumps({ + "access": { + "task1": "asdf", + "task2": "fdsa" + } + }), local=False) + result = self.action.run(job_id=1, type="access", task_id="task1") + self.assertEqual(result, "asdf") + + def test_get_single_task(self): + self.action.action_service.set_value("job_1", json.dumps({ + "access": {"task1": "asdf"}, + "hardware": {"task10": "foobar"} + }), local=False) + result = self.action.run(job_id=1, type="hardware", task_id="task10") + self.assertEqual("foobar", result) diff --git a/laas/tests/test_action_get_task_list.py b/laas/tests/test_action_get_task_list.py new file mode 100644 index 0000000..790e47e --- /dev/null +++ b/laas/tests/test_action_get_task_list.py @@ -0,0 +1,52 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import get_task_list +import json + + +class GetTaskListTestCase(BaseActionTestCase): + action_cls = get_task_list.Task_List_Action + + def setUp(self): + super(GetTaskListTestCase, self).setUp() + self.action = self.get_action_instance() + + def test_tasklist_multiple_tasks(self): + self.action.action_service.set_value("job_1", json.dumps({ + "access": { + "task1": "asdf", + "task2": "fdsa" + } + }), local=False) + result = self.action.run(job_id=1, type="access") + self.assertEqual(set(result), set(["task1", "task2"])) + + def test_tasklist_single_task(self): + self.action.action_service.set_value("job_1", json.dumps({ + "access": {"task1": "asdf"}, + "hardware": {"task10": "asdf"} + }), local=False) + result = self.action.run(job_id=1, type="hardware") + self.assertEqual(set(result), set(["task10"])) + + def test_empty_tasklist(self): + self.action.action_service.set_value("job_1", json.dumps({ + "access": {"task1": "asdf"}, + "hardware": {"task10": "asdf"} + }), local=False) + result = self.action.run(job_id=1, type="unknown") + self.assertFalse(result) diff --git a/laas/tests/test_action_get_xdf.py b/laas/tests/test_action_get_xdf.py new file mode 100644 index 0000000..09ce98f --- /dev/null +++ b/laas/tests/test_action_get_xdf.py @@ -0,0 +1,56 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import get_xdf +import responses + + +class GetXDFTestCase(BaseActionTestCase): + action_cls = get_xdf.XDF_Action + + def setUp(self): + super(GetXDFTestCase, self).setUp() + self.action_service.set_value("lab_auth_token", "my_auth_token", local=False) + self.action = self.get_action_instance(config={ + "dashboard": { + "address": "http://my.dashboard.com", + "lab_name": "my_lab" + } + }) + + @responses.activate + def test_xdf_retrieved(self): + urls = [ + "http://my.dashboard.com/api/some/endpoint/pdf", + "http://my.dashboard.com/api/some/endpoint/idf" + ] + responses.add(responses.GET, urls[0]) + responses.add(responses.GET, urls[1]) + self.action.run(task_data={ + "opnfv": { + "pdf": "/api/some/endpoint/pdf", + "idf": "/api/some/endpoint/idf", + } + }) + self.assertEqual(len(responses.calls), 2) + self.assertEqual(responses.calls[0].request.headers['auth-token'], "my_auth_token") + self.assertEqual(responses.calls[1].request.headers['auth-token'], "my_auth_token") + + try: + urls.remove(responses.calls[0].request.url) + urls.remove(responses.calls[1].request.url) + except ValueError: + self.fail("Requests were sent to the wrong URLs") diff --git a/laas/tests/test_action_network_task.py b/laas/tests/test_action_network_task.py new file mode 100644 index 0000000..5efb8a9 --- /dev/null +++ b/laas/tests/test_action_network_task.py @@ -0,0 +1,110 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import network_task +import json +import mock + + +class NetworkTaskTest(BaseActionTestCase): + action_cls = network_task.PodNetworkManagerAction + + def setUp(self): + super(NetworkTaskTest, self).setUp() + self.action = self.get_action_instance() + host_info = { + "interfaces": { + "mac1": { + "mac": "mac1", + "bus": "bus1", + "switch": "switch1", + "port": "Ethernet1/1", + "name": "ifname1" + }, + "mac2": { + "mac": "mac2", + "bus": "bus2", + "switch": "switch1", + "port": "Ethernet1/2", + "name": "ifname2" + } + } + } + self.action.action_service.set_value("host1", json.dumps(host_info), local=False) + switch_info = {"user": "user", "password": "password"} + self.action.action_service.set_value("switch_switch1", json.dumps(switch_info), local=False) + + def hasConsecutiveCalls(self, args, mock_obj): + """ + args is a list of arguments as tuples. This method asserts that + mock was called with those arguments in that order + """ + if len(args) < 1: + return True + for call_index in range(len(mock_obj.call_args_list)): + arg_index = 0 + while mock_obj.call_args_list[call_index] == (args[arg_index],): + call_index += 1 + arg_index += 1 + if arg_index == len(args): + return True + return False + + def assertInterfaceConfigured(self, mocked_object, config): + all_calls = mocked_object.add_command.call_args_list + all_args = [c[0][0] for c in all_calls] + # first, we set the interface context + self.assertEqual(all_args[0], "interface " + config[0]['port']) + # next, the port must be trunked + self.assertEqual(all_args[1], "switchport mode trunk") + # next, check that the correct vlans are added + vlan_cmd = all_args[2] + self.assertTrue(vlan_cmd.startswith("switchport trunk allowed vlan ")) + parsed = vlan_cmd.split(" ") + self.assertEqual(len(parsed), 5) + expected_vlans = set([98, 99]) + for vlan in config: + expected_vlans.add(vlan['vlan_id']) + + requested_vlans = parsed[-1].split(",") + requested_vlans = set([int(v) for v in requested_vlans]) + self.assertEqual(requested_vlans, expected_vlans) + # TODO: native vlan + # TODO: assert executed + + def test_simple_net_config(self): + with mock.patch('actions.actions.network_task.NXCommand') as Mocked: + mocks = [] + for i in range(2): + mocks.append(mock.Mock()) + Mocked.side_effect = mocks + + net_conf = { + "host1": { + "mac1": [{"tagged": True, "vlan_id": 100}], + "mac2": [{"tagged": False, "vlan_id": 10}], + } + } + self.action.run(net_conf) + + Mocked.assert_any_call("switch1", {"user": "user", "password": "password"}) + config_map = { + "Ethernet1/1": [{"tagged": True, "vlan_id": 100, "port": "Ethernet1/1"}], + "Ethernet1/2": [{"tagged": False, "vlan_id": 10, "port": "Ethernet1/2"}], + } + for mocked_cmd in mocks: + target_iface = mocked_cmd.add_command.call_args_list[0][0][0].split(" ")[-1] + self.assertInterfaceConfigured(mocked_cmd, config_map[target_iface]) diff --git a/laas/tests/test_action_notify_ip_address.py b/laas/tests/test_action_notify_ip_address.py new file mode 100644 index 0000000..c6cd8bf --- /dev/null +++ b/laas/tests/test_action_notify_ip_address.py @@ -0,0 +1,41 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import notify_ip_address +import responses + + +class NotifyIPTestCase(BaseActionTestCase): + action_cls = notify_ip_address.NotifyIPAction + + def setUp(self): + super(NotifyIPTestCase, self).setUp() + self.action_service.set_value("lab_auth_token", "my_auth_token", local=False) + self.action = self.get_action_instance(config={ + "dashboard": { + "address": "http://my.dashboard.com", + "lab_name": "my_lab" + } + }) + + @responses.activate + def test_notify_ip(self): + responses.add(responses.POST, "http://my.dashboard.com/api/labs/my_lab/jobs/1/task1") + self.action.run(addresses="my_address", hostname="my_host", job_id=1, task_id="task1") + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], "my_auth_token") + self.assertTrue("my_address" in responses.calls[0].request.body) + self.assertTrue("my_host" in responses.calls[0].request.body) diff --git a/laas/tests/test_action_notify_ipmi_user.py b/laas/tests/test_action_notify_ipmi_user.py new file mode 100644 index 0000000..a534158 --- /dev/null +++ b/laas/tests/test_action_notify_ipmi_user.py @@ -0,0 +1,43 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import notify_ipmi_user +import responses + + +class IPMINotificationTestCase(BaseActionTestCase): + action_cls = notify_ipmi_user.NotifyIPMIUserAction + + def setUp(self): + super(IPMINotificationTestCase, self).setUp() + self.action_service.set_value("lab_auth_token", "my_auth_token", local=False) + self.action = self.get_action_instance(config={ + "dashboard": { + "address": "http://my.dashboard.com", + "lab_name": "my_lab" + } + }) + + @responses.activate + def test_notify_ipmi_user(self): + responses.add(responses.POST, "http://my.dashboard.com/api/labs/my_lab/jobs/1/task1") + self.action.action_service.set_value("my_ipmi_key", "my_ipmi_password", local=False) + self.action.run(ipmi_key="my_ipmi_key", hostname="my_host", addr="my_ipmi_addr", job_id=1, task_id="task1") + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], "my_auth_token") + self.assertTrue("my_ipmi_addr" in responses.calls[0].request.body) + self.assertTrue("my_host" in responses.calls[0].request.body) + self.assertTrue("my_ipmi_password" in responses.calls[0].request.body) diff --git a/laas/tests/test_action_notify_ssh_access.py b/laas/tests/test_action_notify_ssh_access.py new file mode 100644 index 0000000..65b015e --- /dev/null +++ b/laas/tests/test_action_notify_ssh_access.py @@ -0,0 +1,41 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import notify_ssh_access +import responses + + +class SSHNotificationTestCase(BaseActionTestCase): + action_cls = notify_ssh_access.NotifySSHAction + + def setUp(self): + super(SSHNotificationTestCase, self).setUp() + self.action_service.set_value("lab_auth_token", "my_auth_token", local=False) + self.action = self.get_action_instance(config={ + "dashboard": { + "address": "http://my.dashboard.com", + "lab_name": "my_lab" + } + }) + + @responses.activate + def test_notify_ssh_user(self): + responses.add(responses.POST, "http://my.dashboard.com/api/labs/my_lab/jobs/1/task1") + self.action.run(user="my_ssh_user", hosts=["my_host"], job_id=1, task_id="task1") + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], "my_auth_token") + self.assertTrue("my_ssh_user" in responses.calls[0].request.body) + self.assertTrue("my_host" in responses.calls[0].request.body) diff --git a/laas/tests/test_action_notify_vpn_user.py b/laas/tests/test_action_notify_vpn_user.py new file mode 100644 index 0000000..6e9b687 --- /dev/null +++ b/laas/tests/test_action_notify_vpn_user.py @@ -0,0 +1,46 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import notify_vpn_user +import responses +import json + + +class VPNNotificationTestCase(BaseActionTestCase): + action_cls = notify_vpn_user.NotifyVPNUserAction + + def setUp(self): + super(VPNNotificationTestCase, self).setUp() + self.action_service.set_value("lab_auth_token", "my_auth_token", local=False) + self.action = self.get_action_instance(config={ + "dashboard": { + "address": "http://my.dashboard.com", + "lab_name": "my_lab" + } + }) + + @responses.activate + def test_notify_vpn_user(self): + responses.add(responses.POST, "http://my.dashboard.com/api/labs/my_lab/jobs/1/task1") + self.action_service.set_value("my_vpn_key", json.dumps({ + "username": "vpn_username", + "password": "vpn_password" + }), local=False) + self.action.run(vpn_key="my_vpn_key", job_id=1, task_id="task1") + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], "my_auth_token") + self.assertTrue("vpn_username" in responses.calls[0].request.body) + self.assertTrue("vpn_password" in responses.calls[0].request.body) diff --git a/laas/tests/test_action_parse_network_data.py b/laas/tests/test_action_parse_network_data.py new file mode 100644 index 0000000..fc0e487 --- /dev/null +++ b/laas/tests/test_action_parse_network_data.py @@ -0,0 +1,92 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase + +from actions.actions import parse_network_data + + +class ParseNetworkTestCase(BaseActionTestCase): + action_cls = parse_network_data.ParseNetworkAction + + def setUp(self): + super(ParseNetworkTestCase, self).setUp() + self.action = self.get_action_instance() + self.action.action_service.set_value("default_vlans", "[5, 10]", local=False) + + def test_empty_returns_true(self): + data = { + "host1": { + "mac1": [], + "mac2": [] + } + } + result = self.action.run(data) + self.assertTrue(result['empty']) + + def test_single_vlan(self): + data = { + "host1": { + "mac1": [{ + "tagged": False, + "vlan_id": 10 + }], + "mac2": [] + } + } + result = self.action.run(data) + self.assertFalse(result['empty']) + self.assertEqual("mac1", result['default']) + self.assertEqual("host1", result['host']) + # should be empty string because there are no tagged vlans to map + self.assertFalse(result['mappings']) + + def test_single_tagged_vlan(self): + data = { + "host1": { + "mac1": [{ + "tagged": True, + "vlan_id": 10 + }], + "mac2": [] + } + } + result = self.action.run(data) + self.assertFalse(result['empty']) + self.assertEqual("mac1.10", result['default']) + self.assertEqual("host1", result['host']) + self.assertEqual("mac1-10", result['mappings']) + + def test_complex_case(self): + data = { + "host1": { + "mac1": [ + {"tagged": True, "vlan_id": 50}, + {"tagged": True, "vlan_id": 500}, + {"tagged": False, "vlan_id": 100} + ], + "mac2": [ + {"tagged": True, "vlan_id": 10}, + {"tagged": False, "vlan_id": 1000} + ] + } + } + result = self.action.run(data) + self.assertFalse(result['empty']) + self.assertEqual("mac2.10", result['default']) + self.assertEqual("host1", result['host']) + mapping = set(result['mappings'].split("+")) + expected = set(["mac1-50", "mac1-500", "mac2-10"]) + self.assertTrue(mapping.issubset(expected) and expected.issubset(mapping)) diff --git a/laas/tests/test_action_send_bot_failure.py b/laas/tests/test_action_send_bot_failure.py new file mode 100644 index 0000000..986cb54 --- /dev/null +++ b/laas/tests/test_action_send_bot_failure.py @@ -0,0 +1,45 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import send_bot_failure +import responses +import json + + +class SendBotFailureTestCase(BaseActionTestCase): + action_cls = send_bot_failure.BotFailureAction + + def setUp(self): + super(SendBotFailureTestCase, self).setUp() + self.action_service.set_value("lab_auth_token", "my_auth_token", local=False) + self.action = self.get_action_instance(config={ + "bot": { + "endpoints": { + "failure": "failure", + "notification": "notification" + }, + "address": "http://my.bot.com/endpoint/" + } + }) + + @responses.activate + def test_send_bot_failure(self): + responses.add(responses.POST, "http://my.bot.com/endpoint/failure") + payload = {"k1": "v1", "k2": "v2"} + self.action.run(**payload) + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['Content-Type'], "application/json") + self.assertEqual(json.dumps(payload), responses.calls[0].request.body) diff --git a/laas/tests/test_action_start_task.py b/laas/tests/test_action_start_task.py new file mode 100644 index 0000000..d060c4b --- /dev/null +++ b/laas/tests/test_action_start_task.py @@ -0,0 +1,40 @@ +############################################################################## +# Copyright 2019 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseActionTestCase +from actions.actions import start_task +import responses + + +class FinishTaskTestCase(BaseActionTestCase): + action_cls = start_task.StartTaskAction + + def setUp(self): + super(FinishTaskTestCase, self).setUp() + self.action_service.set_value("lab_auth_token", "my_auth_token", local=False) + self.action = self.get_action_instance(config={ + "dashboard": { + "address": "http://my.dashboard.com", + "lab_name": "my_lab" + } + }) + + @responses.activate + def test_error_task(self): + responses.add(responses.POST, "http://my.dashboard.com/api/labs/my_lab/jobs/1/task1") + self.action.run(job_id=1, task_id="task1") + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], "my_auth_token") + self.assertEqual(responses.calls[0].request.body, "status=100") diff --git a/laas/tests/test_sensor_laas_api.py b/laas/tests/test_sensor_laas_api.py new file mode 100644 index 0000000..c5181b6 --- /dev/null +++ b/laas/tests/test_sensor_laas_api.py @@ -0,0 +1,118 @@ +############################################################################## +# Copyright 2018 Parker Berberian and Others # +# # +# 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. # +############################################################################## +from st2tests.base import BaseSensorTestCase +import laas +import responses +import json + + +class LaaSSensorTest(BaseSensorTestCase): + sensor_cls = laas.LaaS_Sensor + + def setUp(self): + super(LaaSSensorTest, self).setUp() + self.sensor = self.get_sensor_instance(config={ + "dashboard": { + "address": "http://my.dashboard.com", + "lab_name": "my_lab" + } + }) + self.jobs_url = "http://my.dashboard.com/api/labs/my_lab/jobs/new" + self.sensor.sensor_service.set_value("jobs", "[]", local=False) + self.sensor.sensor_service.set_value("lab_auth_token", "my_auth_token", local=False) + self.sensor.setup() + + def clean(self, sensor): + # Removes all existing bookings from the keystore + kvps = sensor.sensor_service.list_values(local=False, prefix="job_") + for kvp in kvps: + sensor.sensor_service.delete_value(local=False, name=kvp.name) + + def get_job(self, job_id): + return { + "id": job_id, + "payload": { + "hardware": "stuff", + "network": "stuff", + "access": "stuff", + } + } + + def assertJobCreated(self, job_id): + self.assertTriggerDispatched( + trigger="laas.start_job_trigger", + payload={"job_id": job_id} + ) + self.assertTrue(json.loads( + self.sensor.sensor_service.get_value("job_" + str(job_id), local=False) + )) + # TODO: solve concurrency issues in job stop / start + # started_jobs = json.loads(self.sensor.sensor_service.get_value("jobs", local=False)) + # self.assertTrue(job_id in started_jobs) + + # Testing Methods + + @responses.activate + def test_empty_throws_no_triggers(self): + responses.add(responses.GET, self.jobs_url, json=[]) + self.sensor.poll() + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], 'my_auth_token') + self.assertEqual(self.get_dispatched_triggers(), []) + + @responses.activate + def test_new_job_throws_trigger(self): + responses.add(responses.GET, self.jobs_url, json=[self.get_job(1)]) + self.sensor.poll() + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], 'my_auth_token') + self.assertJobCreated(1) + + @responses.activate + def test_job_not_restarted(self): + self.sensor.sensor_service.set_value("jobs", "[1]", local=False) + responses.add(responses.GET, self.jobs_url, json=[self.get_job(1)]) + self.sensor.poll() + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], 'my_auth_token') + self.assertEqual(self.get_dispatched_triggers(), []) + + @responses.activate + def test_sensor_does_not_die(self): + # no endpoint added to responses - will throw an error to the sensor + self.sensor.poll() # shouldn't throw, should still work next time + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], 'my_auth_token') + + responses.add(responses.GET, self.jobs_url, json=[self.get_job(1)]) + self.sensor.poll() + self.assertEqual(len(responses.calls), 2) + self.assertEqual(responses.calls[1].request.headers['auth-token'], 'my_auth_token') + self.assertJobCreated(1) + + @responses.activate + def test_multiple_jobs_started(self): + responses.add(responses.GET, self.jobs_url, json=[ + self.get_job(1), + self.get_job(2), + self.get_job(3) + ]) + self.sensor.poll() + self.assertEqual(len(responses.calls), 1) + self.assertEqual(responses.calls[0].request.headers['auth-token'], 'my_auth_token') + self.assertJobCreated(1) + self.assertJobCreated(2) + self.assertJobCreated(3) |