test: Adds test suite for first endpoints

Makefile

@@ -1,13 +1,13 @@
 # this target runs checks on all files
 quality:
+	ruff format --check .
 	ruff check .
 	mypy
-	ruff format --check .
 
 # this target runs checks on all files and potentially modifies some of them
 style:
-	ruff --fix .
 	ruff format .
+	ruff --fix .
 
 # Pin the dependencies
 lock:

pyproject.toml

@@ -39,6 +39,8 @@ pre-commit = { version = "^2.17.0", optional = true }
 
 [tool.poetry.group.dev.dependencies]
 pytest = ">=5.3.2,<8.0.0"
+pytest-asyncio = ">=0.17.0,<1.0.0"
+httpx = ">=0.23.0"
 pytest-cov = ">=3.0.0,<5.0.0"
 pytest-pretty = "^1.0.0"
 

src/app/api/api_v1/endpoints/login.py

@@ -49,7 +49,7 @@
         headers={"Accept": "application/json"},
         timeout=5,
     )
-    if response.status_code != status.HTTP_200_OK:
+    if response.status_code != status.HTTP_200_OK or isinstance(response.json().get("error"), str):
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authorization code.")
     return GHToken(**response.json())
 
@@ -66,12 +66,13 @@
     By default, the token expires after 1 hour.
     """
     # Verify credentials
-    entry = await users.get_by_login(form_data.username)
-    if entry is None or not await verify_password(form_data.password, entry.hashed_password):
+    user = await users.get_by_login(form_data.username)
+    if user is None or not await verify_password(form_data.password, user.hashed_password):
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials.")
     # create access token using user user_id/user_scopes
-    token_data = {"sub": str(entry.id), "scopes": entry.scope.split()}
+    token_data = {"sub": str(user.id), "scopes": user.scope.split()}
     token = await create_access_token(token_data, settings.ACCESS_TOKEN_UNLIMITED_MINUTES)
+    analytics_client.capture(user.id, event="user-login", properties={"login": user.login})
 
     return Token(access_token=token, token_type="bearer")  # nosec B106  # noqa S106
 
@@ -122,5 +123,6 @@
     # create access token using user user_id/user_scopes
     token_data = {"sub": str(user.id), "scopes": user.scope.split()}
     token = await create_access_token(token_data, settings.ACCESS_TOKEN_UNLIMITED_MINUTES)
+    analytics_client.capture(user.id, event="user-login", properties={"login": user.login})
 
     return Token(access_token=token, token_type="bearer")  # nosec B106  # noqa S106

src/app/api/api_v1/endpoints/repos.py

@@ -56,6 +56,7 @@
     user=Security(get_current_user, scopes=[UserScope.USER, UserScope.ADMIN]),
 ) -> List[Repository]:
     entries = await repos.fetch_all() if user.scope == UserScope.ADMIN else await repos.fetch_all(("owner_id", user.id))
+    analytics_client.capture(user.id, event="repo-fetch")
     return [elt for elt in entries]
 
 

src/app/db.py

@@ -3,9 +3,9 @@
 # All rights reserved.
 # Copying and/or distributing is strictly prohibited without the express permission of its copyright owner.
 
-from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.ext.asyncio.engine import AsyncEngine
 from sqlalchemy.orm import sessionmaker
-from sqlmodel import SQLModel, select
+from sqlmodel import SQLModel, create_engine, select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.core.config import settings
@@ -15,7 +15,7 @@
 __all__ = ["get_session", "init_db"]
 
 
-engine = create_async_engine(settings.POSTGRES_URL, echo=settings.DEBUG)
+engine = AsyncEngine(create_engine(settings.POSTGRES_URL, echo=settings.DEBUG))
 
 
 async def get_session() -> AsyncSession:  # type: ignore[misc]

src/tests/conftest.py

@@ -0,0 +1,51 @@
+import asyncio
+from typing import Generator
+
+import pytest
+import pytest_asyncio
+from httpx import AsyncClient
+from sqlalchemy.orm import sessionmaker
+from sqlmodel import SQLModel
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from app.core.config import settings
+from app.db import engine
+from app.main import app
+
+
+@pytest.fixture(scope="session")
+def event_loop(request) -> Generator:
+    loop = asyncio.get_event_loop_policy().new_event_loop()
+    yield loop
+    loop.close()
+
+
+@pytest_asyncio.fixture
+async def async_client():
+    async with AsyncClient(app=app, base_url=f"http://{settings.API_V1_STR}") as client:
+        yield client
+
+
+@pytest_asyncio.fixture(scope="function")
+async def async_session() -> AsyncSession:
+    session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+    async with session() as s:
+        async with engine.begin() as conn:
+            await conn.run_sync(SQLModel.metadata.create_all)
+
+        yield s
+
+    async with engine.begin() as conn:
+        await conn.run_sync(SQLModel.metadata.drop_all)
+
+    await engine.dispose()
+
+
+async def mock_verify_password(plain_password, hashed_password):
+    return hashed_password == f"hashed_{plain_password}"
+
+
+def pytest_configure():
+    # api.security patching
+    pytest.mock_verify_password = mock_verify_password

src/tests/endpoints/test_login.py

@@ -0,0 +1,125 @@
+from typing import Any, Dict, Union
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+import pytest_asyncio
+from httpx import AsyncClient
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from app.core import security
+from app.models import User
+
+USER_TABLE = [
+    {"id": 1, "login": "first_login", "hashed_password": "hashed_first_pwd", "scope": "user"},
+    {"id": 2, "login": "second_login", "hashed_password": "hashed_second_pwd", "scope": "user"},
+]
+
+
+@pytest_asyncio.fixture(scope="function")
+async def session(async_session: AsyncSession, monkeypatch):
+    for entry in USER_TABLE:
+        async_session.add(User(**entry))
+    await async_session.commit()
+    monkeypatch.setattr(security, "verify_password", pytest.mock_verify_password)
+    yield async_session
+
+
+@pytest.mark.parametrize(
+    ("payload", "status_code", "status_detail"),
+    [
+        ({"username": "foo"}, 422, None),
+        ({"github_token": "foo"}, 401, None),
+    ],
+)
+@pytest.mark.asyncio()
+async def test_login_with_github_token(
+    async_client: AsyncClient,
+    session: AsyncSession,
+    payload: Dict[str, Any],
+    status_code: int,
+    status_detail: Union[str, None],
+):
+    response = await async_client.post("/login/token", json=payload)
+    assert response.status_code == status_code
+    if isinstance(status_detail, str):
+        assert response.json()["detail"] == status_detail
+
+
+@pytest.mark.parametrize(
+    ("payload", "status_code", "status_detail"),
+    [
+        ({"username": "foo"}, 422, None),
+        ({"username": "foo", "password": "bar"}, 401, None),
+        # ({"username": "first_login", "password": "first_pwd"}, 200, None),
+    ],
+)
+@pytest.mark.asyncio()
+async def test_login_with_creds(
+    async_client: AsyncClient,
+    session: AsyncSession,
+    payload: Dict[str, Any],
+    status_code: int,
+    status_detail: Union[str, None],
+):
+    response = await async_client.post("/login/creds", data=payload)
+    assert response.status_code == status_code
+    if isinstance(status_detail, str):
+        assert response.json()["detail"] == status_detail
+    if response.status_code // 100 == 2:
+        response_json = response.json()
+        assert response_json["token_type"] == "Bearer"  # noqa: S105
+        assert isinstance(response_json["access_token"], str)
+        assert len(response_json["access_token"]) == 10
+
+
+@pytest.mark.parametrize(
+    ("payload", "status_code", "status_detail", "expected_response"),
+    [
+        ({"code": "foo", "redirect_uri": 0}, 422, None, None),
+        # Github 422
+        ({"code": "foo", "redirect_uri": ""}, 422, None, None),
+        ({"code": "foo", "redirect_uri": "https://quackai.com"}, 401, None, None),
+    ],
+)
+@pytest.mark.asyncio()
+async def test_request_github_token_from_code(
+    async_client: AsyncClient,
+    session: AsyncSession,
+    payload: Dict[str, Any],
+    status_code: int,
+    status_detail: Union[str, None],
+    expected_response: Union[Dict[str, Any], None],
+):
+    response = await async_client.post("/login/github", json=payload)
+    assert response.status_code == status_code
+    if isinstance(status_detail, str):
+        assert response.json()["detail"] == status_detail
+    if isinstance(expected_response, dict):
+        assert response.json() == expected_response
+
+
+@pytest.mark.parametrize(
+    ("scope", "redirect_uri", "status_code"),
+    [
+        ("read:user%20user:email%20repo", "https://app.quack-ai.com", 307),
+    ],
+)
+@pytest.mark.asyncio()
+async def test_authorize_github(
+    async_client: AsyncClient,
+    session: AsyncSession,
+    scope: Any,
+    redirect_uri: Any,
+    status_code: int,
+):
+    response = await async_client.get("/login/authorize", params={"scope": scope, "redirect_uri": redirect_uri})
+    assert response.status_code == status_code
+    for key, _, val in response.headers._list:
+        if key == "location":
+            u = urlparse(val)
+            assert u.scheme == "https"
+            assert u.netloc == "github.com/login/oauth/authorize"
+            q = parse_qs(u.query)
+            assert q.keys() == {"scope", "client_id", "redirect_uri"}
+            assert q["scope"][0] == scope
+            assert q["redirect_uri"][0] == redirect_uri