Merge branch '2024_tt_testing_async'

2024/tuesday_tips/testing_async/.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+    "python.testing.pytestArgs": [
+        "."
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
+}

2024/tuesday_tips/testing_async/data/fictional_events.json

@@ -0,0 +1,38 @@
+{
+    "1": {
+        "title": "Release of the Giant",
+        "description": "A legendary giant, bound in unbreakable chains for millennia, is finally released by a team of four daring adventurers. As the giant stretches his limbs, one of the team members, awed by the gravity of their action, solemnly tells the others to stop fooling around and respect the moment.",
+        "location": "In a world where there is always war between guilds",
+        "date": "2024-02-14"
+    },
+    "2": {
+        "title": "Mysterious Portal in the City",
+        "description": "A swirling portal of vivid colors appears suddenly in the heart of a bustling city. The portal seems to lead to an unknown dimension, sparking both curiosity and fear among the city's inhabitants.",
+        "location": "Downtown Central Park",
+        "date": "2024-03-10"
+    },
+    "3": {
+        "title": "Gravity Reversal Phenomenon",
+        "description": "The small town of Newtonville experiences a bizarre and unexplained phenomenon where gravity reverses for exactly 24 hours, leading to chaos and wonder.",
+        "location": "Newtonville",
+        "date": "2024-04-01"
+    },
+    "4": {
+        "title": "Discovery of an Underground City",
+        "description": "Archaeologists uncover a hidden underground city, perfectly preserved and seemingly untouched for centuries. The city contains unknown symbols and architecture, hinting at a lost civilization.",
+        "location": "Sahara Desert",
+        "date": "2024-05-15"
+    },
+    "5": {
+        "title": "Green Night Sky Event",
+        "description": "For several hours, the night sky turns a bright, eerie green due to a rare celestial event. Astronomers and skywatchers around the world are left in awe.",
+        "location": "Worldwide",
+        "date": "2024-06-21"
+    },
+    "6": {
+        "title": "The Speaking Tree",
+        "description": "An ancient tree in a secluded forest starts to glow mysteriously and speak in various ancient languages. The phenomenon attracts linguists and mystics alike.",
+        "location": "Ancient Forest of Eldwood",
+        "date": "2024-07-07"
+    }
+}

2024/tuesday_tips/testing_async/fetch_event.py

@@ -0,0 +1,24 @@
+import json
+import random
+from time import sleep
+
+from aiohttp import ClientSession
+
+
+async def read_data(
+    path: str = "./data/fictional_events.json",
+) -> dict[str, dict[str, str]]:
+    sleep(random.choice([0.125, 0.5, 0.75, 0.1]))
+    with open(path) as file:
+        data = json.load(file)
+    return data
+
+
+async def fetch_event(event_id: str, session: ClientSession = None) -> dict[str, str]:
+    json_data = await read_data()
+
+    try:
+        event = json_data[event_id]
+        return event
+    except KeyError:
+        return {"error": "Event not found."}

2024/tuesday_tips/testing_async/main.py

@@ -0,0 +1,20 @@
+import aiohttp
+import asyncio
+
+from fetch_event import fetch_event
+
+
+# Example usage
+async def main() -> None:
+    event_id = "1"
+    async with aiohttp.ClientSession() as session:
+        event_info = await fetch_event(session=session, event_id=event_id)
+
+        print(f"Title: {event_info['title']}")
+        print(f"Description: {event_info['description']}")
+        print(f"Location: {event_info['location']}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+

2024/tuesday_tips/testing_async/pyproject.toml

@@ -0,0 +1,17 @@
+[tool.poetry]
+name = "test-async-code"
+version = "0.1.0"
+description = ""
+authors = ["Andreas Bergman <[email protected]>"]
+license = "MIT"
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.12"
+pytest-asyncio = "^0.23.4"
+aiohttp = "^3.9.2"
+
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

2024/tuesday_tips/testing_async/test/fixture_based_fetch_event_test.py

@@ -0,0 +1,32 @@
+# Create a fixture that returns an event loop.
+import asyncio
+from unittest.mock import patch
+from aiohttp import ClientSession
+import pytest
+
+from fetch_event import fetch_event
+
+@pytest.fixture
+async def session():
+    with patch("aiohttp.ClientSession") as mock:
+        yield mock
+
+@pytest.fixture
+def event_loop():
+    asyncio.get_event_loop_policy().set_event_loop(asyncio.new_event_loop())
+    loop = asyncio.get_event_loop()
+    yield loop
+    loop.close()
+
+def test_fetch_event(session: ClientSession, event_loop: asyncio.AbstractEventLoop):
+    results = event_loop.run_until_complete(fetch_event(session=session, event_id="1"))
+    assert len(results) > 0
+
+
+def test_fetch_multiple_events_with_custom_event_loop(
+    session: ClientSession, event_loop: asyncio.AbstractEventLoop
+):
+    tasks = [fetch_event(session=session, event_id=str(i)) for i in range(1, 5)]
+    results = event_loop.run_until_complete(asyncio.gather(*tasks))
+    assert len(results) == 4
+

2024/tuesday_tips/testing_async/test/plugin_based_fetch_event_test.py

@@ -0,0 +1,34 @@
+import asyncio
+import pytest
+from unittest.mock import patch
+from aiohttp import ClientSession
+
+from fetch_event import fetch_event
+
+
+@pytest.fixture
+async def session():
+    with patch("aiohttp.ClientSession") as mock:
+        yield mock
+
+
+@pytest.mark.asyncio
+async def test_fetch_event(session: ClientSession):
+    event = await fetch_event(session=session, event_id="1")
+    event_2 = await fetch_event(session=session, event_id="20")
+
+    assert event is not None
+    assert event_2["error"] == "Event not found."
+
+
+@pytest.mark.asyncio
+async def test_fetch_multiple_events(session: ClientSession):
+    event_ids = ["1", "2", "3", "4", "5", "6"]
+
+    tasks = [
+        asyncio.create_task(fetch_event(session=session, event_id=event_id))
+        for event_id in event_ids
+    ]
+    results = await asyncio.gather(*tasks)
+
+    assert len(results) == 6