"""Offline account-wide model preferences; no HTTP, Docker, or provider calls."""
from concurrent.futures import ThreadPoolExecutor
import importlib.util
from pathlib import Path
import sqlite3
import threading

import pytest

# Load only the stdlib store; control.__init__ imports the optional FastAPI app.
_spec = importlib.util.spec_from_file_location(
    'model_preferences_accounts', Path(__file__).resolve().parents[2] / 'control' / 'accounts.py'
)
_accounts = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_accounts)
Accounts = _accounts.Accounts
public_user = _accounts.public_user


PASSWORD = 'correct-horse-password'
DEFAULTS = {'model': 'auto', 'thinking_level': 'none'}


@pytest.fixture
def accounts(tmp_path):
    store = Accounts(str(tmp_path / 'accounts.sqlite3'))
    try:
        yield store
    finally:
        store.close()


def create_user(store, username='alice'):
    return store.create(username, username.title(), 'user', PASSWORD)['id']


def test_defaults_and_saved_state_survive_reopen(tmp_path):
    path = str(tmp_path / 'accounts.sqlite3')
    store = Accounts(path)
    try:
        user_id = create_user(store)
        assert store.get_model_settings(user_id) == DEFAULTS
        assert store.get(user_id)['status'] == 'provisioning'
        saved = {'model': 'openai/gpt-5.4', 'thinking_level': 'xhigh'}
        assert store.set_model_settings(user_id, **saved) == saved
    finally:
        store.close()
    reopened = Accounts(path)
    try:
        assert reopened.get_model_settings(user_id) == saved
        updated = {'model': 'anthropic/claude-sonnet-4.6', 'thinking_level': 'low'}
        reopened.set_model_settings(user_id, **updated)
        assert reopened.get_model_settings(user_id) == updated
        assert reopened.db.execute('SELECT count(*) FROM account_model_settings').fetchone()[0] == 1
    finally:
        reopened.close()


def test_existing_database_migrates_with_default_preferences(tmp_path):
    path = str(tmp_path / 'legacy.sqlite3')
    with sqlite3.connect(path) as legacy:
        legacy.executescript('''
            CREATE TABLE users (
                id TEXT PRIMARY KEY, username TEXT NOT NULL UNIQUE,
                display_name TEXT NOT NULL, role TEXT NOT NULL,
                status TEXT NOT NULL, container_name TEXT NOT NULL,
                error TEXT, password_hash TEXT NOT NULL,
                hermes_url TEXT, hermes_api_key TEXT,
                desired TEXT NOT NULL DEFAULT 'ready'
            );
            CREATE TABLE rooms (
                room_name TEXT PRIMARY KEY,
                user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
                expires_at REAL NOT NULL
            );
            INSERT INTO users(id,username,display_name,role,status,container_name,password_hash)
                VALUES('existing','existing','Existing','user','error','hope-hermes-existing','unused');
        ''')
    store = Accounts(path)
    try:
        assert store.get_model_settings('existing') == DEFAULTS
        assert store.get('existing')['status'] == 'error'
        assert store.db.execute('SELECT count(*) FROM account_model_settings').fetchone()[0] == 0
        store.set_model_settings('existing', 'openai/gpt-5', 'medium')
        assert store.get_model_settings('existing') == {'model': 'openai/gpt-5', 'thinking_level': 'medium'}
    finally:
        store.close()


def test_tenant_isolation_and_public_schema_unchanged(accounts):
    alice = create_user(accounts)
    bob = create_user(accounts, 'bob')
    before = public_user(accounts.get(alice))
    accounts.set_model_settings(alice, 'openai/gpt-5', 'high')
    assert accounts.get_model_settings(bob) == DEFAULTS
    accounts.set_model_settings(bob, 'local/model:v2', 'minimal')
    assert accounts.get_model_settings(alice) == {'model': 'openai/gpt-5', 'thinking_level': 'high'}
    assert accounts.get_model_settings(bob) == {'model': 'local/model:v2', 'thinking_level': 'minimal'}
    assert public_user(accounts.get(alice)) == before
    assert set(before) == {'id', 'username', 'display_name', 'role', 'status', 'container_name', 'error'}
    result = accounts.get_model_settings(alice)
    result['model'] = 'tampered'
    assert accounts.get_model_settings(alice)['model'] == 'openai/gpt-5'


def test_missing_deleting_and_deleted_accounts_reject_and_cascade(accounts):
    with pytest.raises(KeyError):
        accounts.get_model_settings('missing')
    with pytest.raises(KeyError):
        accounts.set_model_settings('missing', 'auto', 'none')
    user_id = create_user(accounts)
    accounts.set_model_settings(user_id, 'openai/gpt-5', 'high')
    accounts.begin_delete(user_id)
    with pytest.raises(KeyError):
        accounts.get_model_settings(user_id)
    with pytest.raises(KeyError):
        accounts.set_model_settings(user_id, 'auto', 'none')
    accounts.finish_delete(user_id)
    assert accounts.db.execute('SELECT * FROM account_model_settings WHERE user_id=?', (user_id,)).fetchone() is None
    with pytest.raises(KeyError):
        accounts.get_model_settings(user_id)
    with pytest.raises(KeyError):
        accounts.set_model_settings(user_id, 'auto', 'none')
    assert accounts.db.execute('SELECT count(*) FROM account_model_settings').fetchone()[0] == 0


@pytest.mark.parametrize('model', [None, 123, [], '', 'a' * 201, 'two models', ' auto', 'auto ',
                                   'auto\n', 'auto\t', 'auto\r', 'a\x00b', 'a\x1fb', 'a\x7fb', 'café', 'a\u00a0b'])
def test_invalid_models_reject_without_changing_saved_state(accounts, model):
    user_id = create_user(accounts)
    accounts.set_model_settings(user_id, 'valid/model', 'low')
    with pytest.raises(ValueError):
        accounts.set_model_settings(user_id, model, 'high')
    assert accounts.get_model_settings(user_id) == {'model': 'valid/model', 'thinking_level': 'low'}


@pytest.mark.parametrize('thinking_level', [None, 1, [], '', 'NONE', 'disabled', 'max', ' high', 'high ', 'high\n'])
def test_invalid_thinking_levels_reject_without_creating_settings(accounts, thinking_level):
    user_id = create_user(accounts)
    with pytest.raises(ValueError):
        accounts.set_model_settings(user_id, 'auto', thinking_level)
    assert accounts.get_model_settings(user_id) == DEFAULTS
    assert accounts.db.execute('SELECT count(*) FROM account_model_settings').fetchone()[0] == 0


@pytest.mark.parametrize('thinking_level', ['none', 'minimal', 'low', 'medium', 'high', 'xhigh'])
def test_all_thinking_levels_and_model_length_boundary(accounts, thinking_level):
    user_id = create_user(accounts)
    accounts.set_model_settings(user_id, 'a' * 200, thinking_level)
    assert accounts.get_model_settings(user_id) == {'model': 'a' * 200, 'thinking_level': thinking_level}


def test_set_is_guarded_against_deletion_on_another_connection(accounts, tmp_path):
    user_id = create_user(accounts)
    accounts.set_model_settings(user_id, 'original/model', 'none')
    other = Accounts(str(tmp_path / 'accounts.sqlite3'))
    attempted = threading.Event()
    other.db.set_trace_callback(lambda sql: attempted.set() if 'INSERT INTO account_model_settings' in sql else None)
    try:
        with ThreadPoolExecutor(max_workers=1) as executor:
            with accounts.db:
                accounts.db.execute('BEGIN IMMEDIATE')
                accounts.db.execute("UPDATE users SET desired='deleted',status='deleting' WHERE id=?", (user_id,))
                future = executor.submit(other.set_model_settings, user_id, 'new/model', 'high')
                assert attempted.wait(timeout=2), 'Writer did not attempt its guarded statement'
            with pytest.raises(KeyError):
                future.result(timeout=5)
        row = accounts.db.execute('SELECT model,thinking_level FROM account_model_settings WHERE user_id=?', (user_id,)).fetchone()
        assert dict(row) == {'model': 'original/model', 'thinking_level': 'none'}
        accounts.finish_delete(user_id)
        assert accounts.db.execute('SELECT count(*) FROM account_model_settings').fetchone()[0] == 0
    finally:
        other.close()