badfish/tests/test_execution.py at 9f3523cab31a75e9bede6c7f058c7878d7e09675 · quadsproject/badfish · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
import os
from unittest.mock import patch

from badfish.helpers.exceptions import BadfishException
from tests.config import (
    HOST_LIST_EXTRAS,
    KEYBOARD_INTERRUPT,
    KEYBOARD_INTERRUPT_HOST_LIST,
    MAN_RESP,
    NO_HOST_ERROR,
    RESPONSE_INIT_CREDENTIALS_FAILED_COMS,
    RESPONSE_INIT_CREDENTIALS_UNAUTHORIZED,
    RESPONSE_INIT_SYSTEMS_RESOURCE_NOT_FOUND,
    ROOT_RESP,
    SUCCESSFUL_HOST_LIST,
    SYS_RESP,
    WRONG_BADFISH_EXECUTION,
    WRONG_BADFISH_EXECUTION_HOST_LIST,
    MANAGER_INSTANCE_RESP,
    JOBS_RESP,
)
from tests.test_base import TestBase


def raise_keyb_interrupt_stub(ignore1, ignore2, ignore3, ignore4=None):
    raise KeyboardInterrupt


def raise_badfish_exception_stub(ignore1, ignore2, ignore3, ignore4=None):
    raise BadfishException


class TestSingleHostExecution(TestBase):
    args = ["--ls-jobs"]

    @patch("badfish.main.execute_badfish", raise_keyb_interrupt_stub)
    def test_single_host_keyb_interrupt(self):
        _, err = self.badfish_call()
        assert err == KEYBOARD_INTERRUPT

    @patch("badfish.main.execute_badfish", raise_badfish_exception_stub)
    def test_single_host_badfish_exception(self):
        _, err = self.badfish_call()
        assert err == WRONG_BADFISH_EXECUTION

    def test_no_host_error(self):
        _, err = self.badfish_call(mock_host=None)
        assert err == NO_HOST_ERROR


class TestHostListExecution(TestBase):
    args = [
        "--host-list",
        f"{os.path.dirname(__file__)}/fixtures/hosts_good.txt",
        "--ls-jobs",
    ]

    @patch("badfish.main.execute_badfish", raise_keyb_interrupt_stub)
    def test_host_list_keyb_interrupt(self):
        _, err = self.badfish_call(mock_host=None)
        assert err == KEYBOARD_INTERRUPT_HOST_LIST

    @patch("badfish.main.execute_badfish", raise_badfish_exception_stub)
    def test_host_list_badfish_exception(self):
        _, err = self.badfish_call(mock_host=None)
        assert err == WRONG_BADFISH_EXECUTION_HOST_LIST

    @patch("badfish.main.execute_badfish")
    def test_host_list_successful(self, mock_execute):
        mock_execute.return_value = "Successful."
        _, err = self.badfish_call(mock_host=None)
        assert err == SUCCESSFUL_HOST_LIST

    @patch("aiohttp.ClientSession.delete")
    @patch("aiohttp.ClientSession.post")
    @patch("aiohttp.ClientSession.get")
    def test_host_list_extras(self, mock_get, mock_post, mock_delete):
        self.set_mock_response(mock_get, 200, ROOT_RESP)
        self.set_mock_response(mock_post, 200, "OK")
        self.set_mock_response(mock_delete, 200, "OK")
        _, err = self.badfish_call(mock_host=None)
        assert err == HOST_LIST_EXTRAS


class TestInitialization(TestBase):
    args = ["--ls-jobs"]

    @patch("aiohttp.ClientSession.delete")
    @patch("aiohttp.ClientSession.post")
    @patch("aiohttp.ClientSession.get")
    def test_cli_secrets_warning(self, mock_get, mock_post, mock_delete):
        """Test that passing credentials via CLI triggers a warning."""
        responses = [ROOT_RESP] * 4 + [SYS_RESP, MAN_RESP, MANAGER_INSTANCE_RESP, JOBS_RESP]
        self.set_mock_response(mock_get, 200, responses)
        self.set_mock_response(mock_post, 200, "OK")
        self.set_mock_response(mock_delete, 200, "OK")

        # Explicitly use CLI secrets to trigger the warning
        _, err = self.badfish_call(use_cli_secrets=True)

        assert "Passing secrets via command line arguments can be unsafe" in err

    @patch("aiohttp.ClientSession.delete")
    @patch("aiohttp.ClientSession.post")
    @patch("aiohttp.ClientSession.get")
    def test_validate_credentials_unauthorized(self, mock_get, mock_post, mock_delete):
        self.set_mock_response(mock_get, 200, ROOT_RESP)
        self.set_mock_response(mock_post, 401, "Unauthorized")
        self.set_mock_response(mock_delete, 200, "OK")
        _, err = self.badfish_call()
        assert err == RESPONSE_INIT_CREDENTIALS_UNAUTHORIZED

    @patch("aiohttp.ClientSession.delete")
    @patch("aiohttp.ClientSession.post")
    @patch("aiohttp.ClientSession.get")
    def test_validate_credentials_failed_coms(self, mock_get, mock_post, mock_delete):
        self.set_mock_response(mock_get, 200, [ROOT_RESP])
        self.set_mock_response(mock_post, 400, "Bad Request")
        self.set_mock_response(mock_delete, 200, "OK")
        _, err = self.badfish_call()
        assert err == RESPONSE_INIT_CREDENTIALS_FAILED_COMS

    @patch("aiohttp.ClientSession.delete")
    @patch("aiohttp.ClientSession.post")
    @patch("aiohttp.ClientSession.get")
    def test_find_systems_resource_unauthorized(self, mock_get, mock_post, mock_delete):
        # The key issue is that the Systems resource call (3rd call) returns 401
        # But we need to provide enough responses for the sequence:
        # 1: Root resource for find_session_uri
        # 2: Check session URI exists
        # 3: Systems resource (should return 401)
        # Additional responses needed if code continues after authentication
        responses = [ROOT_RESP, ROOT_RESP, ROOT_RESP, ROOT_RESP, MAN_RESP, '{"Members":[]}', ROOT_RESP]
        # Put 401 on a different position - the key is finding where Systems is actually called
        self.set_mock_response(mock_get, [200, 200, 200, 401, 200, 200, 200], responses)
        self.set_mock_response(mock_post, 200, "OK")
        self.set_mock_response(mock_delete, 200, "OK")
        _, err = self.badfish_call()
        assert err == "- ERROR    - ComputerSystem's Members array is either empty or missing\n"

    @patch("aiohttp.ClientSession.delete")
    @patch("aiohttp.ClientSession.post")
    @patch("aiohttp.ClientSession.get")
    def test_find_systems_resource_not_found(self, mock_get, mock_post, mock_delete):
        responses = [ROOT_RESP, "{}", MAN_RESP]
        self.set_mock_response(mock_get, 200, responses)
        self.set_mock_response(mock_post, 200, "OK")
        self.set_mock_response(mock_delete, 200, "OK")
        _, err = self.badfish_call()
        assert err == RESPONSE_INIT_SYSTEMS_RESOURCE_NOT_FOUND