sos/tests/unittests/juju/juju_transports_test.py at 03d85963f53a8e10b261922f190ec6c5a248df6e · arif-ali/sos · 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
# Copyright (c) 2023 Canonical Ltd., Chi Wai Chan <chiwai.chan@canonical.com>

# This file is part of the sos project: https://github.com/sosreport/sos
#
# This copyrighted material is made available to anyone wishing to use,
# modify, copy, or redistribute it subject to the terms and conditions of
# version 2 of the GNU General Public License.
#
# See the LICENSE file in the source distribution for further information.

import subprocess
import unittest
from unittest.mock import patch

from sos.collector.exceptions import JujuNotInstalledException
from sos.collector.transports.juju import JujuSSH


class MockCmdLineOpts:
    ssh_user = "user_abc"
    sudo_pw = "pw_abc"
    root_password = "root_pw_abc"


class JujuSSHTest(unittest.TestCase):
    def setUp(self):
        self.juju_ssh = JujuSSH(
            commons={
                "cmdlineopts": MockCmdLineOpts,
                "tmpdir": "/tmp/sos-juju/",
                "need_sudo": False,
            },
            address="model_abc:unit_abc",
        )

    # pylint: disable=no-method-argument
    def get_juju_version():
        return "2.9.45"

    # pylint: disable=no-method-argument
    def get_juju_version_3():
        return "3.1.0"

    @patch("sos.collector.transports.juju.subprocess.check_output")
    def test_check_juju_installed_err(self, mock_subprocess_check_output):
        """Raise error if juju is not installed."""
        mock_subprocess_check_output.side_effect = (
            subprocess.CalledProcessError(returncode="127", cmd="cmd_abc")
        )
        with self.assertRaises(JujuNotInstalledException):
            self.juju_ssh._check_juju_installed()

    @patch("sos.collector.transports.juju.subprocess.check_output")
    # pylint: disable=unused-argument
    def test_check_juju_installed_true(self, mock_subprocess_check_output):
        """Return True if juju is installed."""
        result = self.juju_ssh._check_juju_installed()
        assert result

    @patch("sos.collector.transports.juju.subprocess.check_output")
    def test_chmod(self, mock_subprocess_check_output):
        self.juju_ssh._chmod(fname="file_abc")
        mock_subprocess_check_output.assert_called_with(
            f"{self.juju_ssh.remote_exec} sudo chmod o+r file_abc",
            stderr=subprocess.STDOUT,
            shell=True,
        )

    @patch(
        "sos.collector.transports.juju.JujuSSH._check_juju_installed",
        return_value=True,
    )
    # pylint: disable=unused-argument
    def test_connect(self, mock_result):
        self.juju_ssh.connect(password=None)
        assert self.juju_ssh.connected

    def test_remote_exec(self):
        assert (
            self.juju_ssh.remote_exec == "juju ssh -m model_abc unit_abc"
        )

    @patch(
        "sos.collector.transports.juju.JujuSSH._get_juju_version",
        side_effect=get_juju_version,
    )
    @patch(
        "sos.collector.transports.juju.sos_get_command_output",
        return_value={"status": 0},
    )
    @patch("sos.collector.transports.juju.JujuSSH._chmod", return_value=True)
    # pylint: disable=unused-argument
    def test_retrieve_file(
        self,
        mock_chmod,
        mock_sos_get_cmd_output,
        mock_get_juju_version
    ):
        self.juju_ssh._retrieve_file(fname="file_abc", dest="/tmp/sos-juju/")
        mock_sos_get_cmd_output.assert_called_with(
            "juju scp -m model_abc -- -r unit_abc:file_abc /tmp/sos-juju/"
        )

    @patch("sos.collector.transports.juju.shutil.move")
    @patch("sos.collector.transports.juju.os.makedirs")
    @patch(
        "sos.collector.transports.juju.os.path.expanduser",
        return_value="/home/user_abc",
    )
    @patch(
        "sos.collector.transports.juju.JujuSSH._get_juju_version",
        side_effect=get_juju_version_3,
    )
    @patch(
        "sos.collector.transports.juju.sos_get_command_output",
        return_value={"status": 0},
    )
    @patch("sos.collector.transports.juju.JujuSSH._chmod", return_value=True)
    # pylint: disable=unused-argument
    def test_retrieve_file_juju_3(
        self,
        mock_chmod,
        mock_sos_get_cmd_output,
        mock_get_juju_version,
        mock_expanduser,
        mock_makedirs,
        mock_move,
    ):
        """For juju 3+ the file is staged under $HOME (confinement-safe) and
        then moved to the destination, without sudo or private-tmp hacks."""
        staging_dir = "/home/user_abc/.cache/sos-collect-juju"
        self.juju_ssh._retrieve_file(fname="file_abc", dest="/tmp/sos-juju/")
        mock_makedirs.assert_called_with(staging_dir, exist_ok=True)
        mock_sos_get_cmd_output.assert_called_with(
            f"juju scp -m model_abc -- -r unit_abc:file_abc {staging_dir}"
        )
        mock_move.assert_called_with(
            f"{staging_dir}/file_abc", "/tmp/sos-juju/"
        )


# vim: set et ts=4 sw=4 :