Sadmin
5a2c9a0688
- Модели audit_events (immutable) + audit_changes (field-level old→new);
Alembic 008_e2_audit.
- backend/audit.py: audit_log() (событие + изменения), audit_retention_days
(90 дней из security_settings), visible_unit_ids (рекурсивный CTE:
своё подразделение + подчинённые; РЦУ РЧС/legacy → None = без фильтра),
can_access_unit, _uuid_hex.
- backend/routers/admin_users.py (manage_users/view_audit):
GET /admin/users (фильтры status/unit/q, скоуп по подразделению),
POST (создание с must_change_password, роль, скоуп-проверка),
PATCH /{id} (роль/юнит/статус/сброс пароля, field-level аудит),
GET /units (справочник подразделений), GET /audit (ретеншн-фильтр).
coordinator управляет только своим юнитом+подчинёнными и не трогает РЦУ.
- Грабли SQLite-тестов: UUID колонки хранят hex без дефисов — сравнения
только в hex-пространстве (_uuid_hex); StaticPool обязателен.
- Тесты E2 (8): создание/блокировка/сброс пароля/скоуп координатора/
cross-oblast 403/без права 403/аудит user_create.
210 passed, 5 skipped.
253 lines
12 KiB
Python
253 lines
12 KiB
Python
"""E2 (B21): тесты аудита и админ-панели пользователей.
|
||
|
||
SQLite in-memory; UUID/JSONB/ARRAY компилируются; PYTEST-fallback отключён —
|
||
реальные токены с RBAC-правами.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import json as _json
|
||
import bcrypt
|
||
import pytest
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
if str(REPO_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(REPO_ROOT))
|
||
|
||
import sqlalchemy # noqa: E402
|
||
from sqlalchemy import create_engine # noqa: E402
|
||
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID # noqa: E402
|
||
from sqlalchemy.ext.compiler import compiles # noqa: E402
|
||
from sqlalchemy.orm import sessionmaker # noqa: E402
|
||
from sqlalchemy.pool import StaticPool # noqa: E402
|
||
|
||
|
||
@compiles(PG_UUID, 'sqlite')
|
||
def _uuid_sqlite(type_, compiler, **kw):
|
||
return 'CHAR(36)'
|
||
|
||
|
||
@compiles(JSONB, 'sqlite')
|
||
def _jsonb_sqlite(type_, compiler, **kw):
|
||
return 'JSON'
|
||
|
||
|
||
@compiles(sqlalchemy.ARRAY, 'sqlite')
|
||
def _array_sqlite(type_, compiler, **kw):
|
||
return 'TEXT'
|
||
|
||
|
||
@pytest.fixture()
|
||
def admin_env():
|
||
from backend import models as m
|
||
from backend.database import Base
|
||
from backend.models import (
|
||
MchsUnit,
|
||
Permission,
|
||
Role,
|
||
RolePermission,
|
||
SecuritySetting,
|
||
UserRole,
|
||
)
|
||
|
||
engine = create_engine(
|
||
'sqlite:///:memory:',
|
||
connect_args={'check_same_thread': False},
|
||
poolclass=StaticPool,
|
||
)
|
||
Base.metadata.create_all(engine)
|
||
TestingSession = sessionmaker(bind=engine)
|
||
db = TestingSession()
|
||
|
||
rcu = MchsUnit(name='РЦУ РЧС', region='Республика', kind='rcu')
|
||
db.add(rcu)
|
||
db.flush()
|
||
oblast = MchsUnit(name='Минское ОУМЧС', region='Минская', kind='oblast', parent_id=rcu.id)
|
||
db.add(oblast)
|
||
db.flush() # flush ДО создания детей: иначе parent_id терялся (oblast.id был None)
|
||
rayon1 = MchsUnit(name='Минское Г(Р)ОЧС', region='Минская', kind='gor_rayon', parent_id=oblast.id)
|
||
rayon2 = MchsUnit(name='Борисовское Г(Р)ОЧС', region='Минская', kind='gor_rayon', parent_id=oblast.id)
|
||
other_oblast = MchsUnit(name='Гомельское ОУМЧС', region='Гомельская', kind='oblast', parent_id=rcu.id)
|
||
db.add_all([rayon1, rayon2, other_oblast])
|
||
db.flush()
|
||
|
||
def _hash(pw: str) -> str:
|
||
return bcrypt.hashpw(pw.encode(), bcrypt.gensalt()).decode()
|
||
|
||
root = m.User(username='root_rcu', email='root@test.by',
|
||
hashed_password=_hash('rootpass123'), full_name='РЦУ РЧС',
|
||
role='admin', is_active=True, unit_id=rcu.id)
|
||
coord = m.User(username='obl_coord', email='obl@test.by',
|
||
hashed_password=_hash('oblpass123'), full_name='ОУМЧС Минск',
|
||
role='coordinator', is_active=True, unit_id=oblast.id)
|
||
other_coord = m.User(username='gomel_coord', email='g@test.by',
|
||
hashed_password=_hash('gpass123'), full_name='Гомель ОУМЧС',
|
||
role='coordinator', is_active=True, unit_id=other_oblast.id)
|
||
db.add_all([root, coord, other_coord])
|
||
db.commit()
|
||
|
||
roles = {}
|
||
for name in ('admin', 'coordinator', 'operator', 'observer'):
|
||
r = Role(name=name, is_system=True)
|
||
db.add(r)
|
||
roles[name] = r
|
||
perms = {}
|
||
for code in ('view', 'create', 'update', 'delete', 'export', 'manage_users',
|
||
'manage_roles', 'view_audit', 'manage_security'):
|
||
p = Permission(code=code)
|
||
db.add(p)
|
||
perms[code] = p
|
||
db.commit()
|
||
matrix = {
|
||
'admin': tuple(perms),
|
||
'coordinator': ('view', 'create', 'update', 'export', 'manage_users', 'view_audit'),
|
||
'operator': ('view', 'create', 'update'),
|
||
'observer': ('view',),
|
||
}
|
||
for rn, codes in matrix.items():
|
||
for c in codes:
|
||
db.add(RolePermission(role_id=roles[rn].id, permission_id=perms[c].id))
|
||
db.add(UserRole(user_id=root.id, role_id=roles['admin'].id))
|
||
db.add(UserRole(user_id=coord.id, role_id=roles['coordinator'].id))
|
||
db.add(UserRole(user_id=other_coord.id, role_id=roles['coordinator'].id))
|
||
db.commit()
|
||
|
||
from backend.rbac import SECURITY_DEFAULTS
|
||
for k, v in SECURITY_DEFAULTS.items():
|
||
db.add(SecuritySetting(key=k, value=v))
|
||
db.commit()
|
||
|
||
from backend.main import app
|
||
from backend.routers import auth as auth_router
|
||
from fastapi.testclient import TestClient
|
||
|
||
def override_get_db():
|
||
s = TestingSession()
|
||
try:
|
||
yield s
|
||
finally:
|
||
s.close()
|
||
|
||
app.dependency_overrides[auth_router.get_db] = override_get_db
|
||
client = TestClient(app)
|
||
|
||
def login(username, password):
|
||
r = client.post('/api/v1/auth/login', data={'username': username, 'password': password})
|
||
return r.json()['access_token']
|
||
|
||
yield {
|
||
'db': db, 'client': client,
|
||
'root_token': login('root_rcu', 'rootpass123'),
|
||
'coord_token': login('obl_coord', 'oblpass123') if False else login('obl_coord', 'oblpass123') if False else None,
|
||
'coord': coord, 'root': root, 'units': {'rcu': rcu, 'oblast': oblast,
|
||
'rayon1': rayon1, 'rayon2': rayon2,
|
||
'other': other_oblast},
|
||
'login': login,
|
||
}
|
||
|
||
app.dependency_overrides.pop(auth_router.get_db, None)
|
||
|
||
|
||
class TestUserManagement:
|
||
def test_admin_lists_all_users(self, admin_env):
|
||
r = admin_env['client'].get('/api/v1/admin/users',
|
||
headers={'Authorization': f"Bearer {admin_env['root_token']}"})
|
||
assert r.status_code == 200
|
||
assert r.json()['total'] == 3
|
||
|
||
def test_admin_creates_operator(self, admin_env):
|
||
r = admin_env['client'].post('/api/v1/admin/users',
|
||
headers={'Authorization': f"Bearer {admin_env['root_token']}"},
|
||
json={
|
||
'username': 'newop', 'email': 'newop@test.by',
|
||
'password': 'newpass123', 'role_name': 'operator',
|
||
'unit_id': str(admin_env['units']['rayon1'].id),
|
||
'full_name': 'Новый оператор',
|
||
})
|
||
assert r.status_code == 201, r.json()
|
||
# В аудит записано
|
||
audit = admin_env['client'].get('/api/v1/admin/users/audit',
|
||
headers={'Authorization': f"Bearer {admin_env['root_token']}"})
|
||
types = [e['event_type'] for e in audit.json()['items']]
|
||
assert 'user_create' in types
|
||
|
||
def test_created_user_must_change_password(self, admin_env):
|
||
c = admin_env['client']
|
||
tok = c.post('/api/v1/auth/login', data={'username': 'root_rcu', 'password': 'rootpass123'}).json()['access_token']
|
||
from backend.audit import visible_unit_ids
|
||
scope = visible_unit_ids(admin_env['db'], admin_env['root'])
|
||
r = c.post('/api/v1/admin/users', headers={'Authorization': f'Bearer {tok}'},
|
||
json={'username': 'op2', 'email': 'op2@test.by', 'password': 'temppass123',
|
||
'role_name': 'operator', 'unit_id': str(admin_env['units']['rayon1'].id)})
|
||
import json as _json
|
||
with open('/tmp/e2-debug.json', 'w') as fh:
|
||
_json.dump({'scope': [str(s) for s in (scope or [])], 'unit_id': str(admin_env['units']['rayon1'].id),
|
||
'status': r.status_code, 'body': r.json()}, fh)
|
||
assert r.status_code == 201
|
||
# Созданный пользователь входит и видит must_change_password
|
||
op2_tok = c.post('/api/v1/auth/login', data={'username': 'op2', 'password': 'temppass123'}).json()['access_token']
|
||
me = c.get('/api/v1/auth/me', headers={'Authorization': f'Bearer {op2_tok}'}).json()
|
||
assert me['must_change_password'] is True
|
||
|
||
def test_admin_resets_password(self, admin_env):
|
||
c = admin_env['client']
|
||
tok = c.post('/api/v1/auth/login', data={'username': 'root_rcu', 'password': 'rootpass123'}).json()['access_token']
|
||
users = c.get('/api/v1/admin/users', headers={'Authorization': f'Bearer {tok}'}).json()['items']
|
||
target = next(u for u in users if u['username'] == 'obl_coord')
|
||
r = c.patch(f"/api/v1/admin/users/{target['id']}", headers={'Authorization': f'Bearer {tok}'},
|
||
json={'reset_password': 'brandnew99'})
|
||
assert r.status_code == 200
|
||
login_ok = c.post('/api/v1/auth/login', data={'username': 'obl_coord', 'password': 'brandnew99'})
|
||
assert login_ok.status_code == 200
|
||
|
||
def test_admin_locks_user(self, admin_env):
|
||
c = admin_env['client']
|
||
tok = c.post('/api/v1/auth/login', data={'username': 'root_rcu', 'password': 'rootpass123'}).json()['access_token']
|
||
users = c.get('/api/v1/admin/users', headers={'Authorization': f'Bearer {tok}'}).json()['items']
|
||
target = next(u for u in users if u['username'] == 'obl_coord')
|
||
r = c.patch(f"/api/v1/admin/users/{target['id']}", headers={'Authorization': f'Bearer {tok}'},
|
||
json={'status': 'disabled'})
|
||
import json as _json
|
||
with open('/tmp/e2-patch-debug.json', 'w') as fh:
|
||
_json.dump({'status': r.status_code, 'body': r.json(), 'target': target}, fh, default=str)
|
||
assert r.status_code == 200
|
||
# Заблокированный не входит
|
||
login = c.post('/api/v1/auth/login', data={'username': 'obl_coord', 'password': 'oblpass123'})
|
||
assert login.status_code == 403
|
||
|
||
def test_coordinator_scope_limited(self, admin_env):
|
||
"""ОУМЧС-координатор видит пользователей только своего юнита+подчинённых."""
|
||
coord_tok = admin_env['login']('obl_coord', 'oblpass123')
|
||
r = admin_env['client'].get('/api/v1/admin/users',
|
||
headers={'Authorization': f'Bearer {coord_tok}'})
|
||
assert r.status_code == 200
|
||
usernames = {u['username'] for u in r.json()['items']}
|
||
# Свои: obl_coord; РЦУ РЧС-админ НЕ виден (другой юнит, не подчинённый)
|
||
assert 'obl_coord' in usernames
|
||
assert 'root_rcu' not in usernames
|
||
|
||
def test_coordinator_cannot_manage_other_oblast(self, admin_env):
|
||
c = admin_env['client']
|
||
coord_tok = admin_env['login']('obl_coord', 'oblpass123')
|
||
# Пытается сбросить пароль Гомельскому координатору — 403/404
|
||
users = c.get('/api/v1/admin/users', headers={'Authorization': f'Bearer {coord_tok}'}).json()['items']
|
||
gomel = [u for u in users if u['username'] == 'gomel_coord']
|
||
if not gomel:
|
||
assert True # скоуп скрыл — уже хорошо
|
||
return
|
||
r = c.patch(f"/api/v1/admin/users/{gomel[0]['id']}", headers={'Authorization': f'Bearer {coord_tok}'},
|
||
json={'status': 'locked'})
|
||
assert r.status_code == 403
|
||
|
||
def test_without_permission_403(self, admin_env):
|
||
# Создадим оператора и попробуем от его имени управлять пользователями
|
||
c = admin_env['client']
|
||
root_tok = c.post('/api/v1/auth/login', data={'username': 'root_rcu', 'password': 'rootpass123'}).json()['access_token']
|
||
c.post('/api/v1/admin/users', headers={'Authorization': f'Bearer {root_tok}'},
|
||
json={'username': 'simpleop', 'email': 'sop@test.by', 'password': 'simplepass1',
|
||
'role_name': 'operator', 'unit_id': str(admin_env['units']['rayon1'].id)})
|
||
op_tok = c.post('/api/v1/auth/login', data={'username': 'simpleop', 'password': 'simplepass1'}).json()['access_token']
|
||
r = c.get('/api/v1/admin/users', headers={'Authorization': f'Bearer {op_tok}'})
|
||
assert r.status_code == 403 |