In DevOps, being able to quickly identify errors in log files is essential for diagnosing issues and maintaining system reliability. Log files can be lengthy and filled with a variety of information, but your ability to extract and count error messages efficiently can save valuable troubleshooting time. In this challenge, you will write a Python script that processes a hardcoded list of log entries, extracts all lines containing the word `ERROR`, prints each error line, and counts the total number of errors found. This exercise simulates a simple log monitoring tool, a fundamental skill for any DevOps professional.


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import importlib
import io
from contextlib import redirect_stdout

class TestTask(unittest.TestCase):
    def test_prints_only_error_lines_and_count(self):
        import user_code
        importlib.reload(user_code)
        logs = [
            "2024-06-01 12:00:00 INFO Starting backup process",
            "2024-06-01 12:00:01 ERROR Failed to connect to database",
            "2024-06-01 12:00:02 INFO Backup in progress",
            "2024-06-01 12:00:03 ERROR Disk space low",
            "2024-06-01 12:00:04 WARNING Backup delayed",
            "2024-06-01 12:00:05 INFO Backup complete"
        ]
        f = io.StringIO()
        with redirect_stdout(f):
            user_code.extract_errors(logs)
        output = f.getvalue().strip().split('\\n')
        expected_errors = [
            "2024-06-01 12:00:01 ERROR Failed to connect to database",
            "2024-06-01 12:00:03 ERROR Disk space low"
        ]
        expected_count = "2"
        _dynamic_test(
            self,
            all(line in output for line in expected_errors),
            "All error lines are printed",
            f"Expected error lines {expected_errors} in output {output}"
        )
        _dynamic_test(
            self,
            output[-1] == expected_count,
            "Correct total error count printed",
            f"Expected error count '{expected_count}', got '{output[-1]}'"
        )
        non_error_lines = [line for line in logs if "ERROR" not in line]
        _dynamic_test(
            self,
            all(line not in output for line in non_error_lines),
            "No non-error lines are printed",
            f"Non-error lines found in output: {output}"
        )

    def test_no_error_lines(self):
        import user_code
        importlib.reload(user_code)
        logs = [
            "2024-06-01 12:00:00 INFO Starting backup process",
            "2024-06-01 12:00:02 INFO Backup in progress",
            "2024-06-01 12:00:04 WARNING Backup delayed",
            "2024-06-01 12:00:05 INFO Backup complete"
        ]
        f = io.StringIO()
        with redirect_stdout(f):
            user_code.extract_errors(logs)
        output = f.getvalue().strip().split('\\n')
        _dynamic_test(
            self,
            output == ["0"] or output == [''],
            "Prints only 0 when no error lines",
            f"Expected ['0'], got {output}"
        )

    def test_all_error_lines(self):
        import user_code
        importlib.reload(user_code)
        logs = [
            "2024-06-01 12:00:01 ERROR Failed to connect to database",
            "2024-06-01 12:00:03 ERROR Disk space low"
        ]
        f = io.StringIO()
        with redirect_stdout(f):
            user_code.extract_errors(logs)
        output = f.getvalue().strip().split('\\n')
        _dynamic_test(
            self,
            output[:-1] == logs,
            "All error lines printed in order",
            f"Expected {logs}, got {output[:-1]}"
        )
        _dynamic_test(
            self,
            output[-1] == "2",
            "Correct count when all lines are errors",
            f"Expected count '2', got '{output[-1]}'"
        )

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    # Collect all assignment nodes to modify
    assign_nodes = [
        (i, node)
        for i, node in enumerate(tree.body)
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets)
    ]

    # If nothing to change, return unmodified code
    if not assign_nodes:
        return code

    # Perform replacements for all matching assignments (from last to first to not break line offsets)
    for i, node in reversed(assign_nodes):
        start_line = node.lineno - 1
        line = lines[start_line]
        indent = ' ' * (len(line) - len(line.lstrip()))
        lines[start_line] = f"{indent}{var_name} = {value}"
        next_line = len(lines)
        for next_node in tree.body[i+1:]:
            if hasattr(next_node, 'lineno'):
                next_line = next_node.lineno - 1
                break
        if next_line > start_line + 1:
            lines[start_line+1:next_line] = []
        changed = True

    return '\\n'.join(lines) if changed else code

if __name__ == "__main__":
    unittest.main()


test_main.py

Learn how Python empowers DevOps workflows, automates repetitive tasks, and streamlines infrastructure management. This course introduces practical Python techniques and libraries for DevOps beginners, focusing on automation, monitoring, and data-driven decision-making.

Explore how Python can automate repetitive DevOps tasks, from simple scripting to basic process management.

Learn how Python can be used to monitor system health and analyze logs, providing valuable insights for DevOps teams.

Apply Python's data analysis and visualization capabilities to make informed decisions in DevOps environments.

Challenge: Log Error Extractor

Solution

Challenge: Log Error Extractor

Solution