I denna utmaning kommer du att tillämpa din förståelse av Pythons **trådhantering** för att simulera ett hektiskt kaféscenario. Din uppgift är att skriva ett program som brygger flera kaffeordrar samtidigt med hjälp av trådar. Du kommer att använda **parallell exekvering** och hänvisa till begreppen kring **trådhantering** som du lärde dig i föregående kapitel. Denna utmaning hjälper dig att befästa din förståelse för att använda **multitrådning** för att hantera flera uppgifter samtidigt och säkerställa korrekt samordning med hjälp av trådsammanfogning.

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

Du kommer att förvärva praktiska färdigheter i Pythons moduler och importer för effektiv kodhantering. Behärska avancerade tekniker för felhantering för mer tillförlitliga applikationer, automatisera filoperationer med filhantering och lär dig robusta teststrategier med Pytest och Unittest för att säkerställa kodens integritet.

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.

Utmaning: Parallell Filbearbetning

Lösning