Sadmin
808ef93a52
- Модель search_operations: title, case_id (unique — операция на карточку),
status (planned/active/paused/completed/archived, CheckConstraint),
unit_id, contour_operation_id (заготовка B20-E4), created_by, closed_at.
- Alembic 009_e3_operations.
- backend/routers/operations.py: POST (создание; unit по умолчанию =
подразделение создателя; скоуп-проверка), GET список (скоуп-фильтр),
GET /{id} (403 вне скоупа), GET /summary (active/total/completed_24h),
PATCH (title/unit/status со статусной машиной: planned→active⇄paused→
completed→archived; недопустимые переходы 400; completed ставит closed_at).
Всё с аудитом operation_create/operation_update.
- Скоуп: unit_id ∈ visible_unit_ids (своё+подчинённые); РЦУ РЧС — все.
- Тесты E3 (12): создание с юнитом/дефолт/дубль кейса/cross-unit 403,
скоуп списков, статусная машина, summary, аудит, cross-scope 403.
222 passed, 5 skipped.
252 lines
11 KiB
Python
252 lines
11 KiB
Python
"""E3 (B21): тесты операций мультипоиска — статусная машина, скоуп, дашборд."""
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
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 ops_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)
|
||
db = sessionmaker(bind=engine)()
|
||
|
||
rcu = MchsUnit(name='РЦУ РЧС', kind='rcu')
|
||
db.add(rcu)
|
||
db.flush()
|
||
oblast = MchsUnit(name='Минское ОУМЧС', kind='oblast', parent_id=rcu.id)
|
||
db.add(oblast)
|
||
db.flush()
|
||
rayon = MchsUnit(name='Минское Г(Р)ОЧС', kind='gor_rayon', parent_id=oblast.id)
|
||
db.add(rayon)
|
||
db.flush()
|
||
other_rayon = MchsUnit(name='Гомельское Г(Р)ОЧС', kind='gor_rayon', parent_id=rcu.id)
|
||
db.add(other_rayon)
|
||
db.flush()
|
||
|
||
def _hash(pw):
|
||
return bcrypt.hashpw(pw.encode(), bcrypt.gensalt()).decode()
|
||
|
||
root = m.User(username='root', email='r@t.by', hashed_password=_hash('rootpass123'),
|
||
role='admin', is_active=True, unit_id=rcu.id)
|
||
coord = m.User(username='coord', email='c@t.by', hashed_password=_hash('coordpass123'),
|
||
role='coordinator', is_active=True, unit_id=oblast.id)
|
||
other_op = m.User(username='other', email='o@t.by', hashed_password=_hash('otherpass123'),
|
||
role='operator', is_active=True, unit_id=other_rayon.id)
|
||
db.add_all([root, coord, other_op])
|
||
db.commit()
|
||
|
||
case1 = m.Case(age_years=10, gender='м', status='active')
|
||
case2 = m.Case(age_years=8, gender='ж', status='active')
|
||
case3 = m.Case(age_years=12, gender='м', status='active')
|
||
db.add_all([case1, case2, case3])
|
||
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_op.id, role_id=roles['operator'].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():
|
||
yield db
|
||
|
||
app.dependency_overrides[auth_router.get_db] = override
|
||
client = TestClient(app)
|
||
|
||
def login(u, p):
|
||
return client.post('/api/v1/auth/login', data={'username': u, 'password': p}).json()['access_token']
|
||
|
||
yield {'db': db, 'client': client,
|
||
'root': login('root', 'rootpass123'),
|
||
'coord': login('coord', 'coordpass123'),
|
||
'other': login('other', 'otherpass123'),
|
||
'cases': {'c1': case1, 'c2': case2, 'c3': case3},
|
||
'units': {'rcu': rcu, 'oblast': oblast, 'rayon': rayon, 'other': other_rayon}}
|
||
|
||
app.dependency_overrides.pop(auth_router.get_db, None)
|
||
|
||
|
||
def _create(client, token, title, case_id, unit_id=None):
|
||
body = {'title': title, 'case_id': case_id}
|
||
if unit_id:
|
||
body['unit_id'] = unit_id
|
||
return client.post('/api/v1/operations', headers={'Authorization': f'Bearer {token}'}, json=body)
|
||
|
||
|
||
class TestCreateOperation:
|
||
def test_admin_creates_with_explicit_unit(self, ops_env):
|
||
r = ops_env['client'].post(
|
||
'/api/v1/operations', headers={'Authorization': f"Bearer {ops_env['root']}"},
|
||
json={'title': 'Поиск: Иванов', 'case_id': str(ops_env['cases']['c1'].id),
|
||
'unit_id': str(ops_env['units']['rayon'].id)})
|
||
assert r.status_code == 201
|
||
assert r.json()['status'] == 'active'
|
||
|
||
def test_operator_defaults_to_own_unit(self, ops_env):
|
||
r = ops_env['client'].post(
|
||
'/api/v1/operations', headers={'Authorization': f"Bearer {ops_env['coord']}"},
|
||
json={'title': 'Поиск 2', 'case_id': str(ops_env['cases']['c2'].id)})
|
||
assert r.status_code == 201
|
||
assert r.json()['unit_id'] == str(ops_env['units']['oblast'].id)
|
||
|
||
def test_duplicate_case_rejected(self, ops_env):
|
||
c = ops_env['client']
|
||
tok = ops_env['root']
|
||
body = {'title': 'Дубль', 'case_id': str(ops_env['cases']['c1'].id),
|
||
'unit_id': str(ops_env['units']['rayon'].id)}
|
||
c.post('/api/v1/operations', headers={'Authorization': f'Bearer {tok}'}, json=body)
|
||
r = c.post('/api/v1/operations', headers={'Authorization': f'Bearer {tok}'}, json=body)
|
||
assert r.status_code == 400
|
||
|
||
def test_cross_unit_create_403(self, ops_env):
|
||
"""Координатор Минска не может создать операцию под Гомельским Г(Р)ОЧС."""
|
||
r = ops_env['client'].post(
|
||
'/api/v1/operations', headers={'Authorization': f"Bearer {ops_env['coord']}"},
|
||
json={'title': 'Чужой', 'case_id': str(ops_env['cases']['c2'].id),
|
||
'unit_id': str(ops_env['units']['other'].id)})
|
||
assert r.status_code == 403
|
||
|
||
|
||
class TestScope:
|
||
def test_coord_sees_only_own_oblast(self, ops_env):
|
||
c = ops_env['client']
|
||
c.post('/api/v1/operations', headers={'Authorization': f"Bearer {ops_env['coord']}"},
|
||
json={'title': 'Минская', 'case_id': str(ops_env['cases']['c1'].id)})
|
||
c.post('/api/v1/operations', headers={'Authorization': f"Bearer {ops_env['other']}"},
|
||
json={'title': 'Гомельская', 'case_id': str(ops_env['cases']['c2'].id)})
|
||
r = c.get('/api/v1/operations', headers={'Authorization': f"Bearer {ops_env['coord']}"})
|
||
titles = {o['title'] for o in r.json()['items']}
|
||
assert 'Минская' in titles
|
||
assert 'Гомельская' not in titles
|
||
|
||
def test_admin_sees_all(self, ops_env):
|
||
c = ops_env['client']
|
||
c.post('/api/v1/operations', headers={'Authorization': f"Bearer {ops_env['coord']}"},
|
||
json={'title': 'Минская', 'case_id': str(ops_env['cases']['c1'].id)})
|
||
c.post('/api/v1/operations', headers={'Authorization': f"Bearer {ops_env['other']}"},
|
||
json={'title': 'Гомельская', 'case_id': str(ops_env['cases']['c2'].id)})
|
||
r = c.get('/api/v1/operations', headers={'Authorization': f"Bearer {ops_env['root']}"})
|
||
assert r.json()['total'] == 2
|
||
|
||
|
||
class TestStateMachine:
|
||
def _create(self, client, token, case_id):
|
||
return client.post('/api/v1/operations', headers={'Authorization': f'Bearer {token}'},
|
||
json={'title': 'S', 'case_id': case_id})
|
||
|
||
def test_active_to_paused_and_back(self, ops_env):
|
||
c = ops_env['client']
|
||
oid = self._create(c, ops_env['root'], str(ops_env['cases']['c1'].id)).json()['id']
|
||
h = {'Authorization': f"Bearer {ops_env['root']}"}
|
||
assert c.patch(f'/api/v1/operations/{oid}', headers=h,
|
||
json={'status': 'paused'}).json()['status'] == 'paused'
|
||
assert c.patch(f'/api/v1/operations/{oid}', headers=h,
|
||
json={'status': 'active'}).json()['status'] == 'active'
|
||
|
||
def test_invalid_transition_rejected(self, ops_env):
|
||
c = ops_env['client']
|
||
oid = self._create(c, ops_env['root'], str(ops_env['cases']['c1'].id)).json()['id']
|
||
h = {'Authorization': f"Bearer {ops_env['root']}"}
|
||
assert c.patch(f'/api/v1/operations/{oid}', headers=h,
|
||
json={'status': 'completed'}).status_code == 200 # active → completed ок
|
||
assert c.patch(f'/api/v1/operations/{oid}', headers=h,
|
||
json={'status': 'active'}).status_code == 400 # completed → active запрещён
|
||
|
||
def test_completed_sets_closed_at(self, ops_env):
|
||
c = ops_env['client']
|
||
oid = self._create(c, ops_env['root'], str(ops_env['cases']['c1'].id)).json()['id']
|
||
h = {'Authorization': f"Bearer {ops_env['root']}"}
|
||
c.patch(f'/api/v1/operations/{oid}', headers=h, json={'status': 'paused'})
|
||
r = c.patch(f'/api/v1/operations/{oid}', headers=h, json={'status': 'completed'})
|
||
assert r.json()['closed_at'] is not None
|
||
|
||
def test_summary_counts(self, ops_env):
|
||
c = ops_env['client']
|
||
self._create(c, ops_env['root'], str(ops_env['cases']['c1'].id))
|
||
r = c.get('/api/v1/operations/summary', headers={'Authorization': f"Bearer {ops_env['root']}"})
|
||
d = r.json()
|
||
assert d['active'] >= 1 and d['total'] >= 1
|
||
|
||
def test_audit_logged(self, ops_env):
|
||
from backend.models import AuditEvent
|
||
c = ops_env['client']
|
||
self._create(c, ops_env['root'], str(ops_env['cases']['c1'].id))
|
||
types = {e.event_type for e in ops_env['db'].query(AuditEvent).all()}
|
||
assert 'operation_create' in types
|
||
|
||
def test_cross_scope_access_403(self, ops_env):
|
||
c = ops_env['client']
|
||
oid = self._create(c, ops_env['other'], str(ops_env['cases']['c3'].id)).json()['id']
|
||
assert c.get(f'/api/v1/operations/{oid}',
|
||
headers={'Authorization': f"Bearer {ops_env['coord']}"}).status_code == 403 |