from concurrent.futures import Future, CancelledError
import logging
import threading
from types import SimpleNamespace

import pytest

from app.trace_commands import completion
from app.stream_lifecycle import StreamLifecycle
from app.multi_roi.projection import StartResult
from app.secure_control import command_error
from contextra_trace.writer import Recorder, DEFAULT


def record(future, emit=None, action='start'):
    rows = []
    job = SimpleNamespace(operation_id='op-1', camera_id='00000003',
                          action=action, _trace_request='request-1')
    completion(job, future, emit=emit or (lambda **r: rows.append(r)),
               error_code=command_error, logger=logging.getLogger('trace-fixture'))
    return rows[0] if rows else None


def resolved(value):
    f = Future(); f.set_result(value)
    return f


def group():
    lifecycle = StreamLifecycle()
    lifecycle.begin(assigned_run_id='run-1', assigned_generation=1)
    return SimpleNamespace(lifecycle=lifecycle)


@pytest.mark.parametrize('wrapped', [False, True])
@pytest.mark.parametrize('action', ['start', 'stop'])
def test_real_lifecycle_and_multi_wrapper_preserve_ids(wrapped, action):
    g = group()
    f = resolved(StartResult(g, {}) if wrapped else g)
    a, b = record(f, action=action), record(f, action=action)
    assert a == b
    assert a['phase'] == 'COMMAND_APPLIED' and a['commandOutcome'] == 'APPLIED'
    assert a['runId'] == 'run-1' and a['observedState'] == 'starting'
    assert a['requestId'] == 'request-1' and a['operationId'] == 'op-1'
    assert a['cameraId'] == '00000003' and a['action'] == action


def test_failure_is_not_applied_and_does_not_log_secret(caplog):
    f = Future(); f.set_exception(RuntimeError('password=DO_NOT_STORE'))
    with caplog.at_level(logging.INFO):
        row = record(f)
    assert row['phase'] == 'COMMAND_FAILED'
    assert row['error'] == 'RUNTIME_COMMAND_REJECTED'
    assert row['runId'] is None and row['observedState'] == 'UNKNOWN'
    assert 'DO_NOT_STORE' not in str(row) + caplog.text
    assert isinstance(f.exception(), RuntimeError)


@pytest.mark.parametrize('as_error', [False, True])
def test_cancelled_future_does_not_raise(as_error):
    f = Future()
    if as_error: f.set_exception(CancelledError())
    else: f.cancel()
    row = record(f)
    assert row['phase'] == 'COMMAND_CANCELLED' and row['observedState'] == 'UNKNOWN'


@pytest.mark.parametrize('action', ['start', 'stop'])
def test_missing_result_does_not_invent_stopped(action):
    row = record(resolved(None), action=action)
    assert row['phase'] == 'COMMAND_RESULT_UNKNOWN'
    assert row['runId'] is None and row['observedState'] == 'UNKNOWN'


def test_incomplete_future_never_waits():
    f = Future()
    row = record(f)
    assert row['observationReason'] == 'FUTURE_NOT_DONE' and not f.done()


def test_stop_observation_distinguishes_cleanup_pending_and_stopped():
    g = group(); generation = g.lifecycle.request_stop()
    assert record(resolved(g), action='stop')['observedState'] == 'stopping'
    g.lifecycle.finish_stop(generation)
    row = record(resolved(g), action='stop')
    assert row['observedState'] == 'stopped' and row['runId'] == 'run-1'


def test_busy_lifecycle_does_not_wait():
    g = group(); entered, release = threading.Event(), threading.Event()
    def hold():
        with g.lifecycle._lock:
            entered.set(); release.wait(2)
    worker = threading.Thread(target=hold); worker.start()
    try:
        assert entered.wait(1)
        row = record(resolved(g))
        assert row['observationReason'] == 'LIFECYCLE_BUSY'
        assert row['observedState'] == 'UNKNOWN' and row['phase'] == 'COMMAND_APPLIED'
    finally:
        release.set(); worker.join(2)
    assert not worker.is_alive()


def test_broken_observer_and_sink_do_not_change_future():
    g = SimpleNamespace(lifecycle=object()); f = resolved(g)
    assert record(f)['observationReason'] == 'STATE_UNAVAILABLE'
    def broken(**_): raise RuntimeError('sink failed')
    assert record(f, emit=broken) is None
    assert f.result() is g


def test_full_queue_and_database_failure_are_not_command_failure():
    r = Recorder(lambda _: (_ for _ in ()).throw(RuntimeError('DB unavailable')),
                 server='fixture', settings=dict(DEFAULT, queue_limit=16), start=False)
    f = resolved(group())
    for _ in range(17): record(f, emit=lambda **row: r.emit(row))
    assert r.queue.qsize() == 16 and r.counts['dropped_full'] == 1
    r.flush([r.queue.get()])
    assert r.error == 'TRACE_SINK_UNAVAILABLE' and f.exception() is None
    r.close()