New unit tests

Author: codeSamuraii

.github/workflows/unit-tests.yml

@@ -25,7 +25,7 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           pip install -r requirements.txt
-          pip install pytest
+          pip install pytest fakeredis
 
       - name: Run tests
         run: pytest --disable-pytest-warnings -v tests/

app.py

@@ -29,8 +29,8 @@ async def lifespan(app: FastAPI):
 
 # App. routes
 app.include_router(misc_router)
+app.include_router(ws_router)  # WebSocket routes first to avoid conflicts
 app.include_router(http_router)
-app.include_router(ws_router)
 
 # Static files
 app.mount('/', StaticFiles(directory='static', html=True), name='static')

tests/conftest.py

@@ -1,19 +1,55 @@
-import asyncio
+import httpx
 import pytest
+import fakeredis
+from typing import AsyncIterator
+from redis import asyncio as redis
 from unittest.mock import AsyncMock, patch
+from starlette.testclient import TestClient
 
+from app import app
+from lib.store import Store
 
-@pytest.fixture(scope="session")
-def event_loop():
-    """Create an instance of the default event loop for the test session."""
-    loop = asyncio.new_event_loop()
-    yield loop
-    loop.close()
-
-@pytest.fixture(autouse=True)
-def setup_test_environment():
-    """Set up test environment before each test."""
-    # Mock Redis client to avoid needing actual Redis instance
-    with patch('app.redis') as mock_redis:
-        mock_redis.close = AsyncMock()
-        yield mock_redis
+
+@pytest.fixture
+def anyio_backend():
+    return 'asyncio'
+
+@pytest.fixture
+async def redis_client() -> AsyncIterator[redis.Redis]:
+    async with fakeredis.FakeAsyncRedis() as client:
+        yield client
+
+@pytest.fixture
+async def test_client(redis_client: redis.Redis) -> AsyncIterator[httpx.AsyncClient]:
+    def get_redis_override(*args, **kwargs) -> redis.Redis:
+        return redis_client
+
+    # Make sure the app has the Redis state set up
+    app.state.redis = redis_client
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        # Patch the `get_redis` method of the `Store` class
+        with patch.object(Store, 'get_redis', new=get_redis_override):
+            print("")
+            yield client
+
+@pytest.fixture
+async def websocket_client(redis_client: redis.Redis):
+    """Alternative WebSocket client using Starlette TestClient."""
+    def get_redis_override(*args, **kwargs) -> redis.Redis:
+        return redis_client
+
+    # Make sure the app has the Redis state set up
+    app.state.redis = redis_client
+
+    # Patch the `get_redis` method of the `Store` class
+    with patch.object(Store, 'get_redis', new=get_redis_override):
+        with TestClient(app) as client:
+            print("")
+            yield client
+
+@pytest.mark.anyio
+async def test_mocks(test_client: httpx.AsyncClient) -> None:
+    response = await test_client.get("/nonexistent-endpoint")
+    assert response.status_code == 404, "Expected 404 for nonexistent endpoint"

tests/helpers.py

@@ -0,0 +1,30 @@
+from string import ascii_letters
+from itertools import islice, repeat, chain
+from typing import Tuple, Iterable, Iterator
+from annotated_types import T
+
+from lib.metadata import FileMetadata
+
+
+def generate_test_file(size_in_kb: int = 10) -> tuple[bytes, FileMetadata]:
+    """Generates a test file with specified size in KB."""
+    chunk_generator = ((letter * 1024).encode() for letter in chain.from_iterable(repeat(ascii_letters)))
+    content = b''.join(next(chunk_generator) for _ in range(size_in_kb))
+
+    metadata = FileMetadata(
+        name="test_file.bin",
+        size=len(content),
+        content_type="application/octet-stream"
+    )
+    return content, metadata
+
+
+def batched(iterable: Iterable[T], chunk_size: int) -> Iterator[Tuple[T, ...]]:
+    "Batch data into lists of length n. The last batch may be shorter."
+    # batched('ABCDEFG', 3) --> ABC DEF G
+    it = iter(iterable)
+    while True:
+        batch = bytes(islice(it, chunk_size))
+        if not batch:
+            return
+        yield batch