From 60766db43cd333a8e91ba79801d0c894fac17f21 Mon Sep 17 00:00:00 2001 From: James Falcon Date: Mon, 6 May 2024 14:19:49 -0500 Subject: [PATCH] fix: Ensure no subp from logs.py import (#5268) --- cloudinit/cmd/devel/logs.py | 57 ++++++++++++++------------ tests/unittests/cmd/devel/test_logs.py | 41 +++++++++--------- tests/unittests/conftest.py | 18 ++++++++ tests/unittests/test_cli.py | 14 +------ 4 files changed, 71 insertions(+), 59 deletions(-) diff --git a/cloudinit/cmd/devel/logs.py b/cloudinit/cmd/devel/logs.py index 3ac3ee4835d..89fe5c97e92 100755 --- a/cloudinit/cmd/devel/logs.py +++ b/cloudinit/cmd/devel/logs.py @@ -13,7 +13,7 @@ import sys from datetime import datetime from pathlib import Path -from typing import NamedTuple +from typing import NamedTuple, Optional from cloudinit.cmd.devel import read_cfg_paths from cloudinit.stages import Init @@ -27,8 +27,23 @@ write_file, ) -PATHS = read_cfg_paths() -CLOUDINIT_RUN_DIR = PATHS.run_dir + +class LogPaths(NamedTuple): + userdata_raw: str + cloud_data: str + run_dir: str + instance_data_sensitive: str + + +def get_log_paths(init: Optional[Init] = None) -> LogPaths: + """Return a Paths object based on the system configuration on disk.""" + paths = init.paths if init else read_cfg_paths() + return LogPaths( + userdata_raw=paths.get_ipath_cur("userdata_raw"), + cloud_data=paths.get_cpath("data"), + run_dir=paths.run_dir, + instance_data_sensitive=paths.lookups["instance_data_sensitive"], + ) class ApportFile(NamedTuple): @@ -81,16 +96,6 @@ class ApportFile(NamedTuple): ] -def _get_user_data_file() -> str: - paths = read_cfg_paths() - return paths.get_ipath_cur("userdata_raw") - - -def _get_cloud_data_path() -> str: - paths = read_cfg_paths() - return paths.get_cpath("data") - - def get_parser(parser=None): """Build or extend and arg parser for collect-logs utility. @@ -122,7 +127,6 @@ def get_parser(parser=None): " Default: cloud-init.tar.gz" ), ) - user_data_file = _get_user_data_file() parser.add_argument( "--include-userdata", "-u", @@ -131,20 +135,20 @@ def get_parser(parser=None): dest="userdata", help=( "Optionally include user-data from {0} which could contain" - " sensitive information.".format(user_data_file) + " sensitive information.".format(get_log_paths().userdata_raw) ), ) return parser -def _copytree_rundir_ignore_files(curdir, files): +def _get_copytree_ignore_files(paths: LogPaths): """Return a list of files to ignore for /run/cloud-init directory""" ignored_files = [ "hook-hotplug-cmd", # named pipe for hotplug ] if os.getuid() != 0: # Ignore root-permissioned files - ignored_files.append(PATHS.lookups["instance_data_sensitive"]) + ignored_files.append(paths.instance_data_sensitive) return ignored_files @@ -216,7 +220,6 @@ def collect_logs(tarfile, include_userdata: bool, verbosity=0): ) return 1 - init = Init(ds_deps=[]) tarfile = os.path.abspath(tarfile) log_dir = datetime.utcnow().date().strftime("cloud-init-logs-%Y-%m-%d") with tempdir(dir="/tmp") as tmp_dir: @@ -250,36 +253,38 @@ def collect_logs(tarfile, include_userdata: bool, verbosity=0): verbosity=verbosity, ) + init = Init(ds_deps=[]) init.read_cfg() + paths = get_log_paths(init) for log in get_config_logfiles(init.cfg): _collect_file(log, log_dir, verbosity) if include_userdata: - user_data_file = _get_user_data_file() + user_data_file = paths.userdata_raw _collect_file(user_data_file, log_dir, verbosity) collect_installer_logs(log_dir, include_userdata, verbosity) run_dir = os.path.join(log_dir, "run") ensure_dir(run_dir) - if os.path.exists(CLOUDINIT_RUN_DIR): + if os.path.exists(paths.run_dir): try: shutil.copytree( - CLOUDINIT_RUN_DIR, + paths.run_dir, os.path.join(run_dir, "cloud-init"), - ignore=_copytree_rundir_ignore_files, + ignore=lambda _, __: _get_copytree_ignore_files(paths), ) except shutil.Error as e: sys.stderr.write("Failed collecting file(s) due to error:\n") sys.stderr.write(str(e) + "\n") - _debug("collected dir %s\n" % CLOUDINIT_RUN_DIR, 1, verbosity) + _debug("collected dir %s\n" % paths.run_dir, 1, verbosity) else: _debug( - "directory '%s' did not exist\n" % CLOUDINIT_RUN_DIR, + "directory '%s' did not exist\n" % paths.run_dir, 1, verbosity, ) - if os.path.exists(os.path.join(CLOUDINIT_RUN_DIR, "disabled")): + if os.path.exists(os.path.join(paths.run_dir, "disabled")): # Fallback to grab previous cloud/data - cloud_data_dir = Path(_get_cloud_data_path()) + cloud_data_dir = Path(paths.cloud_data) if cloud_data_dir.exists(): shutil.copytree( str(cloud_data_dir), diff --git a/tests/unittests/cmd/devel/test_logs.py b/tests/unittests/cmd/devel/test_logs.py index 4c38e034ac7..55a76daf55c 100644 --- a/tests/unittests/cmd/devel/test_logs.py +++ b/tests/unittests/cmd/devel/test_logs.py @@ -35,7 +35,9 @@ def test_collect_logs_with_userdata_requires_root_user( " Try sudo cloud-init collect-logs\n" == m_stderr.getvalue() ) - def test_collect_logs_creates_tarfile(self, m_getuid, mocker, tmpdir): + def test_collect_logs_creates_tarfile( + self, m_getuid, m_log_paths, mocker, tmpdir + ): """collect-logs creates a tarfile with all related cloud-init info.""" m_getuid.return_value = 100 log1 = tmpdir.join("cloud-init.log") @@ -46,12 +48,10 @@ def test_collect_logs_creates_tarfile(self, m_getuid, mocker, tmpdir): write_file(log2, "cloud-init-output-log") log2_rotated = tmpdir.join("cloud-init-output.log.1.gz") write_file(log2_rotated, "cloud-init-output-log-rotated") - run_dir = tmpdir.join("run") - write_file(run_dir.join("results.json"), "results") + run_dir = m_log_paths.run_dir + write_file(str(run_dir / "results.json"), "results") write_file( - run_dir.join( - INSTANCE_JSON_SENSITIVE_FILE, - ), + str(m_log_paths.instance_data_sensitive), "sensitive", ) output_tarfile = str(tmpdir.join("logs.tgz")) @@ -108,7 +108,6 @@ def fake_subprocess_call(cmd, stdout=None, stderr=None): M_PATH + "subprocess.call", side_effect=fake_subprocess_call ) mocker.patch(M_PATH + "sys.stderr", fake_stderr) - mocker.patch(M_PATH + "CLOUDINIT_RUN_DIR", run_dir) mocker.patch(M_PATH + "INSTALLER_APPORT_FILES", []) mocker.patch(M_PATH + "INSTALLER_APPORT_SENSITIVE_FILES", []) logs.collect_logs(output_tarfile, include_userdata=False) @@ -155,7 +154,7 @@ def fake_subprocess_call(cmd, stdout=None, stderr=None): fake_stderr.write.assert_any_call("Wrote %s\n" % output_tarfile) def test_collect_logs_includes_optional_userdata( - self, m_getuid, mocker, tmpdir + self, m_getuid, mocker, tmpdir, m_log_paths ): """collect-logs include userdata when --include-userdata is set.""" m_getuid.return_value = 0 @@ -163,12 +162,12 @@ def test_collect_logs_includes_optional_userdata( write_file(log1, "cloud-init-log") log2 = tmpdir.join("cloud-init-output.log") write_file(log2, "cloud-init-output-log") - userdata = tmpdir.join("user-data.txt") - write_file(userdata, "user-data") - run_dir = tmpdir.join("run") - write_file(run_dir.join("results.json"), "results") + userdata = m_log_paths.userdata_raw + write_file(str(userdata), "user-data") + run_dir = m_log_paths.run_dir + write_file(str(run_dir / "results.json"), "results") write_file( - run_dir.join(INSTANCE_JSON_SENSITIVE_FILE), + str(m_log_paths.instance_data_sensitive), "sensitive", ) output_tarfile = str(tmpdir.join("logs.tgz")) @@ -223,23 +222,21 @@ def fake_subprocess_call(cmd, stdout=None, stderr=None): M_PATH + "subprocess.call", side_effect=fake_subprocess_call ) mocker.patch(M_PATH + "sys.stderr", fake_stderr) - mocker.patch(M_PATH + "CLOUDINIT_RUN_DIR", run_dir) mocker.patch(M_PATH + "INSTALLER_APPORT_FILES", []) mocker.patch(M_PATH + "INSTALLER_APPORT_SENSITIVE_FILES", []) - mocker.patch(M_PATH + "_get_user_data_file", return_value=userdata) logs.collect_logs(output_tarfile, include_userdata=True) # unpack the tarfile and check file contents subp(["tar", "zxvf", output_tarfile, "-C", str(tmpdir)]) out_logdir = tmpdir.join(date_logdir) assert "user-data" == load_text_file( - os.path.join(out_logdir, "user-data.txt") + os.path.join(out_logdir, userdata.name) ) assert "sensitive" == load_text_file( os.path.join( out_logdir, "run", "cloud-init", - INSTANCE_JSON_SENSITIVE_FILE, + m_log_paths.instance_data_sensitive.name, ) ) fake_stderr.write.assert_any_call("Wrote %s\n" % output_tarfile) @@ -400,7 +397,9 @@ def test_include_installer_logs_when_present( class TestParser: - def test_parser_help_has_userdata_file(self, mocker, tmpdir): - userdata = str(tmpdir.join("user-data.txt")) - mocker.patch(M_PATH + "_get_user_data_file", return_value=userdata) - assert userdata in re.sub(r"\s+", "", logs.get_parser().format_help()) + def test_parser_help_has_userdata_file(self, m_log_paths, mocker, tmpdir): + # userdata = str(tmpdir.join("user-data.txt")) + userdata = m_log_paths.userdata_raw + assert str(userdata) in re.sub( + r"\s+", "", logs.get_parser().format_help() + ) diff --git a/tests/unittests/conftest.py b/tests/unittests/conftest.py index b3b678e78eb..e912c90ad28 100644 --- a/tests/unittests/conftest.py +++ b/tests/unittests/conftest.py @@ -1,6 +1,7 @@ import builtins import glob import os +import pathlib import shutil from pathlib import Path from unittest import mock @@ -8,6 +9,7 @@ import pytest from cloudinit import atomic_helper, log, util +from cloudinit.cmd.devel import logs from cloudinit.gpg import GPG from tests.hypothesis import HAS_HYPOTHESIS from tests.unittests.helpers import example_netdev, retarget_many_wrapper @@ -167,3 +169,19 @@ def tmp_path(tmpdir): settings.register_profile("ci", max_examples=1000) settings.load_profile(os.getenv("HYPOTHESIS_PROFILE", "default")) + + +@pytest.fixture +def m_log_paths(mocker, tmp_path): + """Define logs.LogPaths for testing and mock get_log_paths with it.""" + paths = logs.LogPaths( + userdata_raw=tmp_path / "userdata_raw", + cloud_data=tmp_path / "cloud_data", + run_dir=tmp_path / "run_dir", + instance_data_sensitive=tmp_path + / "run_dir" + / "instance_data_sensitive", + ) + pathlib.Path(paths.run_dir).mkdir() + mocker.patch.object(logs, "get_log_paths", return_value=paths) + yield paths diff --git a/tests/unittests/test_cli.py b/tests/unittests/test_cli.py index 9b53db9a8eb..4315d89cd6e 100644 --- a/tests/unittests/test_cli.py +++ b/tests/unittests/test_cli.py @@ -21,14 +21,6 @@ FakeArgs = namedtuple("FakeArgs", ["action", "local", "mode"]) -@pytest.fixture() -def mock_get_user_data_file(mocker, tmpdir): - yield mocker.patch( - "cloudinit.cmd.devel.logs._get_user_data_file", - return_value=tmpdir.join("cloud"), - ) - - @pytest.fixture(autouse=True, scope="module") def disable_setup_logging(): # setup_basic_logging can change the logging level to WARNING, so @@ -255,13 +247,11 @@ def test_modules_subcommand_parser(self, m_status_wrapper, subcommand): "schema", ], ) - @mock.patch("cloudinit.stages.Init._read_cfg", return_value={}) def test_conditional_subcommands_from_entry_point_sys_argv( self, - m_read_cfg, subcommand, capsys, - mock_get_user_data_file, + m_log_paths, mock_status_wrapper, ): """Subcommands from entry-point are properly parsed from sys.argv.""" @@ -284,7 +274,7 @@ def test_conditional_subcommands_from_entry_point_sys_argv( ], ) def test_subcommand_parser( - self, subcommand, mock_get_user_data_file, mock_status_wrapper + self, subcommand, m_log_paths, mock_status_wrapper ): """cloud-init `subcommand` calls its subparser.""" # Provide -h param to `subcommand` to avoid having to mock behavior.