Add tests for Shortify and update docker compose (#1469)

Author: devanshu-ziphq

demos/shortify/config/config.yml

@@ -6,4 +6,4 @@ redis:
   maxsize: 5
 
 host: 127.0.0.1
-port: 9002
+port: 9001

demos/shortify/docker-compose.yml

@@ -1,4 +1,5 @@
-redis:
+services:
+  redis:
     image: redis:6
     ports:
-        - 6379:6379
+      - 6379:6379

demos/shortify/pytest.ini

@@ -12,6 +12,7 @@ addopts =
     # show values of the local vars in errors:
     --showlocals
 asyncio_mode = auto
+asyncio_default_fixture_loop_scope = function
 filterwarnings =
     error
 testpaths = tests/

demos/shortify/requirements.txt

@@ -3,3 +3,5 @@ aiohttp-jinja2==1.6
 pyyaml==6.0.2
 redis==6.1.0
 trafaret==2.1.1
+pytest-aiohttp==1.1.0
+pytest-asyncio==0.26

demos/shortify/shortify/main.py

@@ -5,6 +5,7 @@
 import aiohttp_jinja2
 import jinja2
 from aiohttp import web
+from redis.asyncio import Redis
 
 from shortify.routes import setup_routes
 from shortify.utils import init_redis, load_config
@@ -14,10 +15,13 @@
 PROJ_ROOT = pathlib.Path(__file__).parent.parent
 TEMPLATES_ROOT = pathlib.Path(__file__).parent / 'templates'
 
+# Define AppKey for Redis
+REDIS_KEY = web.AppKey("REDIS_KEY", Redis)
+
 
 async def setup_redis(app, conf):
     redis = await init_redis(conf["redis"])
-    app["redis"] = redis
+    app[REDIS_KEY] = redis
     return redis
 
 
@@ -28,7 +32,7 @@ def setup_jinja(app):
 
 
 async def init():
-    conf = load_config(PROJ_ROOT / 'config' / 'config.yml')
+    conf = load_config(PROJ_ROOT / "config" / "config.yml")
 
     app = web.Application()
     redis = await setup_redis(app, conf)

demos/shortify/shortify/views.py

@@ -19,7 +19,7 @@ async def shortify(self, request):
         long_url = fetch_url(data)
 
         index = await self._redis.incr("shortify:count")
-        path = encode(index)
+        path = encode(index - 1)
         key = "shortify:{}".format(path)
         await self._redis.set(key, long_url)
 
@@ -36,4 +36,4 @@ async def redirect(self, request):
         location = await self._redis.get(key)
         if not location:
             raise web.HTTPNotFound()
-        return web.HTTPFound(location=location.decode())
+        raise web.HTTPFound(location=location.decode())

demos/shortify/tests/__init__.py

@@ -0,0 +1,37 @@
+import asyncio
+
+import pytest
+from redis import asyncio as aioredis
+
+from shortify.main import init, PROJ_ROOT
+from shortify.utils import load_config
+
+TEST_CONFIG_PATH = PROJ_ROOT / "config" / "config.yml"
+TEST_CONFIG = load_config(TEST_CONFIG_PATH.as_posix())
+
+
+@pytest.fixture
+async def cli(aiohttp_client):
+    app, _, _ = await init()
+    return await aiohttp_client(app)
+
+
+@pytest.fixture
+async def redis():
+    """Create a Redis connection for testing."""
+    redis_config = TEST_CONFIG["redis"]
+    redis = await aioredis.from_url(
+        f"redis://{redis_config['host']}:{redis_config['port']}",
+    )
+    yield redis
+    await redis.aclose()
+    # Give time for all connections to close properly
+    await asyncio.sleep(0.1)
+
+
+@pytest.fixture(autouse=True)
+async def clean_redis(redis):
+    """Clean Redis database before each test."""
+    await redis.flushdb()
+    # Give time for all connections to close properly
+    await asyncio.sleep(0.1)

demos/shortify/tests/conftest.py

@@ -1,3 +1,37 @@
-async def test_something():
-    # TODO: Create some actual tests.
-    assert True
+async def test_create_short_url(cli, clean_redis):
+    """Test creating a short URL for a long URL."""
+    long_url = "https://example.com/very/long/url/that/needs/shortening"
+
+    async with cli.post("/shortify", json={"url": long_url}) as resp:
+        await resp.read()
+        assert resp.status == 200
+
+    data = await resp.json()
+    assert "url" in data
+    assert data["url"] == "http://127.0.0.1:9001/a"
+
+
+async def test_redirect_to_long_url(cli, clean_redis):
+    """Test redirecting from short URL to original long URL."""
+    long_url = "https://example.com/very/long/url/that/needs/shortening"
+
+    # Create a short URL
+    async with cli.post("/shortify", json={"url": long_url}) as resp:
+        await resp.read()
+        assert resp.status == 200
+        data = await resp.json()
+        short_url = data["url"]
+        short_code = short_url.split("/")[-1]
+
+    # Test the redirect
+    async with cli.get(f"/{short_code}", allow_redirects=False) as resp:
+        await resp.read()
+        assert resp.status == 302
+        assert resp.headers["Location"] == long_url
+
+
+async def test_invalid_short_url(cli, clean_redis):
+    """Test accessing a non-existent short URL."""
+    async with cli.get("/nonexistent", allow_redirects=False) as resp:
+        await resp.read()
+        assert resp.status == 404