このチャレンジでは、Pythonの**スレッド**の概念を活用して、忙しいコーヒーショップのシナリオをシミュレートします。複数のコーヒー注文をスレッドを使って同時に抽出するプログラムを作成してください。前章で学んだ**スレッド管理**の概念を参照しながら、**並列実行**を活用します。このチャレンジを通じて、**マルチスレッド**を用いて複数のタスクを同時に処理し、スレッドの結合による適切な調整を行う方法を確実に理解できます。

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

Pythonのモジュールとインポートによる効率的なコード管理の実践的スキルを習得します。より信頼性の高いアプリケーションのための高度なエラーハンドリング技術をマスターし、ファイル操作の自動化をファイルハンドリングで実現し、PytestとUnittestによる堅牢なテスト戦略を学ぶことで、コードの完全性を確保します。

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.

チャレンジ：並列ファイル処理

解答