Dans ce défi, vous appliquerez votre compréhension des concepts de **threading** en Python pour simuler le scénario d'un café très fréquenté. Votre tâche consiste à écrire un programme qui prépare plusieurs commandes de café simultanément à l'aide de threads. Vous exploiterez l'**exécution parallèle**, en vous référant aux notions de **gestion des threads** abordées dans le chapitre précédent. Ce défi vous aidera à consolider votre maîtrise de l'utilisation du **multithreading** pour gérer plusieurs tâches en même temps et à assurer une bonne coordination grâce à la jonction des threads.

import unittest
import user_code
import importlib
import io
import sys
import time

class TestTask(unittest.TestCase):
    def test_brew_coffee_prints_message(self):
        importlib.reload(user_code)
        f = io.StringIO()
        sys_stdout = sys.stdout
        sys.stdout = f
        try:
            user_code.brew_coffee("Mocha")
        finally:
            sys.stdout = sys_stdout
        output = normalize_text(f.getvalue())
        expected = normalize_text("Mocha is ready!")
        _dynamic_test(
            self,
            expected in output,
            "brew_coffee prints correct ready message",
            f"Expected output '{expected}' not found in '{output}'",
        )

    def test_brew_all_coffees_starts_threads_and_joins(self):
        importlib.reload(user_code)
        orders = ["Americano", "Flat White", "Macchiato"]
        f = io.StringIO()
        sys_stdout = sys.stdout
        sys.stdout = f
        start = time.time()
        try:
            user_code.brew_all_coffees(orders)
        finally:
            sys.stdout = sys_stdout
        elapsed = time.time() - start
        output = normalize_text(f.getvalue())
        for order in orders:
            expected = normalize_text(f"{order} is ready!")
            _dynamic_test(
                self,
                expected in output,
                f"Output contains ready message for {order}",
                f"Expected ready message for '{order}' not found in output: '{output}'",
            )
        _dynamic_test(
            self,
            elapsed >= 0.2 and elapsed < 0.5,
            "brew_all_coffees runs in parallel (not sequentially)",
            f"brew_all_coffees did not run in parallel. Elapsed: {elapsed}",
        )

    def test_brew_all_coffees_handles_empty_orders(self):
        importlib.reload(user_code)
        orders = []
        f = io.StringIO()
        sys_stdout = sys.stdout
        sys.stdout = f
        try:
            user_code.brew_all_coffees(orders)
        finally:
            sys.stdout = sys_stdout
        output = f.getvalue().strip()
        _dynamic_test(
            self,
            output == "",
            "No output when orders is empty",
            f"Expected no output, got: '{output}'",
        )

    def test_brew_coffee_sleep_time(self):
        importlib.reload(user_code)
        start = time.time()
        user_code.brew_coffee("TestCoffee")
        elapsed = time.time() - start
        _dynamic_test(
            self,
            elapsed >= 0.19,
            "brew_coffee sleeps for at least 0.2 seconds",
            f"brew_coffee did not sleep long enough. Elapsed: {elapsed}",
        )

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):
    import re
    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:
    import ast
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    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 not assign_nodes:
        return code
    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


test_main.py

Vous acquerrez des compétences pratiques dans les modules et les imports de Python pour une gestion efficace du code. Maîtrisez des techniques avancées de gestion des erreurs pour des applications plus fiables, automatisez les opérations de fichiers avec la gestion des fichiers, et apprenez des stratégies de test robustes avec Pytest et Unittest pour assurer l'intégrité de votre code.

Master Python's exception system for robust error handling and debugging.

Explore advanced techniques for file operations and resource management using context managers.

Compare and apply Python's multithreading and multiprocessing modules for concurrent programming.

Master asynchronous programming in Python using the asyncio library for scalable I/O-bound applications.

Explore how to write, organize, and run tests using Python's unittest and pytest frameworks.

Défi : Traitement Parallèle de Fichiers

Solution