A traffic light system is a classic example of a process that cycles through a set of states—typically **"Red," "Green," and "Yellow"**—each for a fixed period. In a real-world intersection, the light changes states in a predictable order to manage the flow of vehicles and pedestrians. Modeling this behavior asynchronously allows you to simulate the timed transitions between states without blocking the rest of your application. By using Python's `asyncio` library, you can create a coroutine that switches the light's state after a delay, mimicking how a real traffic signal operates. This approach demonstrates how **asynchronous programming** can handle time-based events in a non-blocking way.


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

class TestTask(unittest.TestCase):
    def test_traffic_light_prints_states_in_order(self):
        import user_code
        importlib.reload(user_code)

        async def run_and_capture():
            f = io.StringIO()
            with redirect_stdout(f):
                await user_code.traffic_light()
            return f.getvalue()

        output = asyncio.run(run_and_capture())
        output_norm = normalize_text(output)
        expected_order = ["red", "green", "yellow"]
        for state in expected_order:
            _dynamic_test(
                self,
                state in output_norm,
                f"Output includes state: {state}",
                f"Expected state '{state}' not found in output: '{output_norm}'",
            )
        # Check order
        idx_red = output_norm.find("red")
        idx_green = output_norm.find("green")
        idx_yellow = output_norm.find("yellow")
        _dynamic_test(
            self,
            idx_red < idx_green < idx_yellow,
            "States are printed in correct order (Red, Green, Yellow)",
            f"States are not printed in correct order: {output_norm}",
        )

    def test_traffic_light_completes_one_cycle(self):
        import user_code
        importlib.reload(user_code)

        async def run_and_capture():
            f = io.StringIO()
            with redirect_stdout(f):
                await user_code.traffic_light()
            return f.getvalue()

        output = asyncio.run(run_and_capture())
        output_norm = normalize_text(output)
        for state in ["red", "green", "yellow"]:
            state_count = output_norm.count(state)
            _dynamic_test(
                self,
                state_count == 1,
                f"State '{state}' appears exactly once in the output.",
                f"State '{state}' appears {state_count} times in the output (should be 1).",
            )

    def test_traffic_light_uses_asyncio_sleep(self):
        import user_code
        importlib.reload(user_code)
        import ast
        import inspect
        source = inspect.getsource(user_code.traffic_light)
        tree = ast.parse(source)
        found_sleep = False
        for node in ast.walk(tree):
            if isinstance(node, ast.Await) and isinstance(node.value, ast.Call):
                func = node.value.func
                if isinstance(func, ast.Attribute) and func.attr == "sleep":
                    found_sleep = True
        _dynamic_test(
            self,
            found_sleep,
            "Function uses await asyncio.sleep for delays.",
            "Function does not use await asyncio.sleep for delays.",
        )

    def test_traffic_light_is_async_function(self):
        import user_code
        importlib.reload(user_code)
        is_async = asyncio.iscoroutinefunction(getattr(user_code, "traffic_light", None))
        _dynamic_test(
            self,
            is_async,
            "traffic_light is an async function.",
            "traffic_light is not an async function.",
        )

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: Simulated Traffic Light

解答