Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,14 @@
)
from main import app
from datetime import datetime, UTC, timedelta
from utils.core.rate_limit import clear_all_rate_limiters


@pytest.fixture(autouse=True)
def reset_rate_limiters() -> Generator[None, None, None]:
clear_all_rate_limiters()
yield
clear_all_rate_limiters()


# Define a custom exception for test setup errors
Expand Down
22 changes: 3 additions & 19 deletions tests/routers/core/test_account.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
import pytest
from fastapi.testclient import TestClient
from starlette.datastructures import URLPath
from sqlmodel import Session, select
Expand Down Expand Up @@ -28,28 +27,13 @@
get_password_hash,
)
from utils.core.rate_limit import (
login_ip_limiter,
forgot_password_email_limiter,
forgot_password_ip_limiter,
login_email_limiter,
login_ip_limiter,
register_ip_limiter,
forgot_password_ip_limiter,
forgot_password_email_limiter,
)


@pytest.fixture(autouse=True)
def _reset_rate_limiters():
"""Reset all rate limiter state between tests to avoid cross-test pollution."""
yield
for limiter in (
login_ip_limiter,
login_email_limiter,
register_ip_limiter,
forgot_password_ip_limiter,
forgot_password_email_limiter,
):
limiter._attempts.clear()


# --- API Endpoint Tests ---


Expand Down
21 changes: 1 addition & 20 deletions tests/test_htmx.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,34 +8,15 @@
- Non-HTMX paths remain unchanged (303 RedirectResponse or full-page error).
"""

import pytest
from starlette.requests import Request
from fastapi.templating import Jinja2Templates
from tests.conftest import htmx_headers
from utils.core.htmx import is_htmx_request, toast_response, append_toast
from utils.core.rate_limit import (
login_ip_limiter,
login_email_limiter,
register_ip_limiter,
forgot_password_ip_limiter,
forgot_password_email_limiter,
login_ip_limiter,
)


@pytest.fixture(autouse=True)
def _reset_rate_limiters():
"""Reset all rate limiter state between tests."""
yield
for limiter in (
login_ip_limiter,
login_email_limiter,
register_ip_limiter,
forgot_password_ip_limiter,
forgot_password_email_limiter,
):
limiter._attempts.clear()


# ---------------------------------------------------------------------------
# 1.3 — is_htmx_request helper
# ---------------------------------------------------------------------------
Expand Down
28 changes: 18 additions & 10 deletions utils/core/organizations.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
from typing import Any, cast

from sqlmodel import Session, select
from sqlalchemy.orm import selectinload
from sqlalchemy.orm import InstrumentedAttribute, selectinload

from utils.core.models import Organization, Role, User, Invitation

Expand All @@ -21,13 +23,15 @@ def load_org_for_members_partial(
select(Organization)
.where(Organization.id == organization_id)
.options(
selectinload(Organization.roles)
.selectinload(Role.users)
.selectinload(User.account),
selectinload(Organization.roles)
.selectinload(Role.users)
.selectinload(User.roles),
selectinload(Organization.roles).selectinload(Role.permissions),
selectinload(cast(InstrumentedAttribute[Any], Organization.roles))
.selectinload(cast(InstrumentedAttribute[Any], Role.users))
.selectinload(cast(InstrumentedAttribute[Any], User.account)),
selectinload(cast(InstrumentedAttribute[Any], Organization.roles))
.selectinload(cast(InstrumentedAttribute[Any], Role.users))
.selectinload(cast(InstrumentedAttribute[Any], User.roles)),
selectinload(
cast(InstrumentedAttribute[Any], Organization.roles)
).selectinload(cast(InstrumentedAttribute[Any], Role.permissions)),
)
).first()
user_permissions = _user_permissions_for_org(user, organization_id)
Expand All @@ -43,8 +47,12 @@ def load_org_for_roles_partial(
select(Organization)
.where(Organization.id == organization_id)
.options(
selectinload(Organization.roles).selectinload(Role.users),
selectinload(Organization.roles).selectinload(Role.permissions),
selectinload(
cast(InstrumentedAttribute[Any], Organization.roles)
).selectinload(cast(InstrumentedAttribute[Any], Role.users)),
selectinload(
cast(InstrumentedAttribute[Any], Organization.roles)
).selectinload(cast(InstrumentedAttribute[Any], Role.permissions)),
)
).first()
user_permissions = _user_permissions_for_org(user, organization_id)
Expand Down
14 changes: 14 additions & 0 deletions utils/core/rate_limit.py
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,20 @@ def _int_env(name: str, default: int) -> int:
window_seconds=_int_env("FORGOT_PASSWORD_EMAIL_WINDOW_SECONDS", 60),
)

_ALL_LIMITERS = (
login_ip_limiter,
login_email_limiter,
register_ip_limiter,
forgot_password_ip_limiter,
forgot_password_email_limiter,
)


def clear_all_rate_limiters() -> None:
"""Clear all in-memory rate limiter state."""
for limiter in _ALL_LIMITERS:
limiter._attempts.clear()


# --- Dependency helpers ---

Expand Down
Loading