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
|
diff --git a/libcloud/__init__.py b/libcloud/__init__.py
index 39e5507bc1..1f4e9c22e6 100644
--- a/libcloud/__init__.py
+++ b/libcloud/__init__.py
@@ -65,6 +65,19 @@ def close_file(fd):
atexit.register(close_file, fo)
+def reset_debug():
+ """
+ Reset debugging functionality (if set).
+
+ NOTE: This function is only meant to be used in the tests.
+ """
+ from libcloud.common.base import Connection, LibcloudConnection
+ from libcloud.utils.loggingconnection import LoggingConnection
+
+ LoggingConnection.log = None
+ Connection.conn_class = LibcloudConnection
+
+
def _init_once():
"""
Utility function that is ran once on Library import.
diff --git a/libcloud/test/test_init.py b/libcloud/test/test_init.py
index 418dcdf4ca..2d63ca29dc 100644
--- a/libcloud/test/test_init.py
+++ b/libcloud/test/test_init.py
@@ -17,12 +17,14 @@
import sys
import logging
import tempfile
+from unittest import mock
from unittest.mock import patch
import libcloud
-from libcloud import _init_once
+from libcloud import _init_once, reset_debug
from libcloud.base import DriverTypeNotFoundError
from libcloud.test import unittest
+from libcloud.common.base import LibcloudConnection, Connection
from libcloud.utils.loggingconnection import LoggingConnection
try:
@@ -32,32 +34,46 @@
except ImportError:
have_paramiko = False
+_, TEMP_LOGFILE_PATH = tempfile.mkstemp()
+
class TestUtils(unittest.TestCase):
- def tearDown(self):
- if "LIBCLOUD_DEBUG" in os.environ:
- del os.environ["LIBCLOUD_DEBUG"]
+ def setUp(self):
+ reset_debug()
- def test_init_once_and_debug_mode(self):
+ @mock.patch.dict(os.environ, {"LIBCLOUD_DEBUG": ""}, clear=True)
+ def test_init_once_and_no_debug_mode(self):
if have_paramiko:
paramiko_logger = logging.getLogger("paramiko")
paramiko_logger.setLevel(logging.INFO)
+ self.assertIsNone(LoggingConnection.log)
+ self.assertEqual(Connection.conn_class, LibcloudConnection)
+
# Debug mode is disabled
_init_once()
self.assertIsNone(LoggingConnection.log)
+ self.assertEqual(Connection.conn_class, LibcloudConnection)
if have_paramiko:
paramiko_log_level = paramiko_logger.getEffectiveLevel()
self.assertEqual(paramiko_log_level, logging.INFO)
- # Enable debug mode
- _, tmp_path = tempfile.mkstemp()
- os.environ["LIBCLOUD_DEBUG"] = tmp_path
+ @mock.patch.dict(os.environ, {"LIBCLOUD_DEBUG": TEMP_LOGFILE_PATH}, clear=True)
+ def test_init_once_and_debug_mode(self):
+ if have_paramiko:
+ paramiko_logger = logging.getLogger("paramiko")
+ paramiko_logger.setLevel(logging.INFO)
+
+ self.assertIsNone(LoggingConnection.log)
+ self.assertEqual(Connection.conn_class, LibcloudConnection)
+
+ # Debug mode is enabled
_init_once()
self.assertTrue(LoggingConnection.log is not None)
+ self.assertEqual(Connection.conn_class, LoggingConnection)
if have_paramiko:
paramiko_log_level = paramiko_logger.getEffectiveLevel()
|
