Skip to content

Commit

Permalink
[Monitor] Adjusts diagnostics tests (Azure#32001)
Browse files Browse the repository at this point in the history
  • Loading branch information
@pvaneck
pvaneck authored Sep 12, 2023
1 parent 0ab203d commit 3220dc2
Show file tree
Hide file tree
Showing 3 changed files with 137 additions and 288 deletions.
20 changes: 20 additions & 0 deletions sdk/monitor/azure-monitor-opentelemetry/tests/conftest.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
#-------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
#--------------------------------------------------------------------------
import os
from tempfile import mkstemp

import pytest


@pytest.fixture
def temp_file_path():
f, path = mkstemp()
os.close(f)
yield path
try:
os.unlink(path)
except:
pass
122 changes: 54 additions & 68 deletions sdk/monitor/azure-monitor-opentelemetry/tests/diagnostics/test_diagnostic_logging.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,17 +5,14 @@
# --------------------------------------------------------------------------

import logging
import os
from importlib import reload
from json import loads
from os.path import join
from pathlib import Path
from unittest import TestCase
from unittest.mock import patch

import azure.monitor.opentelemetry._diagnostics.diagnostic_logging as diagnostic_logger

TEST_LOGGER_PATH = str(Path.home())
TEST_DIAGNOSTIC_LOGGER_FILE_NAME_TEMPLATE = "test-applicationinsights-extension-%s.log"

TEST_SITE_NAME = "TEST_SITE_NAME"
TEST_CUSTOMER_IKEY = "TEST_CUSTOMER_IKEY"
TEST_EXTENSION_VERSION = "TEST_EXTENSION_VERSION"
Expand All @@ -31,24 +28,14 @@
TEST_LOGGER_SUB_MODULE = logging.getLogger(TEST_LOGGER_NAME_SUB_MODULE)


def get_test_file_name(test_id):
return TEST_DIAGNOSTIC_LOGGER_FILE_NAME_TEMPLATE % test_id

def get_test_file_location(test_id):
return join(
TEST_LOGGER_PATH, get_test_file_name(test_id)
)

def clear_file(test_id):
with open(get_test_file_location(test_id), "w") as f:
def clear_file(file_path):
with open(file_path, "w") as f:
f.seek(0)
f.truncate()


def check_file_for_messages(
level, messages, test_id, logger_name=TEST_LOGGER_NAME_SUB_MODULE
):
with open(get_test_file_location(test_id), "r") as f:
def check_file_for_messages(file_path, level, messages, logger_name=TEST_LOGGER_NAME_SUB_MODULE):
with open(file_path, "r") as f:
f.seek(0)
for message in messages:
json = loads(f.readline())
Expand All @@ -66,20 +53,18 @@ def check_file_for_messages(
assert not f.read()


def check_file_is_empty(test_id):
with open(get_test_file_location(test_id), "r") as f:
def check_file_is_empty(file_path):
with open(file_path, "r") as f:
f.seek(0)
assert not f.read()


def set_up(
file_path,
is_diagnostics_enabled,
test_id,
logger=TEST_LOGGER,
subscription_id_env_var=TEST_SUBSCRIPTION_ID_ENV_VAR,
) -> None:
clear_file(test_id)
check_file_is_empty(test_id)
diagnostic_logger._logger.handlers.clear()
logger.handlers.clear()
TEST_LOGGER.handlers.clear()
Expand All @@ -96,11 +81,11 @@ def set_up(
assert not diagnostic_logger.AzureDiagnosticLogging._initialized
patch(
"azure.monitor.opentelemetry._diagnostics.diagnostic_logging._DIAGNOSTIC_LOG_PATH",
TEST_LOGGER_PATH,
os.path.dirname(file_path),
).start()
patch(
"azure.monitor.opentelemetry._diagnostics.diagnostic_logging._DIAGNOSTIC_LOGGER_FILE_NAME",
get_test_file_name(test_id),
os.path.basename(file_path),
).start()
patch(
"azure.monitor.opentelemetry._diagnostics.diagnostic_logging._get_customer_ikey_from_env_var",
Expand All @@ -121,99 +106,100 @@ def set_up(
diagnostic_logger.AzureDiagnosticLogging.enable(logger)


class TestDiagnosticLogger(TestCase):
def test_initialized(self):
set_up(test_id=self.id(), is_diagnostics_enabled=True)
self.assertTrue(diagnostic_logger.AzureDiagnosticLogging._initialized)
class TestDiagnosticLogger:

def test_initialized(self, temp_file_path):
set_up(temp_file_path, is_diagnostics_enabled=True)
assert diagnostic_logger.AzureDiagnosticLogging._initialized is True

def test_uninitialized(self):
set_up(test_id=self.id(), is_diagnostics_enabled=False)
self.assertFalse(diagnostic_logger.AzureDiagnosticLogging._initialized)
def test_uninitialized(self, temp_file_path):
set_up(temp_file_path, is_diagnostics_enabled=False)
assert diagnostic_logger.AzureDiagnosticLogging._initialized is False

def test_info(self):
set_up(test_id=self.id(), is_diagnostics_enabled=True)
def test_info(self, temp_file_path):
set_up(temp_file_path, is_diagnostics_enabled=True)
TEST_LOGGER_SUB_MODULE.info(MESSAGE1)
TEST_LOGGER_SUB_MODULE.info(MESSAGE2)
check_file_is_empty(test_id=self.id())
check_file_is_empty(temp_file_path)

def test_info_with_info_log_level(self):
set_up(test_id=self.id(), is_diagnostics_enabled=True)
def test_info_with_info_log_level(self, temp_file_path):
set_up(temp_file_path, is_diagnostics_enabled=True)
TEST_LOGGER_SUB_MODULE.setLevel(logging.INFO)
TEST_LOGGER_SUB_MODULE.info(MESSAGE1)
TEST_LOGGER_SUB_MODULE.info(MESSAGE2)
TEST_LOGGER_SUB_MODULE.setLevel(logging.NOTSET)
check_file_for_messages("INFO", (MESSAGE1, MESSAGE2), test_id=self.id())
check_file_for_messages(temp_file_path, "INFO", (MESSAGE1, MESSAGE2))

def test_info_with_sub_module_info_log_level(self):
set_up(test_id=self.id(), is_diagnostics_enabled=True)
def test_info_with_sub_module_info_log_level(self, temp_file_path):
set_up(temp_file_path, is_diagnostics_enabled=True)
TEST_LOGGER_SUB_MODULE.setLevel(logging.INFO)
TEST_LOGGER_SUB_MODULE.info(MESSAGE1)
TEST_LOGGER_SUB_MODULE.info(MESSAGE2)
TEST_LOGGER_SUB_MODULE.setLevel(logging.NOTSET)
check_file_for_messages("INFO", (MESSAGE1, MESSAGE2), test_id=self.id())
check_file_for_messages(temp_file_path, "INFO", (MESSAGE1, MESSAGE2))

def test_warning(self):
set_up(test_id=self.id(), is_diagnostics_enabled=True)
def test_warning(self, temp_file_path):
set_up(temp_file_path, is_diagnostics_enabled=True)
TEST_LOGGER_SUB_MODULE.warning(MESSAGE1)
TEST_LOGGER_SUB_MODULE.warning(MESSAGE2)
check_file_for_messages("WARNING", (MESSAGE1, MESSAGE2), test_id=self.id())
check_file_for_messages(temp_file_path, "WARNING", (MESSAGE1, MESSAGE2))

def test_warning_multiple_enable(self):
set_up(test_id=self.id(), is_diagnostics_enabled=True)
def test_warning_multiple_enable(self, temp_file_path):
set_up(temp_file_path, is_diagnostics_enabled=True)
diagnostic_logger.AzureDiagnosticLogging.enable(TEST_LOGGER)
diagnostic_logger.AzureDiagnosticLogging.enable(TEST_LOGGER)
TEST_LOGGER_SUB_MODULE.warning(MESSAGE1)
TEST_LOGGER_SUB_MODULE.warning(MESSAGE2)
check_file_for_messages("WARNING", (MESSAGE1, MESSAGE2), test_id=self.id())
check_file_for_messages(temp_file_path, "WARNING", (MESSAGE1, MESSAGE2))

def test_error(self):
set_up(test_id=self.id(), is_diagnostics_enabled=True)
def test_error(self, temp_file_path):
set_up(temp_file_path, is_diagnostics_enabled=True)
TEST_LOGGER_SUB_MODULE.error(MESSAGE1)
TEST_LOGGER_SUB_MODULE.error(MESSAGE2)
check_file_for_messages("ERROR", (MESSAGE1, MESSAGE2), test_id=self.id())
check_file_for_messages(temp_file_path, "ERROR", (MESSAGE1, MESSAGE2))

def test_off_app_service_info(self):
set_up(test_id=self.id(), is_diagnostics_enabled=False)
def test_off_app_service_info(self, temp_file_path):
set_up(temp_file_path, is_diagnostics_enabled=False)
TEST_LOGGER.info(MESSAGE1)
TEST_LOGGER.info(MESSAGE2)
TEST_LOGGER_SUB_MODULE.info(MESSAGE1)
TEST_LOGGER_SUB_MODULE.info(MESSAGE2)
check_file_is_empty(test_id=self.id())
check_file_is_empty(temp_file_path)

def test_off_app_service_warning(self):
set_up(test_id=self.id(), is_diagnostics_enabled=False)
def test_off_app_service_warning(self, temp_file_path):
set_up(temp_file_path, is_diagnostics_enabled=False)
TEST_LOGGER.warning(MESSAGE1)
TEST_LOGGER.warning(MESSAGE2)
TEST_LOGGER_SUB_MODULE.warning(MESSAGE1)
TEST_LOGGER_SUB_MODULE.warning(MESSAGE2)
check_file_is_empty(test_id=self.id())
check_file_is_empty(temp_file_path)

def test_off_app_service_error(self):
set_up(test_id=self.id(), is_diagnostics_enabled=False)
def test_off_app_service_error(self, temp_file_path):
set_up(temp_file_path, is_diagnostics_enabled=False)
TEST_LOGGER.error(MESSAGE1)
TEST_LOGGER.error(MESSAGE2)
TEST_LOGGER_SUB_MODULE.error(MESSAGE1)
TEST_LOGGER_SUB_MODULE.error(MESSAGE2)
check_file_is_empty(test_id=self.id())
check_file_is_empty(temp_file_path)

def test_subscription_id_plus(self):
def test_subscription_id_plus(self, temp_file_path):
set_up(
test_id=self.id(),
temp_file_path,
is_diagnostics_enabled=True,
subscription_id_env_var=TEST_SUBSCRIPTION_ID_ENV_VAR,
)
self.assertEqual(diagnostic_logger._SUBSCRIPTION_ID, TEST_SUBSCRIPTION_ID)
assert diagnostic_logger._SUBSCRIPTION_ID == TEST_SUBSCRIPTION_ID
TEST_LOGGER_SUB_MODULE.warning(MESSAGE1)
TEST_LOGGER_SUB_MODULE.warning(MESSAGE2)
check_file_for_messages("WARNING", (MESSAGE1, MESSAGE2), test_id=self.id())
check_file_for_messages(temp_file_path, "WARNING", (MESSAGE1, MESSAGE2))

def test_subscription_id_no_plus(self):
def test_subscription_id_no_plus(self, temp_file_path):
set_up(
test_id=self.id(),
temp_file_path,
is_diagnostics_enabled=True,
subscription_id_env_var=TEST_SUBSCRIPTION_ID,
)
self.assertEqual(diagnostic_logger._SUBSCRIPTION_ID, TEST_SUBSCRIPTION_ID)
assert diagnostic_logger._SUBSCRIPTION_ID == TEST_SUBSCRIPTION_ID
TEST_LOGGER_SUB_MODULE.warning(MESSAGE1)
TEST_LOGGER_SUB_MODULE.warning(MESSAGE2)
check_file_for_messages("WARNING", (MESSAGE1, MESSAGE2), test_id=self.id())
check_file_for_messages(temp_file_path, "WARNING", (MESSAGE1, MESSAGE2))
Loading

0 comments on commit 3220dc2

Please sign in to comment.