diff --git a/shopdb/core/api/users.py b/shopdb/core/api/users.py index 84f9dac..e8eab95 100644 --- a/shopdb/core/api/users.py +++ b/shopdb/core/api/users.py @@ -1,389 +1,399 @@ -"""User management API routes.""" - -from flask import Blueprint, request -from flask_jwt_extended import jwt_required, current_user -from werkzeug.security import generate_password_hash - -from shopdb.extensions import db -from shopdb.core.models import ( - User, Role, Permission, AuditLog, full_permission_catalog) -from shopdb.utils.responses import success_response, error_response, ErrorCodes -from shopdb.utils.authz import require_role - -users_bp = Blueprint('users', __name__) - - -@users_bp.route('', methods=['GET']) -@jwt_required() -@require_role('admin') -def list_users(): - """List all users.""" - users = User.query.order_by(User.username).all() - return success_response([user_to_dict(u) for u in users]) - - -@users_bp.route('/', methods=['GET']) -@jwt_required() -def get_user(userid: int): - """Get a single user.""" - # inline: decorators cannot express admin-or-self - if not current_user.hasrole('admin') and current_user.userid != userid: - return error_response(ErrorCodes.FORBIDDEN, 'Access denied', http_code=403) - - user = db.session.get(User, userid) - if not user: - return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404) - - return success_response(user_to_dict(user)) - - -@users_bp.route('', methods=['POST']) -@jwt_required() -@require_role('admin') -def create_user(): - """Create a new user.""" - data = request.get_json() - if not data: - return error_response(ErrorCodes.VALIDATION_ERROR, 'Request body required') - - # Validate required fields - if not data.get('username'): - return error_response(ErrorCodes.VALIDATION_ERROR, 'Username is required') - if not data.get('email'): - return error_response(ErrorCodes.VALIDATION_ERROR, 'Email is required') - if not data.get('password'): - return error_response(ErrorCodes.VALIDATION_ERROR, 'Password is required') - - # Check uniqueness - if User.query.filter_by(username=data['username']).first(): - return error_response(ErrorCodes.CONFLICT, 'Username already exists', http_code=409) - if User.query.filter_by(email=data['email']).first(): - return error_response(ErrorCodes.CONFLICT, 'Email already exists', http_code=409) - - # Admin-created accounts are forced to change the password on first login - # unless the admin explicitly opts out. - mustchange = data.get('mustchangepassword', True) - - user = User( - username=data['username'], - email=data['email'], - passwordhash=generate_password_hash(data['password']), - firstname=data.get('firstname'), - lastname=data.get('lastname'), - isactive=data.get('isactive', True), - mustchangepassword=bool(mustchange) - ) - - # Assign roles - role_ids = data.get('roles', []) - if role_ids: - roles = Role.query.filter(Role.roleid.in_(role_ids)).all() - user.roles = roles - - db.session.add(user) - - # Audit log - AuditLog.log('created', 'User', entityname=user.username) - - db.session.commit() - - # Best-effort welcome email. The account exists regardless of mail outcome; - # a failure is surfaced as a warning in the response, never a hard error. - warning = None - if data.get('sendwelcome', True) and user.email: - sent = _send_welcome_email(user, data['password']) - if not sent: - warning = 'User created but the welcome email could not be sent.' - - payload = user_to_dict(user) - if warning: - payload['warning'] = warning - return success_response(payload, message='User created', http_code=201) - - -def _send_welcome_email(user, temp_password): - """Send a new-user welcome email with sign-in details. Returns True on send. - - Best-effort: any failure (including email being disabled) returns False so - the caller can surface a soft warning without failing user creation. - """ - from shopdb.core.api.settings import get_cached_settings - from shopdb.utils.mailer import render_email, send_email - - settings = get_cached_settings() or {} - facility = settings.get('facility_name') or 'ShopDB' - base_url = (settings.get('site_base_url') or '').rstrip('/') - login_link = f'{base_url}/login' if base_url else 'the ShopDB sign-in page' - - body = ( - f'

An account has been created for you at {facility}.

' - '' - f'' - f'' - f'' - f'' - '
Username{user.username}
Temporary password{temp_password}
' - f'

Sign in at {login_link}. You will be asked to set a new password ' - 'the first time you log in.

' - ) - html, text = render_email(f'Welcome to {facility}', body) - return send_email(user.email, f'Your {facility} account', html, text=text) - - -@users_bp.route('/', methods=['PUT']) -@jwt_required() -def update_user(userid: int): - """Update a user.""" - # inline: decorators cannot express admin-or-self - if not current_user.hasrole('admin') and current_user.userid != userid: - return error_response(ErrorCodes.FORBIDDEN, 'Access denied', http_code=403) - - user = db.session.get(User, userid) - if not user: - return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404) - - data = request.get_json() - if not data: - return error_response(ErrorCodes.VALIDATION_ERROR, 'Request body required') - - changes = {} - - # Update fields - if 'email' in data and data['email'] != user.email: - if User.query.filter(User.email == data['email'], User.userid != userid).first(): - return error_response(ErrorCodes.CONFLICT, 'Email already in use', http_code=409) - changes['email'] = {'old': user.email, 'new': data['email']} - user.email = data['email'] - - if 'firstname' in data: - if data['firstname'] != user.firstname: - changes['firstname'] = {'old': user.firstname, 'new': data['firstname']} - user.firstname = data['firstname'] - - if 'lastname' in data: - if data['lastname'] != user.lastname: - changes['lastname'] = {'old': user.lastname, 'new': data['lastname']} - user.lastname = data['lastname'] - - # Admin-only fields (inline: gates a subset of fields on a shared route) - if current_user.hasrole('admin'): - if 'isactive' in data: - if data['isactive'] != user.isactive: - changes['isactive'] = {'old': user.isactive, 'new': data['isactive']} - user.isactive = data['isactive'] - - if 'roles' in data: - old_roles = [r.rolename for r in user.roles] - roles = Role.query.filter(Role.roleid.in_(data['roles'])).all() - new_roles = [r.rolename for r in roles] - if set(old_roles) != set(new_roles): - changes['roles'] = {'old': old_roles, 'new': new_roles} - user.roles = roles - - # Unlock user - if data.get('unlock'): - user.lockeduntil = None - user.failedlogins = 0 - changes['unlocked'] = {'old': True, 'new': False} - - # Password change - if 'password' in data and data['password']: - user.passwordhash = generate_password_hash(data['password']) - changes['password'] = {'old': '***', 'new': '***'} - - if changes: - AuditLog.log('updated', 'User', entityid=user.userid, entityname=user.username, changes=changes) - - db.session.commit() - - return success_response(user_to_dict(user), message='User updated') - - -@users_bp.route('/', methods=['DELETE']) -@jwt_required() -@require_role('admin') -def delete_user(userid: int): - """Delete a user.""" - if current_user.userid == userid: - return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot delete your own account') - - user = db.session.get(User, userid) - if not user: - return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404) - - username = user.username - db.session.delete(user) - - AuditLog.log('deleted', 'User', entityid=userid, entityname=username) - - db.session.commit() - - return success_response(None, message='User deleted') - - -# Permissions endpoints -@users_bp.route('/permissions', methods=['GET']) -@jwt_required() -def list_permissions(): - """List assignable permissions grouped by category. - - Driven by full_permission_catalog() (core plus ENABLED plugins) so a - disabled plugin's permissions drop out of the role grid. Roles assign by - name; the permissionid comes from the seeded Permission row when present. - """ - idbyname = {p.name: p.permissionid for p in Permission.query.all()} - - catalog = full_permission_catalog() - catalog.sort(key=lambda e: (e[2], e[0])) - - grouped = {} - flat = [] - for name, description, category in catalog: - entry = { - 'permissionid': idbyname.get(name), - 'name': name, - 'description': description, - } - grouped.setdefault(category, []).append(entry) - flat.append({**entry, 'category': category}) - - return success_response({'permissions': flat, 'grouped': grouped}) - - -# Roles endpoints -@users_bp.route('/roles', methods=['GET']) -@jwt_required() -def list_roles(): - """List all roles with their permissions.""" - roles = Role.query.order_by(Role.rolename).all() - return success_response([{ - 'roleid': r.roleid, - 'rolename': r.rolename, - 'description': r.description, - 'usercount': r.users.count(), - 'permissions': [p.name for p in r.permissions], - 'isadmin': r.rolename == 'admin' - } for r in roles]) - - -@users_bp.route('/roles', methods=['POST']) -@jwt_required() -@require_role('admin') -def create_role(): - """Create a new role.""" - data = request.get_json() - if not data or not data.get('rolename'): - return error_response(ErrorCodes.VALIDATION_ERROR, 'Role name is required') - - if Role.query.filter_by(rolename=data['rolename']).first(): - return error_response(ErrorCodes.CONFLICT, 'Role already exists', http_code=409) - - role = Role( - rolename=data['rolename'], - description=data.get('description') - ) - - # Assign permissions - if 'permissions' in data: - perms = Permission.query.filter(Permission.name.in_(data['permissions'])).all() - role.permissions = perms - - db.session.add(role) - - AuditLog.log('created', 'Role', entityname=role.rolename) - - db.session.commit() - - return success_response({ - 'roleid': role.roleid, - 'rolename': role.rolename, - 'description': role.description, - 'permissions': [p.name for p in role.permissions] - }, message='Role created', http_code=201) - - -@users_bp.route('/roles/', methods=['PUT']) -@jwt_required() -@require_role('admin') -def update_role(roleid: int): - """Update a role.""" - role = db.session.get(Role, roleid) - if not role: - return error_response(ErrorCodes.NOT_FOUND, 'Role not found', http_code=404) - - # Cannot modify admin role permissions - if role.rolename == 'admin' and 'permissions' in request.get_json(): - return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot modify admin role permissions') - - data = request.get_json() - changes = {} - - if 'description' in data: - if data['description'] != role.description: - changes['description'] = {'old': role.description, 'new': data['description']} - role.description = data['description'] - - # Update permissions - if 'permissions' in data and role.rolename != 'admin': - old_perms = [p.name for p in role.permissions] - perms = Permission.query.filter(Permission.name.in_(data['permissions'])).all() - new_perms = [p.name for p in perms] - if set(old_perms) != set(new_perms): - changes['permissions'] = {'old': old_perms, 'new': new_perms} - role.permissions = perms - - if changes: - AuditLog.log('updated', 'Role', entityid=role.roleid, entityname=role.rolename, changes=changes) - - db.session.commit() - - return success_response({ - 'roleid': role.roleid, - 'rolename': role.rolename, - 'description': role.description, - 'permissions': [p.name for p in role.permissions] - }, message='Role updated') - - -@users_bp.route('/roles/', methods=['DELETE']) -@jwt_required() -@require_role('admin') -def delete_role(roleid: int): - """Delete a role.""" - role = db.session.get(Role, roleid) - if not role: - return error_response(ErrorCodes.NOT_FOUND, 'Role not found', http_code=404) - - if role.rolename == 'admin': - return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot delete the admin role') - - if role.users.count() > 0: - return error_response(ErrorCodes.VALIDATION_ERROR, f'Role is assigned to {role.users.count()} user(s)') - - rolename = role.rolename - db.session.delete(role) - - AuditLog.log('deleted', 'Role', entityid=roleid, entityname=rolename) - - db.session.commit() - - return success_response(None, message='Role deleted') - - -def user_to_dict(user: User) -> dict: - """Convert user to dict for API response.""" - return { - 'userid': user.userid, - 'username': user.username, - 'email': user.email, - 'firstname': user.firstname, - 'lastname': user.lastname, - 'isactive': user.isactive, - 'islocked': user.islocked, - 'mustchangepassword': bool(user.mustchangepassword), - 'lastlogindate': user.lastlogindate.isoformat() + 'Z' if user.lastlogindate else None, - 'failedlogins': user.failedlogins, - 'roles': [{'roleid': r.roleid, 'rolename': r.rolename} for r in user.roles], - 'createddate': user.createddate.isoformat() + 'Z' if user.createddate else None, - 'modifieddate': user.modifieddate.isoformat() + 'Z' if user.modifieddate else None - } +"""User management API routes.""" + +from flask import Blueprint, request +from flask_jwt_extended import jwt_required, current_user +from werkzeug.security import generate_password_hash + +from shopdb.extensions import db +from shopdb.core.models import ( + User, Role, Permission, AuditLog, full_permission_catalog) +from shopdb.utils.responses import success_response, error_response, ErrorCodes +from shopdb.utils.authz import require_role + +users_bp = Blueprint('users', __name__) + + +@users_bp.route('', methods=['GET']) +@jwt_required() +@require_role('admin') +def list_users(): + """List all users.""" + users = User.query.order_by(User.username).all() + return success_response([user_to_dict(u) for u in users]) + + +@users_bp.route('/', methods=['GET']) +@jwt_required() +def get_user(userid: int): + """Get a single user.""" + # inline: decorators cannot express admin-or-self + if not current_user.hasrole('admin') and current_user.userid != userid: + return error_response(ErrorCodes.FORBIDDEN, 'Access denied', http_code=403) + + user = db.session.get(User, userid) + if not user: + return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404) + + return success_response(user_to_dict(user)) + + +@users_bp.route('', methods=['POST']) +@jwt_required() +@require_role('admin') +def create_user(): + """Create a new user.""" + data = request.get_json() + if not data: + return error_response(ErrorCodes.VALIDATION_ERROR, 'Request body required') + + # Validate required fields + if not data.get('username'): + return error_response(ErrorCodes.VALIDATION_ERROR, 'Username is required') + if not data.get('email'): + return error_response(ErrorCodes.VALIDATION_ERROR, 'Email is required') + if not data.get('password'): + return error_response(ErrorCodes.VALIDATION_ERROR, 'Password is required') + + # Check uniqueness + if User.query.filter_by(username=data['username']).first(): + return error_response(ErrorCodes.CONFLICT, 'Username already exists', http_code=409) + if User.query.filter_by(email=data['email']).first(): + return error_response(ErrorCodes.CONFLICT, 'Email already exists', http_code=409) + + # Admin-created accounts are forced to change the password on first login + # unless the admin explicitly opts out. + mustchange = data.get('mustchangepassword', True) + + user = User( + username=data['username'], + email=data['email'], + passwordhash=generate_password_hash(data['password']), + firstname=data.get('firstname'), + lastname=data.get('lastname'), + isactive=data.get('isactive', True), + mustchangepassword=bool(mustchange) + ) + + # Assign roles + role_ids = data.get('roles', []) + if role_ids: + roles = Role.query.filter(Role.roleid.in_(role_ids)).all() + user.roles = roles + + db.session.add(user) + + # Audit log + AuditLog.log('created', 'User', entityname=user.username) + + db.session.commit() + + # Best-effort welcome email. The account exists regardless of mail outcome; + # a failure is surfaced as a warning in the response, never a hard error. + warning = None + if data.get('sendwelcome', True) and user.email: + sent = _send_welcome_email(user, data['password']) + if not sent: + warning = 'User created but the welcome email could not be sent.' + + payload = user_to_dict(user) + if warning: + payload['warning'] = warning + return success_response(payload, message='User created', http_code=201) + + +def _send_welcome_email(user, temp_password): + """Send a new-user welcome email with sign-in details. Returns True on send. + + Best-effort: any failure (including email being disabled) returns False so + the caller can surface a soft warning without failing user creation. + """ + from shopdb.core.api.settings import get_cached_settings + from shopdb.utils.mailer import render_email, send_email + + settings = get_cached_settings() or {} + facility = settings.get('facility_name') or 'ShopDB' + base_url = (settings.get('site_base_url') or '').rstrip('/') + login_link = f'{base_url}/login' if base_url else 'the ShopDB sign-in page' + + body = ( + f'

An account has been created for you at {facility}.

' + '' + f'' + f'' + f'' + f'' + '
Username{user.username}
Temporary password{temp_password}
' + f'

Sign in at {login_link}. You will be asked to set a new password ' + 'the first time you log in.

' + ) + html, text = render_email(f'Welcome to {facility}', body) + return send_email(user.email, f'Your {facility} account', html, text=text) + + +@users_bp.route('/', methods=['PUT']) +@jwt_required() +def update_user(userid: int): + """Update a user.""" + # inline: decorators cannot express admin-or-self + if not current_user.hasrole('admin') and current_user.userid != userid: + return error_response(ErrorCodes.FORBIDDEN, 'Access denied', http_code=403) + + user = db.session.get(User, userid) + if not user: + return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404) + + data = request.get_json() + if not data: + return error_response(ErrorCodes.VALIDATION_ERROR, 'Request body required') + + changes = {} + + # Update fields + if 'email' in data and data['email'] != user.email: + if User.query.filter(User.email == data['email'], User.userid != userid).first(): + return error_response(ErrorCodes.CONFLICT, 'Email already in use', http_code=409) + changes['email'] = {'old': user.email, 'new': data['email']} + user.email = data['email'] + + if 'firstname' in data: + if data['firstname'] != user.firstname: + changes['firstname'] = {'old': user.firstname, 'new': data['firstname']} + user.firstname = data['firstname'] + + if 'lastname' in data: + if data['lastname'] != user.lastname: + changes['lastname'] = {'old': user.lastname, 'new': data['lastname']} + user.lastname = data['lastname'] + + # Admin-only fields (inline: gates a subset of fields on a shared route) + if current_user.hasrole('admin'): + if 'isactive' in data: + if data['isactive'] != user.isactive: + changes['isactive'] = {'old': user.isactive, 'new': data['isactive']} + user.isactive = data['isactive'] + + if 'roles' in data: + old_roles = [r.rolename for r in user.roles] + roles = Role.query.filter(Role.roleid.in_(data['roles'])).all() + new_roles = [r.rolename for r in roles] + if set(old_roles) != set(new_roles): + changes['roles'] = {'old': old_roles, 'new': new_roles} + user.roles = roles + + # Unlock user + if data.get('unlock'): + user.lockeduntil = None + user.failedlogins = 0 + changes['unlocked'] = {'old': True, 'new': False} + + # Password change + if 'password' in data and data['password']: + user.passwordhash = generate_password_hash(data['password']) + changes['password'] = {'old': '***', 'new': '***'} + + if changes: + AuditLog.log('updated', 'User', entityid=user.userid, entityname=user.username, changes=changes) + + db.session.commit() + + return success_response(user_to_dict(user), message='User updated') + + +@users_bp.route('/', methods=['DELETE']) +@jwt_required() +@require_role('admin') +def delete_user(userid: int): + """Delete a user.""" + if current_user.userid == userid: + return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot delete your own account') + + user = db.session.get(User, userid) + if not user: + return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404) + + username = user.username + + # Rows that reference the user would otherwise block the delete: + # revoke their API tokens outright, and DETACH their audit history + # (userid -> NULL) - the log rows themselves are kept, entityname and + # details still tell the story. + from shopdb.core.models import ApiToken + ApiToken.query.filter_by(userid=userid).delete(synchronize_session=False) + AuditLog.query.filter_by(userid=userid).update( + {'userid': None}, synchronize_session=False) + + db.session.delete(user) + + AuditLog.log('deleted', 'User', entityid=userid, entityname=username) + + db.session.commit() + + return success_response(None, message='User deleted') + + +# Permissions endpoints +@users_bp.route('/permissions', methods=['GET']) +@jwt_required() +def list_permissions(): + """List assignable permissions grouped by category. + + Driven by full_permission_catalog() (core plus ENABLED plugins) so a + disabled plugin's permissions drop out of the role grid. Roles assign by + name; the permissionid comes from the seeded Permission row when present. + """ + idbyname = {p.name: p.permissionid for p in Permission.query.all()} + + catalog = full_permission_catalog() + catalog.sort(key=lambda e: (e[2], e[0])) + + grouped = {} + flat = [] + for name, description, category in catalog: + entry = { + 'permissionid': idbyname.get(name), + 'name': name, + 'description': description, + } + grouped.setdefault(category, []).append(entry) + flat.append({**entry, 'category': category}) + + return success_response({'permissions': flat, 'grouped': grouped}) + + +# Roles endpoints +@users_bp.route('/roles', methods=['GET']) +@jwt_required() +def list_roles(): + """List all roles with their permissions.""" + roles = Role.query.order_by(Role.rolename).all() + return success_response([{ + 'roleid': r.roleid, + 'rolename': r.rolename, + 'description': r.description, + 'usercount': r.users.count(), + 'permissions': [p.name for p in r.permissions], + 'isadmin': r.rolename == 'admin' + } for r in roles]) + + +@users_bp.route('/roles', methods=['POST']) +@jwt_required() +@require_role('admin') +def create_role(): + """Create a new role.""" + data = request.get_json() + if not data or not data.get('rolename'): + return error_response(ErrorCodes.VALIDATION_ERROR, 'Role name is required') + + if Role.query.filter_by(rolename=data['rolename']).first(): + return error_response(ErrorCodes.CONFLICT, 'Role already exists', http_code=409) + + role = Role( + rolename=data['rolename'], + description=data.get('description') + ) + + # Assign permissions + if 'permissions' in data: + perms = Permission.query.filter(Permission.name.in_(data['permissions'])).all() + role.permissions = perms + + db.session.add(role) + + AuditLog.log('created', 'Role', entityname=role.rolename) + + db.session.commit() + + return success_response({ + 'roleid': role.roleid, + 'rolename': role.rolename, + 'description': role.description, + 'permissions': [p.name for p in role.permissions] + }, message='Role created', http_code=201) + + +@users_bp.route('/roles/', methods=['PUT']) +@jwt_required() +@require_role('admin') +def update_role(roleid: int): + """Update a role.""" + role = db.session.get(Role, roleid) + if not role: + return error_response(ErrorCodes.NOT_FOUND, 'Role not found', http_code=404) + + # Cannot modify admin role permissions + if role.rolename == 'admin' and 'permissions' in request.get_json(): + return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot modify admin role permissions') + + data = request.get_json() + changes = {} + + if 'description' in data: + if data['description'] != role.description: + changes['description'] = {'old': role.description, 'new': data['description']} + role.description = data['description'] + + # Update permissions + if 'permissions' in data and role.rolename != 'admin': + old_perms = [p.name for p in role.permissions] + perms = Permission.query.filter(Permission.name.in_(data['permissions'])).all() + new_perms = [p.name for p in perms] + if set(old_perms) != set(new_perms): + changes['permissions'] = {'old': old_perms, 'new': new_perms} + role.permissions = perms + + if changes: + AuditLog.log('updated', 'Role', entityid=role.roleid, entityname=role.rolename, changes=changes) + + db.session.commit() + + return success_response({ + 'roleid': role.roleid, + 'rolename': role.rolename, + 'description': role.description, + 'permissions': [p.name for p in role.permissions] + }, message='Role updated') + + +@users_bp.route('/roles/', methods=['DELETE']) +@jwt_required() +@require_role('admin') +def delete_role(roleid: int): + """Delete a role.""" + role = db.session.get(Role, roleid) + if not role: + return error_response(ErrorCodes.NOT_FOUND, 'Role not found', http_code=404) + + if role.rolename == 'admin': + return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot delete the admin role') + + if role.users.count() > 0: + return error_response(ErrorCodes.VALIDATION_ERROR, f'Role is assigned to {role.users.count()} user(s)') + + rolename = role.rolename + db.session.delete(role) + + AuditLog.log('deleted', 'Role', entityid=roleid, entityname=rolename) + + db.session.commit() + + return success_response(None, message='Role deleted') + + +def user_to_dict(user: User) -> dict: + """Convert user to dict for API response.""" + return { + 'userid': user.userid, + 'username': user.username, + 'email': user.email, + 'firstname': user.firstname, + 'lastname': user.lastname, + 'isactive': user.isactive, + 'islocked': user.islocked, + 'mustchangepassword': bool(user.mustchangepassword), + 'lastlogindate': user.lastlogindate.isoformat() + 'Z' if user.lastlogindate else None, + 'failedlogins': user.failedlogins, + 'roles': [{'roleid': r.roleid, 'rolename': r.rolename} for r in user.roles], + 'createddate': user.createddate.isoformat() + 'Z' if user.createddate else None, + 'modifieddate': user.modifieddate.isoformat() + 'Z' if user.modifieddate else None + } diff --git a/tests/test_core/test_users_api.py b/tests/test_core/test_users_api.py new file mode 100644 index 0000000..8158ee1 --- /dev/null +++ b/tests/test_core/test_users_api.py @@ -0,0 +1,29 @@ + + +def test_delete_user_with_tokens_and_audit_history(client, auth_headers, app, db): + """Deleting a user revokes their API tokens and detaches (not deletes) + their audit rows - the importer-user case.""" + from werkzeug.security import generate_password_hash + from shopdb.core.models import User, ApiToken, AuditLog + + with app.app_context(): + user = User(username='importer2', email='importer2@test.local', + passwordhash=generate_password_hash('x'), isactive=True) + db.session.add(user) + db.session.flush() + db.session.add(ApiToken(userid=user.userid, name='import token', + tokenprefix='deadbeef', tokenhash='x' * 64)) + db.session.add(AuditLog(userid=user.userid, action='created', + entitytype='Asset', entityid=1, + entityname='imported thing')) + db.session.commit() + userid = user.userid + + response = client.delete(f'/api/users/{userid}', headers=auth_headers) + assert response.status_code == 200, response.get_json() + + with app.app_context(): + assert db.session.get(User, userid) is None + assert ApiToken.query.filter_by(userid=userid).count() == 0 + detached = AuditLog.query.filter_by(entityname='imported thing').one() + assert detached.userid is None