summaryrefslogtreecommitdiff
path: root/tests/common
diff options
context:
space:
mode:
Diffstat (limited to 'tests/common')
-rw-r--r--tests/common/osutil/test_default.py127
-rw-r--r--tests/common/test_conf.py7
-rw-r--r--tests/common/test_errorstate.py69
-rw-r--r--tests/common/test_event.py15
-rw-r--r--tests/common/test_logger.py40
5 files changed, 245 insertions, 13 deletions
diff --git a/tests/common/osutil/test_default.py b/tests/common/osutil/test_default.py
index 08125ae..c9fa1de 100644
--- a/tests/common/osutil/test_default.py
+++ b/tests/common/osutil/test_default.py
@@ -489,6 +489,62 @@ Match host 192.168.1.2\n\
print("WRITING TO {0}".format(waagent_sudoers))
self.assertEqual(1, count)
+ def test_get_firewall_dropped_packets_returns_zero_if_firewall_disabled(self):
+ osutil._enable_firewall = False
+ util = osutil.DefaultOSUtil()
+
+ self.assertEqual(0, util.get_firewall_dropped_packets("not used"))
+
+ @patch('azurelinuxagent.common.utils.shellutil.run_get_output')
+ def test_get_firewall_dropped_packets_returns_negative_if_error(self, mock_output):
+ osutil._enable_firewall = True
+ util = osutil.DefaultOSUtil()
+
+ mock_output.side_effect = [
+ (0, "iptables v{0}".format(osutil.IPTABLES_LOCKING_VERSION)),
+ (1, "not used")]
+ self.assertEqual(-1, util.get_firewall_dropped_packets("not used"))
+
+ @patch('azurelinuxagent.common.utils.shellutil.run_get_output')
+ def test_get_firewall_dropped_packets_returns_negative_if_exception(self, mock_output):
+ osutil._enable_firewall = True
+ util = osutil.DefaultOSUtil()
+
+ mock_output.side_effect = [
+ (0, "iptables v{0}".format(osutil.IPTABLES_LOCKING_VERSION)),
+ (1, Exception)]
+ self.assertEqual(-1, util.get_firewall_dropped_packets("not used"))
+
+ @patch('azurelinuxagent.common.utils.shellutil.run_get_output')
+ def test_get_firewall_dropped_packets_transient_error_ignored(self, mock_output):
+ osutil._enable_firewall = True
+ util = osutil.DefaultOSUtil()
+
+ mock_output.side_effect = [
+ (0, "iptables v{0}".format(osutil.IPTABLES_LOCKING_VERSION)),
+ (3, "can't initialize iptables table `security': iptables who? (do you need to insmod?)")]
+ self.assertEqual(0, util.get_firewall_dropped_packets("not used"))
+
+ @patch('azurelinuxagent.common.utils.shellutil.run_get_output')
+ def test_get_firewall_dropped_packets(self, mock_output):
+ osutil._enable_firewall = True
+ util = osutil.DefaultOSUtil()
+
+ mock_output.side_effect = [
+ (0, "iptables v{0}".format(osutil.IPTABLES_LOCKING_VERSION)),
+ (0,
+'''
+
+Chain OUTPUT (policy ACCEPT 104 packets, 43628 bytes)
+ pkts bytes target prot opt in out source destination
+ 0 0 ACCEPT tcp -- any any anywhere 168.63.129.16 owner UID match daemon
+ 32 1920 DROP tcp -- any any anywhere 168.63.129.16
+
+''')]
+ dst = '168.63.129.16'
+
+ self.assertEqual(32, util.get_firewall_dropped_packets(dst))
+
@patch('os.getuid', return_value=42)
@patch('azurelinuxagent.common.utils.shellutil.run_get_output')
@patch('azurelinuxagent.common.utils.shellutil.run')
@@ -592,6 +648,34 @@ Match host 192.168.1.2\n\
])
self.assertFalse(osutil._enable_firewall)
+ @patch('azurelinuxagent.common.utils.shellutil.run_get_output')
+ @patch('azurelinuxagent.common.utils.shellutil.run')
+ def test_enable_firewall_checks_for_invalid_iptables_options(self, mock_run, mock_output):
+ osutil._enable_firewall = True
+ util = osutil.DefaultOSUtil()
+
+ dst = '1.2.3.4'
+ version = "iptables v{0}".format(osutil.IPTABLES_LOCKING_VERSION)
+ wait = "-w"
+
+ # iptables uses the following exit codes
+ # 0 - correct function
+ # 1 - other errors
+ # 2 - errors which appear to be caused by invalid or abused command
+ # line parameters
+ mock_run.side_effect = [2]
+ mock_output.return_value = (0, version)
+
+ self.assertFalse(util.enable_firewall(dst_ip='1.2.3.4', uid=42))
+ self.assertFalse(osutil._enable_firewall)
+
+ mock_run.assert_has_calls([
+ call(osutil.FIREWALL_DROP.format(wait, "C", dst), chk_err=False),
+ ])
+ mock_output.assert_has_calls([
+ call(osutil.IPTABLES_VERSION)
+ ])
+
@patch('os.getuid', return_value=42)
@patch('azurelinuxagent.common.utils.shellutil.run_get_output')
@patch('azurelinuxagent.common.utils.shellutil.run')
@@ -624,17 +708,54 @@ Match host 192.168.1.2\n\
version = "iptables v{0}".format(osutil.IPTABLES_LOCKING_VERSION)
wait = "-w"
- mock_run.side_effect = [0, 0]
+ mock_run.side_effect = [0, 1, 0, 1]
mock_output.side_effect = [(0, version), (0, "Output")]
- self.assertTrue(util.remove_firewall())
+ self.assertTrue(util.remove_firewall(dst, uid))
mock_run.assert_has_calls([
- call(osutil.FIREWALL_FLUSH.format(wait), chk_err=False)
+ call(osutil.FIREWALL_DELETE_CONNTRACK.format(wait, dst), chk_err=False),
+ call(osutil.FIREWALL_DELETE_CONNTRACK.format(wait, dst), chk_err=False),
+ call(osutil.FIREWALL_DELETE_OWNER.format(wait, dst, uid), chk_err=False),
+ call(osutil.FIREWALL_DELETE_OWNER.format(wait, dst, uid), chk_err=False),
])
mock_output.assert_has_calls([
call(osutil.IPTABLES_VERSION)
])
self.assertTrue(osutil._enable_firewall)
+ @patch('os.getuid', return_value=42)
+ @patch('azurelinuxagent.common.utils.shellutil.run_get_output')
+ @patch('azurelinuxagent.common.utils.shellutil.run')
+ def test_remove_firewall_does_not_repeat(self, mock_run, mock_output, _):
+ osutil._enable_firewall = True
+ util = osutil.DefaultOSUtil()
+
+ dst_ip='1.2.3.4'
+ uid=42
+ version = "iptables v{0}".format(osutil.IPTABLES_LOCKING_VERSION)
+ wait = "-w"
+
+ mock_run.side_effect = [2]
+ mock_output.side_effect = [(0, version), (1, "Output")]
+ self.assertFalse(util.remove_firewall(dst_ip, uid))
+
+ mock_run.assert_has_calls([
+ call(osutil.FIREWALL_DELETE_CONNTRACK.format(wait, dst_ip), chk_err=False),
+ ])
+ mock_output.assert_has_calls([
+ call(osutil.IPTABLES_VERSION)
+ ])
+ self.assertFalse(osutil._enable_firewall)
+
+ self.assertTrue(mock_run.call_count == 1)
+ self.assertTrue(mock_output.call_count == 1)
+
+ self.assertFalse(util.remove_firewall())
+ self.assertFalse(util.remove_firewall())
+
+ self.assertTrue(mock_run.call_count == 1)
+ self.assertTrue(mock_output.call_count == 1)
+
+
if __name__ == '__main__':
unittest.main()
diff --git a/tests/common/test_conf.py b/tests/common/test_conf.py
index 93759de..057c83b 100644
--- a/tests/common/test_conf.py
+++ b/tests/common/test_conf.py
@@ -48,6 +48,7 @@ class TestConf(AgentTestCase):
"OS.EnableFIPS" : True,
"OS.RootDeviceScsiTimeout" : '300',
"OS.OpensslPath" : '/usr/bin/openssl',
+ "OS.SshClientAliveInterval" : 42,
"OS.SshDir" : "/notareal/path",
"HttpProxy.Host" : None,
"HttpProxy.Port" : None,
@@ -64,7 +65,7 @@ class TestConf(AgentTestCase):
"AutoUpdate.GAFamily" : "Prod",
"EnableOverProvisioning" : False,
"OS.AllowHTTP" : False,
- "OS.EnableFirewall" : True
+ "OS.EnableFirewall" : False
}
def setUp(self):
@@ -77,6 +78,7 @@ class TestConf(AgentTestCase):
def test_key_value_handling(self):
self.assertEqual("Value1", self.conf.get("FauxKey1", "Bad"))
self.assertEqual("Value2 Value2", self.conf.get("FauxKey2", "Bad"))
+ self.assertEqual("delalloc,rw,noatime,nobarrier,users,mode=777", self.conf.get("FauxKey3", "Bad"))
def test_get_ssh_dir(self):
self.assertTrue(get_ssh_dir(self.conf).startswith("/notareal/path"))
@@ -109,4 +111,5 @@ class TestConf(AgentTestCase):
for k in TestConf.EXPECTED_CONFIGURATION.keys():
self.assertEqual(
TestConf.EXPECTED_CONFIGURATION[k],
- configuration[k])
+ configuration[k],
+ k)
diff --git a/tests/common/test_errorstate.py b/tests/common/test_errorstate.py
new file mode 100644
index 0000000..7513fe5
--- /dev/null
+++ b/tests/common/test_errorstate.py
@@ -0,0 +1,69 @@
+from datetime import timedelta
+
+from azurelinuxagent.common.errorstate import *
+from tests.tools import *
+
+
+class TestErrorState(unittest.TestCase):
+ def test_errorstate00(self):
+ """
+ If ErrorState is never incremented, it will never trigger.
+ """
+ test_subject = ErrorState(timedelta(seconds=10000))
+ self.assertFalse(test_subject.is_triggered())
+ self.assertEqual(0, test_subject.count)
+
+ def test_errorstate01(self):
+ """
+ If ErrorState is never incremented, and the timedelta is zero it will
+ not trigger.
+ """
+ test_subject = ErrorState(timedelta(seconds=0))
+ self.assertFalse(test_subject.is_triggered())
+ self.assertEqual(0, test_subject.count)
+
+ def test_errorstate02(self):
+ """
+ If ErrorState is triggered, and the current time is within timedelta
+ of now it will trigger.
+ """
+ test_subject = ErrorState(timedelta(seconds=0))
+ test_subject.incr()
+
+
+ self.assertTrue(test_subject.is_triggered())
+ self.assertEqual(1, test_subject.count)
+
+ @patch('azurelinuxagent.common.errorstate.datetime')
+ def test_errorstate03(self, mock_time):
+ """
+ ErrorState will not trigger until
+ 1. ErrorState has been incr() at least once.
+ 2. The timedelta from the first incr() has elapsed.
+ """
+ test_subject = ErrorState(timedelta(minutes=15))
+
+ for x in range(1, 10):
+ mock_time.utcnow = Mock(return_value=datetime.utcnow() + timedelta(minutes=x))
+
+ test_subject.incr()
+ self.assertFalse(test_subject.is_triggered())
+
+ mock_time.utcnow = Mock(return_value=datetime.utcnow() + timedelta(minutes=30))
+ test_subject.incr()
+ self.assertTrue(test_subject.is_triggered())
+
+ def test_errorstate04(self):
+ """
+ If ErrorState is reset the timestamp of the last incr() is reset to
+ None.
+ """
+
+ test_subject = ErrorState(timedelta(minutes=15))
+ self.assertTrue(test_subject.timestamp is None)
+
+ test_subject.incr()
+ self.assertTrue(test_subject.timestamp is not None)
+
+ test_subject.reset()
+ self.assertTrue(test_subject.timestamp is None)
diff --git a/tests/common/test_event.py b/tests/common/test_event.py
index 55a99c4..01bcd7b 100644
--- a/tests/common/test_event.py
+++ b/tests/common/test_event.py
@@ -23,7 +23,8 @@ import azurelinuxagent.common.event as event
import azurelinuxagent.common.logger as logger
from azurelinuxagent.common.event import add_event, \
- mark_event_status, should_emit_event
+ mark_event_status, should_emit_event, \
+ WALAEventOperation
from azurelinuxagent.common.future import ustr
from azurelinuxagent.common.version import CURRENT_VERSION
@@ -48,8 +49,7 @@ class TestEvent(AgentTestCase):
self.assertTrue(es.event_succeeded("Foo", "1.2", "FauxOperation"))
def test_event_status_records_status(self):
- d = tempfile.mkdtemp()
- es = event.EventStatus(tempfile.mkdtemp())
+ es = event.EventStatus()
es.mark_event_status("Foo", "1.2", "FauxOperation", True)
self.assertTrue(es.event_succeeded("Foo", "1.2", "FauxOperation"))
@@ -69,7 +69,7 @@ class TestEvent(AgentTestCase):
self.assertFalse(es.event_succeeded("Foo", "1.2", "FauxOperation"))
def test_should_emit_event_ignores_unknown_operations(self):
- event.__event_status__ = event.EventStatus(tempfile.mkdtemp())
+ event.__event_status__ = event.EventStatus()
self.assertTrue(event.should_emit_event("Foo", "1.2", "FauxOperation", True))
self.assertTrue(event.should_emit_event("Foo", "1.2", "FauxOperation", False))
@@ -82,7 +82,7 @@ class TestEvent(AgentTestCase):
def test_should_emit_event_handles_known_operations(self):
- event.__event_status__ = event.EventStatus(tempfile.mkdtemp())
+ event.__event_status__ = event.EventStatus()
# Known operations always initially "fire"
for op in event.__event_status_operations__:
@@ -144,7 +144,7 @@ class TestEvent(AgentTestCase):
event.add_periodic(logger.EVERY_DAY, "FauxEvent")
self.assertEqual(1, mock_event.call_count)
- h = hash("FauxEvent"+""+ustr(True)+"")
+ h = hash("FauxEvent"+WALAEventOperation.Unknown+ustr(True))
event.__event_logger__.periodic_events[h] = \
datetime.now() - logger.EVERY_DAY - logger.EVERY_HOUR
event.add_periodic(logger.EVERY_DAY, "FauxEvent")
@@ -158,7 +158,8 @@ class TestEvent(AgentTestCase):
mock_event.assert_called_once_with(
"FauxEvent",
duration=0, evt_type='', is_internal=False, is_success=True,
- log_event=True, message='', op='', version=str(CURRENT_VERSION))
+ log_event=True, message='', op=WALAEventOperation.Unknown,
+ version=str(CURRENT_VERSION))
def test_save_event(self):
add_event('test', message='test event')
diff --git a/tests/common/test_logger.py b/tests/common/test_logger.py
index 9e298b3..005c429 100644
--- a/tests/common/test_logger.py
+++ b/tests/common/test_logger.py
@@ -15,17 +15,20 @@
# Requires Python 2.4+ and Openssl 1.0+
#
+import json
from datetime import datetime
import azurelinuxagent.common.logger as logger
+from azurelinuxagent.common.event import add_log_event
+from azurelinuxagent.common.version import CURRENT_AGENT, CURRENT_VERSION
from tests.tools import *
_MSG = "This is our test logging message {0} {1}"
_DATA = ["arg1", "arg2"]
-class TestLogger(AgentTestCase):
+class TestLogger(AgentTestCase):
@patch('azurelinuxagent.common.logger.Logger.info')
def test_periodic_emits_if_not_previously_sent(self, mock_info):
logger.reset_periodic()
@@ -64,3 +67,38 @@ class TestLogger(AgentTestCase):
logger.periodic(logger.EVERY_DAY, _MSG, *_DATA)
mock_info.assert_called_once_with(_MSG, *_DATA)
+
+ def test_telemetry_logger(self):
+ mock = MagicMock()
+ appender = logger.TelemetryAppender(logger.LogLevel.WARNING, mock)
+ appender.write(logger.LogLevel.WARNING, "--unit-test--")
+
+ mock.assert_called_once_with(logger.LogLevel.WARNING, "--unit-test--")
+
+ @patch('azurelinuxagent.common.event.EventLogger.save_event')
+ def test_telemetry_logger1(self, mock_save):
+ appender = logger.TelemetryAppender(logger.LogLevel.WARNING, add_log_event)
+ appender.write(logger.LogLevel.WARNING, "--unit-test--")
+
+ self.assertEqual(1, mock_save.call_count)
+ telemetry_json = json.loads(mock_save.call_args[0][0])
+
+ self.assertEqual('FFF0196F-EE4C-4EAF-9AA5-776F622DEB4F', telemetry_json['providerId'])
+ self.assertEqual(7, telemetry_json['eventId'])
+
+ self.assertEqual(5, len(telemetry_json['parameters']))
+ for x in telemetry_json['parameters']:
+ if x['name'] == 'EventName':
+ self.assertEqual(x['value'], 'Log')
+
+ elif x['name'] == 'CapabilityUsed':
+ self.assertEqual(x['value'], 'WARNING')
+
+ elif x['name'] == 'Context1':
+ self.assertEqual(x['value'], '--unit-test--')
+
+ elif x['name'] == 'Context2':
+ self.assertEqual(x['value'], '')
+
+ elif x['name'] == 'Context3':
+ self.assertEqual(x['value'], '')
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-10 01:13:43 +0000