vector/backend/tests/test_e1_rbac.py

"""E1 (B21): тесты RBAC-ядра — сессии, блокировка, права, журнал.

SQLite in-memory с RBAC-таблицами; UUID/JSONB подменяются компиляторами.
"""
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
from sqlalchemy import create_engine
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool


@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'


@compiles(sqlalchemy.ARRAY, 'sqlite')
def _array_sqlite(type_, compiler, **kw):
    return 'TEXT'


@pytest.fixture()
def rbac_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,  # одна connection на все threads (иначе per-thread БД)
    )
    Base.metadata.create_all(engine)
    TestingSession = sessionmaker(bind=engine)
    db = TestingSession()

    unit = MchsUnit(name='Минское Г(Р)ОЧС', region='Минская', kind='gor_rayon')
    rcu = MchsUnit(name='РЦУ РЧС', region='Республика', kind='rcu')
    db.add_all([unit, rcu])
    db.flush()

    def _hash(pw: str) -> str:
        return bcrypt.hashpw(pw.encode(), bcrypt.gensalt()).decode()

    user = m.User(username='op1', email='op1@test.by', hashed_password=_hash('oldpass123'),
                  full_name='Оператор Г(Р)ОЧС', role='operator', is_active=True,
                  unit_id=unit.id)
    admin = m.User(username='root_rcu', email='root@test.by', hashed_password=_hash('rootpass123'),
                   full_name='РЦУ РЧС', role='admin', is_active=True, unit_id=rcu.id)
    db.add_all([user, admin])
    db.commit()

    roles = {}
    for name, desc in [('admin', ''), ('coordinator', ''), ('operator', ''), ('observer', '')]:
        r = Role(name=name, description=desc, 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=user.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_get_db():
        s = TestingSession()
        try:
            yield s
        finally:
            s.close()

    app.dependency_overrides[auth_router.get_db] = override_get_db
    client = TestClient(app)

    yield {'db': db, 'client': client, 'user': user, 'admin': admin}

    app.dependency_overrides.pop(auth_router.get_db, None)


def _login(client, username: str, password: str):
    return client.post('/api/v1/auth/login', data={'username': username, 'password': password})


class TestLogin:
    def test_login_creates_session_and_token(self, rbac_env):
        r = _login(rbac_env['client'], 'op1', 'oldpass123')
        assert r.status_code == 200
        assert 'access_token' in r.json()

    def test_wrong_password_fails(self, rbac_env):
        assert _login(rbac_env['client'], 'op1', 'wrong').status_code == 401

    def test_lockout_after_max_failures(self, rbac_env):
        c = rbac_env['client']
        for _ in range(5):
            c.post('/api/v1/auth/login', data={'username': 'op1', 'password': 'wrong'})
        r = _login(c, 'op1', 'oldpass123')
        assert r.status_code == 403
        assert 'locked' in r.json()['detail'].lower()

    def test_me_has_permissions(self, rbac_env):
        c = rbac_env['client']
        tok = _login(c, 'op1', 'oldpass123').json()['access_token']
        me = c.get('/api/v1/auth/me', headers={'Authorization': f'Bearer {tok}'}).json()
        assert me['permissions'] == ['create', 'update', 'view']
        assert me['unit_id'] is not None

    def test_admin_permissions_full(self, rbac_env):
        c = rbac_env['client']
        tok = _login(c, 'root_rcu', 'rootpass123').json()['access_token']
        me = c.get('/api/v1/auth/me', headers={'Authorization': f'Bearer {tok}'}).json()
        assert 'manage_users' in me['permissions']
        assert len(me['permissions']) == 9

    def test_logout_revokes_session(self, rbac_env):
        c = rbac_env['client']
        tok = _login(c, 'op1', 'oldpass123').json()['access_token']
        h = {'Authorization': f'Bearer {tok}'}
        assert c.get('/api/v1/auth/me', headers=h).status_code == 200
        c.post('/api/v1/auth/logout', headers=h)
        assert c.get('/api/v1/auth/me', headers=h).status_code == 401

    def test_change_password_revokes_sessions(self, rbac_env):
        c = rbac_env['client']
        tok = _login(c, 'op1', 'oldpass123').json()['access_token']
        h = {'Authorization': f'Bearer {tok}'}
        r = c.post('/api/v1/auth/change-password', headers=h,
                   json={'current_password': 'oldpass123', 'new_password': 'newpass456'})
        assert r.status_code == 200
        assert c.get('/api/v1/auth/me', headers=h).status_code == 401
        assert _login(c, 'op1', 'newpass456').status_code == 200

    def test_short_password_rejected(self, rbac_env):
        c = rbac_env['client']
        tok = _login(c, 'op1', 'oldpass123').json()['access_token']
        h = {'Authorization': f'Bearer {tok}'}
        r = c.post('/api/v1/auth/change-password', headers=h,
                   json={'current_password': 'oldpass123', 'new_password': 'abc'})
        assert r.status_code == 400

    def test_auth_events_logged(self, rbac_env):
        from backend.models import AuthEvent
        c = rbac_env['client']
        _login(c, 'op1', 'oldpass123')
        _login(c, 'op1', 'WRONG')
        types = {e.event_type for e in rbac_env['db'].query(AuthEvent).all()}
        assert 'login' in types and 'login_failed' in types