You are building a rate-limited API client that fetches comment data from `jsonplaceholder.typicode.com`. The client must limit concurrency to avoid overwhelming the server, apply a per-request timeout, and collect results without failing on individual errors.

The endpoint `https://jsonplaceholder.typicode.com/comments/{comment_id}` returns a JSON object with the following fields:

- `postId`: the ID of the post the comment belongs to;
- `name`: the comment title;
- `email`: the commenter's email address.



import unittest
import asyncio
import re

# Helper function to dynamically generate test names and assertions
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)


class TestUserCode(unittest.TestCase):

    def test_fetch_comment_is_coroutine(self):
        import user_code
        _dynamic_test(
            self,
            asyncio.iscoroutinefunction(user_code.fetch_comment),
            "The `fetch_comment` function is defined with `async def`.",
            "Expected `fetch_comment` to be defined with `async def`."
        )

    def test_main_is_coroutine(self):
        import user_code
        _dynamic_test(
            self,
            asyncio.iscoroutinefunction(user_code.main),
            "The `main` function is defined with `async def`.",
            "Expected `main` to be defined with `async def`."
        )

    def test_semaphore_is_used(self):
        with open("user_code.py", "r") as file:
            source = file.read()
        _dynamic_test(
            self,
            bool(re.search(r"asyncio\.Semaphore\s*\(\s*5\s*\)", source)),
            "`asyncio.Semaphore(5)` is used to limit concurrency.",
            "Expected `asyncio.Semaphore(5)` to be used."
        )

    def test_wait_for_is_used(self):
        with open("user_code.py", "r") as file:
            source = file.read()
        _dynamic_test(
            self,
            bool(re.search(r"asyncio\.wait_for\s*\(", source)),
            "`asyncio.wait_for()` is used to apply a timeout.",
            "Expected `asyncio.wait_for()` to be used in `fetch_comment()`."
        )

    def test_gather_is_used(self):
        with open("user_code.py", "r") as file:
            source = file.read()
        _dynamic_test(
            self,
            bool(re.search(r"asyncio\.gather\s*\(", source)),
            "`asyncio.gather()` is used to fetch comments concurrently.",
            "Expected `asyncio.gather()` to be used in `main()`."
        )

    def test_fetch_comment_returns_string(self):
        import user_code
        import httpx

        async def run():
            semaphore = asyncio.Semaphore(5)
            async with httpx.AsyncClient() as client:
                result = await user_code.fetch_comment(client, semaphore, 1)
            return result

        result = asyncio.run(run())
        _dynamic_test(
            self,
            isinstance(result, str) and "[Post" in result,
            f"`fetch_comment()` returns a correctly formatted string.",
            f"Expected a string starting with `[Post`, but got `{result}`."
        )

    def test_all_15_comments_fetched(self):
        with open("user_code.py", "r") as file:
            source = file.read()
        _dynamic_test(
            self,
            bool(re.search(r"range\s*\(\s*1\s*,\s*16\s*\)", source)),
            "Comments with IDs `1` through `15` are fetched.",
            "Expected `range(1, 16)` to be used to generate comment IDs `1` through `15`."
        )


if __name__ == "__main__":
    unittest.main()

test_main.py

Master the skills needed to build efficient, concurrent Python applications using asyncio – from coroutines and event loop fundamentals through tasks, synchronization primitives, and real-world debugging techniques.

Gain a solid understanding of asynchronous programming concepts and the core mechanisms behind Python's asyncio library.

Dive into asyncio's task scheduling mechanisms and learn how to run multiple coroutines concurrently.

Get hands-on experience with HTTP clients, synchronization primitives, and queue-based pipelines using asyncio.

Explore advanced asyncio features for controlling task groups, running blocking code, and debugging asynchronous applications.

Challenge: Building an Async API Client

Рішення