When running several asynchronous tasks in Python using `asyncio`, it is common for some tasks to succeed while others might raise exceptions. If you do not handle these exceptions properly, a single error can interrupt the flow of your program or cause you to miss important information about what went wrong. By logging errors as they occur, you can ensure that all tasks are given a chance to complete, and you can review any issues afterward. This approach is especially important in production systems, where reliability and clear error reporting are critical.


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

class TestTask(unittest.TestCase):
    def test_all_tasks_run_and_errors_logged(self):
        import user_code
        importlib.reload(user_code)
        
        f = io.StringIO()
        with redirect_stdout(f):
            importlib.reload(user_code)
        output = normalize_text(f.getvalue())
        # Check that errors from B and D are logged
        expected_b = normalize_text("Error in B: Task B failed with an error.")
        expected_d = normalize_text("Error in D: Task D failed with an error.")
        _dynamic_test(
            self,
            expected_b in output,
            "Error for task B is logged correctly.",
            f"Expected error log for B not found in output: {output}",
        )
        _dynamic_test(
            self,
            expected_d in output,
            "Error for task D is logged correctly.",
            f"Expected error log for D not found in output: {output}",
        )
        # Ensure no error logs for tasks A and C
        not_expected_a = normalize_text("Error in A")
        not_expected_c = normalize_text("Error in C")
        _dynamic_test(
            self,
            not_expected_a not in output,
            "No error log for successful task A.",
            f"Unexpected error log for A found in output: {output}",
        )
        _dynamic_test(
            self,
            not_expected_c not in output,
            "No error log for successful task C.",
            f"Unexpected error log for C found in output: {output}",
        )

    def test_tasks_are_awaited_individually(self):
        # This checks that exception in one task does not prevent others from running (all errors logged)
        import user_code
        importlib.reload(user_code)
        f = io.StringIO()
        with redirect_stdout(f):
            importlib.reload(user_code)
        output = normalize_text(f.getvalue())
        expected_b = normalize_text("Error in B: Task B failed with an error.")
        expected_d = normalize_text("Error in D: Task D failed with an error.")
        _dynamic_test(
            self,
            expected_b in output and expected_d in output,
            "All errors are logged, indicating each task is awaited individually.",
            f"Not all error logs present. Output: {output}",
        )

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()
    for i, node in enumerate(tree.body):
        if isinstance(node, ast.Assign):
            for target in node.targets:
                if isinstance(target, ast.Name) and target.id == var_name:
                    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] = []
                    
                    return '\\n'.join(lines)
    return code

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


test_main.py

A beginner-friendly course introducing Python's asyncio library, focusing on asynchronous programming concepts, practical usage, and hands-on challenges with real-world scenarios.

Introduction to asynchronous programming and the basics of Python's asyncio library.

Delve into creating, running, and managing multiple asyncio tasks concurrently.

Explore more advanced asyncio patterns such as task cancellation and error handling.

Challenge: Async Error Logger

Solução

Challenge: Async Error Logger

Solução