Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 2 additions & 7 deletions conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -95,8 +95,8 @@ def disable_subp_usage(request, fixture_utils):

Note that this can only catch invocations where the ``subp`` module is
imported and ``subp.subp(...)`` is called. ``from cloudinit.subp import
subp`` imports happen before the patching here (or the CiTestCase
monkey-patching) happens, so are left untouched.
subp`` is left untouched because those imports happen before the patching
happens here.

While ``disable_subp_usage`` unconditionally patches
``cloudinit.subp.subp``, any test-local patching will override this
Expand Down Expand Up @@ -128,11 +128,6 @@ def test_bash(self):
def test_several_things(self):
subp.subp(["bash"])
subp.subp(["whoami"])

This fixture (roughly) mirrors the functionality of
``CiTestCase.allowed_subp``. N.B. While autouse fixtures do affect
non-pytest tests, CiTestCase's ``allowed_subp`` does take precedence (and
we have ``TestDisableSubpUsageInTestSubclass`` to confirm that).
"""
allow_subp_for = fixture_utils.closest_marker_args_or(
request, "allow_subp_for", None
Expand Down
110 changes: 1 addition & 109 deletions tests/unittests/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,25 +2,20 @@
# pylint: disable=attribute-defined-outside-init

import copy
import functools
import io
import logging
import os
import random
import shutil
import string
import tempfile
import time
import unittest
from contextlib import contextmanager
from typing import ClassVar, List, Union
from unittest import mock
from unittest.util import strclass
from urllib.parse import urlsplit, urlunsplit

import responses

from cloudinit import distros, helpers, settings, subp, util
from cloudinit import distros, helpers, settings, util
from cloudinit.config.schema import (
SchemaValidationError,
validate_cloudconfig_schema,
Expand All @@ -30,8 +25,6 @@
from tests.helpers import cloud_init_project_dir
from tests.hypothesis_jsonschema import HAS_HYPOTHESIS_JSONSCHEMA

_real_subp = subp.subp

# Used for skipping tests
SkipTest = unittest.SkipTest
skipIf = unittest.skipIf
Expand Down Expand Up @@ -164,107 +157,6 @@ def add_patch(self, target, attr, *args, **kwargs):
setattr(self, attr, p)


class CiTestCase(TestCase):
"""This is the preferred test case base class unless user
needs other test case classes below."""

# Subclass overrides for specific test behavior
# Whether or not a unit test needs logfile setup
with_logs = False
allowed_subp: ClassVar[Union[List, bool]] = False
SUBP_SHELL_TRUE = "shell=true"

@contextmanager
def allow_subp(self, allowed_subp):
orig = self.allowed_subp
try:
self.allowed_subp = allowed_subp
yield
finally:
self.allowed_subp = orig

def setUp(self):
super(CiTestCase, self).setUp()
if self.with_logs:
# Create a log handler so unit tests can search expected logs.
self.logger = logging.getLogger()
self.logs = io.StringIO()
formatter = logging.Formatter("%(levelname)s: %(message)s")
handler = logging.StreamHandler(self.logs)
handler.setFormatter(formatter)
self.old_handlers = self.logger.handlers
self.logger.handlers = [handler]
self.old_level = logging.root.level
self.logger.level = logging.DEBUG
if self.allowed_subp is True:
subp.subp = _real_subp
else:
subp.subp = self._fake_subp

def _fake_subp(self, *args, **kwargs):
if "args" in kwargs:
cmd = kwargs["args"]
else:
if not args:
raise TypeError(
"subp() missing 1 required positional argument: 'args'"
)
cmd = args[0]

if not isinstance(cmd, str):
cmd = cmd[0]
pass_through = False
if not isinstance(self.allowed_subp, (list, bool)):
raise TypeError("self.allowed_subp supports list or bool.")
if isinstance(self.allowed_subp, bool):
pass_through = self.allowed_subp
else:
pass_through = (cmd in self.allowed_subp) or (
self.SUBP_SHELL_TRUE in self.allowed_subp
and kwargs.get("shell")
)
if pass_through:
return _real_subp(*args, **kwargs)
raise RuntimeError(
"called subp. set self.allowed_subp=True to allow\n subp(%s)"
% ", ".join(
[str(repr(a)) for a in args]
+ ["%s=%s" % (k, repr(v)) for k, v in kwargs.items()]
)
)

def tearDown(self):
if self.with_logs:
# Remove the handler we setup
logging.getLogger().handlers = self.old_handlers
logging.getLogger().setLevel(self.old_level)
subp.subp = _real_subp
super(CiTestCase, self).tearDown()

def tmp_dir(self, dir=None, cleanup=True):
# return a full path to a temporary directory that will be cleaned up.
if dir is None:
tmpd = tempfile.mkdtemp(prefix="ci-%s." % self.__class__.__name__)
else:
tmpd = tempfile.mkdtemp(dir=dir)
self.addCleanup(
functools.partial(shutil.rmtree, tmpd, ignore_errors=True)
)
return tmpd

def tmp_path(self, path, dir=None):
# return an absolute path to 'path' under dir.
# if dir is None, one will be created with tmp_dir()
# the file is not created or modified.
if dir is None:
dir = self.tmp_dir()
return os.path.normpath(os.path.abspath(os.path.join(dir, path)))

@classmethod
def random_string(cls, length=8):
return random_string(length)


def replicate_test_root(example_root, target_root):
real_root = resourceLocation()
real_root = os.path.join(real_root, "roots", example_root)
Expand Down
6 changes: 3 additions & 3 deletions tests/unittests/sources/test_init.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
redact_sensitive_keys,
)
from cloudinit.user_data import UserDataProcessor
from tests.unittests.helpers import CiTestCase, assert_count_equal, mock
from tests.unittests.helpers import assert_count_equal, mock


class DataSourceTestSubclassNet(DataSource):
Expand Down Expand Up @@ -933,7 +933,7 @@ def fake_get_data():
) in caplog.record_tuples


class TestRedactSensitiveData(CiTestCase):
class TestRedactSensitiveData:
def test_redact_sensitive_data_noop_when_no_sensitive_keys_present(self):
"""When sensitive_keys is absent or empty from metadata do nothing."""
md = {"my": "data"}
Expand Down Expand Up @@ -962,7 +962,7 @@ def test_redact_sensitive_data_does_redacts_with_default_string(self):
assert secure_md == redact_sensitive_keys(md)


class TestCanonicalCloudID(CiTestCase):
class TestCanonicalCloudID:
def test_cloud_id_returns_platform_on_unknowns(self):
"""When region and cloud_name are unknown, return platform."""
assert "platform" == canonical_cloud_id(
Expand Down
25 changes: 0 additions & 25 deletions tests/unittests/test_conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

from cloudinit import subp
from conftest import UnexpectedSubpError
from tests.unittests.helpers import CiTestCase


class TestDisableSubpUsage:
Expand Down Expand Up @@ -51,27 +50,3 @@ def test_subp_usage_can_be_conditionally_reenabled_for_multiple_cmds(self):
def test_both_marks_raise_an_error(self):
with pytest.raises(UnexpectedSubpError, match="marked both"):
subp.subp(["sh"])


class TestDisableSubpUsageInTestSubclass(CiTestCase):
"""Test that disable_subp_usage doesn't impact CiTestCase's subp logic.

Once the rest of the CiTestCase tests are removed, this class
should be removed as well.
"""

def test_using_subp_raises_exception(self):
with pytest.raises(Exception):
subp.subp(["some", "args"])

def test_typeerrors_on_incorrect_usage(self):
with pytest.raises(TypeError):
subp.subp()

def test_subp_usage_can_be_reenabled(self):
_old_allowed_subp = self.allow_subp
self.allowed_subp = True
try:
subp.subp(["sh", "-c", "true"])
finally:
self.allowed_subp = _old_allowed_subp