try UserState enum

This commit is contained in:
Kenneth Kehl
2025-08-28 11:12:49 -07:00
parent 99110608aa
commit 1cef8881a8
13 changed files with 94 additions and 27 deletions
+2 -2
View File
@@ -169,7 +169,7 @@
"filename": "app/enums.py",
"hashed_secret": "12322e07b94ee3c7cd65a2952ece441538b53eb3",
"is_verified": false,
"line_number": 123,
"line_number": 129,
"is_secret": false
}
],
@@ -374,5 +374,5 @@
}
]
},
"generated_at": "2025-08-12T18:08:49Z"
"generated_at": "2025-08-28T18:12:29Z"
}
+2 -1
View File
@@ -3,6 +3,7 @@ from sqlalchemy.sql.expression import func
from app import db
from app.dao.dao_utils import VersionOptions, autocommit, version_class
from app.enums import UserState
from app.models import Domain, Organization, Service, User
@@ -125,7 +126,7 @@ def dao_get_users_for_organization(organization_id):
return (
db.session.query(User)
.join(User.organizations)
.where(Organization.id == organization_id, User.state == "active")
.where(Organization.id == organization_id, User.state == UserState.ACTIVE)
.order_by(User.created_at)
.all()
)
+2 -1
View File
@@ -2,6 +2,7 @@ from sqlalchemy import select
from app import db
from app.dao.dao_utils import autocommit
from app.enums import UserState
from app.models import ServiceUser, User
@@ -17,7 +18,7 @@ def dao_get_active_service_users(service_id):
stmt = (
select(ServiceUser)
.join(User, User.id == ServiceUser.user_id)
.where(User.state == "active", ServiceUser.service_id == service_id)
.where(User.state == UserState.ACTIVE, ServiceUser.service_id == service_id)
)
return db.session.execute(stmt).scalars().all()
+4 -1
View File
@@ -22,6 +22,7 @@ from app.enums import (
NotificationStatus,
NotificationType,
ServicePermissionType,
UserState,
)
from app.models import (
AnnualBilling,
@@ -738,7 +739,9 @@ def dao_resume_service(service_id):
def dao_fetch_active_users_for_service(service_id):
stmt = select(User).where(User.services.any(id=service_id), User.state == "active")
stmt = select(User).where(
User.services.any(id=service_id), User.state == UserState.ACTIVE
)
result = db.session.execute(stmt)
return result.scalars().all()
+3 -3
View File
@@ -11,7 +11,7 @@ from app import db
from app.dao.dao_utils import autocommit
from app.dao.permissions_dao import permission_dao
from app.dao.service_user_dao import dao_get_service_users_by_user_id
from app.enums import AuthType, PermissionType
from app.enums import AuthType, PermissionType, UserState
from app.errors import InvalidRequest
from app.models import Organization, Service, User, VerifyCode
from app.utils import escape_special_characters, get_archived_db_column_value, utc_now
@@ -233,7 +233,7 @@ def dao_archive_user(user):
user.password = str(uuid.uuid4())
# Changing the current_session_id signs the user out
user.current_session_id = "00000000-0000-0000-0000-000000000000"
user.state = "inactive"
user.state = UserState.INACTIVE
db.session.add(user)
@@ -243,7 +243,7 @@ def user_can_be_archived(user):
for service in active_services:
other_active_users = [
x for x in service.users if x.state == "active" and x != user
x for x in service.users if x.state == UserState.ACTIVE and x != user
]
if not other_active_users:
+6
View File
@@ -25,6 +25,12 @@ class AuthType(StrEnum):
WEBAUTHN = "webauthn_auth"
class UserState(StrEnum):
ACTIVE = "active"
INACTIVE = "inactive"
PENDING = "pending"
class CallbackType(StrEnum):
DELIVERY_STATUS = "delivery_status"
COMPLAINT = "complaint"
+5 -1
View File
@@ -28,6 +28,7 @@ from app.enums import (
ServicePermissionType,
TemplateProcessType,
TemplateType,
UserState,
)
from app.hashing import check_hash, hashpw
from app.history_meta import Versioned
@@ -70,6 +71,7 @@ _enum_column_names = {
PermissionType: "permission_types",
AgreementType: "agreement_types",
AgreementStatus: "agreement_statuses",
UserState: "user_states",
}
@@ -135,7 +137,9 @@ class User(db.Model):
)
logged_in_at = db.Column(db.DateTime, nullable=True)
failed_login_count = db.Column(db.Integer, nullable=False, default=0)
state = db.Column(db.String, nullable=False, default="pending")
state = enum_column(
UserState, index=True, nullable=False, default=UserState.PENDING
)
platform_admin = db.Column(db.Boolean, nullable=False, default=False)
current_session_id = db.Column(UUID(as_uuid=True), nullable=True)
auth_type = enum_column(AuthType, index=True, nullable=False, default=AuthType.SMS)
+5 -5
View File
@@ -32,7 +32,7 @@ from app.dao.users_dao import (
save_user_attribute,
use_user_code,
)
from app.enums import CodeType, KeyType, NotificationType, TemplateType
from app.enums import CodeType, KeyType, NotificationType, TemplateType, UserState
from app.errors import InvalidRequest, register_errors
from app.models import Permission, Service
from app.notifications.process_notifications import (
@@ -179,10 +179,10 @@ def archive_user(user_id):
def activate_user(user_id):
check_suspicious_id(user_id)
user = get_user_by_id(user_id=user_id)
if user.state == "active":
if user.state == UserState.ACTIVE:
raise InvalidRequest("User already active", status_code=400)
user.state = "active"
user.state = UserState.ACTIVE
save_model_user(user)
return jsonify(data=user.serialize()), 200
@@ -191,10 +191,10 @@ def activate_user(user_id):
def deactivate_user(user_id):
check_suspicious_id(user_id)
user = get_user_by_id(user_id=user_id)
if user.state == "pending":
if user.state == UserState.PENDING:
raise InvalidRequest("User already inactive", status_code=400)
user.state = "pending"
user.state = UserState.PENDING
save_model_user(user)
return jsonify(data=user.serialize()), 200
@@ -0,0 +1,51 @@
"""
Revision ID: 0418_user_state_enum
Revises: 0417_change_total_message_limit
Create Date: 2025-08-28 12:34:32.857422
"""
from contextlib import contextmanager
from enum import Enum
from re import I
from typing import Iterator, TypedDict
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
from app.enums import (
AuthType,
)
revision = "0410_enums_for_everything"
down_revision = "0409_fix_service_name"
user_state_enum = postgresql.ENUM(
"active", "pending", "inactive", name="user_states", create_type=False
)
def upgrade():
user_state_enum.create(op.get_bind(), checkfirst=True)
op.alter_column(
"user",
"state",
existing_type=sa.String(),
type_=user_state_enum,
existing_nullable=False,
)
def downgrade():
op.alter_column(
"user",
"state",
existing_type=user_state_enum,
type_=sa.String,
existing_nullable=False,
)
user_state_enum.drop(op.get_bind(), checkfirst=True)
+2 -2
View File
@@ -16,7 +16,7 @@ from app.dao.organization_dao import (
dao_get_users_for_organization,
dao_update_organization,
)
from app.enums import OrganizationType
from app.enums import OrganizationType, UserState
from app.models import Organization, Service
from app.utils import utc_now
from tests.app.db import (
@@ -310,7 +310,7 @@ def test_dao_get_users_for_organization_only_returns_active_users(sample_organiz
organization_id=sample_organization.id, user_id=second.id
)
second.state = "inactive"
second.state = UserState.INACTIVE
results = dao_get_users_for_organization(organization_id=sample_organization.id)
assert len(results) == 1
+4 -4
View File
@@ -26,7 +26,7 @@ from app.dao.users_dao import (
update_user_password,
user_can_be_archived,
)
from app.enums import AuthType, CodeType, PermissionType
from app.enums import AuthType, CodeType, PermissionType, UserState
from app.errors import InvalidRequest
from app.models import User, VerifyCode
from app.utils import utc_now
@@ -271,7 +271,7 @@ def test_dao_archive_user(sample_user, sample_organization, fake_uuid):
assert sample_user.current_session_id == uuid.UUID(
"00000000-0000-0000-0000-000000000000"
)
assert sample_user.state == "inactive"
assert sample_user.state == UserState.INACTIVE
assert not sample_user.check_password("password")
@@ -329,8 +329,8 @@ def test_user_cannot_be_archived_if_they_belong_to_a_service_with_no_other_activ
sample_service,
):
active_user = create_user(email="1@test.com")
pending_user = create_user(email="2@test.com", state="pending")
inactive_user = create_user(email="3@test.com", state="inactive")
pending_user = create_user(email="2@test.com", state=UserState.PENDING)
inactive_user = create_user(email="3@test.com", state=UserState.INACTIVE)
sample_service.users = [active_user, pending_user, inactive_user]
+2 -1
View File
@@ -44,6 +44,7 @@ from app.enums import (
NotificationType,
OrganizationType,
TemplateType,
UserState,
)
from app.models import (
AnnualBilling,
@@ -273,7 +274,7 @@ def test_create_test_user_command(notify_db_session, notify_api):
user = db.session.execute(stmt).scalars().first()
assert user.email_address == "somebody@fake.gov"
assert user.auth_type == AuthType.SMS
assert user.state == "active"
assert user.state == UserState.ACTIVE
def test_insert_inbound_numbers_from_file(notify_db_session, notify_api, tmpdir):
+6 -6
View File
@@ -10,7 +10,7 @@ from sqlalchemy import delete, func, select
from app import db
from app.dao.service_user_dao import dao_get_service_user, dao_update_service_user
from app.enums import AuthType, KeyType, NotificationType, PermissionType
from app.enums import AuthType, KeyType, NotificationType, PermissionType, UserState
from app.models import Notification, Permission, User
from tests.app.db import (
create_organization,
@@ -766,23 +766,23 @@ def test_send_user_confirm_new_email_returns_400_when_email_missing(
def test_activate_user(admin_request, sample_user):
sample_user.state = "pending"
sample_user.state = UserState.PENDING
resp = admin_request.post("user.activate_user", user_id=sample_user.id)
assert resp["data"]["id"] == str(sample_user.id)
assert resp["data"]["state"] == "active"
assert sample_user.state == "active"
assert sample_user.state == UserState.ACTIVE
def test_deactivate_user(admin_request, sample_user):
sample_user.state = "active"
sample_user.state = UserState.ACTIVE
resp = admin_request.post("user.deactivate_user", user_id=sample_user.id)
assert resp["data"]["id"] == str(sample_user.id)
assert resp["data"]["state"] == "pending"
assert sample_user.state == "pending"
assert sample_user.state == UserState.PENDING
def test_activate_user_fails_if_already_active(admin_request, sample_user):
@@ -790,7 +790,7 @@ def test_activate_user_fails_if_already_active(admin_request, sample_user):
"user.activate_user", user_id=sample_user.id, _expected_status=400
)
assert resp["message"] == "User already active"
assert sample_user.state == "active"
assert sample_user.state == UserState.ACTIVE
def test_update_user_auth_type(admin_request, sample_user):