Imagine you are developing a tool to help users download several large files from the internet. Downloading them one after another would take a lot of time. To speed things up, you want to download all the files at the same time. By starting a separate thread for each file download, you can make the most of your computer's resources and save significant time, especially when network delays are involved.


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

def mock_requests_get(url):
    class Response:
        def __init__(self, content):
            self.content = content
    # Use filename from URL as content for uniqueness
    filename = url.split('/')[-1]
    return Response(content=(f"content of {filename}").encode())

class TestTask(unittest.TestCase):
    def setUp(self):
        # Patch requests.get
        import user_code
        self.requests = importlib.import_module('requests')
        self.orig_get = self.requests.get
        self.requests.get = mock_requests_get
        # Clean up files if they exist
        self.files = ["file1.txt", "file2.txt"]
        for fname in self.files:
            if os.path.exists(fname):
                os.remove(fname)
    def tearDown(self):
        # Restore requests.get
        self.requests.get = self.orig_get
        # Clean up files
        for fname in self.files:
            if os.path.exists(fname):
                os.remove(fname)

    def test_parallel_download_creates_files(self):
        import user_code
        importlib.reload(user_code)
        urls = [
            "https://example.com/file1.txt",
            "https://example.com/file2.txt"
        ]
        f = io.StringIO()
        with redirect_stdout(f):
            user_code.parallel_download(urls)
        # Check files created
        file1_exists = os.path.exists("file1.txt")
        file2_exists = os.path.exists("file2.txt")
        _dynamic_test(
            self,
            file1_exists and file2_exists,
            "Both files are downloaded and exist.",
            f"Expected files to exist: file1.txt={file1_exists}, file2.txt={file2_exists}.",
        )

    def test_parallel_download_file_content(self):
        import user_code
        importlib.reload(user_code)
        urls = [
            "https://example.com/file1.txt",
            "https://example.com/file2.txt"
        ]
        user_code.parallel_download(urls)
        # Check content
        with open("file1.txt", "rb") as f1:
            content1 = f1.read()
        with open("file2.txt", "rb") as f2:
            content2 = f2.read()
        _dynamic_test(
            self,
            content1 == b"content of file1.txt" and content2 == b"content of file2.txt",
            "Files have the correct content.",
            f"Expected file1.txt and file2.txt to have correct content, got: {content1}, {content2}",
        )

    def test_parallel_download_prints_downloaded(self):
        import user_code
        importlib.reload(user_code)
        urls = [
            "https://example.com/file1.txt",
            "https://example.com/file2.txt"
        ]
        f = io.StringIO()
        with redirect_stdout(f):
            user_code.parallel_download(urls)
        output = normalize_text(f.getvalue())
        expected1 = normalize_text("Downloaded file1.txt")
        expected2 = normalize_text("Downloaded file2.txt")
        _dynamic_test(
            self,
            expected1 in output and expected2 in output,
            "Output includes correct download messages for all files.",
            f"Expected download messages not found in 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 the concepts, techniques, and practical applications of multithreading and multiprocessing in Python. Learn how to write concurrent programs, manage threads and processes, and solve real-world problems using parallel execution.

Explore the foundational concepts of concurrency, parallelism, and the differences between threads and processes in Python.

Learn how to create, start, and manage threads in Python, and understand thread synchronization.

Explore how to create and manage processes in Python, and understand inter-process communication.

Delve into advanced concurrency patterns, performance considerations, and best practices for writing robust concurrent code.

Challenge: Parallel File Downloader

解答