E3 (B21): операции мультипоиска — статусная машина, скоуп, дашборд

- Модель 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.

backend/tests/test_e3_operations.py

@@ -0,0 +1,252 @@
+"""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