import re
import os
from sqlalchemy import create_engine
from sqlalchemy.engine import make_url
from sqlalchemy.orm import sessionmaker
from app.db import Base
from datetime import timedelta

import pytest
from bs4 import BeautifulSoup
from fastapi.testclient import TestClient
from sqlalchemy import select

from app.config import Settings, get_settings
from app.db import get_db
from app.main import app
from app.models import SystemUser, UserAction, Employee, EmployeePublication, CouncilCorrections, utcnow
from app.security import (SESSION_COOKIE, _session_data, ensure_bootstrap_admin, hash_password,
                          sign_session, temporary_password, verify_password)
from app.services import council


@pytest.fixture()
def accounts(db_session, monkeypatch):
    test_engine = None
    native_url = os.getenv('USERS_TEST_DATABASE_URL')
    if native_url:
        assert make_url(native_url).database == 'miem_users_test', 'Use a disposable miem_users_test database only'
        test_engine = create_engine(native_url)
        Base.metadata.create_all(test_engine)
        db_session = sessionmaker(bind=test_engine)()
    settings = Settings(_env_file=None, admin_username='admin', admin_password='initial-admin-test', session_secret='users-test-secret')
    admin = ensure_bootstrap_admin(db_session, settings)
    user = SystemUser(username='reader', full_name='Тестовый пользователь', role='user',
                      password_hash=hash_password('Initial user password'), must_change_password=False,
                      first_login_at=utcnow())
    db_session.add(user)
    employee = Employee(profile_key='users:employee', canonical_url='https://www.hse.ru/staff/test', full_name='Тестовый сотрудник',
        current_data={'sections': [{'title': 'Учёные степени', 'items': ['Доктор технических наук']}]})
    db_session.add(employee)
    db_session.flush()
    db_session.add(EmployeePublication(employee_id=employee.id, title='Исходная статья', year=2024,
        publication_type='ARTICLE', source_hash='users-publication', raw_data={'journal': 'Журнал'}))
    db_session.commit()
    monkeypatch.setattr(council, '_articles_by_author', lambda: {})
    app.dependency_overrides[get_db] = lambda: db_session
    app.dependency_overrides[get_settings] = lambda: settings
    client = TestClient(app)
    try:
        yield client, db_session, settings, admin, user, employee
    finally:
        client.close()
        app.dependency_overrides.clear()
        if test_engine is not None:
            db_session.close()
            Base.metadata.drop_all(test_engine)
            test_engine.dispose()


def act_as(client, user, settings):
    client.cookies.set(SESSION_COOKIE, sign_session(user.username, settings, user.session_version))
    csrf = _session_data(client.cookies.get(SESSION_COOKIE), settings)['csrf']
    return {'X-CSRF-Token': csrf}


def login(client, username, password):
    page = client.get('/admin/login')
    csrf = BeautifulSoup(page.text, 'html.parser').select_one('[name=csrf_token]')['value']
    return client.post('/admin/login', data={'username': username, 'password': password, 'csrf_token': csrf}, follow_redirects=False)


def test_password_hash_and_eight_character_generator():
    for _ in range(100):
        password = temporary_password()
        assert len(password) == 8
        assert all(re.search(pattern, password) for pattern in (r'[A-Z]', r'[a-z]', r'[0-9]', r'[!@#$%+_\-]'))
    encoded = hash_password('Секретная фраза с пробелами')
    assert verify_password('Секретная фраза с пробелами', encoded)
    assert not verify_password('Другой пароль', encoded)
    assert not verify_password('anything', 'broken hash')
    assert hash_password('same password') != hash_password('same password')


def test_create_user_first_login_forces_password_change(accounts):
    client, db, settings, admin, _, _ = accounts
    headers = act_as(client, admin, settings)
    response = client.post('/admin/users', data={'username': 'new.reader', 'full_name': 'Новый пользователь', 'role': 'user'}, headers=headers)
    assert response.status_code == 200 and response.headers['cache-control'] == 'no-store'
    password = BeautifulSoup(response.text, 'html.parser').select_one('[data-temporary-password]')['value']
    user = db.scalar(select(SystemUser).where(SystemUser.username == 'new.reader'))
    assert verify_password(password, user.password_hash) and user.must_change_password
    assert password not in str(db.scalars(select(UserAction.changes)).all())
    client.cookies.clear()
    signed_in = login(client, user.username, password)
    assert signed_in.status_code == 303 and signed_in.headers['location'] == '/admin/account'
    db.refresh(user)
    assert user.first_login_at and user.last_login_at and user.password_changed_at is None
    first_login = user.first_login_at
    assert 'data-required-password' in client.get('/admin/account').text
    assert client.get('/admin/directory', follow_redirects=False).headers['location'] == '/admin/account'
    assert client.get('/api/employees', follow_redirects=False).status_code == 403
    old_token = client.cookies.get(SESSION_COOKIE)
    csrf = _session_data(old_token, settings)['csrf']
    changed = client.post('/admin/account/password', data={'current_password': password,
        'new_password': 'A new personal passphrase', 'confirmation': 'A new personal passphrase'}, headers={'X-CSRF-Token': csrf}, follow_redirects=False)
    assert changed.status_code == 303
    db.refresh(user)
    assert not user.must_change_password and user.password_changed_at and user.session_version == 1
    assert user.first_login_at == first_login
    assert 'data-required-password' not in client.get('/admin/account').text
    assert 'data-guide-pending="true"' in client.get('/admin/account').text
    assert client.get('/admin/directory').status_code == 200
    assert client.get('/admin/users').status_code == 403
    client.cookies.clear()
    client.cookies.set(SESSION_COOKIE, old_token)
    assert client.get('/api/employees', follow_redirects=False).status_code == 401


@pytest.mark.parametrize('path', ['/admin/crawl-now', '/admin/runs', '/admin/dismissed/refresh', '/admin/employees/1/refresh', '/api/crawl-runs'])
def test_regular_user_cannot_start_scraping(accounts, path):
    client, db, settings, admin, user, _ = accounts
    response = client.post(path, headers=act_as(client, user, settings), follow_redirects=False)
    assert response.status_code == 403
    assert 'Запустить парсинг' not in client.get('/admin').text
    assert 'Обновить данные' not in client.get('/admin/employees/1').text
    assert client.get('/api/stats').status_code == 200


def test_council_user_edit_audit_and_personal_history_isolation(accounts):
    client, db, settings, admin, user, employee = accounts
    candidate = council.council_candidates(db)['candidates'][0]
    article = candidate['articles'][0]
    data = {'employee_id': employee.id, 'edit_key': article['edit_key'], 'revision': 0,
        'title': 'Изменённая статья', 'journal': 'Новый журнал', 'quartile': 'Q2', 'specialty_codes': ['1.2.2']}
    act_as(client, user, settings)
    assert client.post('/admin/council/publications', json=data, headers={'X-Council-Edit': '1'}).status_code == 403
    assert client.post(f'/admin/users/{user.id}', data={'role': 'council', 'is_active': 'true'},
                       headers=act_as(client, admin, settings), follow_redirects=False).status_code == 303
    db.refresh(user)
    assert user.role == 'council' and user.session_version == 1
    act_as(client, user, settings)
    response = client.post('/admin/council/publications', json=data, headers={'X-Council-Edit': '1'})
    assert response.status_code == 200
    action = db.scalar(select(UserAction).where(UserAction.action == 'card_edit'))
    assert action.user_id == user.id and action.employee_id == employee.id and action.publication_year == 2024
    assert action.changes['title'] == {'before': 'Исходная статья', 'after': 'Изменённая статья'}
    assert action.changes['quartile'] == {'before': '', 'after': 'Q2'}
    assert action.changes['specialty_codes']['after']['codes'] == ['1.2.2']
    db.add(UserAction(user_id=admin.id, action='card_edit', article_title='Чужое действие', changes={}))
    db.commit()
    profile = client.get(f'/admin/account?user_id={admin.id}')
    assert 'Изменённая статья' in profile.text and 'Исходная статья' in profile.text
    assert 'Чужое действие' not in profile.text
    assert client.get(f'/admin/users?user_id={admin.id}').status_code == 403
    assert client.post('/admin/council/publications', json=data, headers={'X-Council-Edit': '1'}).status_code == 409
    assert db.scalars(select(UserAction).where(UserAction.user_id == user.id)).all() == [action]


@pytest.mark.parametrize('role', ['user', 'council', 'admin'])
def test_navigation_and_run_access_match_role(accounts, role):
    client, db, settings, _, user, _ = accounts
    user.role = role
    db.commit()
    act_as(client, user, settings)
    dashboard = BeautifulSoup(client.get('/admin').text, 'html.parser')
    links = {link['href'] for link in dashboard.select('nav a[href]')}
    assert {'/admin', '/admin/directory', '/admin/account', '/admin/guide'} <= links
    assert ('/admin/council' in links) == (role != 'user')
    assert ('/admin/runs' in links) == (role == 'admin')
    assert ('/admin/users' in links) == (role == 'admin')
    assert bool(dashboard.select('[data-progress-panel]')) == (role == 'admin')
    assert ('latest_run' in client.get('/api/stats').json()) == (role == 'admin')
    for path in ('/admin/runs', '/api/crawl-runs', '/api/crawl-runs/latest', '/api/crawl-runs/1'):
        assert client.get(path).status_code == (200 if role == 'admin' else 403)
    assert client.get('/admin/runs/1', follow_redirects=False).status_code == (303 if role == 'admin' else 403)
    client.cookies.clear()
    assert client.get('/admin/directory', follow_redirects=False).headers['location'] == '/admin/login'
    assert client.get('/admin/council').status_code == 200
    assert client.get('/docs').status_code == 404


def test_admin_reset_invalidates_sessions_and_preserves_password_change_date(accounts):
    client, db, settings, admin, user, _ = accounts
    user.password_changed_at = utcnow()
    db.commit()
    previous_changed = user.password_changed_at
    old_token = sign_session(user.username, settings, user.session_version)
    response = client.post(f'/admin/users/{user.id}/reset-password', headers=act_as(client, admin, settings))
    assert response.status_code == 200
    password = BeautifulSoup(response.text, 'html.parser').select_one('[data-temporary-password]')['value']
    db.refresh(user)
    assert user.must_change_password and user.password_reset_at and verify_password(password, user.password_hash)
    assert user.password_changed_at == previous_changed
    assert password not in client.get('/admin/users').text
    client.cookies.clear()
    client.cookies.set(SESSION_COOKIE, old_token)
    assert client.get('/api/employees', follow_redirects=False).status_code == 401
    assert login(client, user.username, password).headers['location'] == '/admin/account'
    assert db.scalar(select(UserAction).where(UserAction.action == 'password_reset')).target_user_id == user.id


def test_role_changes_disable_and_protection_against_locking_out_self(accounts):
    client, db, settings, admin, user, _ = accounts
    old_token = sign_session(user.username, settings, user.session_version)
    headers = act_as(client, admin, settings)
    assert client.post(f'/admin/users/{admin.id}', data={'role': 'user', 'is_active': 'false'}, headers=headers).status_code == 422
    assert client.post(f'/admin/users/{admin.id}/reset-password', headers=headers).status_code == 422
    assert client.post(f'/admin/users/{user.id}', data={'role': 'admin', 'is_active': 'true'}, headers=headers, follow_redirects=False).status_code == 303
    db.refresh(user)
    assert user.role == 'admin'
    client.cookies.clear()
    client.cookies.set(SESSION_COOKIE, old_token)
    assert client.get('/admin/users', follow_redirects=False).status_code == 303
    headers = act_as(client, admin, settings)
    assert client.post(f'/admin/users/{user.id}', data={'role': 'user', 'is_active': 'false'}, headers=headers, follow_redirects=False).status_code == 303
    db.refresh(user)
    client.cookies.clear()
    assert login(client, user.username, 'Initial user password').status_code == 401
    assert not user.is_active


def test_csrf_duplicate_user_validation_and_login_throttling(accounts):
    client, db, settings, admin, user, _ = accounts
    headers = act_as(client, admin, settings)
    assert client.post('/admin/users', data={'username': 'unprotected'}).status_code == 403
    assert client.post('/admin/account/password', data={'current_password': 'anything'}).status_code == 403
    assert client.post('/admin/login', data={'username': user.username, 'password': 'anything'}).status_code == 403
    for username, role, code in [('reader', 'user', 409), ('bad login', 'user', 422), ('newuser', 'superuser', 422)]:
        assert client.post('/admin/users', data={'username': username, 'role': role}, headers=headers).status_code == code
    client.cookies.clear()
    for _ in range(5):
        assert login(client, user.username, 'wrong password').status_code == 401
    db.refresh(user)
    assert user.locked_until and user.failed_login_attempts == 5
    assert login(client, user.username, 'Initial user password').status_code == 401
    user.locked_until = utcnow() - timedelta(minutes=1)
    db.commit()
    for attempt in range(1, 6):
        assert login(client, user.username, 'wrong password').status_code == 401
        db.refresh(user)
        assert user.failed_login_attempts == attempt
        assert bool(user.locked_until) == (attempt == 5)
    assert login(client, user.username, 'Initial user password').status_code == 401
    db.refresh(user)
    assert user.failed_login_attempts == 5
    user.locked_until = utcnow() - timedelta(minutes=1)
    db.commit()
    assert login(client, user.username, 'Initial user password').status_code == 303
    db.refresh(user)
    assert user.failed_login_attempts == 0 and user.locked_until is None


def test_bootstrap_admin_first_login_and_changed_password_no_env_fallback(accounts):
    client, db, settings, admin, _, _ = accounts
    assert login(client, admin.username, 'initial-admin-test').headers['location'] == '/admin/account'
    csrf = _session_data(client.cookies.get(SESSION_COOKIE), settings)['csrf']
    assert client.post('/admin/account/password', data={'current_password': 'initial-admin-test',
        'new_password': 'New administrator passphrase', 'confirmation': 'New administrator passphrase'}, headers={'X-CSRF-Token': csrf}, follow_redirects=False).status_code == 303
    client.cookies.clear()
    assert login(client, admin.username, 'initial-admin-test').status_code == 401
    assert login(client, admin.username, 'New administrator passphrase').status_code == 303


def test_cross_origin_and_session_expiry_fail_closed(accounts, monkeypatch):
    import time
    from app import security
    client, db, settings, admin, user, _ = accounts
    headers = act_as(client, admin, settings)
    assert client.post('/admin/users', data={'username': 'attacker'}, headers={**headers, 'Origin': 'https://evil.example'}).status_code == 403
    assert client.post('/admin/users', data={'username': 'attacker'}, headers={**headers, 'Origin': 'http://['}).status_code == 403
    assert db.scalar(select(SystemUser).where(SystemUser.username == 'attacker')) is None
    token = sign_session(user.username, settings, user.session_version)
    now = int(time.time())
    monkeypatch.setattr(security.time, 'time', lambda: now + security.SESSION_SECONDS + 1)
    assert security.read_session(token, settings) is None


def test_password_errors_and_noop_card_edit_are_not_audited(accounts):
    client, db, settings, _, user, employee = accounts
    user.role = 'council'
    db.commit()
    headers = act_as(client, user, settings)
    before = user.password_hash
    for current, new, confirmation in [('wrong', 'New valid personal password', 'New valid personal password'),
                                      ('Initial user password', 'short', 'short'),
                                      ('Initial user password', 'New valid personal password', 'Different valid password')]:
        response = client.post('/admin/account/password', headers=headers,
            data={'current_password': current, 'new_password': new, 'confirmation': confirmation})
        assert response.status_code == 422 and user.password_hash == before
    article = council.council_candidates(db)['candidates'][0]['articles'][0]
    response = client.post('/admin/council/publications', headers={'X-Council-Edit': '1'}, json={
        'employee_id': employee.id, 'edit_key': article['edit_key'], 'revision': 0, 'title': article['title'],
        'journal': article['edit_journal'], 'quartile': article['edit_quartile'], 'specialty_codes': None})
    assert response.status_code == 200
    assert db.scalars(select(UserAction).where(UserAction.user_id == user.id)).all() == []


def test_logout_revokes_replayed_cookie(accounts):
    client, db, settings, _, user, _ = accounts
    headers = act_as(client, user, settings)
    old_token = client.cookies.get(SESSION_COOKIE)
    assert client.post('/admin/logout', headers=headers, follow_redirects=False).status_code == 303
    client.cookies.set(SESSION_COOKIE, old_token)
    assert client.get('/api/employees', follow_redirects=False).status_code == 401


def test_password_change_and_audit_rollback_atomically(accounts, monkeypatch):
    from sqlalchemy.exc import OperationalError
    client, db, settings, _, user, _ = accounts
    headers = act_as(client, user, settings)
    before = user.password_hash
    def unavailable():
        raise OperationalError('UPDATE', {}, Exception('private database details'))
    monkeypatch.setattr(db, 'commit', unavailable)
    response = client.post('/admin/account/password', data={'current_password': 'Initial user password',
        'new_password': 'A different personal passphrase', 'confirmation': 'A different personal passphrase'}, headers=headers)
    assert response.status_code == 503 and 'private database details' not in response.text
    db.refresh(user)
    assert user.password_hash == before and user.session_version == 0
    assert db.scalar(select(UserAction).where(UserAction.action == 'password_change')) is None
    assert db.scalar(select(UserAction).where(UserAction.action == 'guide_start')) is None


@pytest.mark.parametrize('role', ['user', 'council', 'admin'])
def test_guide_first_password_dismissal_and_repeat_change(accounts, role):
    client, db, settings, _, user, _ = accounts
    user.role = role
    db.commit()
    headers = act_as(client, user, settings)
    assert 'data-guide-pending="false"' in client.get('/admin/account').text
    data = {'current_password': 'Initial user password', 'new_password': 'New onboarding passphrase',
            'confirmation': 'New onboarding passphrase'}
    assert client.post('/admin/account/password', data=data, headers=headers, follow_redirects=False).status_code == 303
    page = client.get('/admin/account')
    assert 'data-guide-pending="true"' in page.text
    assert 'data-required-password' not in page.text
    assert page.headers['Cache-Control'] == 'no-store'
    guide = client.get('/admin/guide')
    parsed = BeautifulSoup(guide.text, 'html.parser')
    assert len(parsed.select('[data-guide-step]')) == (2 if role == 'user' else 6)
    assert not parsed.select('[data-guide]')
    assert ('Прочитайте показатели сотрудника' in guide.text) == (role != 'user')
    assert ('Проверьте строку публикации' in guide.text) == (role != 'user')
    assert ('управлять доступом' in guide.text) == (role == 'admin')
    assert ('не ограничивает «ВСЕГО»' in guide.text) == (role != 'user')
    csrf = BeautifulSoup(page.text, 'html.parser').select_one('[data-guide]')['data-guide-csrf']
    assert client.post('/admin/guide/dismiss', headers={'X-CSRF-Token': 'wrong'}, follow_redirects=False).status_code == 403
    assert 'data-guide-pending="true"' in client.get('/admin/directory').text
    for _ in range(2):
        response = client.post('/admin/guide/dismiss', headers={'X-CSRF-Token': csrf, 'Accept': 'application/json'}, follow_redirects=False)
        assert response.status_code == 204
    assert len(db.scalars(select(UserAction).where(UserAction.user_id == user.id, UserAction.action == 'guide_finish')).all()) == 1
    client.cookies.clear()
    headers = act_as(client, user, settings)
    assert 'data-guide-pending="false"' in client.get('/admin').text
    data = {'current_password': data['new_password'], 'new_password': 'Second onboarding passphrase',
            'confirmation': 'Second onboarding passphrase'}
    assert client.post('/admin/account/password', data=data, headers=headers, follow_redirects=False).status_code == 303
    assert 'data-guide-pending="false"' in client.get('/admin/account').text
    assert client.get('/admin/guide').status_code == 200
    assert len(db.scalars(select(UserAction).where(UserAction.user_id == user.id, UserAction.action == 'guide_start')).all()) == 1


def test_guide_access_and_no_javascript_completion(accounts):
    client, db, settings, _, user, _ = accounts
    assert not BeautifulSoup(client.get('/admin/council').text, 'html.parser').select('[data-guide]')
    assert client.get('/admin/guide', follow_redirects=False).headers['location'] == '/admin/login'
    user.must_change_password = True
    db.commit()
    headers = act_as(client, user, settings)
    assert client.get('/admin/guide', follow_redirects=False).headers['location'] == '/admin/account'
    assert client.post('/admin/guide/dismiss', headers=headers, follow_redirects=False).headers['location'] == '/admin/account'
    user.must_change_password = False
    db.add(UserAction(user_id=user.id, action='guide_start'))
    db.commit()
    page = BeautifulSoup(client.get('/admin').text, 'html.parser')
    assert page.select_one('[data-guide-fallback] a[href="/admin/guide"]')
    response = client.post('/admin/guide/dismiss', headers=headers, follow_redirects=False)
    assert response.status_code == 303 and response.headers['location'] == '/admin'
    assert 'data-guide-pending="false"' in client.get('/admin/account').text


def test_guide_dismissal_failure_does_not_mark_it_finished(accounts, monkeypatch):
    from sqlalchemy.exc import OperationalError
    client, db, settings, _, user, _ = accounts
    db.add(UserAction(user_id=user.id, action='guide_start'))
    db.commit()
    headers = act_as(client, user, settings)
    def unavailable():
        raise OperationalError('INSERT', {}, Exception('private database details'))
    monkeypatch.setattr(db, 'commit', unavailable)
    response = client.post('/admin/guide/dismiss', headers={**headers, 'Accept': 'application/json'})
    assert response.status_code == 503 and 'private database details' not in response.text
    assert db.scalar(select(UserAction).where(UserAction.action == 'guide_finish')) is None
    assert 'data-guide-pending="true"' in client.get('/admin/account').text


def test_postgres_concurrent_admin_demotions_keep_an_admin(accounts):
    from concurrent.futures import ThreadPoolExecutor
    client, db, settings, admin, _, _ = accounts
    if db.get_bind().dialect.name != 'postgresql':
        pytest.skip('Requires the disposable PostgreSQL test database')
    second = SystemUser(username='admin-two', role='admin', password_hash=hash_password('Second administrator passphrase'),
                        must_change_password=False, first_login_at=utcnow())
    db.add(second)
    db.commit()
    factory = sessionmaker(bind=db.get_bind())
    def sessions():
        with factory() as session:
            yield session
    app.dependency_overrides[get_db] = sessions
    clients = [TestClient(app), TestClient(app)]
    actors = [admin, second]
    targets = [second.id, admin.id]
    def demote(index):
        headers = act_as(clients[index], actors[index], settings)
        response = clients[index].post(f'/admin/users/{targets[index]}',
            data={'role': 'user', 'is_active': 'true'}, headers=headers, follow_redirects=False)
        return response.status_code, response.headers.get('location')
    try:
        with ThreadPoolExecutor(max_workers=2) as executor:
            responses = list(executor.map(demote, [0, 1]))
        assert sum(location == '/admin/users' for status, location in responses) == 1
        assert all(status in {303, 403} for status, location in responses)
        remaining = db.scalars(select(SystemUser).where(SystemUser.role == 'admin', SystemUser.is_active.is_(True))).all()
        assert len(remaining) == 1
    finally:
        for other in clients:
            other.close()