feat(backend): added auth, reviews, users modules

also provided tests
This commit is contained in:
ITQ
2026-02-12 20:48:29 +03:00
parent cb9692089f
commit 613c99dce2
60 changed files with 5101 additions and 127 deletions
+55
View File
@@ -0,0 +1,55 @@
# Lotty A/B Platform
[![wakatime](https://wakatime.com/badge/user/cb406c1c-8eb4-4829-b9f9-816a0d284d7e/project/072eef64-ba2e-4a95-a5bb-62e22186eb22.svg)](https://wakatime.com/badge/user/cb406c1c-8eb4-4829-b9f9-816a0d284d7e/project/072eef64-ba2e-4a95-a5bb-62e22186eb22)
Service for managing A/B testing experiments. Drive your tests without breaking user experience!_)
## 📋 Instructions
### Dedicated services setup
[Backend](./src/backend/README.md)
### Setup with docker compose
#### Warning
Please note that by default containers will use ports 80 (reverse proxy) and range 14601-14611 (for direct access to containers), so there is must be no listeners on this ports range. You can customize the ports in [.env.template](./.env.template).
#### 0. Prerequisites
- [Docker](https://www.docker.com/) (latest version recommended)
- [Docker compose](https://www.docker.com/) (latest version recommended)
- Cloned repository
#### 1. Configuration
- Docker compose configuration files are stored in [deploy/compose](./deploy/compose).
- Configuration files for containers are stored in [infrastrucutre/configs](./infrastrucutre/configs).
Env could be customized by creating `.env` file in each service config directory, it will automatically override the default values from `.env.template`.
- Ports on which containers will be accessible are defined in [.env.template](./.env.template). This could be customized by creating `.env` file in the root directory and patching the following lines compose you are running:
```yaml
x-defaults: &defaults
project_directory: ./
env_file:
- ./.env.template
- ./.env # add this
```
#### 2. Choosing compose configuration
- [compose.yaml](./compose.yaml) - default configuration for with base services included.
- [compose.prod.yaml](./compose.prod.yaml) - configuration for production environment with full observability stack.
#### 3. Running compose configuration
To run the compose configuration, use the following command:
```bash
docker compose -f compose.yaml up
# OR
docker compose -f compose.prod.yaml up
```
Thats it, project is already preconfigured for running, so no changes before running this are required.
+3 -3
View File
@@ -110,14 +110,14 @@ Backend will be available on [127.0.0.1:8080](http://127.0.0.1:8080).
uv sync --all-extras uv sync --all-extras
``` ```
### Run tests ### Run tests (with coverage)
```bash ```bash
uv run coverage run --source="." manage.py test just test-coverage
``` ```
### Check coverage ### Check coverage
```bash ```bash
uv run coverage report just show-coverage
``` ```
View File
+7
View File
@@ -0,0 +1,7 @@
from django.apps import AppConfig
class AuthConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "api.v1.auth"
label = "api_v1_auth"
+100
View File
@@ -0,0 +1,100 @@
from http import HTTPStatus
from django.contrib.auth import authenticate
from django.http import HttpRequest
from ninja import Router
from ninja.errors import AuthenticationError
from api.v1.auth.schemas import (
LoginIn,
MeOut,
TokenPairOut,
TokenRefreshIn,
TokenRefreshOut,
)
from apps.users.auth.bearer import jwt_bearer
from apps.users.auth.jwt import (
TokenError,
create_access_token,
create_token_pair,
decode_refresh_token,
)
from apps.users.models import User
router = Router(tags=["auth"])
@router.post(
"/login",
response={HTTPStatus.OK: TokenPairOut},
summary="Obtain JWT token pair",
description=(
"Authenticate with username and password. "
"Returns an access token and a refresh token."
),
)
def login(
request: HttpRequest,
payload: LoginIn,
) -> tuple[int, TokenPairOut]:
user = authenticate(
request,
username=payload.username,
password=payload.password,
)
if user is None or not isinstance(user, User):
raise AuthenticationError
if not user.is_active:
raise AuthenticationError
tokens = create_token_pair(user.pk, user.role)
return HTTPStatus.OK, TokenPairOut(
access=tokens["access"],
refresh=tokens["refresh"],
)
@router.post(
"/refresh",
response={HTTPStatus.OK: TokenRefreshOut},
summary="Refresh access token",
description="Exchange a valid refresh token for a new access token.",
)
def refresh(
request: HttpRequest,
payload: TokenRefreshIn,
) -> tuple[int, TokenRefreshOut]:
try:
claims = decode_refresh_token(payload.refresh)
except TokenError:
raise AuthenticationError from TokenError
user_id: str | None = claims.get("sub")
if not user_id:
raise AuthenticationError
try:
user = User.objects.get(pk=user_id, is_active=True)
except User.DoesNotExist:
raise AuthenticationError from User.DoesNotExist
access = create_access_token(user.pk, user.role)
return HTTPStatus.OK, TokenRefreshOut(access=access)
@router.get(
"/me",
response={HTTPStatus.OK: MeOut},
auth=jwt_bearer,
summary="Current user profile",
description="Return the profile of the currently authenticated user.",
)
def me(request: HttpRequest) -> tuple[int, MeOut]:
user = getattr(request, "auth", None)
if not isinstance(user, User):
raise AuthenticationError
return HTTPStatus.OK, MeOut.model_validate(user)
+64
View File
@@ -0,0 +1,64 @@
from typing import ClassVar
from ninja import ModelSchema, Schema
from pydantic import Field
from apps.users.models import User
class LoginIn(Schema):
username: str = Field(
...,
min_length=1,
max_length=150,
description="Username of the account",
)
password: str = Field(
...,
min_length=1,
description="Account password",
)
class TokenPairOut(Schema):
access: str = Field(
...,
description=(
"Short-lived JWT access token (used in Authorization header)"
),
)
refresh: str = Field(
...,
description=(
"Long-lived JWT refresh token (used to obtain a new access token)"
),
)
class TokenRefreshIn(Schema):
refresh: str = Field(
...,
min_length=1,
description="A valid, non-expired refresh token",
)
class TokenRefreshOut(Schema):
access: str = Field(
...,
description="Newly issued JWT access token",
)
class MeOut(ModelSchema):
class Meta:
model = User
fields: ClassVar[tuple[str, ...]] = (
User.id.field.name,
User.username.field.name,
User.email.field.name,
User.role.field.name,
User.first_name.field.name,
User.last_name.field.name,
User.is_active.field.name,
)
+104
View File
@@ -0,0 +1,104 @@
import json
from django.test import Client, TestCase
from django.urls import reverse
from apps.users.auth.jwt import (
create_token_pair,
)
from apps.users.models import User, UserRole
from apps.users.tests._helpers import _auth_header, _make_user
class AuthAPITest(TestCase):
def setUp(self) -> None:
self.client = Client()
self.user: User = _make_user(
username="api_auth",
email="api_auth@x.com",
password="testpass123",
role=UserRole.ADMIN,
)
def test_login_success(self) -> None:
resp = self.client.post(
reverse("api-1:login"),
data=json.dumps(
{"username": "api_auth", "password": "testpass123"}
),
content_type="application/json",
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertIn("access", data)
self.assertIn("refresh", data)
def test_login_wrong_password(self) -> None:
resp = self.client.post(
reverse("api-1:login"),
data=json.dumps({"username": "api_auth", "password": "wrong"}),
content_type="application/json",
)
self.assertEqual(resp.status_code, 401)
def test_login_nonexistent_user(self) -> None:
resp = self.client.post(
reverse("api-1:login"),
data=json.dumps({"username": "ghost", "password": "nope"}),
content_type="application/json",
)
self.assertEqual(resp.status_code, 401)
def test_login_inactive_user(self) -> None:
self.user.is_active = False
self.user.save()
resp = self.client.post(
reverse("api-1:login"),
data=json.dumps(
{"username": "api_auth", "password": "testpass123"}
),
content_type="application/json",
)
self.assertEqual(resp.status_code, 401)
def test_refresh_success(self) -> None:
pair: dict[str, str] = create_token_pair(self.user.pk, self.user.role)
resp = self.client.post(
reverse("api-1:refresh"),
data=json.dumps({"refresh": pair["refresh"]}),
content_type="application/json",
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertIn("access", data)
def test_refresh_invalid_token(self) -> None:
resp = self.client.post(
reverse("api-1:refresh"),
data=json.dumps({"refresh": "invalid.token.here"}),
content_type="application/json",
)
self.assertEqual(resp.status_code, 401)
def test_refresh_with_access_token_fails(self) -> None:
pair: dict[str, str] = create_token_pair(self.user.pk, self.user.role)
resp = self.client.post(
reverse("api-1:refresh"),
data=json.dumps({"refresh": pair["access"]}),
content_type="application/json",
)
self.assertEqual(resp.status_code, 401)
def test_me_authenticated(self) -> None:
resp = self.client.get(
reverse("api-1:me"),
HTTP_AUTHORIZATION=_auth_header(self.user),
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertEqual(data["username"], "api_auth")
self.assertEqual(data["role"], "admin")
def test_me_unauthenticated(self) -> None:
resp = self.client.get(reverse("api-1:me"))
self.assertEqual(resp.status_code, 401)
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class ReviewsConfig(AppConfig):
name = "api.v1.reviews"
label = "api_v1_reviews"
+344
View File
@@ -0,0 +1,344 @@
from http import HTTPStatus
from django.core.exceptions import ValidationError
from django.http import Http404, HttpRequest
from django.shortcuts import get_object_or_404
from ninja import Router
from api.v1.reviews.schemas import (
ApproverGroupAddApproverIn,
ApproverGroupCreateIn,
ApproverGroupListOut,
ApproverGroupOut,
ApproverGroupRemoveApproverIn,
ApproverGroupUpdateIn,
ApproverOut,
EffectiveReviewPolicyOut,
ReviewSettingsOut,
ReviewSettingsUpdateIn,
)
from apps.reviews.models import ApproverGroup
from apps.reviews.selectors import (
approver_group_get_by_experimenter_id,
approver_group_list,
get_effective_approvers_for_experimenter,
review_settings_load,
)
from apps.reviews.services import (
approver_group_add_approver,
approver_group_create,
approver_group_delete,
approver_group_remove_approver,
approver_group_update,
review_settings_update,
)
from apps.users.auth.bearer import jwt_bearer, require_admin
from apps.users.models import User, UserRole
router = Router(tags=["reviews"], auth=jwt_bearer)
@router.get(
"/approver-groups",
response={HTTPStatus.OK: ApproverGroupListOut},
summary="List approver groups",
description=(
"Return a paginated list of all approver groups. Admin only."
),
)
@require_admin
def list_approver_groups(
request: HttpRequest,
limit: int = 50,
offset: int = 0,
) -> tuple[int, ApproverGroupListOut]:
qs = approver_group_list()
total = qs.count()
items = [
ApproverGroupOut.model_validate(item)
for item in qs[offset : offset + limit]
]
return HTTPStatus.OK, ApproverGroupListOut(count=total, items=items)
@router.post(
"/approver-groups",
response={HTTPStatus.CREATED: ApproverGroupOut},
summary="Create approver group",
description=(
"Create a new approver group for an experimenter. "
"Each experimenter can have at most one group. Admin only."
),
)
@require_admin
def create_approver_group(
request: HttpRequest,
payload: ApproverGroupCreateIn,
) -> tuple[int, ApproverGroupOut]:
experimenter = get_object_or_404(User, pk=payload.experimenter_id)
group = approver_group_create(
experimenter=experimenter,
approver_ids=payload.approver_ids or None,
min_approvals=payload.min_approvals,
)
response_group = get_object_or_404(
ApproverGroup.objects.select_related("experimenter").prefetch_related(
"approvers"
),
pk=group.pk,
)
return HTTPStatus.CREATED, ApproverGroupOut.model_validate(response_group)
@router.get(
"/approver-groups/{group_id}",
response={HTTPStatus.OK: ApproverGroupOut},
summary="Get approver group",
description="Retrieve a single approver group by its UUID. Admin only.",
)
@require_admin
def get_approver_group(
request: HttpRequest,
group_id: str,
) -> tuple[int, ApproverGroupOut]:
group = get_object_or_404(
ApproverGroup.objects.select_related("experimenter").prefetch_related(
"approvers"
),
pk=group_id,
)
return HTTPStatus.OK, ApproverGroupOut.model_validate(group)
@router.patch(
"/approver-groups/{group_id}",
response={HTTPStatus.OK: ApproverGroupOut},
summary="Update approver group",
description=(
"Partially update an existing approver group. "
"Only non-null fields in the payload are applied. Admin only."
),
)
@require_admin
def update_approver_group(
request: HttpRequest,
group_id: str,
payload: ApproverGroupUpdateIn,
) -> tuple[int, ApproverGroupOut]:
group = get_object_or_404(
ApproverGroup.objects.select_related("experimenter").prefetch_related(
"approvers"
),
pk=group_id,
)
updated = approver_group_update(
group=group,
approver_ids=payload.approver_ids,
min_approvals=payload.min_approvals,
)
response_group = get_object_or_404(
ApproverGroup.objects.select_related("experimenter").prefetch_related(
"approvers"
),
pk=updated.pk,
)
return HTTPStatus.OK, ApproverGroupOut.model_validate(response_group)
@router.delete(
"/approver-groups/{group_id}",
response={HTTPStatus.NO_CONTENT: None},
summary="Delete approver group",
description=(
"Delete an approver group. After deletion the experimenter "
"falls back to the global review-settings policy. Admin only."
),
)
@require_admin
def delete_approver_group(
request: HttpRequest,
group_id: str,
) -> tuple[int, None]:
group = get_object_or_404(
ApproverGroup.objects.select_related("experimenter").prefetch_related(
"approvers"
),
pk=group_id,
)
approver_group_delete(group=group)
return HTTPStatus.NO_CONTENT, None
@router.post(
"/approver-groups/{group_id}/approvers/add",
response={HTTPStatus.OK: ApproverGroupOut},
summary="Add approver to group",
description=(
"Add a single approver user to an existing approver group. "
"The user must have the 'approver' role. Admin only."
),
)
@require_admin
def add_approver_to_group(
request: HttpRequest,
group_id: str,
payload: ApproverGroupAddApproverIn,
) -> tuple[int, ApproverGroupOut]:
group = get_object_or_404(
ApproverGroup.objects.select_related("experimenter").prefetch_related(
"approvers"
),
pk=group_id,
)
approver = get_object_or_404(User, pk=payload.approver_id)
approver_group_add_approver(group=group, approver=approver)
response_group = get_object_or_404(
ApproverGroup.objects.select_related("experimenter").prefetch_related(
"approvers"
),
pk=group.pk,
)
return HTTPStatus.OK, ApproverGroupOut.model_validate(response_group)
@router.post(
"/approver-groups/{group_id}/approvers/remove",
response={HTTPStatus.OK: ApproverGroupOut},
summary="Remove approver from group",
description=(
"Remove a single approver user from an existing approver group. "
"Fails if removal would leave fewer approvers than the "
"minimum approval threshold. Admin only."
),
)
@require_admin
def remove_approver_from_group(
request: HttpRequest,
group_id: str,
payload: ApproverGroupRemoveApproverIn,
) -> tuple[int, ApproverGroupOut]:
group = get_object_or_404(
ApproverGroup.objects.select_related("experimenter").prefetch_related(
"approvers"
),
pk=group_id,
)
approver = get_object_or_404(User, pk=payload.approver_id)
approver_group_remove_approver(group=group, approver=approver)
response_group = get_object_or_404(
ApproverGroup.objects.select_related("experimenter").prefetch_related(
"approvers"
),
pk=group.pk,
)
return HTTPStatus.OK, ApproverGroupOut.model_validate(response_group)
@router.get(
"/approver-groups/by-experimenter/{experimenter_id}",
response={HTTPStatus.OK: ApproverGroupOut},
summary="Get approver group by experimenter",
description=(
"Retrieve the approver group assigned to a specific experimenter. "
"Returns 404 if no explicit group exists. "
"Available to any authenticated user."
),
)
@require_admin
def get_approver_group_by_experimenter(
request: HttpRequest,
experimenter_id: str,
) -> tuple[int, ApproverGroupOut]:
group = approver_group_get_by_experimenter_id(experimenter_id)
if group is None:
raise Http404
return HTTPStatus.OK, ApproverGroupOut.model_validate(group)
@router.get(
"/settings",
response={HTTPStatus.OK: ReviewSettingsOut},
summary="Get review settings",
description=(
"Retrieve the global fallback review settings. "
"Available to any authenticated user."
),
)
def get_review_settings(
request: HttpRequest,
) -> tuple[int, ReviewSettingsOut]:
settings = review_settings_load()
return HTTPStatus.OK, ReviewSettingsOut.model_validate(settings)
@router.put(
"/settings",
response={HTTPStatus.OK: ReviewSettingsOut},
summary="Update review settings",
description=(
"Update the global fallback review settings. "
"Only non-null fields in the payload are applied. Admin only."
),
)
@require_admin
def update_review_settings(
request: HttpRequest,
payload: ReviewSettingsUpdateIn,
) -> tuple[int, ReviewSettingsOut]:
settings = review_settings_update(
default_min_approvals=payload.default_min_approvals,
allow_any_approver=payload.allow_any_approver,
)
return HTTPStatus.OK, ReviewSettingsOut.model_validate(settings)
@router.get(
"/effective-policy/{experimenter_id}",
response={HTTPStatus.OK: EffectiveReviewPolicyOut},
summary="Get effective review policy for an experimenter",
description=(
"Resolve and return the effective review policy for a given "
"experimenter, taking into account both an explicit approver "
"group (if any) and the global fallback settings. "
"Available to any authenticated user."
),
)
def get_effective_policy(
request: HttpRequest,
experimenter_id: str,
) -> tuple[int, EffectiveReviewPolicyOut]:
experimenter = get_object_or_404(User, pk=experimenter_id)
if experimenter.role != UserRole.EXPERIMENTER:
raise ValidationError(
{
"experimenter_id": (
f"User '{experimenter.username}' does not have the "
f"'{UserRole.EXPERIMENTER}' role."
)
}
)
approvers_qs, min_approvals = get_effective_approvers_for_experimenter(
experimenter,
)
has_group = (
approver_group_get_by_experimenter_id(experimenter_id) is not None
)
return HTTPStatus.OK, EffectiveReviewPolicyOut(
experimenter_id=str(experimenter.pk),
min_approvals=min_approvals,
approvers=[
ApproverOut.model_validate(approver) for approver in approvers_qs
],
source="approver_group" if has_group else "global_fallback",
has_explicit_group=has_group,
)
+166
View File
@@ -0,0 +1,166 @@
from datetime import datetime
from typing import ClassVar
from ninja import ModelSchema, Schema
from pydantic import Field
from apps.reviews.models import ApproverGroup, ReviewSettings
from apps.users.models import User
class ApproverOut(ModelSchema):
first_name: str = Field("", alias="firstName")
last_name: str = Field("", alias="lastName")
class Meta:
model = User
fields: ClassVar[tuple[str, ...]] = (
User.id.field.name,
User.username.field.name,
User.email.field.name,
User.first_name.field.name,
User.last_name.field.name,
)
class ExperimenterOut(ModelSchema):
class Meta:
model = User
fields: ClassVar[tuple[str, ...]] = (
User.id.field.name,
User.username.field.name,
User.email.field.name,
)
class ApproverGroupOut(ModelSchema):
experimenter: ExperimenterOut
approvers: list[ApproverOut]
min_approvals: int
created_at: datetime
updated_at: datetime
class Meta:
model = ApproverGroup
fields: ClassVar[tuple[str, ...]] = (
ApproverGroup.id.field.name,
ApproverGroup.experimenter.field.name,
ApproverGroup.approvers.field.name,
ApproverGroup.min_approvals.field.name,
ApproverGroup.created_at.field.name,
ApproverGroup.updated_at.field.name,
)
class ApproverGroupCreateIn(Schema):
experimenter_id: str = Field(
...,
alias="experimenterId",
description="UUID of the experimenter user this group belongs to.",
)
approver_ids: list[str] = Field(
default_factory=list,
alias="approverIds",
description=(
"List of user UUIDs to add as approvers. "
"Each user must have the 'approver' role."
),
)
min_approvals: int = Field(
1,
alias="minApprovals",
ge=1,
description=(
"Number of distinct approvals required. "
"Must be >= 1 and <= number of approvers (if any are provided)."
),
)
class ApproverGroupUpdateIn(Schema):
approver_ids: list[str] | None = Field(
None,
alias="approverIds",
description=(
"If provided, replaces the current set of approvers. "
"Each user must have the 'approver' role."
),
)
min_approvals: int | None = Field(
None,
alias="minApprovals",
ge=1,
description="New minimum approval threshold.",
)
class ApproverGroupListOut(Schema):
count: int
items: list[ApproverGroupOut]
class ApproverGroupAddApproverIn(Schema):
approver_id: str = Field(
...,
alias="approverId",
description="UUID of the user to add as an approver.",
)
class ApproverGroupRemoveApproverIn(Schema):
approver_id: str = Field(
...,
alias="approverId",
description="UUID of the approver to remove.",
)
class ReviewSettingsOut(ModelSchema):
default_min_approvals: int
allow_any_approver: bool
updated_at: datetime
class Meta:
model = ReviewSettings
fields: ClassVar[tuple[str, ...]] = (
ReviewSettings.id.field.name,
ReviewSettings.default_min_approvals.field.name,
ReviewSettings.allow_any_approver.field.name,
ReviewSettings.updated_at.field.name,
)
class ReviewSettingsUpdateIn(Schema):
default_min_approvals: int | None = Field(
None,
alias="defaultMinApprovals",
ge=1,
description="New default minimum approval threshold.",
)
allow_any_approver: bool | None = Field(
None,
alias="allowAnyApprover",
description="New fallback policy for approver eligibility.",
)
class EffectiveReviewPolicyOut(Schema):
experimenter_id: str = Field(..., description="UUID of the experimenter.")
min_approvals: int = Field(
..., description="Effective number of approvals required."
)
approvers: list[ApproverOut] = Field(
...,
description="List of users who are eligible to approve.",
)
source: str = Field(
...,
description=(
"Where the policy comes from: "
"'approver_group' or 'global_fallback'."
),
)
has_explicit_group: bool = Field(
...,
description="Whether the experimenter has an explicit approver group.",
)
@@ -0,0 +1,787 @@
import json
import uuid
from django.test import Client, TestCase
from django.urls import reverse
from apps.reviews.models import ApproverGroup
from apps.reviews.services import approver_group_create, review_settings_update
from apps.reviews.tests._helpers import (
_get,
_make_admin,
_make_approver,
_make_experimenter,
_make_viewer,
)
from apps.users.models import User
from apps.users.tests._helpers import _auth_header
class ApproverGroupAPITest(TestCase):
def setUp(self) -> None:
self.client = Client()
self.admin: User = _make_admin("_api")
self.viewer: User = _make_viewer("_api")
self.experimenter: User = _make_experimenter("_api")
self.approver1: User = _make_approver("_api1")
self.approver2: User = _make_approver("_api2")
self.approver3: User = _make_approver("_api3")
self.admin_auth: str = _auth_header(self.admin)
self.viewer_auth: str = _auth_header(self.viewer)
self.exp_auth: str = _auth_header(self.experimenter)
def test_list_groups_admin(self) -> None:
approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver1.pk)],
)
resp = self.client.get(
reverse("api-1:list_approver_groups"),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertIn("count", data)
self.assertIn("items", data)
self.assertEqual(data["count"], 1)
def test_list_groups_viewer_denied(self) -> None:
resp = self.client.get(
reverse("api-1:list_approver_groups"),
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_list_groups_unauthenticated(self) -> None:
resp = self.client.get(reverse("api-1:list_approver_groups"))
self.assertEqual(resp.status_code, 401)
def test_list_groups_pagination(self) -> None:
exp2: User = _make_experimenter("_api2")
approver_group_create(experimenter=self.experimenter)
approver_group_create(experimenter=exp2)
resp = self.client.get(
reverse("api-1:list_approver_groups"),
data={"limit": 1, "offset": 0},
HTTP_AUTHORIZATION=self.admin_auth,
)
data = resp.json()
self.assertEqual(data["count"], 2)
self.assertEqual(len(data["items"]), 1)
def test_create_group_admin(self) -> None:
resp = self.client.post(
reverse("api-1:create_approver_group"),
data=json.dumps(
{
"experimenterId": str(self.experimenter.pk),
"approverIds": [
str(self.approver1.pk),
str(self.approver2.pk),
],
"minApprovals": 2,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 201)
data = resp.json()
self.assertEqual(_get(data, "minApprovals", "min_approvals"), 2)
self.assertEqual(len(data["approvers"]), 2)
self.assertEqual(data["experimenter"]["id"], str(self.experimenter.pk))
def test_create_group_without_approvers(self) -> None:
resp = self.client.post(
reverse("api-1:create_approver_group"),
data=json.dumps(
{
"experimenterId": str(self.experimenter.pk),
"minApprovals": 1,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 201)
data = resp.json()
self.assertEqual(len(data["approvers"]), 0)
def test_create_group_viewer_denied(self) -> None:
resp = self.client.post(
reverse("api-1:create_approver_group"),
data=json.dumps(
{
"experimenterId": str(self.experimenter.pk),
"minApprovals": 1,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_create_group_experimenter_denied(self) -> None:
resp = self.client.post(
reverse("api-1:create_approver_group"),
data=json.dumps(
{
"experimenterId": str(self.experimenter.pk),
"minApprovals": 1,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.exp_auth,
)
self.assertEqual(resp.status_code, 403)
def test_create_group_nonexistent_experimenter(self) -> None:
resp = self.client.post(
reverse("api-1:create_approver_group"),
data=json.dumps(
{
"experimenterId": str(uuid.uuid4()),
"minApprovals": 1,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
def test_create_group_experimenter_wrong_role(self) -> None:
resp = self.client.post(
reverse("api-1:create_approver_group"),
data=json.dumps(
{
"experimenterId": str(self.viewer.pk),
"minApprovals": 1,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
def test_create_group_duplicate_raises(self) -> None:
self.client.post(
reverse("api-1:create_approver_group"),
data=json.dumps(
{
"experimenterId": str(self.experimenter.pk),
"minApprovals": 1,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
resp = self.client.post(
reverse("api-1:create_approver_group"),
data=json.dumps(
{
"experimenterId": str(self.experimenter.pk),
"minApprovals": 1,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400, 409])
def test_create_group_approver_wrong_role(self) -> None:
resp = self.client.post(
reverse("api-1:create_approver_group"),
data=json.dumps(
{
"experimenterId": str(self.experimenter.pk),
"approverIds": [str(self.viewer.pk)],
"minApprovals": 1,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
def test_get_group_admin(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver1.pk)],
)
resp = self.client.get(
reverse(
"api-1:get_approver_group",
kwargs={"group_id": str(group.pk)},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertEqual(data["id"], str(group.pk))
self.assertEqual(len(data["approvers"]), 1)
def test_get_group_not_found(self) -> None:
resp = self.client.get(
reverse(
"api-1:get_approver_group",
kwargs={"group_id": str(uuid.uuid4())},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
def test_get_group_viewer_denied(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter
)
resp = self.client.get(
reverse(
"api-1:get_approver_group",
kwargs={"group_id": str(group.pk)},
),
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_update_group_admin(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver1.pk)],
min_approvals=1,
)
resp = self.client.patch(
reverse(
"api-1:update_approver_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps(
{
"approverIds": [
str(self.approver1.pk),
str(self.approver2.pk),
],
"minApprovals": 2,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertEqual(_get(data, "minApprovals", "min_approvals"), 2)
self.assertEqual(len(data["approvers"]), 2)
def test_update_group_partial_min_approvals(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[
str(self.approver1.pk),
str(self.approver2.pk),
],
min_approvals=1,
)
resp = self.client.patch(
reverse(
"api-1:update_approver_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps({"minApprovals": 2}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertEqual(_get(data, "minApprovals", "min_approvals"), 2)
def test_update_group_not_found(self) -> None:
resp = self.client.patch(
reverse(
"api-1:update_approver_group",
kwargs={"group_id": str(uuid.uuid4())},
),
data=json.dumps({"minApprovals": 1}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
def test_update_group_viewer_denied(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter
)
resp = self.client.patch(
reverse(
"api-1:update_approver_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps({"minApprovals": 1}),
content_type="application/json",
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_delete_group_admin(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter
)
pk = group.pk
resp = self.client.delete(
reverse(
"api-1:delete_approver_group",
kwargs={"group_id": str(pk)},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 204)
self.assertFalse(ApproverGroup.objects.filter(pk=pk).exists())
def test_delete_group_not_found(self) -> None:
resp = self.client.delete(
reverse(
"api-1:delete_approver_group",
kwargs={"group_id": str(uuid.uuid4())},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
def test_delete_group_viewer_denied(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter
)
resp = self.client.delete(
reverse(
"api-1:delete_approver_group",
kwargs={"group_id": str(group.pk)},
),
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_add_approver_to_group_admin(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver1.pk)],
)
resp = self.client.post(
reverse(
"api-1:add_approver_to_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps({"approverId": str(self.approver2.pk)}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
self.assertEqual(len(resp.json()["approvers"]), 2)
def test_add_approver_not_found_group(self) -> None:
resp = self.client.post(
reverse(
"api-1:add_approver_to_group",
kwargs={"group_id": str(uuid.uuid4())},
),
data=json.dumps({"approverId": str(self.approver1.pk)}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
def test_add_approver_not_found_user(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter
)
resp = self.client.post(
reverse(
"api-1:add_approver_to_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps({"approverId": str(uuid.uuid4())}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
def test_add_approver_wrong_role(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter
)
resp = self.client.post(
reverse(
"api-1:add_approver_to_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps({"approverId": str(self.viewer.pk)}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
def test_add_approver_duplicate(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver1.pk)],
)
resp = self.client.post(
reverse(
"api-1:add_approver_to_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps({"approverId": str(self.approver1.pk)}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
def test_add_approver_viewer_denied(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter
)
resp = self.client.post(
reverse(
"api-1:add_approver_to_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps({"approverId": str(self.approver1.pk)}),
content_type="application/json",
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_remove_approver_from_group_admin(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[
str(self.approver1.pk),
str(self.approver2.pk),
],
min_approvals=1,
)
resp = self.client.post(
reverse(
"api-1:remove_approver_from_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps({"approverId": str(self.approver2.pk)}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
self.assertEqual(len(resp.json()["approvers"]), 1)
def test_remove_approver_below_min_raises(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver1.pk)],
min_approvals=1,
)
resp = self.client.post(
reverse(
"api-1:remove_approver_from_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps({"approverId": str(self.approver1.pk)}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
def test_remove_approver_not_in_group(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver1.pk)],
)
resp = self.client.post(
reverse(
"api-1:remove_approver_from_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps({"approverId": str(self.approver2.pk)}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
def test_remove_approver_viewer_denied(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[
str(self.approver1.pk),
str(self.approver2.pk),
],
)
resp = self.client.post(
reverse(
"api-1:remove_approver_from_group",
kwargs={"group_id": str(group.pk)},
),
data=json.dumps({"approverId": str(self.approver1.pk)}),
content_type="application/json",
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
class ApproverGroupByExperimenterAPITest(TestCase):
def setUp(self) -> None:
self.client = Client()
self.admin: User = _make_admin("_byexp")
self.experimenter: User = _make_experimenter("_byexp")
self.approver: User = _make_approver("_byexp")
self.admin_auth: str = _auth_header(self.admin)
def test_get_by_experimenter_found(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver.pk)],
)
resp = self.client.get(
reverse(
"api-1:get_approver_group_by_experimenter",
kwargs={"experimenter_id": str(self.experimenter.pk)},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp.json()["id"], str(group.pk))
def test_get_by_experimenter_not_found(self) -> None:
exp2: User = _make_experimenter("_byexp2")
resp = self.client.get(
reverse(
"api-1:get_approver_group_by_experimenter",
kwargs={"experimenter_id": str(exp2.pk)},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
def test_get_by_experimenter_invalid_id(self) -> None:
resp = self.client.get(
reverse(
"api-1:get_approver_group_by_experimenter",
kwargs={"experimenter_id": str(uuid.uuid4())},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
def test_get_by_experimenter_unauthenticated(self) -> None:
resp = self.client.get(
reverse(
"api-1:get_approver_group_by_experimenter",
kwargs={"experimenter_id": str(self.experimenter.pk)},
)
)
self.assertEqual(resp.status_code, 401)
class ReviewSettingsAPITest(TestCase):
def setUp(self) -> None:
self.client = Client()
self.admin: User = _make_admin("_rs")
self.viewer: User = _make_viewer("_rs")
self.experimenter: User = _make_experimenter("_rs")
self.approver: User = _make_approver("_rs")
self.admin_auth: str = _auth_header(self.admin)
self.viewer_auth: str = _auth_header(self.viewer)
self.exp_auth: str = _auth_header(self.experimenter)
self.appr_auth: str = _auth_header(self.approver)
def test_get_settings_admin(self) -> None:
resp = self.client.get(
reverse("api-1:get_review_settings"),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
min_val = _get(data, "defaultMinApprovals", "default_min_approvals")
any_val = _get(data, "allowAnyApprover", "allow_any_approver")
self.assertEqual(min_val, 1)
self.assertFalse(any_val)
def test_get_settings_any_authenticated_role(self) -> None:
for auth in [self.viewer_auth, self.exp_auth, self.appr_auth]:
resp = self.client.get(
reverse("api-1:get_review_settings"),
HTTP_AUTHORIZATION=auth,
)
self.assertEqual(resp.status_code, 200)
def test_get_settings_unauthenticated(self) -> None:
resp = self.client.get(reverse("api-1:get_review_settings"))
self.assertEqual(resp.status_code, 401)
def test_update_settings_admin(self) -> None:
resp = self.client.put(
reverse("api-1:update_review_settings"),
data=json.dumps(
{"defaultMinApprovals": 3, "allowAnyApprover": False}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
min_val = _get(data, "defaultMinApprovals", "default_min_approvals")
any_val = _get(data, "allowAnyApprover", "allow_any_approver")
self.assertEqual(min_val, 3)
self.assertFalse(any_val)
def test_update_settings_partial(self) -> None:
resp = self.client.put(
reverse("api-1:update_review_settings"),
data=json.dumps({"defaultMinApprovals": 5}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
min_val = _get(data, "defaultMinApprovals", "default_min_approvals")
self.assertEqual(min_val, 5)
def test_update_settings_viewer_denied(self) -> None:
resp = self.client.put(
reverse("api-1:update_review_settings"),
data=json.dumps({"defaultMinApprovals": 2}),
content_type="application/json",
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_update_settings_experimenter_denied(self) -> None:
resp = self.client.put(
reverse("api-1:update_review_settings"),
data=json.dumps({"defaultMinApprovals": 2}),
content_type="application/json",
HTTP_AUTHORIZATION=self.exp_auth,
)
self.assertEqual(resp.status_code, 403)
def test_update_settings_approver_denied(self) -> None:
resp = self.client.put(
reverse("api-1:update_review_settings"),
data=json.dumps({"defaultMinApprovals": 2}),
content_type="application/json",
HTTP_AUTHORIZATION=self.appr_auth,
)
self.assertEqual(resp.status_code, 403)
def test_update_settings_invalid_min_approvals(self) -> None:
resp = self.client.put(
reverse("api-1:update_review_settings"),
data=json.dumps({"defaultMinApprovals": 0}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
class EffectivePolicyAPITest(TestCase):
def setUp(self) -> None:
self.client = Client()
self.admin: User = _make_admin("_pol")
self.experimenter: User = _make_experimenter("_pol")
self.exp_no_group: User = _make_experimenter("_pol2")
self.approver: User = _make_approver("_pol")
self.admin_auth: str = _auth_header(self.admin)
self.viewer: User = _make_viewer("_pol")
self.viewer_auth: str = _auth_header(self.viewer)
self.group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver.pk)],
min_approvals=1,
)
def test_effective_policy_with_group(self) -> None:
resp = self.client.get(
reverse(
"api-1:get_effective_policy",
kwargs={"experimenter_id": str(self.experimenter.pk)},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
exp_id = _get(data, "experimenterId", "experimenter_id")
min_val = _get(data, "minApprovals", "min_approvals")
has_group = _get(data, "hasExplicitGroup", "has_explicit_group")
self.assertEqual(exp_id, str(self.experimenter.pk))
self.assertEqual(min_val, 1)
self.assertEqual(data["source"], "approver_group")
self.assertTrue(has_group)
self.assertEqual(len(data["approvers"]), 1)
self.assertEqual(data["approvers"][0]["id"], str(self.approver.pk))
def test_effective_policy_fallback(self) -> None:
review_settings_update(
default_min_approvals=2, allow_any_approver=True
)
resp = self.client.get(
reverse(
"api-1:get_effective_policy",
kwargs={"experimenter_id": str(self.exp_no_group.pk)},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
min_val = _get(data, "minApprovals", "min_approvals")
has_group = _get(data, "hasExplicitGroup", "has_explicit_group")
self.assertEqual(data["source"], "global_fallback")
self.assertFalse(has_group)
self.assertEqual(min_val, 2)
def test_effective_policy_fallback_deny(self) -> None:
review_settings_update(allow_any_approver=False)
resp = self.client.get(
reverse(
"api-1:get_effective_policy",
kwargs={"experimenter_id": str(self.exp_no_group.pk)},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertEqual(data["source"], "global_fallback")
self.assertEqual(len(data["approvers"]), 0)
def test_effective_policy_non_experimenter_user(self) -> None:
resp = self.client.get(
reverse(
"api-1:get_effective_policy",
kwargs={"experimenter_id": str(self.viewer.pk)},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
def test_effective_policy_nonexistent_user(self) -> None:
resp = self.client.get(
reverse(
"api-1:get_effective_policy",
kwargs={"experimenter_id": str(uuid.uuid4())},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
def test_effective_policy_unauthenticated(self) -> None:
resp = self.client.get(
reverse(
"api-1:get_effective_policy",
kwargs={"experimenter_id": str(self.experimenter.pk)},
)
)
self.assertEqual(resp.status_code, 401)
def test_effective_policy_any_authenticated_can_read(self) -> None:
resp = self.client.get(
reverse(
"api-1:get_effective_policy",
kwargs={"experimenter_id": str(self.experimenter.pk)},
),
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 200)
@@ -0,0 +1,317 @@
import json
from django.core.exceptions import ValidationError
from django.test import Client, TestCase
from django.urls import reverse
from apps.reviews.models import ApproverGroup
from apps.reviews.selectors import (
get_effective_approvers_for_experimenter,
get_min_approvals_for_experimenter,
)
from apps.reviews.services import (
approver_group_create,
approver_group_delete,
approver_group_update,
review_settings_update,
)
from apps.reviews.tests._helpers import (
_make_admin,
_make_approver,
_make_experimenter,
_make_viewer,
)
from apps.users.models import User
from apps.users.tests._helpers import _auth_header
class ReviewRBACTest(TestCase):
def setUp(self) -> None:
self.client = Client()
self.admin: User = _make_admin("_rbac")
self.experimenter: User = _make_experimenter("_rbac")
self.approver: User = _make_approver("_rbac")
self.viewer: User = _make_viewer("_rbac")
self.non_admin_users: dict[str, str] = {
"experimenter": _auth_header(self.experimenter),
"approver": _auth_header(self.approver),
"viewer": _auth_header(self.viewer),
}
self.group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver.pk)],
)
def test_list_groups_denied_for_non_admins(self) -> None:
for role_name, auth in self.non_admin_users.items():
resp = self.client.get(
reverse("api-1:list_approver_groups"),
HTTP_AUTHORIZATION=auth,
)
self.assertEqual(resp.status_code, 403, role_name)
def test_create_group_denied_for_non_admins(self) -> None:
for role_name, auth in self.non_admin_users.items():
exp2: User = _make_experimenter(f"_rbac_cr_{role_name}")
resp = self.client.post(
reverse("api-1:create_approver_group"),
data=json.dumps(
{
"experimenterId": str(exp2.pk),
"minApprovals": 1,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=auth,
)
self.assertEqual(resp.status_code, 403, role_name)
def test_get_group_denied_for_non_admins(self) -> None:
for role_name, auth in self.non_admin_users.items():
resp = self.client.get(
reverse(
"api-1:get_approver_group",
kwargs={"group_id": str(self.group.pk)},
),
HTTP_AUTHORIZATION=auth,
)
self.assertEqual(resp.status_code, 403, role_name)
def test_update_group_denied_for_non_admins(self) -> None:
for role_name, auth in self.non_admin_users.items():
resp = self.client.patch(
reverse(
"api-1:update_approver_group",
kwargs={"group_id": str(self.group.pk)},
),
data=json.dumps({"minApprovals": 1}),
content_type="application/json",
HTTP_AUTHORIZATION=auth,
)
self.assertEqual(resp.status_code, 403, role_name)
def test_delete_group_denied_for_non_admins(self) -> None:
for role_name, auth in self.non_admin_users.items():
resp = self.client.delete(
reverse(
"api-1:delete_approver_group",
kwargs={"group_id": str(self.group.pk)},
),
HTTP_AUTHORIZATION=auth,
)
self.assertEqual(resp.status_code, 403, role_name)
def test_add_approver_denied_for_non_admins(self) -> None:
appr2: User = _make_approver("_rbac_add")
for role_name, auth in self.non_admin_users.items():
resp = self.client.post(
reverse(
"api-1:add_approver_to_group",
kwargs={"group_id": str(self.group.pk)},
),
data=json.dumps({"approverId": str(appr2.pk)}),
content_type="application/json",
HTTP_AUTHORIZATION=auth,
)
self.assertEqual(resp.status_code, 403, role_name)
def test_remove_approver_denied_for_non_admins(self) -> None:
for role_name, auth in self.non_admin_users.items():
resp = self.client.post(
reverse(
"api-1:remove_approver_from_group",
kwargs={"group_id": str(self.group.pk)},
),
data=json.dumps({"approverId": str(self.approver.pk)}),
content_type="application/json",
HTTP_AUTHORIZATION=auth,
)
self.assertEqual(resp.status_code, 403, role_name)
def test_update_settings_denied_for_non_admins(self) -> None:
for role_name, auth in self.non_admin_users.items():
resp = self.client.put(
reverse("api-1:update_review_settings"),
data=json.dumps({"defaultMinApprovals": 99}),
content_type="application/json",
HTTP_AUTHORIZATION=auth,
)
self.assertEqual(resp.status_code, 403, role_name)
class ReviewEdgeCasesTest(TestCase):
def setUp(self) -> None:
self.client = Client()
self.admin: User = _make_admin("_edge")
self.admin_auth: str = _auth_header(self.admin)
def test_delete_group_then_fallback_applies(self) -> None:
exp: User = _make_experimenter("_edge1")
appr: User = _make_approver("_edge1")
group: ApproverGroup = approver_group_create(
experimenter=exp,
approver_ids=[str(appr.pk)],
min_approvals=1,
)
approver_group_delete(group=group)
review_settings_update(
default_min_approvals=3, allow_any_approver=True
)
min_app: int = get_min_approvals_for_experimenter(exp)
self.assertEqual(min_app, 3)
def test_inactive_approver_excluded_from_effective_policy(self) -> None:
exp: User = _make_experimenter("_edge2")
appr: User = _make_approver("_edge2")
approver_group_create(
experimenter=exp,
approver_ids=[str(appr.pk)],
)
appr.is_active = False
appr.save()
approvers, _ = get_effective_approvers_for_experimenter(exp)
self.assertEqual(approvers.count(), 0)
def test_create_group_with_all_three_approvers(self) -> None:
exp: User = _make_experimenter("_edge3")
apprs: list[User] = [_make_approver(f"_edge3_{i}") for i in range(3)]
group: ApproverGroup = approver_group_create(
experimenter=exp,
approver_ids=[str(a.pk) for a in apprs],
min_approvals=2,
)
self.assertEqual(group.approvers.count(), 3)
self.assertEqual(group.min_approvals, 2)
def test_update_group_to_empty_approvers(self) -> None:
exp: User = _make_experimenter("_edge4")
appr: User = _make_approver("_edge4")
group: ApproverGroup = approver_group_create(
experimenter=exp,
approver_ids=[str(appr.pk)],
min_approvals=1,
)
with self.assertRaises(ValidationError):
approver_group_update(
group=group, approver_ids=[], min_approvals=1
)
def test_api_output_format_approver_group(self) -> None:
exp: User = _make_experimenter("_edge5")
appr: User = _make_approver("_edge5")
resp = self.client.post(
reverse("api-1:create_approver_group"),
data=json.dumps(
{
"experimenterId": str(exp.pk),
"approverIds": [str(appr.pk)],
"minApprovals": 1,
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 201)
data = resp.json()
self.assertIn("id", data)
self.assertIn("experimenter", data)
self.assertIn("approvers", data)
self.assertTrue("minApprovals" in data or "min_approvals" in data)
self.assertTrue("createdAt" in data or "created_at" in data)
self.assertTrue("updatedAt" in data or "updated_at" in data)
self.assertIn("id", data["experimenter"])
self.assertIn("username", data["experimenter"])
self.assertIn("email", data["experimenter"])
self.assertEqual(len(data["approvers"]), 1)
appr_data = data["approvers"][0]
self.assertIn("id", appr_data)
self.assertIn("username", appr_data)
self.assertIn("email", appr_data)
def test_api_output_format_review_settings(self) -> None:
resp = self.client.get(
reverse("api-1:get_review_settings"),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertIn("id", data)
self.assertTrue(
"defaultMinApprovals" in data or "default_min_approvals" in data
)
self.assertTrue(
"allowAnyApprover" in data or "allow_any_approver" in data
)
self.assertTrue("updatedAt" in data or "updated_at" in data)
def test_api_output_format_effective_policy(self) -> None:
exp: User = _make_experimenter("_edge6")
appr: User = _make_approver("_edge6")
approver_group_create(
experimenter=exp,
approver_ids=[str(appr.pk)],
)
resp = self.client.get(
reverse(
"api-1:get_effective_policy",
kwargs={"experimenter_id": str(exp.pk)},
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertTrue("experimenterId" in data or "experimenter_id" in data)
self.assertTrue("minApprovals" in data or "min_approvals" in data)
self.assertIn("approvers", data)
self.assertIn("source", data)
self.assertTrue(
"hasExplicitGroup" in data or "has_explicit_group" in data
)
def test_multiple_experimenters_independent_groups(self) -> None:
exp1: User = _make_experimenter("_edge_m1")
exp2: User = _make_experimenter("_edge_m2")
appr1: User = _make_approver("_edge_m1")
appr2: User = _make_approver("_edge_m2")
g1: ApproverGroup = approver_group_create(
experimenter=exp1,
approver_ids=[str(appr1.pk)],
min_approvals=1,
)
g2: ApproverGroup = approver_group_create(
experimenter=exp2,
approver_ids=[str(appr2.pk)],
min_approvals=1,
)
self.assertNotEqual(g1.pk, g2.pk)
self.assertTrue(g1.can_approve(appr1))
self.assertFalse(g1.can_approve(appr2))
self.assertTrue(g2.can_approve(appr2))
self.assertFalse(g2.can_approve(appr1))
def test_concurrent_fallback_and_explicit_group(self) -> None:
exp_with: User = _make_experimenter("_edge_c1")
exp_without: User = _make_experimenter("_edge_c2")
appr: User = _make_approver("_edge_c")
approver_group_create(
experimenter=exp_with,
approver_ids=[str(appr.pk)],
min_approvals=1,
)
review_settings_update(
default_min_approvals=5, allow_any_approver=True
)
approvers1, min1 = get_effective_approvers_for_experimenter(exp_with)
self.assertEqual(min1, 1)
self.assertEqual(approvers1.count(), 1)
approvers2, min2 = get_effective_approvers_for_experimenter(
exp_without
)
self.assertEqual(min2, 5)
self.assertGreaterEqual(approvers2.count(), 1)
+17 -4
View File
@@ -7,6 +7,9 @@ from ninja import NinjaAPI, Schema
from ninja.renderers import BaseRenderer from ninja.renderers import BaseRenderer
from api.v1 import handlers from api.v1 import handlers
from api.v1.auth.endpoints import router as auth_router
from api.v1.reviews.endpoints import router as reviews_router
from api.v1.users.endpoints import router as users_router
class ORJSONRenderer(BaseRenderer): class ORJSONRenderer(BaseRenderer):
@@ -33,10 +36,20 @@ router = NinjaAPI(
) )
# router.add_router( router.add_router(
# "health", "auth",
# health_router, auth_router,
# ) )
router.add_router(
"users",
users_router,
)
router.add_router(
"reviews",
reviews_router,
)
for exception, handler in handlers.exception_handlers: for exception, handler in handlers.exception_handlers:
+1 -5
View File
@@ -2,12 +2,10 @@ from datetime import datetime
from typing import Any from typing import Any
from ninja import Schema from ninja import Schema
from pydantic import ConfigDict, Field from pydantic import Field
class FieldError(Schema): class FieldError(Schema):
model_config = ConfigDict(populate_by_name=True)
field: str = Field( field: str = Field(
..., ...,
description="Field name with error (can be nested)", description="Field name with error (can be nested)",
@@ -19,8 +17,6 @@ class FieldError(Schema):
class ApiError(Schema): class ApiError(Schema):
model_config = ConfigDict(populate_by_name=True)
code: str code: str
message: str message: str
trace_id: str = Field(..., alias="traceId") trace_id: str = Field(..., alias="traceId")
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class UsersConfig(AppConfig):
name = "api.v1.users"
label = "api_v1_users"
+192
View File
@@ -0,0 +1,192 @@
from http import HTTPStatus
from django.core.exceptions import ValidationError
from django.http import HttpRequest
from django.shortcuts import get_object_or_404
from ninja import Router
from api.v1.users.schemas import (
UserCreateIn,
UserListOut,
UserOut,
UserRoleAssignIn,
UserUpdateIn,
)
from apps.users.auth.bearer import jwt_bearer, require_admin
from apps.users.models import User, UserRole
from apps.users.selectors import user_list
from apps.users.services import (
user_assign_role,
user_create,
user_delete,
user_update,
)
router = Router(tags=["users"])
@router.get(
"",
response={HTTPStatus.OK: UserListOut},
auth=jwt_bearer,
summary="List users",
description=(
"Return a filtered, paginated list of platform users. Admin only."
),
)
@require_admin
def list_users(
request: HttpRequest,
role: str | None = None,
is_active: bool | None = None,
search: str | None = None,
limit: int = 50,
offset: int = 0,
) -> tuple[int, UserListOut]:
qs = user_list(role=role, is_active=is_active, search=search)
total = qs.count()
items = [
UserOut.model_validate(item) for item in qs[offset : offset + limit]
]
return HTTPStatus.OK, UserListOut(count=total, items=items)
@router.post(
"",
response={201: UserOut},
auth=jwt_bearer,
summary="Create user",
description=(
"Create a new platform user with the specified role. Admin only."
),
)
@require_admin
def create_user(
request: HttpRequest,
payload: UserCreateIn,
) -> tuple[int, UserOut]:
valid_roles = {choice[0] for choice in UserRole.choices}
if payload.role not in valid_roles:
raise ValidationError(
{
"role": (
f"Invalid role '{payload.role}'. "
f"Must be one of: {', '.join(sorted(valid_roles))}"
)
}
)
user = user_create(
username=payload.username,
email=payload.email,
password=payload.password,
role=payload.role,
first_name=payload.first_name,
last_name=payload.last_name,
)
return HTTPStatus.CREATED, UserOut.model_validate(user)
@router.get(
"/{user_id}",
response={HTTPStatus.OK: UserOut},
auth=jwt_bearer,
summary="Get user",
description="Retrieve a single user by their UUID. Admin only.",
)
@require_admin
def get_user(
request: HttpRequest,
user_id: str,
) -> tuple[int, UserOut]:
user = get_object_or_404(User, pk=user_id)
return HTTPStatus.OK, UserOut.model_validate(user)
@router.patch(
"/{user_id}",
response={HTTPStatus.OK: UserOut},
auth=jwt_bearer,
summary="Update user",
description=(
"Partially update an existing user. "
"Only non-null fields in the payload are applied. Admin only."
),
)
@require_admin
def update_user(
request: HttpRequest,
user_id: str,
payload: UserUpdateIn,
) -> tuple[int, UserOut]:
user = get_object_or_404(User, pk=user_id)
if payload.role is not None:
valid_roles = {choice[0] for choice in UserRole.choices}
if payload.role not in valid_roles:
raise ValidationError(
{
"role": (
f"Invalid role '{payload.role}'. "
f"Must be one of: {', '.join(sorted(valid_roles))}"
)
}
)
updated_user = user_update(
user=user,
username=payload.username,
email=payload.email,
password=payload.password,
role=payload.role,
is_active=payload.is_active,
first_name=payload.first_name,
last_name=payload.last_name,
)
return HTTPStatus.OK, UserOut.model_validate(updated_user)
@router.delete(
"/{user_id}",
response={204: None},
auth=jwt_bearer,
summary="Delete user",
description="Permanently delete a user. Admin only.",
)
@require_admin
def delete_user(
request: HttpRequest,
user_id: str,
) -> tuple[int, None]:
user = get_object_or_404(User, pk=user_id)
current_user = getattr(request, "auth", None)
if not isinstance(current_user, User):
raise ValidationError({"user": "Authentication required."})
if str(user.pk) == str(current_user.pk):
raise ValidationError({"user": "You cannot delete your own account."})
user_delete(user=user)
return HTTPStatus.NO_CONTENT, None
@router.post(
"/{user_id}/role",
response={HTTPStatus.OK: UserOut},
auth=jwt_bearer,
summary="Assign role",
description="Change the platform role of an existing user. Admin only.",
)
@require_admin
def assign_role(
request: HttpRequest,
user_id: str,
payload: UserRoleAssignIn,
) -> tuple[int, UserOut]:
user = get_object_or_404(User, pk=user_id)
updated_user = user_assign_role(user=user, role=payload.role)
return HTTPStatus.OK, UserOut.model_validate(updated_user)
+123
View File
@@ -0,0 +1,123 @@
from typing import ClassVar
from ninja import ModelSchema, Schema
from pydantic import Field
from apps.users.models import User
class UserOut(ModelSchema):
first_name: str = Field("", alias="firstName")
last_name: str = Field("", alias="lastName")
is_active: bool
class Meta:
model = User
fields: ClassVar[tuple[str, ...]] = (
User.id.field.name,
User.username.field.name,
User.email.field.name,
User.role.field.name,
User.first_name.field.name,
User.last_name.field.name,
User._meta.get_field("is_active").name,
)
class UserCreateIn(Schema):
username: str = Field(
...,
min_length=1,
max_length=150,
description="Unique username for the new account.",
)
email: str = Field(
...,
min_length=1,
max_length=254,
description="Email address.",
)
password: str = Field(
...,
min_length=8,
max_length=128,
description="Account password (min 8 characters).",
)
role: str = Field(
"viewer",
description=(
"Platform role to assign. "
"One of: admin, experimenter, approver, viewer."
),
)
first_name: str = Field(
"",
alias="firstName",
max_length=150,
description="First name.",
)
last_name: str = Field(
"",
alias="lastName",
max_length=150,
description="Last name.",
)
class UserUpdateIn(Schema):
username: str | None = Field(
None,
min_length=1,
max_length=150,
description="New username.",
)
email: str | None = Field(
None,
min_length=1,
max_length=254,
description="New email address.",
)
password: str | None = Field(
None,
min_length=8,
max_length=128,
description="New password (min 8 characters).",
)
role: str | None = Field(
None,
description=(
"New platform role. One of: admin, experimenter, approver, viewer."
),
)
first_name: str | None = Field(
None,
alias="firstName",
max_length=150,
description="New first name.",
)
last_name: str | None = Field(
None,
alias="lastName",
max_length=150,
description="New last name.",
)
is_active: bool | None = Field(
None,
alias="isActive",
description="Set active/inactive status.",
)
class UserRoleAssignIn(Schema):
role: str = Field(
...,
description=(
"Platform role to assign. "
"One of: admin, experimenter, approver, viewer."
),
)
class UserListOut(Schema):
count: int
items: list[UserOut]
@@ -0,0 +1,23 @@
from django.test import Client, TestCase
from apps.users.models import User, UserRole
from apps.users.tests._helpers import _auth_header, _make_user
class BaseUsersAPITest(TestCase):
def setUp(self) -> None:
self.client = Client()
self.admin: User = _make_user(
username="mgmt_admin",
email="mgmt_admin@x.com",
password="adminpass1",
role=UserRole.ADMIN,
)
self.viewer: User = _make_user(
username="mgmt_viewer",
email="mgmt_viewer@x.com",
password="viewerpass",
role=UserRole.VIEWER,
)
self.admin_auth: str = _auth_header(self.admin)
self.viewer_auth: str = _auth_header(self.viewer)
@@ -0,0 +1,96 @@
import json
import uuid
from django.urls import reverse
from apps.users.models import User, UserRole
from apps.users.tests._helpers import _make_user
from ._crud_base import BaseUsersAPITest
class UsersAPIDeleteAssignRoleTest(BaseUsersAPITest):
def test_delete_user_admin(self) -> None:
target: User = _make_user(
username="to_delete",
email="del@lotty.local",
role=UserRole.VIEWER,
)
resp = self.client.delete(
reverse("api-1:delete_user", kwargs={"user_id": str(target.pk)}),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 204)
self.assertFalse(User.objects.filter(pk=target.pk).exists())
def test_delete_self_denied(self) -> None:
resp = self.client.delete(
reverse(
"api-1:delete_user", kwargs={"user_id": str(self.admin.pk)}
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
def test_delete_user_viewer_denied(self) -> None:
resp = self.client.delete(
reverse(
"api-1:delete_user", kwargs={"user_id": str(self.admin.pk)}
),
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_delete_user_not_found(self) -> None:
resp = self.client.delete(
reverse(
"api-1:delete_user", kwargs={"user_id": str(uuid.uuid4())}
),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
def test_assign_role_admin(self) -> None:
resp = self.client.post(
reverse(
"api-1:assign_role", kwargs={"user_id": str(self.viewer.pk)}
),
data=json.dumps({"role": "experimenter"}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp.json()["role"], "experimenter")
def test_assign_role_invalid(self) -> None:
resp = self.client.post(
reverse(
"api-1:assign_role", kwargs={"user_id": str(self.viewer.pk)}
),
data=json.dumps({"role": "megaboss"}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
def test_assign_role_viewer_denied(self) -> None:
resp = self.client.post(
reverse(
"api-1:assign_role", kwargs={"user_id": str(self.admin.pk)}
),
data=json.dumps({"role": "viewer"}),
content_type="application/json",
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_assign_role_not_found(self) -> None:
resp = self.client.post(
reverse(
"api-1:assign_role", kwargs={"user_id": str(uuid.uuid4())}
),
data=json.dumps({"role": "admin"}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
@@ -0,0 +1,125 @@
import json
from django.urls import reverse
from ._crud_base import BaseUsersAPITest
class UsersAPIListCreateTest(BaseUsersAPITest):
def test_list_users_admin(self) -> None:
resp = self.client.get(
reverse("api-1:list_users"),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertIn("count", data)
self.assertIn("items", data)
self.assertEqual(data["count"], 2)
def test_list_users_viewer_denied(self) -> None:
resp = self.client.get(
reverse("api-1:list_users"),
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_list_users_unauthenticated(self) -> None:
resp = self.client.get(reverse("api-1:list_users"))
self.assertEqual(resp.status_code, 401)
def test_list_users_filter_by_role(self) -> None:
resp = self.client.get(
reverse("api-1:list_users"),
data={"role": "admin"},
HTTP_AUTHORIZATION=self.admin_auth,
)
data = resp.json()
self.assertEqual(data["count"], 1)
self.assertEqual(data["items"][0]["role"], "admin")
def test_list_users_search(self) -> None:
resp = self.client.get(
reverse("api-1:list_users"),
data={"search": "mgmt_viewer"},
HTTP_AUTHORIZATION=self.admin_auth,
)
data = resp.json()
self.assertEqual(data["count"], 1)
def test_list_users_pagination(self) -> None:
resp = self.client.get(
reverse("api-1:list_users"),
data={"limit": 1, "offset": 0},
HTTP_AUTHORIZATION=self.admin_auth,
)
data = resp.json()
self.assertEqual(len(data["items"]), 1)
self.assertEqual(data["count"], 2)
def test_create_user_admin(self) -> None:
resp = self.client.post(
reverse("api-1:create_user"),
data=json.dumps(
{
"username": "newuser",
"email": "new@lotty.local",
"password": "newpass123",
"role": "experimenter",
"firstName": "New",
"lastName": "User",
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 201)
data = resp.json()
self.assertEqual(data["username"], "newuser")
self.assertEqual(data["role"], "experimenter")
def test_create_user_invalid_role(self) -> None:
resp = self.client.post(
reverse("api-1:create_user"),
data=json.dumps(
{
"username": "bad_role",
"email": "bad@lotty.local",
"password": "password1",
"role": "superadmin",
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
def test_create_user_viewer_denied(self) -> None:
resp = self.client.post(
reverse("api-1:create_user"),
data=json.dumps(
{
"username": "denied",
"email": "denied@lotty.local",
"password": "password1",
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_create_user_duplicate_username(self) -> None:
resp = self.client.post(
reverse("api-1:create_user"),
data=json.dumps(
{
"username": "mgmt_admin",
"email": "other@lotty.local",
"password": "password1",
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [409, 422, 400, 500])
@@ -0,0 +1,93 @@
import json
import uuid
from django.urls import reverse
from ._crud_base import BaseUsersAPITest
class UsersAPIReadUpdateTest(BaseUsersAPITest):
def test_get_user_admin(self) -> None:
resp = self.client.get(
reverse("api-1:get_user", kwargs={"user_id": str(self.viewer.pk)}),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp.json()["username"], "mgmt_viewer")
def test_get_user_not_found(self) -> None:
resp = self.client.get(
reverse("api-1:get_user", kwargs={"user_id": str(uuid.uuid4())}),
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
def test_get_user_viewer_denied(self) -> None:
resp = self.client.get(
reverse("api-1:get_user", kwargs={"user_id": str(self.admin.pk)}),
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_update_user_admin(self) -> None:
resp = self.client.patch(
reverse(
"api-1:update_user", kwargs={"user_id": str(self.viewer.pk)}
),
data=json.dumps(
{"username": "renamed_viewer", "role": "approver"}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertEqual(data["username"], "renamed_viewer")
self.assertEqual(data["role"], "approver")
def test_update_user_partial(self) -> None:
original_role = self.viewer.role
resp = self.client.patch(
reverse(
"api-1:update_user", kwargs={"user_id": str(self.viewer.pk)}
),
data=json.dumps({"firstName": "Updated"}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertEqual(data["role"], original_role)
def test_update_user_invalid_role(self) -> None:
resp = self.client.patch(
reverse(
"api-1:update_user", kwargs={"user_id": str(self.viewer.pk)}
),
data=json.dumps({"role": "superadmin"}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertIn(resp.status_code, [422, 400])
def test_update_user_viewer_denied(self) -> None:
resp = self.client.patch(
reverse(
"api-1:update_user", kwargs={"user_id": str(self.admin.pk)}
),
data=json.dumps({"firstName": "Hacked"}),
content_type="application/json",
HTTP_AUTHORIZATION=self.viewer_auth,
)
self.assertEqual(resp.status_code, 403)
def test_update_user_not_found(self) -> None:
resp = self.client.patch(
reverse(
"api-1:update_user", kwargs={"user_id": str(uuid.uuid4())}
),
data=json.dumps({"firstName": "Ghost"}),
content_type="application/json",
HTTP_AUTHORIZATION=self.admin_auth,
)
self.assertEqual(resp.status_code, 404)
@@ -0,0 +1,100 @@
import json
from django.test import Client, TestCase
from django.urls import reverse
from apps.users.models import User, UserRole
from apps.users.tests._helpers import _auth_header, _make_user
class RoleBasedAccessControlTest(TestCase):
def setUp(self) -> None:
self.client = Client()
self.roles = {}
for role_val in [
UserRole.ADMIN,
UserRole.EXPERIMENTER,
UserRole.APPROVER,
UserRole.VIEWER,
]:
user: User = _make_user(
username=f"rbac_{role_val}",
email=f"rbac_{role_val}@x.com",
password="password1",
role=role_val,
)
self.roles[role_val] = {
"user": user,
"auth": _auth_header(user),
}
def test_admin_can_list(self) -> None:
resp = self.client.get(
reverse("api-1:list_users"),
HTTP_AUTHORIZATION=self.roles[UserRole.ADMIN]["auth"],
)
self.assertEqual(resp.status_code, 200)
def test_experimenter_cannot_list(self) -> None:
resp = self.client.get(
reverse("api-1:list_users"),
HTTP_AUTHORIZATION=self.roles[UserRole.EXPERIMENTER]["auth"],
)
self.assertEqual(resp.status_code, 403)
def test_approver_cannot_list(self) -> None:
resp = self.client.get(
reverse("api-1:list_users"),
HTTP_AUTHORIZATION=self.roles[UserRole.APPROVER]["auth"],
)
self.assertEqual(resp.status_code, 403)
def test_viewer_cannot_list(self) -> None:
resp = self.client.get(
reverse("api-1:list_users"),
HTTP_AUTHORIZATION=self.roles[UserRole.VIEWER]["auth"],
)
self.assertEqual(resp.status_code, 403)
def test_experimenter_cannot_create(self) -> None:
resp = self.client.post(
reverse("api-1:create_user"),
data=json.dumps(
{
"username": "blocked",
"email": "blocked@x.com",
"password": "password1",
}
),
content_type="application/json",
HTTP_AUTHORIZATION=self.roles[UserRole.EXPERIMENTER]["auth"],
)
self.assertEqual(resp.status_code, 403)
def test_approver_cannot_delete(self) -> None:
target = self.roles[UserRole.VIEWER]["user"]
resp = self.client.delete(
reverse("api-1:delete_user", kwargs={"user_id": str(target.pk)}),
HTTP_AUTHORIZATION=self.roles[UserRole.APPROVER]["auth"],
)
self.assertEqual(resp.status_code, 403)
def test_viewer_cannot_assign_role(self) -> None:
target = self.roles[UserRole.EXPERIMENTER]["user"]
resp = self.client.post(
reverse("api-1:assign_role", kwargs={"user_id": str(target.pk)}),
data=json.dumps({"role": "admin"}),
content_type="application/json",
HTTP_AUTHORIZATION=self.roles[UserRole.VIEWER]["auth"],
)
self.assertEqual(resp.status_code, 403)
def test_experimenter_cannot_update(self) -> None:
target = self.roles[UserRole.VIEWER]["user"]
resp = self.client.patch(
reverse("api-1:update_user", kwargs={"user_id": str(target.pk)}),
data=json.dumps({"firstName": "Nope"}),
content_type="application/json",
HTTP_AUTHORIZATION=self.roles[UserRole.EXPERIMENTER]["auth"],
)
self.assertEqual(resp.status_code, 403)
+108
View File
@@ -0,0 +1,108 @@
from django.contrib import admin
from django.utils.translation import gettext_lazy as _
from apps.reviews.models import ApproverGroup, ReviewSettings
@admin.register(ReviewSettings)
class ReviewSettingsAdmin(admin.ModelAdmin):
list_display = (
ReviewSettings.default_min_approvals.field.name,
ReviewSettings.allow_any_approver.field.name,
ReviewSettings.updated_at.field.name,
)
readonly_fields = (ReviewSettings.updated_at.field.name,)
fieldsets = (
(
_("Fallback review policy"),
{
"fields": (
ReviewSettings.default_min_approvals.field.name,
ReviewSettings.allow_any_approver.field.name,
),
"description": _(
"Global fallback settings applied when an experimenter "
"has no explicit approver group assigned."
),
},
),
(
_("Metadata"),
{
"fields": (ReviewSettings.updated_at.field.name,),
},
),
)
def has_add_permission(self, request) -> bool:
if ReviewSettings.objects.exists():
return False
return super().has_add_permission(request)
def has_delete_permission(self, request, obj=None) -> bool:
return False
@admin.register(ApproverGroup)
class ApproverGroupAdmin(admin.ModelAdmin):
list_display = (
ApproverGroup.experimenter.field.name,
ApproverGroup.min_approvals.field.name,
"approver_count",
ApproverGroup.created_at.field.name,
ApproverGroup.updated_at.field.name,
)
list_filter = (ApproverGroup.min_approvals.field.name,)
search_fields = (
f"{ApproverGroup.experimenter.field.name}__username",
f"{ApproverGroup.experimenter.field.name}__email",
f"{ApproverGroup.approvers.field.name}__username",
f"{ApproverGroup.approvers.field.name}__email",
)
raw_id_fields = (ApproverGroup.experimenter.field.name,)
filter_horizontal = (ApproverGroup.approvers.field.name,)
readonly_fields = (
ApproverGroup.created_at.field.name,
ApproverGroup.updated_at.field.name,
)
ordering = (f"-{ApproverGroup.created_at.field.name}",)
fieldsets = (
(
_("Experimenter"),
{
"fields": (ApproverGroup.experimenter.field.name,),
"description": _(
"The experimenter whose experiments this group governs. "
"Each experimenter can have at most one approver group."
),
},
),
(
_("Approvers"),
{
"fields": (
ApproverGroup.approvers.field.name,
ApproverGroup.min_approvals.field.name,
),
"description": _(
"Select which approver-role users may approve this "
"experimenter's experiments and how many approvals "
"are required before an experiment can be launched."
),
},
),
(
_("Metadata"),
{
"fields": (
ApproverGroup.created_at.field.name,
ApproverGroup.updated_at.field.name,
),
},
),
)
@admin.display(description=_("Approvers"))
def approver_count(self, obj):
return obj.approvers.count()
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class ReviewsConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "apps.reviews"
@@ -0,0 +1,47 @@
# Generated by Django 5.2.11 on 2026-02-12 10:14
import django.core.validators
import django.db.models.deletion
import uuid
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='ReviewSettings',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('default_min_approvals', models.PositiveIntegerField(default=1, help_text='Fallback number of approvals required when no approver group is explicitly assigned to an experimenter.', validators=[django.core.validators.MinValueValidator(1)], verbose_name='default minimum approvals')),
('allow_any_approver', models.BooleanField(default=True, help_text='When True, any user with the Approver role can approve experiments that have no explicit approver group. When False, experiments without an approver group cannot proceed to review.', verbose_name='allow any approver')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='updated at')),
],
options={
'verbose_name': 'review settings',
'verbose_name_plural': 'review settings',
},
),
migrations.CreateModel(
name='ApproverGroup',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('min_approvals', models.PositiveIntegerField(default=1, help_text="Number of distinct approvals required before an experiment can transition to 'approved'.", validators=[django.core.validators.MinValueValidator(1)], verbose_name='minimum approvals')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='updated at')),
('approvers', models.ManyToManyField(blank=True, help_text="Approver-role users who may approve this experimenter's experiments.", limit_choices_to={'role': 'approver'}, related_name='approvable_groups', to=settings.AUTH_USER_MODEL, verbose_name='approvers')),
('experimenter', models.OneToOneField(help_text='The experimenter whose experiments this group governs.', limit_choices_to={'role': 'experimenter'}, on_delete=django.db.models.deletion.CASCADE, related_name='approver_group', to=settings.AUTH_USER_MODEL, verbose_name='experimenter')),
],
options={
'verbose_name': 'approver group',
'verbose_name_plural': 'approver groups',
},
),
]
+114
View File
@@ -0,0 +1,114 @@
from typing import Any
from django.conf import settings
from django.core.validators import MinValueValidator
from django.db import models
from django.utils.translation import gettext_lazy as _
from apps.core.models import BaseModel
from apps.users.models import User
class ReviewSettings(BaseModel):
default_min_approvals: models.PositiveIntegerField[Any, Any] = (
models.PositiveIntegerField(
default=1,
validators=[MinValueValidator(1)],
verbose_name=_("default minimum approvals"),
help_text=_(
"Fallback number of approvals required when no "
"approver group is explicitly assigned to an experimenter."
),
)
)
allow_any_approver: models.BooleanField[Any, Any] = models.BooleanField(
default=False,
verbose_name=_("allow any approver"),
help_text=_(
"When True, any user with the Approver role can approve "
"experiments that have no explicit approver group. "
"When False, experiments without an approver group "
"cannot proceed to review."
),
)
updated_at: models.DateTimeField = models.DateTimeField(
auto_now=True,
verbose_name=_("updated at"),
)
class Meta:
verbose_name = _("review settings")
verbose_name_plural = _("review settings")
def __str__(self) -> str:
return (
f"ReviewSettings(min_approvals={self.default_min_approvals}, "
f"allow_any_approver={self.allow_any_approver})"
)
def save(self, *args, **kwargs) -> None:
existing: ReviewSettings | None = ReviewSettings.objects.first()
if existing and existing.pk != self.pk:
self.pk = existing.pk
super().save(*args, **kwargs)
ReviewSettings.objects.exclude(pk=self.pk).delete()
@classmethod
def load(cls) -> "ReviewSettings":
obj, _ = cls.objects.get_or_create(
defaults={"default_min_approvals": 1, "allow_any_approver": False},
)
return obj
class ApproverGroup(BaseModel):
experimenter: models.OneToOneField[Any, Any] = models.OneToOneField(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name="approver_group",
limit_choices_to={"role": "experimenter"},
verbose_name=_("experimenter"),
help_text=_("The experimenter whose experiments this group governs."),
)
approvers: models.ManyToManyField = models.ManyToManyField(
settings.AUTH_USER_MODEL,
related_name="approvable_groups",
blank=True,
limit_choices_to={"role": "approver"},
verbose_name=_("approvers"),
help_text=_(
"Approver-role users who may approve this experimenter's experiments."
),
)
min_approvals: models.PositiveIntegerField[Any, Any] = (
models.PositiveIntegerField(
default=1,
validators=[MinValueValidator(1)],
verbose_name=_("minimum approvals"),
help_text=_(
"Number of distinct approvals required before "
"an experiment can transition to 'approved'."
),
)
)
created_at: models.DateTimeField = models.DateTimeField(
auto_now_add=True,
verbose_name=_("created at"),
)
updated_at: models.DateTimeField = models.DateTimeField(
auto_now=True,
verbose_name=_("updated at"),
)
class Meta:
verbose_name = _("approver group")
verbose_name_plural = _("approver groups")
def __str__(self) -> str:
return (
f"ApproverGroup(experimenter={self.experimenter.pk}, "
f"min_approvals={self.min_approvals})"
)
def can_approve(self, user: User) -> bool:
return self.approvers.filter(pk=user.pk).exists()
+118
View File
@@ -0,0 +1,118 @@
import uuid
from django.db.models import QuerySet
from apps.reviews.models import ApproverGroup, ReviewSettings
from apps.users.models import User, UserRole
def review_settings_load() -> ReviewSettings:
return ReviewSettings.load()
def approver_group_get_by_id(group_id: str) -> ApproverGroup | None:
try:
uuid.UUID(group_id)
except ValueError:
return None
return (
ApproverGroup.objects.select_related("experimenter")
.prefetch_related("approvers")
.filter(id=group_id)
.first()
)
def approver_group_get_by_experimenter(
experimenter: User,
) -> ApproverGroup | None:
return (
ApproverGroup.objects.select_related("experimenter")
.prefetch_related("approvers")
.filter(experimenter=experimenter)
.first()
)
def approver_group_get_by_experimenter_id(
experimenter_id: str,
) -> ApproverGroup | None:
try:
uuid.UUID(experimenter_id)
except ValueError:
return None
return (
ApproverGroup.objects.select_related("experimenter")
.prefetch_related("approvers")
.filter(experimenter_id=experimenter_id)
.first()
)
def approver_group_list() -> QuerySet[ApproverGroup]:
return (
ApproverGroup.objects.select_related("experimenter")
.prefetch_related("approvers")
.order_by("-created_at")
)
def approver_group_list_for_approver(
approver: User,
) -> QuerySet[ApproverGroup]:
return (
ApproverGroup.objects.filter(approvers=approver)
.select_related("experimenter")
.prefetch_related("approvers")
.order_by("-created_at")
)
def get_effective_approvers_for_experimenter(
experimenter: User,
) -> tuple[QuerySet[User], int]:
group: ApproverGroup | None = approver_group_get_by_experimenter(
experimenter
)
if group is not None:
return group.approvers.filter(is_active=True), group.min_approvals
settings: ReviewSettings = review_settings_load()
if settings.allow_any_approver:
all_approvers: QuerySet[User] = User.objects.filter(
role=UserRole.APPROVER,
is_active=True,
)
return all_approvers, settings.default_min_approvals
return User.objects.none(), settings.default_min_approvals
def can_user_approve_experimenter(
approver: User,
experimenter: User,
) -> bool:
if approver.role != UserRole.APPROVER:
return False
if not approver.is_active:
return False
group: ApproverGroup | None = approver_group_get_by_experimenter(
experimenter
)
if group is not None:
return group.approvers.filter(pk=approver.pk).exists()
settings: ReviewSettings = review_settings_load()
return settings.allow_any_approver
def get_min_approvals_for_experimenter(experimenter: User) -> int:
group: ApproverGroup | None = approver_group_get_by_experimenter(
experimenter
)
if group is not None:
return group.min_approvals
return review_settings_load().default_min_approvals
+236
View File
@@ -0,0 +1,236 @@
from typing import Any
from django.core.exceptions import ValidationError
from django.db import transaction
from apps.reviews.models import ApproverGroup, ReviewSettings
from apps.users.models import User, UserRole
def review_settings_update(
*,
default_min_approvals: int | None = None,
allow_any_approver: bool | None = None,
) -> ReviewSettings:
settings = ReviewSettings.load()
if default_min_approvals is not None:
if default_min_approvals < 1:
raise ValidationError(
{
"default_min_approvals": (
"Minimum approvals must be at least 1."
)
}
)
settings.default_min_approvals = default_min_approvals
if allow_any_approver is not None:
settings.allow_any_approver = allow_any_approver
settings.save()
return settings
def _validate_experimenter(user: User) -> None:
if user.role != UserRole.EXPERIMENTER:
raise ValidationError(
{
"experimenter": (
f"User '{user.username}' has role '{user.role}'. "
f"Only users with the '{UserRole.EXPERIMENTER}' role "
f"can be assigned an approver group."
)
}
)
if not user.is_active:
raise ValidationError(
{"experimenter": "The experimenter must be an active user."}
)
def _validate_approvers(approvers: list[User]) -> None:
invalid = [u for u in approvers if u.role != UserRole.APPROVER]
if invalid:
names = ", ".join(u.username for u in invalid)
raise ValidationError(
{
"approvers": (
f"The following users do not have the "
f"'{UserRole.APPROVER}' role: {names}"
)
}
)
def _validate_min_approvals(
min_approvals: int,
approver_count: int | None = None,
) -> None:
if min_approvals < 1:
raise ValidationError(
{"min_approvals": "Minimum approvals must be at least 1."}
)
if approver_count is not None and min_approvals > approver_count:
raise ValidationError(
{
"min_approvals": (
f"min_approvals ({min_approvals}) cannot exceed the "
f"number of assigned approvers ({approver_count})."
)
}
)
@transaction.atomic
def approver_group_create(
*,
experimenter: User,
approver_ids: list[Any] | None = None,
min_approvals: int = 1,
) -> ApproverGroup:
_validate_experimenter(experimenter)
if ApproverGroup.objects.filter(experimenter=experimenter).exists():
raise ValidationError(
{
"experimenter": (
f"An approver group already exists for "
f"experimenter '{experimenter.username}'."
)
}
)
approvers: list[User] = []
if approver_ids:
approvers = list(
User.objects.filter(pk__in=approver_ids, is_active=True)
)
found_ids = {str(u.pk) for u in approvers}
missing = [
str(aid) for aid in approver_ids if str(aid) not in found_ids
]
if missing:
raise ValidationError(
{
"approvers": (
f"Users not found or inactive: {', '.join(missing)}"
)
}
)
_validate_approvers(approvers)
_validate_min_approvals(
min_approvals, len(approvers) if approvers else None
)
group = ApproverGroup(
experimenter=experimenter,
min_approvals=min_approvals,
)
group.save()
if approvers:
group.approvers.set(approvers)
return group
@transaction.atomic
def approver_group_update(
*,
group: ApproverGroup,
approver_ids: list[Any] | None = None,
min_approvals: int | None = None,
) -> ApproverGroup:
approvers: list[User] | None = None
if approver_ids is not None:
approvers = list(
User.objects.filter(pk__in=approver_ids, is_active=True)
)
found_ids = {str(u.pk) for u in approvers}
missing = [
str(aid) for aid in approver_ids if str(aid) not in found_ids
]
if missing:
raise ValidationError(
{
"approvers": (
f"Users not found or inactive: {', '.join(missing)}"
)
}
)
_validate_approvers(approvers)
if min_approvals is not None:
approver_count = (
len(approvers)
if approvers is not None
else group.approvers.count()
)
_validate_min_approvals(min_approvals, approver_count)
group.min_approvals = min_approvals
group.save()
if approvers is not None:
group.approvers.set(approvers)
return group
@transaction.atomic
def approver_group_delete(*, group: ApproverGroup) -> None:
group.delete()
@transaction.atomic
def approver_group_add_approver(
*,
group: ApproverGroup,
approver: User,
) -> ApproverGroup:
_validate_approvers([approver])
if group.approvers.filter(pk=approver.pk).exists():
raise ValidationError(
{
"approver": (
f"User '{approver.username}' is already in this "
f"approver group."
)
}
)
group.approvers.add(approver)
return group
@transaction.atomic
def approver_group_remove_approver(
*,
group: ApproverGroup,
approver: User,
) -> ApproverGroup:
if not group.approvers.filter(pk=approver.pk).exists():
raise ValidationError(
{
"approver": (
f"User '{approver.username}' is not in this "
f"approver group."
)
}
)
remaining = group.approvers.count() - 1
if remaining < group.min_approvals:
raise ValidationError(
{
"approver": (
f"Cannot remove approver: would leave {remaining} "
f"approver(s), but {group.min_approvals} required."
)
}
)
group.approvers.remove(approver)
return group
@@ -0,0 +1,40 @@
from apps.users.models import User, UserRole
from apps.users.tests._helpers import _make_user
def _make_experimenter(suffix="") -> User:
return _make_user(
username=f"exp{suffix}",
email=f"exp{suffix}@lotty.local",
role=UserRole.EXPERIMENTER,
)
def _make_approver(suffix="") -> User:
return _make_user(
username=f"appr{suffix}",
email=f"appr{suffix}@lotty.local",
role=UserRole.APPROVER,
)
def _make_admin(suffix="") -> User:
return _make_user(
username=f"admin{suffix}",
email=f"admin{suffix}@lotty.local",
role=UserRole.ADMIN,
)
def _make_viewer(suffix="") -> User:
return _make_user(
username=f"viewer{suffix}",
email=f"viewer{suffix}@lotty.local",
role=UserRole.VIEWER,
)
def _get(data, camel_key, snake_key):
if camel_key in data:
return data[camel_key]
return data[snake_key]
@@ -0,0 +1,377 @@
import uuid
from typing import Any
from django.core.exceptions import ValidationError
from django.db.models import QuerySet
from django.test import TestCase
from apps.reviews.models import ApproverGroup
from apps.reviews.selectors import (
approver_group_get_by_experimenter,
approver_group_get_by_experimenter_id,
approver_group_get_by_id,
approver_group_list,
approver_group_list_for_approver,
)
from apps.reviews.services import (
approver_group_add_approver,
approver_group_create,
approver_group_delete,
approver_group_remove_approver,
approver_group_update,
)
from apps.users.models import User
from ._helpers import (
_make_admin,
_make_approver,
_make_experimenter,
_make_viewer,
)
class ApproverGroupModelTest(TestCase):
def setUp(self) -> None:
self.experimenter: User = _make_experimenter("_model")
self.approver: User = _make_approver("_model")
def test_create_group(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver.pk)],
min_approvals=1,
)
self.assertIsNotNone(group.pk)
self.assertEqual(group.experimenter, self.experimenter)
self.assertEqual(group.min_approvals, 1)
self.assertEqual(group.approvers.count(), 1)
def test_str_representation(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
min_approvals=2,
)
result = str(group)
self.assertIn("ApproverGroup", result)
self.assertIn("min_approvals=2", result)
def test_can_approve_true(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver.pk)],
)
self.assertTrue(group.can_approve(self.approver))
def test_can_approve_false_not_in_group(self) -> None:
other_approver: User = _make_approver("_other")
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver.pk)],
)
self.assertFalse(group.can_approve(other_approver))
def test_one_to_one_constraint(self) -> None:
approver_group_create(experimenter=self.experimenter)
with self.assertRaises(ValidationError):
approver_group_create(
experimenter=self.experimenter,
min_approvals=1,
)
class ApproverGroupCreateServiceTest(TestCase):
def setUp(self) -> None:
self.experimenter: User = _make_experimenter("_create")
self.approver1: User = _make_approver("_create1")
self.approver2: User = _make_approver("_create2")
def test_create_with_approvers(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver1.pk), str(self.approver2.pk)],
min_approvals=2,
)
self.assertEqual(group.approvers.count(), 2)
self.assertEqual(group.min_approvals, 2)
def test_create_without_approvers(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
min_approvals=1,
)
self.assertEqual(group.approvers.count(), 0)
self.assertEqual(group.min_approvals, 1)
def test_create_default_min_approvals(self) -> None:
group: ApproverGroup = approver_group_create(
experimenter=self.experimenter
)
self.assertEqual(group.min_approvals, 1)
def test_create_rejects_non_experimenter_user(self) -> None:
admin: User = _make_admin("_cre_admin")
with self.assertRaises(ValidationError):
approver_group_create(experimenter=admin)
def test_create_rejects_viewer_as_experimenter(self) -> None:
viewer: User = _make_viewer("_cre_viewer")
with self.assertRaises(ValidationError):
approver_group_create(experimenter=viewer)
def test_create_rejects_approver_as_experimenter(self) -> None:
approver: User = _make_approver("_cre_as_exp")
with self.assertRaises(ValidationError):
approver_group_create(experimenter=approver)
def test_create_rejects_inactive_experimenter(self) -> None:
self.experimenter.is_active = False
self.experimenter.save()
with self.assertRaises(ValidationError):
approver_group_create(experimenter=self.experimenter)
def test_create_rejects_non_approver_in_approver_list(self) -> None:
viewer: User = _make_viewer("_cre_bad_appr")
with self.assertRaises(ValidationError):
approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(viewer.pk)],
)
def test_create_rejects_missing_approver_ids(self) -> None:
with self.assertRaises(ValidationError):
approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(uuid.uuid4())],
)
def test_create_rejects_duplicate_group(self) -> None:
approver_group_create(experimenter=self.experimenter)
with self.assertRaises(ValidationError):
approver_group_create(experimenter=self.experimenter)
def test_create_min_approvals_zero_raises(self) -> None:
with self.assertRaises(ValidationError):
approver_group_create(
experimenter=self.experimenter,
min_approvals=0,
)
def test_create_min_approvals_exceeds_approvers_raises(self) -> None:
with self.assertRaises(ValidationError):
approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver1.pk)],
min_approvals=3,
)
class ApproverGroupUpdateServiceTest(TestCase):
def setUp(self) -> None:
self.experimenter: User = _make_experimenter("_upd")
self.approver1: User = _make_approver("_upd1")
self.approver2: User = _make_approver("_upd2")
self.approver3: User = _make_approver("_upd3")
self.group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver1.pk), str(self.approver2.pk)],
min_approvals=1,
)
def test_update_min_approvals(self) -> None:
updated: ApproverGroup = approver_group_update(
group=self.group, min_approvals=2
)
self.assertEqual(updated.min_approvals, 2)
def test_update_approver_ids_replaces_set(self) -> None:
updated: ApproverGroup = approver_group_update(
group=self.group,
approver_ids=[str(self.approver3.pk)],
)
approver_pks: set[Any] = set(
updated.approvers.values_list("pk", flat=True)
)
self.assertEqual(approver_pks, {self.approver3.pk})
def test_update_both_fields(self) -> None:
updated: ApproverGroup = approver_group_update(
group=self.group,
approver_ids=[
str(self.approver1.pk),
str(self.approver2.pk),
str(self.approver3.pk),
],
min_approvals=3,
)
self.assertEqual(updated.approvers.count(), 3)
self.assertEqual(updated.min_approvals, 3)
def test_update_no_op(self) -> None:
updated: ApproverGroup = approver_group_update(group=self.group)
self.assertEqual(updated.min_approvals, self.group.min_approvals)
def test_update_min_approvals_exceeds_new_approver_count_raises(
self,
) -> None:
with self.assertRaises(ValidationError):
approver_group_update(
group=self.group,
approver_ids=[str(self.approver1.pk)],
min_approvals=5,
)
def test_update_min_approvals_exceeds_existing_count_raises(self) -> None:
with self.assertRaises(ValidationError):
approver_group_update(group=self.group, min_approvals=10)
def test_update_rejects_non_approver_role(self) -> None:
viewer: User = _make_viewer("_upd_bad")
with self.assertRaises(ValidationError):
approver_group_update(
group=self.group,
approver_ids=[str(viewer.pk)],
)
def test_update_rejects_missing_user(self) -> None:
with self.assertRaises(ValidationError):
approver_group_update(
group=self.group,
approver_ids=[str(uuid.uuid4())],
)
class ApproverGroupDeleteServiceTest(TestCase):
def test_delete_removes_group(self) -> None:
exp: User = _make_experimenter("_del")
group: ApproverGroup = approver_group_create(experimenter=exp)
pk = group.pk
approver_group_delete(group=group)
self.assertFalse(ApproverGroup.objects.filter(pk=pk).exists())
def test_delete_allows_recreating_group(self) -> None:
exp: User = _make_experimenter("_del2")
group: ApproverGroup = approver_group_create(experimenter=exp)
approver_group_delete(group=group)
new_group: ApproverGroup = approver_group_create(experimenter=exp)
self.assertIsNotNone(new_group.pk)
class ApproverGroupAddRemoveServiceTest(TestCase):
def setUp(self) -> None:
self.experimenter: User = _make_experimenter("_ar")
self.approver1: User = _make_approver("_ar1")
self.approver2: User = _make_approver("_ar2")
self.group: ApproverGroup = approver_group_create(
experimenter=self.experimenter,
approver_ids=[str(self.approver1.pk)],
min_approvals=1,
)
def test_add_approver(self) -> None:
approver_group_add_approver(group=self.group, approver=self.approver2)
self.assertEqual(self.group.approvers.count(), 2)
self.assertTrue(
self.group.approvers.filter(pk=self.approver2.pk).exists()
)
def test_add_approver_rejects_non_approver_role(self) -> None:
viewer: User = _make_viewer("_ar_bad")
with self.assertRaises(ValidationError):
approver_group_add_approver(group=self.group, approver=viewer)
def test_add_approver_rejects_duplicate(self) -> None:
with self.assertRaises(ValidationError):
approver_group_add_approver(
group=self.group, approver=self.approver1
)
def test_remove_approver(self) -> None:
approver_group_add_approver(group=self.group, approver=self.approver2)
approver_group_remove_approver(
group=self.group, approver=self.approver1
)
self.assertEqual(self.group.approvers.count(), 1)
self.assertFalse(
self.group.approvers.filter(pk=self.approver1.pk).exists()
)
def test_remove_approver_not_in_group_raises(self) -> None:
with self.assertRaises(ValidationError):
approver_group_remove_approver(
group=self.group, approver=self.approver2
)
def test_remove_approver_below_min_raises(self) -> None:
with self.assertRaises(ValidationError):
approver_group_remove_approver(
group=self.group, approver=self.approver1
)
class ApproverGroupSelectorsTest(TestCase):
def setUp(self) -> None:
self.exp1: User = _make_experimenter("_sel1")
self.exp2: User = _make_experimenter("_sel2")
self.appr1: User = _make_approver("_sel1")
self.appr2: User = _make_approver("_sel2")
self.group1: ApproverGroup = approver_group_create(
experimenter=self.exp1,
approver_ids=[str(self.appr1.pk)],
min_approvals=1,
)
self.group2: ApproverGroup = approver_group_create(
experimenter=self.exp2,
approver_ids=[str(self.appr1.pk), str(self.appr2.pk)],
min_approvals=2,
)
def test_get_by_id(self) -> None:
found: ApproverGroup | None = approver_group_get_by_id(
str(self.group1.pk)
)
self.assertEqual(found, self.group1)
def test_get_by_id_invalid_uuid(self) -> None:
self.assertIsNone(approver_group_get_by_id("not-a-uuid"))
def test_get_by_id_nonexistent(self) -> None:
self.assertIsNone(approver_group_get_by_id(str(uuid.uuid4())))
def test_get_by_experimenter(self) -> None:
found: ApproverGroup | None = approver_group_get_by_experimenter(
self.exp1
)
self.assertEqual(found, self.group1)
def test_get_by_experimenter_no_group(self) -> None:
exp3: User = _make_experimenter("_sel3")
self.assertIsNone(approver_group_get_by_experimenter(exp3))
def test_get_by_experimenter_id(self) -> None:
found: ApproverGroup | None = approver_group_get_by_experimenter_id(
str(self.exp2.pk)
)
self.assertEqual(found, self.group2)
def test_get_by_experimenter_id_invalid(self) -> None:
self.assertIsNone(approver_group_get_by_experimenter_id("bad"))
def test_get_by_experimenter_id_nonexistent(self) -> None:
self.assertIsNone(
approver_group_get_by_experimenter_id(str(uuid.uuid4()))
)
def test_list_all(self) -> None:
qs: QuerySet[ApproverGroup] = approver_group_list()
self.assertEqual(qs.count(), 2)
def test_list_for_approver(self) -> None:
qs: QuerySet[ApproverGroup] = approver_group_list_for_approver(
self.appr1
)
self.assertEqual(qs.count(), 2)
qs2: QuerySet[ApproverGroup] = approver_group_list_for_approver(
self.appr2
)
self.assertEqual(qs2.count(), 1)
@@ -0,0 +1,100 @@
from django.test import TestCase
from apps.reviews.models import ApproverGroup
from apps.reviews.selectors import (
can_user_approve_experimenter,
get_effective_approvers_for_experimenter,
get_min_approvals_for_experimenter,
)
from apps.reviews.services import approver_group_create, review_settings_update
from apps.users.models import User
from ._helpers import _make_admin, _make_approver, _make_experimenter
class EffectiveReviewPolicyTest(TestCase):
def setUp(self) -> None:
self.exp_with_group: User = _make_experimenter("_eff1")
self.exp_without_group: User = _make_experimenter("_eff2")
self.appr1: User = _make_approver("_eff1")
self.appr2: User = _make_approver("_eff2")
self.group: ApproverGroup = approver_group_create(
experimenter=self.exp_with_group,
approver_ids=[str(self.appr1.pk)],
min_approvals=1,
)
def test_with_explicit_group(self) -> None:
approvers, min_approvals = get_effective_approvers_for_experimenter(
self.exp_with_group
)
self.assertEqual(min_approvals, 1)
self.assertEqual(approvers.count(), 1)
self.assertEqual(approvers.first(), self.appr1)
def test_fallback_allow_any_approver(self) -> None:
review_settings_update(
default_min_approvals=2, allow_any_approver=True
)
approvers, min_approvals = get_effective_approvers_for_experimenter(
self.exp_without_group
)
self.assertEqual(min_approvals, 2)
self.assertIn(self.appr1, approvers)
self.assertIn(self.appr2, approvers)
def test_fallback_deny_any_approver(self) -> None:
review_settings_update(allow_any_approver=False)
approvers, _ = get_effective_approvers_for_experimenter(
self.exp_without_group
)
self.assertEqual(approvers.count(), 0)
def test_get_min_approvals_with_group(self) -> None:
result: int = get_min_approvals_for_experimenter(self.exp_with_group)
self.assertEqual(result, 1)
def test_get_min_approvals_fallback(self) -> None:
review_settings_update(default_min_approvals=5)
result: int = get_min_approvals_for_experimenter(
self.exp_without_group
)
self.assertEqual(result, 5)
class CanUserApproveTest(TestCase):
def setUp(self) -> None:
self.exp: User = _make_experimenter("_can")
self.appr_in: User = _make_approver("_can_in")
self.appr_out: User = _make_approver("_can_out")
self.group: ApproverGroup = approver_group_create(
experimenter=self.exp,
approver_ids=[str(self.appr_in.pk)],
)
def test_approver_in_group_can_approve(self) -> None:
self.assertTrue(can_user_approve_experimenter(self.appr_in, self.exp))
def test_approver_not_in_group_cannot_approve(self) -> None:
self.assertFalse(
can_user_approve_experimenter(self.appr_out, self.exp)
)
def test_non_approver_role_cannot_approve(self) -> None:
admin: User = _make_admin("_can_a")
self.assertFalse(can_user_approve_experimenter(admin, self.exp))
def test_inactive_approver_cannot_approve(self) -> None:
self.appr_in.is_active = False
self.appr_in.save()
self.assertFalse(can_user_approve_experimenter(self.appr_in, self.exp))
def test_fallback_any_approver_can_approve(self) -> None:
exp2: User = _make_experimenter("_can2")
review_settings_update(allow_any_approver=True)
self.assertTrue(can_user_approve_experimenter(self.appr_out, exp2))
def test_fallback_deny_blocks_approval(self) -> None:
exp2: User = _make_experimenter("_can3")
review_settings_update(allow_any_approver=False)
self.assertFalse(can_user_approve_experimenter(self.appr_out, exp2))
@@ -0,0 +1,100 @@
from django.core.exceptions import ValidationError
from django.test import TestCase
from apps.reviews.models import ReviewSettings
from apps.reviews.selectors import review_settings_load
from apps.reviews.services import review_settings_update
class ReviewSettingsModelTest(TestCase):
def test_load_creates_default_if_missing(self) -> None:
self.assertEqual(ReviewSettings.objects.count(), 0)
settings: ReviewSettings = ReviewSettings.load()
self.assertEqual(ReviewSettings.objects.count(), 1)
self.assertEqual(settings.default_min_approvals, 1)
self.assertFalse(settings.allow_any_approver)
def test_load_returns_existing_singleton(self) -> None:
s1: ReviewSettings = ReviewSettings.load()
s2: ReviewSettings = ReviewSettings.load()
self.assertEqual(s1.pk, s2.pk)
self.assertEqual(ReviewSettings.objects.count(), 1)
def test_save_enforces_singleton(self) -> None:
s1: ReviewSettings = ReviewSettings.load()
s1.default_min_approvals = 3
s1.allow_any_approver = False
s1.save()
self.assertEqual(ReviewSettings.objects.count(), 1)
s1.refresh_from_db()
self.assertEqual(s1.default_min_approvals, 3)
self.assertFalse(s1.allow_any_approver)
def test_str_representation(self) -> None:
settings: ReviewSettings = ReviewSettings.load()
result = str(settings)
self.assertIn("ReviewSettings", result)
self.assertIn("min_approvals=1", result)
def test_updated_at_changes_on_save(self) -> None:
settings: ReviewSettings = ReviewSettings.load()
original_updated = settings.updated_at
settings.default_min_approvals = 5
settings.save()
settings.refresh_from_db()
self.assertGreaterEqual(settings.updated_at, original_updated)
class ReviewSettingsServiceTest(TestCase):
def test_update_min_approvals(self) -> None:
settings: ReviewSettings = review_settings_update(
default_min_approvals=3
)
self.assertEqual(settings.default_min_approvals, 3)
def test_update_allow_any_approver(self) -> None:
settings: ReviewSettings = review_settings_update(
allow_any_approver=False
)
self.assertFalse(settings.allow_any_approver)
def test_update_both_fields(self) -> None:
settings: ReviewSettings = review_settings_update(
default_min_approvals=5,
allow_any_approver=False,
)
self.assertEqual(settings.default_min_approvals, 5)
self.assertFalse(settings.allow_any_approver)
def test_update_partial_leaves_other_field(self) -> None:
review_settings_update(
default_min_approvals=3, allow_any_approver=False
)
settings: ReviewSettings = review_settings_update(
default_min_approvals=7
)
self.assertEqual(settings.default_min_approvals, 7)
self.assertFalse(settings.allow_any_approver)
def test_update_min_approvals_zero_raises(self) -> None:
with self.assertRaises(ValidationError):
review_settings_update(default_min_approvals=0)
def test_update_min_approvals_negative_raises(self) -> None:
with self.assertRaises(ValidationError):
review_settings_update(default_min_approvals=-1)
def test_no_op_update_returns_settings(self) -> None:
settings: ReviewSettings = review_settings_update()
self.assertIsInstance(settings, ReviewSettings)
class ReviewSettingsSelectorsTest(TestCase):
def test_review_settings_load_returns_instance(self) -> None:
settings: ReviewSettings = review_settings_load()
self.assertIsInstance(settings, ReviewSettings)
def test_review_settings_load_is_idempotent(self) -> None:
s1: ReviewSettings = review_settings_load()
s2: ReviewSettings = review_settings_load()
self.assertEqual(s1.pk, s2.pk)
+92
View File
@@ -0,0 +1,92 @@
from django.contrib import admin
from django.contrib.auth.admin import UserAdmin as BaseUserAdmin
from django.utils.translation import gettext_lazy as _
from apps.users.models import User
@admin.register(User)
class UserAdmin(BaseUserAdmin):
list_display = (
User.username.field.name,
User.email.field.name,
User.role.field.name,
User.first_name.field.name,
User.last_name.field.name,
User._meta.get_field("is_active").name,
User.is_staff.field.name,
)
list_filter = (
User.role.field.name,
User._meta.get_field("is_active").name,
User.is_staff.field.name,
User.is_superuser.field.name,
)
search_fields = (
User.username.field.name,
User.email.field.name,
User.first_name.field.name,
User.last_name.field.name,
)
ordering = (User.username.field.name,)
fieldsets = (
(None, {"fields": (User.username.field.name, "password")}),
(
_("Personal info"),
{
"fields": (
User.first_name.field.name,
User.last_name.field.name,
User.email.field.name,
)
},
),
(
_("Platform role"),
{
"fields": (User.role.field.name,),
"description": _(
"Platform role that defines user permissions: "
"admin, experimenter, approver, or viewer."
),
},
),
(
_("Permissions"),
{
"fields": (
User._meta.get_field("is_active").name,
User.is_staff.field.name,
User.is_superuser.field.name,
User.groups.field.name,
User.user_permissions.field.name,
),
},
),
(
_("Important dates"),
{
"fields": (
User.last_login.field.name,
User.date_joined.field.name,
)
},
),
)
add_fieldsets = (
(
None,
{
"classes": ("wide",),
"fields": (
User.username.field.name,
User.email.field.name,
"password1",
"password2",
User.role.field.name,
),
},
),
)
+85
View File
@@ -0,0 +1,85 @@
import logging
from collections.abc import Callable
from functools import wraps
from typing import Any
from django.http import HttpRequest
from ninja.security import HttpBearer
from apps.users.auth.jwt import TokenError, decode_access_token
from apps.users.models import User
from config.errors import ForbiddenError
logger: logging.Logger = logging.getLogger("lotty")
class JWTBearer(HttpBearer):
def authenticate(
self,
request: HttpRequest,
token: str,
) -> User | None:
try:
payload: dict[str, Any] = decode_access_token(token)
except TokenError:
logger.debug("JWT authentication failed: invalid token")
return None
user_id: str | None = payload.get("sub")
if not user_id:
logger.debug("JWT authentication failed: missing 'sub' claim")
return None
try:
user: User = User.objects.get(pk=user_id, is_active=True)
except User.DoesNotExist:
logger.debug(
"JWT authentication failed: user %s not found or inactive",
user_id,
)
return None
return user
# Singleton is not the best way, yep
jwt_bearer = JWTBearer()
def require_roles(*allowed_roles: str) -> Callable:
def checker(request: HttpRequest) -> User:
user: User | None = getattr(request, "auth", None)
if user is None:
raise ForbiddenError("Authentication required")
if user.role not in allowed_roles:
raise ForbiddenError(
f"Role '{user.role}' is not permitted for this action. "
f"Required one of: {', '.join(allowed_roles)}"
)
return user
def guard(arg: HttpRequest | Callable) -> Callable | User:
if isinstance(arg, HttpRequest):
return checker(arg)
if callable(arg):
@wraps(arg)
def wrapped(request: HttpRequest, *args, **kwargs) -> Callable:
checker(request)
return arg(request, *args, **kwargs)
return wrapped
raise TypeError("Role guard expects a request or a callable")
return guard
require_admin = require_roles("admin")
require_experimenter = require_roles("experimenter")
require_approver = require_roles("approver")
require_admin_or_experimenter = require_roles("admin", "experimenter")
require_admin_or_approver = require_roles("admin", "approver")
+104
View File
@@ -0,0 +1,104 @@
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import UUID
import jwt
from django.conf import settings
_ALGORITHM = "HS256"
_ACCESS_TOKEN_LIFETIME = timedelta(hours=24)
_REFRESH_TOKEN_LIFETIME = timedelta(days=7)
def _get_secret() -> str:
return settings.SECRET_KEY
def create_access_token(
user_id: UUID | str,
role: str,
*,
extra_claims: dict[str, Any] | None = None,
lifetime: timedelta | None = None,
) -> str:
now: datetime = datetime.now(tz=UTC)
exp: datetime = now + (lifetime or _ACCESS_TOKEN_LIFETIME)
payload: dict[str, Any] = {
"sub": str(user_id),
"role": role,
"type": "access",
"iat": now,
"exp": exp,
}
if extra_claims:
payload.update(extra_claims)
return jwt.encode(payload, _get_secret(), algorithm=_ALGORITHM)
def create_refresh_token(
user_id: UUID | str,
*,
lifetime: timedelta | None = None,
) -> str:
now: datetime = datetime.now(tz=UTC)
exp: datetime = now + (lifetime or _REFRESH_TOKEN_LIFETIME)
payload: dict[str, Any] = {
"sub": str(user_id),
"type": "refresh",
"iat": now,
"exp": exp,
}
return jwt.encode(payload, _get_secret(), algorithm=_ALGORITHM)
def create_token_pair(
user_id: UUID | str,
role: str,
) -> dict[str, str]:
return {
"access": create_access_token(user_id, role),
"refresh": create_refresh_token(user_id),
}
class TokenError(Exception):
def __init__(self, detail: str = "Invalid token") -> None:
self.detail: str = detail
super().__init__(detail)
def decode_token(
token: str,
*,
expected_type: str | None = None,
) -> dict[str, Any]:
try:
payload: dict[str, Any] = jwt.decode(
token,
_get_secret(),
algorithms=[_ALGORITHM],
)
except jwt.ExpiredSignatureError:
raise TokenError("Token has expired") from None
except jwt.InvalidTokenError as exc:
raise TokenError(f"Invalid token: {exc}") from None
if expected_type and payload.get("type") != expected_type:
raise TokenError(
f"Expected token type '{expected_type}', "
f"got '{payload.get('type')}'"
)
return payload
def decode_access_token(token: str) -> dict[str, Any]:
return decode_token(token, expected_type="access")
def decode_refresh_token(token: str) -> dict[str, Any]:
return decode_token(token, expected_type="refresh")
@@ -0,0 +1,152 @@
from django.core.management.base import BaseCommand, CommandParser
from apps.users.models import User, UserRole
SEED_USERS = [
{
"username": "admin",
"email": "admin@lotty.local",
"role": UserRole.ADMIN,
"first_name": "Admin",
"last_name": "User",
"is_staff": True,
"is_superuser": True,
},
{
"username": "experimenter",
"email": "experimenter@lotty.local",
"role": UserRole.EXPERIMENTER,
"first_name": "Experimenter",
"last_name": "User",
"is_staff": False,
"is_superuser": False,
},
{
"username": "approver",
"email": "approver@lotty.local",
"role": UserRole.APPROVER,
"first_name": "Approver",
"last_name": "User",
"is_staff": False,
"is_superuser": False,
},
{
"username": "approver2",
"email": "approver2@lotty.local",
"role": UserRole.APPROVER,
"first_name": "Approver Two",
"last_name": "User",
"is_staff": False,
"is_superuser": False,
},
{
"username": "viewer",
"email": "viewer@lotty.local",
"role": UserRole.VIEWER,
"first_name": "Viewer",
"last_name": "User",
"is_staff": False,
"is_superuser": False,
},
]
DEFAULT_PASSWORD = "password123" # noqa: S105
class Command(BaseCommand):
help = (
"Seed the database with demo users for each platform role "
"(admin, experimenter, approver, viewer)."
)
def add_arguments(self, parser: CommandParser) -> None:
parser.add_argument(
"--password",
type=str,
default=DEFAULT_PASSWORD,
help=(
f"Password to set for all seeded users "
f"(default: {DEFAULT_PASSWORD})."
),
)
parser.add_argument(
"--force",
action="store_true",
default=False,
help=(
"Overwrite existing users with the same username "
"(resets their password and fields)."
),
)
def handle(self, *args, **options) -> None:
password: str = options["password"]
force: bool = options["force"]
created_count = 0
updated_count = 0
skipped_count = 0
for user_data in SEED_USERS:
username = user_data["username"]
existing = User.objects.filter(username=username).first()
if existing and not force:
self.stdout.write(
self.style.WARNING(
f" SKIP {username} (already exists, "
f"use --force to overwrite)"
)
)
skipped_count += 1
continue
if existing and force:
existing.email = user_data["email"]
existing.role = user_data["role"]
existing.first_name = user_data["first_name"]
existing.last_name = user_data["last_name"]
existing.is_staff = user_data["is_staff"]
existing.is_superuser = user_data["is_superuser"]
existing.is_active = True
existing.set_password(password)
existing.save()
self.stdout.write(
self.style.WARNING(
f" UPDATE {username} (role={user_data['role']})"
)
)
updated_count += 1
continue
user = User(
username=username,
email=user_data["email"],
role=user_data["role"],
first_name=user_data["first_name"],
last_name=user_data["last_name"],
is_staff=user_data["is_staff"],
is_superuser=user_data["is_superuser"],
is_active=True,
)
user.set_password(password)
user.save()
self.stdout.write(
self.style.SUCCESS(
f" CREATE {username} (role={user_data['role']})"
)
)
created_count += 1
self.stdout.write("")
self.stdout.write(
self.style.SUCCESS(
f"Done: {created_count} created, "
f"{updated_count} updated, "
f"{skipped_count} skipped."
)
)
self.stdout.write(
self.style.NOTICE(f"All seeded users have password: {password}")
)
+21 -108
View File
@@ -1,4 +1,4 @@
# Generated by Django 5.2.11 on 2026-02-10 20:37 # Generated by Django 5.2.11 on 2026-02-12 13:05
import django.contrib.auth.models import django.contrib.auth.models
import django.contrib.auth.validators import django.contrib.auth.validators
@@ -12,122 +12,35 @@ class Migration(migrations.Migration):
initial = True initial = True
dependencies = [ dependencies = [
("auth", "0012_alter_user_first_name_max_length"), ('auth', '0012_alter_user_first_name_max_length'),
] ]
operations = [ operations = [
migrations.CreateModel( migrations.CreateModel(
name="User", name='User',
fields=[ fields=[
("password", models.CharField(max_length=128, verbose_name="password")), ('password', models.CharField(max_length=128, verbose_name='password')),
( ('last_login', models.DateTimeField(blank=True, null=True, verbose_name='last login')),
"last_login", ('is_superuser', models.BooleanField(default=False, help_text='Designates that this user has all permissions without explicitly assigning them.', verbose_name='superuser status')),
models.DateTimeField( ('username', models.CharField(error_messages={'unique': 'A user with that username already exists.'}, help_text='Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only.', max_length=150, unique=True, validators=[django.contrib.auth.validators.UnicodeUsernameValidator()], verbose_name='username')),
blank=True, null=True, verbose_name="last login" ('first_name', models.CharField(blank=True, max_length=150, verbose_name='first name')),
), ('last_name', models.CharField(blank=True, max_length=150, verbose_name='last name')),
), ('email', models.EmailField(blank=True, max_length=254, verbose_name='email address')),
( ('is_staff', models.BooleanField(default=False, help_text='Designates whether the user can log into this admin site.', verbose_name='staff status')),
"is_superuser", ('is_active', models.BooleanField(default=True, help_text='Designates whether this user should be treated as active. Unselect this instead of deleting accounts.', verbose_name='active')),
models.BooleanField( ('date_joined', models.DateTimeField(default=django.utils.timezone.now, verbose_name='date joined')),
default=False, ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
help_text="Designates that this user has all permissions without explicitly assigning them.", ('role', models.CharField(choices=[('admin', 'Admin'), ('experimenter', 'Experimenter'), ('approver', 'Approver'), ('viewer', 'Viewer')], db_index=True, default='viewer', help_text='Platform role that defines user permissions.', max_length=20, verbose_name='role')),
verbose_name="superuser status", ('groups', models.ManyToManyField(blank=True, help_text='The groups this user belongs to. A user will get all permissions granted to each of their groups.', related_name='user_set', related_query_name='user', to='auth.group', verbose_name='groups')),
), ('user_permissions', models.ManyToManyField(blank=True, help_text='Specific permissions for this user.', related_name='user_set', related_query_name='user', to='auth.permission', verbose_name='user permissions')),
),
(
"username",
models.CharField(
error_messages={
"unique": "A user with that username already exists."
},
help_text="Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only.",
max_length=150,
unique=True,
validators=[
django.contrib.auth.validators.UnicodeUsernameValidator()
],
verbose_name="username",
),
),
(
"first_name",
models.CharField(
blank=True, max_length=150, verbose_name="first name"
),
),
(
"last_name",
models.CharField(
blank=True, max_length=150, verbose_name="last name"
),
),
(
"email",
models.EmailField(
blank=True, max_length=254, verbose_name="email address"
),
),
(
"is_staff",
models.BooleanField(
default=False,
help_text="Designates whether the user can log into this admin site.",
verbose_name="staff status",
),
),
(
"is_active",
models.BooleanField(
default=True,
help_text="Designates whether this user should be treated as active. Unselect this instead of deleting accounts.",
verbose_name="active",
),
),
(
"date_joined",
models.DateTimeField(
default=django.utils.timezone.now, verbose_name="date joined"
),
),
(
"id",
models.UUIDField(
default=uuid.uuid4,
editable=False,
primary_key=True,
serialize=False,
),
),
(
"groups",
models.ManyToManyField(
blank=True,
help_text="The groups this user belongs to. A user will get all permissions granted to each of their groups.",
related_name="user_set",
related_query_name="user",
to="auth.group",
verbose_name="groups",
),
),
(
"user_permissions",
models.ManyToManyField(
blank=True,
help_text="Specific permissions for this user.",
related_name="user_set",
related_query_name="user",
to="auth.permission",
verbose_name="user permissions",
),
),
], ],
options={ options={
"verbose_name": "user", 'verbose_name': 'user',
"verbose_name_plural": "users", 'verbose_name_plural': 'users',
"swappable": "AUTH_USER_MODEL", 'swappable': 'AUTH_USER_MODEL',
}, },
managers=[ managers=[
("objects", django.contrib.auth.models.UserManager()), ('objects', django.contrib.auth.models.UserManager()),
], ],
), ),
] ]
+33
View File
@@ -1,11 +1,44 @@
from django.contrib.auth.models import AbstractUser from django.contrib.auth.models import AbstractUser
from django.db import models
from django.utils.translation import gettext_lazy as _ from django.utils.translation import gettext_lazy as _
from apps.core.models import BaseModel from apps.core.models import BaseModel
class UserRole(models.TextChoices):
ADMIN = "admin", _("Admin")
EXPERIMENTER = "experimenter", _("Experimenter")
APPROVER = "approver", _("Approver")
VIEWER = "viewer", _("Viewer")
class User(AbstractUser, BaseModel): class User(AbstractUser, BaseModel):
role = models.CharField(
max_length=20,
choices=UserRole.choices,
default=UserRole.VIEWER,
db_index=True,
verbose_name=_("role"),
help_text=_("Platform role that defines user permissions"),
)
class Meta: class Meta:
swappable = "AUTH_USER_MODEL" swappable = "AUTH_USER_MODEL"
verbose_name = _("user") verbose_name = _("user")
verbose_name_plural = _("users") verbose_name_plural = _("users")
@property
def is_admin_role(self) -> bool:
return self.role == UserRole.ADMIN
@property
def is_experimenter(self) -> bool:
return self.role == UserRole.EXPERIMENTER
@property
def is_approver(self) -> bool:
return self.role == UserRole.APPROVER
@property
def is_viewer(self) -> bool:
return self.role == UserRole.VIEWER
+71 -4
View File
@@ -1,8 +1,8 @@
import uuid import uuid
from django.db.models import QuerySet from django.db.models import Q, QuerySet
from apps.users.models import User from apps.users.models import User, UserRole
def user_get_by_id(user_id: str) -> User | None: def user_get_by_id(user_id: str) -> User | None:
@@ -13,5 +13,72 @@ def user_get_by_id(user_id: str) -> User | None:
return User.objects.filter(id=user_id).first() return User.objects.filter(id=user_id).first()
def user_list() -> QuerySet[User]: def user_get_by_username(username: str) -> User | None:
return User.objects.all() return User.objects.filter(username=username).first()
def user_get_by_email(email: str) -> User | None:
return User.objects.filter(email=email).first()
def user_list(
*,
is_active: bool | None = None,
role: str | None = None,
search: str | None = None,
) -> QuerySet[User]:
qs = User.objects.all()
if is_active is not None:
qs = qs.filter(is_active=is_active)
if role is not None:
qs = qs.filter(role=role)
if search:
qs = qs.filter(
Q(username__icontains=search)
| Q(email__icontains=search)
| Q(first_name__icontains=search)
| Q(last_name__icontains=search)
)
return qs.order_by("username")
def user_list_by_role(role: str) -> QuerySet[User]:
return User.objects.filter(role=role, is_active=True).order_by("username")
def user_list_admins() -> QuerySet[User]:
return user_list_by_role(UserRole.ADMIN)
def user_list_experimenters() -> QuerySet[User]:
return user_list_by_role(UserRole.EXPERIMENTER)
def user_list_approvers() -> QuerySet[User]:
return user_list_by_role(UserRole.APPROVER)
def user_list_viewers() -> QuerySet[User]:
return user_list_by_role(UserRole.VIEWER)
def user_exists_with_username(
username: str, *, exclude_id: str | None = None
) -> bool:
qs = User.objects.filter(username=username)
if exclude_id is not None:
qs = qs.exclude(id=exclude_id)
return qs.exists()
def user_exists_with_email(
email: str, *, exclude_id: str | None = None
) -> bool:
qs = User.objects.filter(email=email)
if exclude_id is not None:
qs = qs.exclude(id=exclude_id)
return qs.exists()
+69 -2
View File
@@ -1,6 +1,9 @@
from typing import Any from typing import Any
from apps.users.models import User from django.core.exceptions import ValidationError
from django.db import transaction
from apps.users.models import User, UserRole
def user_create( def user_create(
@@ -8,9 +11,10 @@ def user_create(
username: str, username: str,
email: str, email: str,
password: str | None = None, password: str | None = None,
role: str = UserRole.VIEWER,
**extra_fields: Any, **extra_fields: Any,
) -> User: ) -> User:
user = User(username=username, email=email, **extra_fields) user = User(username=username, email=email, role=role, **extra_fields)
if password is not None: if password is not None:
user.set_password(password) user.set_password(password)
else: else:
@@ -18,3 +22,66 @@ def user_create(
user.save() user.save()
return user return user
def user_update(
*,
user: User,
username: str | None = None,
email: str | None = None,
password: str | None = None,
role: str | None = None,
is_active: bool | None = None,
first_name: str | None = None,
last_name: str | None = None,
) -> User:
if username is not None:
user.username = username
if email is not None:
user.email = email
if role is not None:
user.role = role
if is_active is not None:
user.is_active = is_active
if first_name is not None:
user.first_name = first_name
if last_name is not None:
user.last_name = last_name
if password is not None:
user.set_password(password)
user.save()
return user
def user_assign_role(*, user: User, role: str) -> User:
valid_roles = {choice[0] for choice in UserRole.choices}
if role not in valid_roles:
raise ValidationError(
{
"role": f"Invalid role '{role}'. "
f"Must be one of: {', '.join(sorted(valid_roles))}"
}
)
user.role = role
user.save()
return user
@transaction.atomic
def user_delete(*, user: User) -> None:
user.delete()
def user_deactivate(*, user: User) -> User:
user.is_active = False
user.save()
return user
def user_activate(*, user: User) -> User:
user.is_active = True
user.save()
return user
+24
View File
@@ -0,0 +1,24 @@
from apps.users.auth.jwt import create_access_token
from apps.users.models import User, UserRole
from apps.users.services import user_create
def _make_user(
username="testuser",
email="test@lotty.local",
password="testpass123", # noqa: S107
role=UserRole.VIEWER,
**kwargs,
) -> User:
return user_create(
username=username,
email=email,
password=password,
role=role,
**kwargs,
)
def _auth_header(user) -> str:
token: str = create_access_token(user.pk, user.role)
return f"Bearer {token}"
+116
View File
@@ -0,0 +1,116 @@
import uuid
from datetime import timedelta
from typing import Any
from django.core.handlers.wsgi import WSGIRequest
from django.test import RequestFactory, TestCase
from apps.users.auth.bearer import JWTBearer
from apps.users.auth.jwt import (
TokenError,
create_access_token,
create_refresh_token,
create_token_pair,
decode_access_token,
decode_refresh_token,
decode_token,
)
from apps.users.models import User, UserRole
from ._helpers import _make_user
class JWTCreateTest(TestCase):
def test_create_access_token(self) -> None:
token: str = create_access_token(uuid.uuid4(), "admin")
self.assertIsInstance(token, str)
self.assertTrue(len(token) > 0)
def test_create_refresh_token(self) -> None:
token: str = create_refresh_token(uuid.uuid4())
self.assertIsInstance(token, str)
def test_create_token_pair(self) -> None:
pair: dict[str, str] = create_token_pair(uuid.uuid4(), "viewer")
self.assertIn("access", pair)
self.assertIn("refresh", pair)
class JWTDecodeTest(TestCase):
def setUp(self) -> None:
self.uid: uuid.UUID = uuid.uuid4()
def test_decode_access_token(self) -> None:
token: str = create_access_token(self.uid, "experimenter")
payload: dict[str, Any] = decode_access_token(token)
self.assertEqual(payload["sub"], str(self.uid))
self.assertEqual(payload["role"], "experimenter")
self.assertEqual(payload["type"], "access")
def test_decode_refresh_token(self) -> None:
token: str = create_refresh_token(self.uid)
payload: dict[str, Any] = decode_refresh_token(token)
self.assertEqual(payload["sub"], str(self.uid))
self.assertEqual(payload["type"], "refresh")
def test_decode_wrong_type_raises(self) -> None:
token: str = create_refresh_token(self.uid)
with self.assertRaises(TokenError):
decode_access_token(token)
def test_decode_expired_token_raises(self) -> None:
token: str = create_access_token(
self.uid, "admin", lifetime=timedelta(seconds=-1)
)
with self.assertRaises(TokenError):
decode_access_token(token)
def test_decode_invalid_token_raises(self) -> None:
with self.assertRaises(TokenError):
decode_token("not.a.jwt")
def test_extra_claims(self) -> None:
token: str = create_access_token(
self.uid, "admin", extra_claims={"org": "lotty"}
)
payload: dict[str, Any] = decode_access_token(token)
self.assertEqual(payload["org"], "lotty")
class JWTBearerTest(TestCase):
def setUp(self) -> None:
self.bearer = JWTBearer()
self.user: User = _make_user(
username="bearer_user",
email="bearer@x.com",
role=UserRole.ADMIN,
)
def test_valid_token_returns_user(self) -> None:
token: str = create_access_token(self.user.pk, self.user.role)
request: WSGIRequest = RequestFactory().get("/")
result: User | None = self.bearer.authenticate(request, token)
self.assertEqual(result, self.user)
def test_invalid_token_returns_none(self) -> None:
request: WSGIRequest = RequestFactory().get("/")
result: User | None = self.bearer.authenticate(request, "garbage")
self.assertIsNone(result)
def test_nonexistent_user_returns_none(self) -> None:
token: str = create_access_token(uuid.uuid4(), "admin")
request: WSGIRequest = RequestFactory().get("/")
result: User | None = self.bearer.authenticate(request, token)
self.assertIsNone(result)
def test_inactive_user_returns_none(self) -> None:
self.user.is_active = False
self.user.save()
token: str = create_access_token(self.user.pk, self.user.role)
request: WSGIRequest = RequestFactory().get("/")
result: User | None = self.bearer.authenticate(request, token)
self.assertIsNone(result)
@@ -0,0 +1,56 @@
import uuid
from django.test import TestCase
from apps.users.models import User, UserRole
from ._helpers import _make_user
class UserRoleChoicesTest(TestCase):
def test_choices_count(self) -> None:
self.assertEqual(len(UserRole.choices), 4)
def test_choice_values(self) -> None:
values = {c[0] for c in UserRole.choices}
self.assertEqual(
values, {"admin", "experimenter", "approver", "viewer"}
)
class UserModelTest(TestCase):
def test_default_role_is_viewer(self) -> None:
user: User = _make_user()
self.assertEqual(user.role, UserRole.VIEWER)
def test_role_properties(self) -> None:
admin: User = _make_user(
username="a", email="a@x.com", role=UserRole.ADMIN
)
self.assertTrue(admin.is_admin_role)
self.assertFalse(admin.is_experimenter)
self.assertFalse(admin.is_approver)
self.assertFalse(admin.is_viewer)
exp: User = _make_user(
username="e", email="e@x.com", role=UserRole.EXPERIMENTER
)
self.assertTrue(exp.is_experimenter)
appr: User = _make_user(
username="ap", email="ap@x.com", role=UserRole.APPROVER
)
self.assertTrue(appr.is_approver)
viewer: User = _make_user(
username="v", email="v@x.com", role=UserRole.VIEWER
)
self.assertTrue(viewer.is_viewer)
def test_uuid_primary_key(self) -> None:
user: User = _make_user()
self.assertIsInstance(user.pk, uuid.UUID)
def test_str_representation(self) -> None:
user: User = _make_user(username="hello")
self.assertEqual(str(user), "hello")
@@ -0,0 +1,44 @@
from django.core.handlers.wsgi import WSGIRequest
from django.test import RequestFactory, TestCase
from apps.users.auth.bearer import require_admin, require_roles
from apps.users.models import User, UserRole
from config.errors import ForbiddenError
from ._helpers import _make_user
class RequireRolesTest(TestCase):
def setUp(self) -> None:
self.admin: User = _make_user(
username="rr_admin", email="rr_admin@x.com", role=UserRole.ADMIN
)
self.viewer: User = _make_user(
username="rr_viewer", email="rr_viewer@x.com", role=UserRole.VIEWER
)
def _make_request(self, user) -> WSGIRequest:
request: WSGIRequest = RequestFactory().get("/")
request.auth = user
return request
def test_require_admin_passes(self) -> None:
request: WSGIRequest = self._make_request(self.admin)
result = require_admin(request)
self.assertEqual(result, self.admin)
def test_require_admin_denies_viewer(self) -> None:
request: WSGIRequest = self._make_request(self.viewer)
with self.assertRaises(ForbiddenError):
require_admin(request)
def test_require_roles_multiple(self) -> None:
checker = require_roles("admin", "viewer")
request: WSGIRequest = self._make_request(self.viewer)
result = checker(request)
self.assertEqual(result, self.viewer)
def test_require_roles_no_auth_raises(self) -> None:
request: WSGIRequest = RequestFactory().get("/")
with self.assertRaises(ForbiddenError):
require_admin(request)
@@ -0,0 +1,123 @@
import uuid
from django.db.models import QuerySet
from django.test import TestCase
from apps.users.models import User, UserRole
from apps.users.selectors import (
user_exists_with_email,
user_exists_with_username,
user_get_by_email,
user_get_by_id,
user_get_by_username,
user_list,
user_list_admins,
user_list_approvers,
user_list_by_role,
user_list_experimenters,
user_list_viewers,
)
from ._helpers import _make_user
class UserSelectorsTest(TestCase):
def setUp(self) -> None:
self.admin: User = _make_user(
username="sel_admin",
email="sel_admin@x.com",
role=UserRole.ADMIN,
)
self.exp: User = _make_user(
username="sel_exp",
email="sel_exp@x.com",
role=UserRole.EXPERIMENTER,
)
self.appr: User = _make_user(
username="sel_appr",
email="sel_appr@x.com",
role=UserRole.APPROVER,
)
self.viewer: User = _make_user(
username="sel_viewer",
email="sel_viewer@x.com",
role=UserRole.VIEWER,
)
def test_get_by_id(self) -> None:
found: User | None = user_get_by_id(str(self.admin.pk))
self.assertEqual(found, self.admin)
def test_get_by_id_invalid_uuid(self) -> None:
self.assertIsNone(user_get_by_id("not-a-uuid"))
def test_get_by_id_nonexistent(self) -> None:
self.assertIsNone(user_get_by_id(str(uuid.uuid4())))
def test_get_by_username(self) -> None:
found: User | None = user_get_by_username("sel_exp")
self.assertEqual(found, self.exp)
def test_get_by_username_nonexistent(self) -> None:
self.assertIsNone(user_get_by_username("ghost"))
def test_get_by_email(self) -> None:
found: User | None = user_get_by_email("sel_appr@x.com")
self.assertEqual(found, self.appr)
def test_list_all(self) -> None:
qs: QuerySet[User] = user_list()
self.assertEqual(qs.count(), 4)
def test_list_filter_by_role(self) -> None:
qs: QuerySet[User] = user_list(role=UserRole.ADMIN)
self.assertEqual(qs.count(), 1)
self.assertEqual(qs.first(), self.admin)
def test_list_filter_by_is_active(self) -> None:
self.viewer.is_active = False
self.viewer.save()
qs: QuerySet[User] = user_list(is_active=True)
self.assertEqual(qs.count(), 3)
def test_list_filter_by_search(self) -> None:
qs: QuerySet[User] = user_list(search="sel_exp")
self.assertEqual(qs.count(), 1)
def test_list_by_role(self) -> None:
qs: QuerySet[User] = user_list_by_role(UserRole.APPROVER)
self.assertEqual(qs.count(), 1)
def test_list_admins(self) -> None:
self.assertEqual(user_list_admins().count(), 1)
def test_list_experimenters(self) -> None:
self.assertEqual(user_list_experimenters().count(), 1)
def test_list_approvers(self) -> None:
self.assertEqual(user_list_approvers().count(), 1)
def test_list_viewers(self) -> None:
self.assertEqual(user_list_viewers().count(), 1)
def test_exists_with_username(self) -> None:
self.assertTrue(user_exists_with_username("sel_admin"))
self.assertFalse(user_exists_with_username("ghost"))
def test_exists_with_username_exclude(self) -> None:
self.assertFalse(
user_exists_with_username(
"sel_admin", exclude_id=str(self.admin.pk)
)
)
def test_exists_with_email(self) -> None:
self.assertTrue(user_exists_with_email("sel_viewer@x.com"))
self.assertFalse(user_exists_with_email("ghost@x.com"))
def test_exists_with_email_exclude(self) -> None:
self.assertFalse(
user_exists_with_email(
"sel_viewer@x.com", exclude_id=str(self.viewer.pk)
)
)
@@ -0,0 +1,128 @@
from django.core.exceptions import ValidationError
from django.test import TestCase
from apps.users.models import User, UserRole
from apps.users.services import (
user_activate,
user_assign_role,
user_create,
user_deactivate,
user_delete,
user_update,
)
from ._helpers import _make_user
class UserCreateServiceTest(TestCase):
def test_create_with_defaults(self) -> None:
user: User = user_create(
username="svc_user",
email="svc@lotty.local",
password="pass1234",
)
self.assertEqual(user.role, UserRole.VIEWER)
self.assertTrue(user.check_password("pass1234"))
self.assertTrue(user.is_active)
def test_create_with_role(self) -> None:
user: User = user_create(
username="admin_svc",
email="admin_svc@lotty.local",
password="pass1234",
role=UserRole.ADMIN,
)
self.assertEqual(user.role, UserRole.ADMIN)
def test_create_without_password(self) -> None:
user: User = user_create(
username="nopw",
email="nopw@lotty.local",
)
self.assertFalse(user.has_usable_password())
def test_create_with_extra_fields(self) -> None:
user: User = user_create(
username="extra",
email="extra@lotty.local",
password="pass1234",
first_name="First",
last_name="Last",
)
self.assertEqual(user.first_name, "First")
self.assertEqual(user.last_name, "Last")
class UserUpdateServiceTest(TestCase):
def setUp(self) -> None:
self.user: User = _make_user()
def test_update_username(self) -> None:
updated: User = user_update(user=self.user, username="newname")
self.assertEqual(updated.username, "newname")
def test_update_email(self) -> None:
updated: User = user_update(user=self.user, email="new@lotty.local")
self.assertEqual(updated.email, "new@lotty.local")
def test_update_role(self) -> None:
updated: User = user_update(user=self.user, role=UserRole.ADMIN)
self.assertEqual(updated.role, UserRole.ADMIN)
def test_update_password(self) -> None:
updated: User = user_update(user=self.user, password="newpass99")
self.assertTrue(updated.check_password("newpass99"))
def test_update_is_active(self) -> None:
updated: User = user_update(user=self.user, is_active=False)
self.assertFalse(updated.is_active)
def test_partial_update_leaves_other_fields(self) -> None:
original_email = self.user.email
updated: User = user_update(user=self.user, username="changed")
self.assertEqual(updated.email, original_email)
def test_update_names(self) -> None:
updated: User = user_update(
user=self.user, first_name="Jane", last_name="Doe"
)
self.assertEqual(updated.first_name, "Jane")
self.assertEqual(updated.last_name, "Doe")
class UserAssignRoleServiceTest(TestCase):
def setUp(self) -> None:
self.user: User = _make_user()
def test_assign_valid_role(self) -> None:
updated: User = user_assign_role(
user=self.user, role=UserRole.EXPERIMENTER
)
self.assertEqual(updated.role, UserRole.EXPERIMENTER)
def test_assign_invalid_role_raises(self) -> None:
with self.assertRaises(ValidationError):
user_assign_role(user=self.user, role="superadmin")
class UserDeleteServiceTest(TestCase):
def test_hard_delete(self) -> None:
user: User = _make_user()
pk = user.pk
user_delete(user=user)
self.assertFalse(User.objects.filter(pk=pk).exists())
class UserActivateDeactivateServiceTest(TestCase):
def setUp(self) -> None:
self.user: User = _make_user()
def test_deactivate(self) -> None:
updated: User = user_deactivate(user=self.user)
self.assertFalse(updated.is_active)
def test_activate(self) -> None:
self.user.is_active = False
self.user.save()
updated: User = user_activate(user=self.user)
self.assertTrue(updated.is_active)
+6 -1
View File
@@ -429,8 +429,13 @@ INSTALLED_APPS = [
"storages", "storages",
# Internal apps # Internal apps
"apps.core", "apps.core",
"apps.flags",
"apps.users", "apps.users",
# API apps "apps.reviews",
# API v1 apps
"api.v1.auth",
"api.v1.users",
"api.v1.reviews",
] ]
# GUID # GUID
+2
View File
@@ -172,6 +172,8 @@ ignore = [
"PT022", "PT022",
"RUF001", "RUF001",
"RUF012", "RUF012",
"PT009", # django docs recommends using unittest-style
"PT027",
] ]
[tool.ruff.lint.isort] [tool.ruff.lint.isort]