from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
from datetime import datetime, timedelta, timezone
import json
import sqlite3
import time
from urllib.parse import urlencode
import uuid

import pytest

from informed_decision.approval_http import ApprovalHTTPClient
from informed_decision.disposition import DispositionRefused, Verb
from informed_decision.http_transport import TransportError
from informed_decision.memo import BindingLevel, Highlight, PacketItem, Principal, Scope, StepKind
from informed_decision.policy import PolicyError
from informed_decision.review import ReviewController, ReviewError
from informed_decision.store import Conflict, Store
from informed_decision.web import App, SESSION_COOKIE
from informed_decision.oidc import ORIGIN
from test_approval_component import component, signing_key
from test_browser_auth import call
from test_policy_client import policy
from test_skeleton import make_memo


@pytest.fixture
def review(component, tmp_path):
    client,engine,transport,session,_=component
    private=tmp_path/'private';private.mkdir(mode=0o700)
    store=Store(private/'review.sqlite')
    packet=store.put_document(b'<script>private-packet-sentinel</script>\nFull fixture terms.')
    memo=make_memo(approval_id='fixture',approval_binding_digest=client.get_approval('fixture')['binding']['digest'],
        question='Approve the synthetic factory delivery?', requested_act='deliver',
        highlights=(Highlight('h-1','doc-1','Disposable test target; no production action',required_ack=True),),
        packet=(PacketItem('doc-1','Whole fixture document',packet),),
        brief='Disposable browser/component exercise. private-brief-sentinel <img src=x onerror=alert(1)>')
    memo=replace(memo,binding=replace(memo.binding,principal=Principal(session.subject,'person','Fixture reviewer'),
        target=Scope('resource','fixture','Synthetic factory target','test')))
    store.save_memo(memo)
    controller=ReviewController(store,policy(),lambda s:ApprovalHTTPClient('https://approval.test',s,transport=transport))
    class Login:
        def session(self,sid):return session if sid=='fixture-session' else None
    app=App(Login(),controller,readiness=lambda:True)
    return controller,session,app,memo,engine,transport


def opened(review):
    controller,session,*_=review
    return controller.open(session,'memo-1')


def post(review,p,route,**params):
    _,session,app,*_=review
    return call(app,'/presentations/'+p.id+'/'+route,'POST',cookie=SESSION_COOKIE+'=fixture-session',
                origin=ORIGIN,body=urlencode({'csrf':session.csrf,**params}).encode())


def test_read_denial_precedes_disclosure_engine_read_and_presentation(review):
    controller,session,app,memo,engine,transport=review
    controller.policy.transport.change=lambda d:d.update(effect='deny')
    before=len(transport.calls)
    r=call(app,'/review',query='memo_id=memo-1',cookie=SESSION_COOKIE+'=fixture-session')
    assert r['status']==403 and 'private-brief-sentinel' not in r['body']
    assert len(transport.calls)==before and controller.store.evidence()==[]
    assert controller.store.policy_observations()[0]['outcome']=='policy_denied'


def test_no_session_or_wrong_named_recipient_cannot_check_or_render(review):
    controller,session,app,*_=review
    assert call(app,'/review',query='memo_id=memo-1')['status']==401
    with pytest.raises(ReviewError):controller.open(replace(session,subject='other'),'memo-1')
    assert controller.policy.transport.calls==[] and controller.store.policy_observations()==[]


@pytest.mark.parametrize('change', [{'locale':'de'}, {'binding_level':BindingLevel.AES},
    {'binding_level':BindingLevel.QES}, {'step_kind':StepKind.REVIEW}])
def test_unsupported_profile_cannot_misstate_locale_or_binding_level(review,change):
    controller,session,_,memo,*_=review
    controller.store.save_memo(memo.next_version(**change))
    with pytest.raises(ReviewError,match='unsupported_review_profile'):
        controller.open(session,memo.id)
    assert controller.policy.transport.calls==[] and controller.store.evidence()==[]


def test_open_persists_exact_policy_and_binding_but_not_browser_credential(review):
    controller,session,app,*_=review
    page=opened(review);content=controller.store.presentation_content(page.presentation.id)
    observed=controller.store.policy_observations()
    assert observed[0]['decision_attributable']==0
    assert content['policy_observation_id']==observed[0]['id']
    assert content['approval_binding']['digest']==page.memo.approval_binding_digest
    assert session.access_token not in json.dumps(observed)
    assert all('policy_observation_id' not in r['envelope'] for r in controller.store.evidence())
    response=call(app,'/presentations/'+page.presentation.id,cookie=SESSION_COOKIE+'=fixture-session')
    assert response['status']==200 and '&lt;img' in response['body'] and '<img src=x' not in response['body']
    assert '&lt;script&gt;private-packet-sentinel' in response['body']


@pytest.mark.parametrize('fault',['origin','csrf','duplicate','content-type','field'])
def test_post_form_refusals_precede_policy_and_side_effects(review,fault):
    controller,session,app,*_=review;p=opened(review).presentation
    body=urlencode({'csrf':session.csrf,'h0':'h-1'}).encode();origin=ORIGIN;ctype='application/x-www-form-urlencoded'
    if fault=='origin':origin='https://other.test'
    if fault=='csrf':body=b'csrf=wrong&h0=h-1'
    if fault=='duplicate':body+=b'&h0=other'
    if fault=='content-type':ctype='application/json'
    if fault=='field':body+=b'&subject=other'
    before=len(controller.policy.transport.calls)
    r=call(app,'/presentations/'+p.id+'/ack','POST',cookie=SESSION_COOKIE+'=fixture-session',body=body,origin=origin,content_type=ctype)
    assert r['status'] in (400,403) and len(controller.policy.transport.calls)==before
    assert not controller.store.presentation(p.id).acked_highlight_ids


def test_highlights_gate_accept_then_one_real_entry_and_original_view(review):
    controller,session,app,memo,engine,transport=review;p=opened(review).presentation
    op=str(uuid.uuid4())
    assert post(review,p,'act',verb='accept',operation_id=op)['status']==409
    assert not engine.get('fixture').entries
    assert post(review,p,'ack',h0='h-1')['status']==303
    assert post(review,p,'act',verb='accept',operation_id=op)['status']==303
    assert len(engine.get('fixture').entries)==1
    posts=sum(method=='POST' for method,_ in transport.calls)
    assert post(review,p,'act',verb='accept',operation_id=op)['status']==303
    assert sum(method=='POST' for method,_ in transport.calls)==posts==1
    page=controller.open(session,memo.id)
    assert page.presentation.id==p.id and page.intent['state']=='confirmed'
    response=call(app,'/presentations/'+p.id,cookie=SESSION_COOKIE+'=fixture-session')
    assert 'Approval entry recorded.' in response['body']
    assert 'Accept the complete request</button>' not in response['body']
    assert not any(url.endswith('/consume') for _,url in transport.calls)


@pytest.mark.parametrize('verb',[Verb.RETURN,Verb.DISCUSS,Verb.DECLINE])
def test_nonaccept_actions_preserve_vocabulary_without_engine_mutation(review,verb):
    controller,session,app,memo,engine,transport=review;p=opened(review).presentation
    controller.acknowledge(session,p.id,['h-1'])
    result=post(review,p,'act',verb=verb.value,operation_id=str(uuid.uuid4()),reason='clarification_needed' if verb is Verb.RETURN else '',note='Private response')
    assert result['status']==303 and not engine.get('fixture').entries
    assert controller.store.dispositions_for(p.id)[0].verb is verb
    response=call(app,'/presentations/'+p.id,cookie=SESSION_COOKIE+'=fixture-session')
    assert 'Private response' in response['body']
    assert all('Private response' not in r['envelope'] for r in controller.store.evidence())


def test_fresh_act_refusal_after_allowed_render_cannot_submit(review):
    controller,session,app,memo,engine,transport=review;p=opened(review).presentation
    controller.acknowledge(session,p.id,['h-1'])
    controller.policy.transport.change=lambda d:d.update(effect='deny')
    assert post(review,p,'act',verb='accept',operation_id=str(uuid.uuid4()))['status']==403
    assert not engine.get('fixture').entries and controller.store.dispositions_for(p.id)==[]


def test_policy_storage_failure_prevents_presentation(review):
    controller,session,*_=review
    with sqlite3.connect(controller.store.path) as db:
        db.execute("CREATE TRIGGER fail_observation BEFORE INSERT ON policy_observations BEGIN SELECT RAISE(ABORT,'disk failure'); END")
    with pytest.raises(sqlite3.IntegrityError):opened(review)
    assert controller.store.evidence()==[]


def test_revision_between_policy_and_present_cannot_render_unchecked_version(review):
    controller,session,_,memo,*_=review
    controller.policy.transport.change=lambda d:controller.store.save_memo(memo.next_version(brief='changed'))
    with pytest.raises(Conflict):opened(review)
    assert controller.store.evidence()==[]


def test_stale_presentation_and_wrong_actor_cannot_ack_or_act(review):
    controller,session,_,memo,engine,_=review;p=opened(review).presentation
    with pytest.raises(ReviewError):controller.acknowledge(replace(session,subject='other'),p.id,['h-1'])
    controller.store.save_memo(memo.next_version(brief='new version'))
    with pytest.raises(ReviewError):controller.acknowledge(session,p.id,['h-1'])
    with pytest.raises(ReviewError):controller.act(session,p.id,Verb.ACCEPT,operation_id='stale')
    assert not engine.get('fixture').entries


def test_expiry_after_last_engine_read_prevents_reservation_and_post(review):
    controller,session,_,memo,engine,transport=review;p=opened(review).presentation
    controller.acknowledge(session,p.id,['h-1'])
    original=transport.request;reads=0
    def request(method,url,**kwargs):
        nonlocal reads
        result=original(method,url,**kwargs)
        if method=='GET':
            reads+=1
            if reads==2:controller.clock=lambda:time.time()+40
        return result
    transport.request=request
    with pytest.raises(PolicyError):controller.act(session,p.id,Verb.ACCEPT,operation_id='late')
    assert not engine.get('fixture').entries
    assert controller.store.intent_for('fixture',session.subject)['state']=='prepared'


def test_parallel_clicks_submit_once(review):
    controller,session,_,memo,engine,transport=review;p=opened(review).presentation
    controller.acknowledge(session,p.id,['h-1'])
    def click(_):return controller.act(session,p.id,Verb.ACCEPT,operation_id='one-click')
    with ThreadPoolExecutor(max_workers=2) as pool:assert list(pool.map(click,range(2)))==[p.id,p.id]
    assert len(engine.get('fixture').entries)==1
    assert sum(method=='POST' for method,_ in transport.calls)==1


def test_lost_entry_reply_is_visible_and_never_retried(review):
    controller,session,app,memo,engine,transport=review;p=opened(review).presentation
    controller.acknowledge(session,p.id,['h-1'])
    original=transport.request
    def lose_reply(method,url,**kwargs):
        result=original(method,url,**kwargs)
        if method=='POST':raise TransportError('lost response')
        return result
    transport.request=lose_reply
    controller.act(session,p.id,Verb.ACCEPT,operation_id='lost')
    assert controller.store.intent_for('fixture',session.subject)['state']=='unresolved'
    response=call(app,'/presentations/'+p.id,cookie=SESSION_COOKIE+'=fixture-session')
    assert 'Submission outcome is not confirmed.' in response['body']
    assert 'Accept the complete request</button>' not in response['body']
    controller.act(session,p.id,Verb.ACCEPT,operation_id='new-click')
    assert len(engine.get('fixture').entries)==1 and sum(m=='POST' for m,_ in transport.calls)==1


def test_unknown_existing_entry_never_acquires_a_new_presentation_link(review):
    controller,session,_,memo,engine,transport=review;p=opened(review).presentation
    ApprovalHTTPClient('https://approval.test',session,transport=transport).add_entry('fixture')
    with pytest.raises(ReviewError,match='existing_entry_unlinked'):
        controller.act(session,p.id,Verb.ACCEPT,operation_id='unlinked')
    assert controller.store.intent_for('fixture',session.subject) is None
    assert sum(m=='POST' for m,_ in transport.calls)==1


def test_packet_download_rechecks_entitlement_and_forces_attachment(review):
    controller,session,app,*_=review;p=opened(review).presentation
    path='/presentations/'+p.id+'/packet/0'
    response=call(app,path,cookie=SESSION_COOKIE+'=fixture-session')
    assert response['status']==200
    assert dict(response['headers'])['Content-Type']=='application/octet-stream'
    assert dict(response['headers'])['Content-Disposition'].startswith('attachment;')
    controller.policy.transport.change=lambda d:d.update(effect='deny')
    denied=call(app,path,cookie=SESSION_COOKIE+'=fixture-session')
    assert denied['status']==403 and 'private-packet-sentinel' not in denied['body']


def test_audit_not_ready_blocks_accept_before_policy_or_post(review):
    controller,session,app,memo,engine,_=review;p=opened(review).presentation
    app.readiness=lambda:False
    assert post(review,p,'act',verb='accept',operation_id=str(uuid.uuid4()))['status']==503
    assert not engine.get('fixture').entries
    assert call(app,'/readyz')['status']==503


def test_renderer_release_cannot_silently_change_under_a_memo(review):
    controller,session,_,memo,*_=review
    controller.store.save_memo(memo.next_version(ui_release='informed-decision@0.1.0'))
    with pytest.raises(ReviewError,match='renderer_changed'):opened(review)
    assert controller.store.evidence()==[]


@pytest.mark.parametrize("padding", ["", " ", "\t\r\n", "\u00a0"])
def test_pasted_memo_id_opens_current_version(review, padding):
    controller, session, app, memo, engine, _ = review
    controller.store.save_memo(memo.next_version())
    response = call(app, '/review', cookie=SESSION_COOKIE+'=fixture-session',
                    query=urlencode({'memo_id': padding+memo.id+padding}))
    assert response['status'] == 200
    observations = controller.store.policy_observations()
    assert json.loads(observations[-1]['request'])['context']['memo_version'] == 2
    assert not engine.get('fixture').entries


@pytest.mark.parametrize("query", ["memo_id=", "memo_id=+", "memo_id=%60memo-1%60",
    "memo_id=memo-1&memo_id=memo-1", "memo_id=memo-1&extra=x", "memo_id=memo+1",
    "memo_id=https%3A%2F%2Fdecisions.coulomb.social%2Freview", "memo_id=%FF"])
def test_invalid_memo_input_explains_correction_before_policy(review, query):
    controller, _, app, _, engine, _ = review
    response = call(app, '/review', cookie=SESSION_COOKIE+'=fixture-session', query=query)
    assert response['status'] == 400
    assert 'Enter the memo identifier only' in response['body']
    assert not controller.store.policy_observations()
    assert not engine.get('fixture').entries


def add_memo(review, approval_id):
    controller,session,app,memo,engine,transport=review
    now=datetime.now(timezone.utc)
    engine.create({"actor":"synthetic-requester","principal":"factory-fixture","action":"deliver",
                   "purpose":"local-component-proof","target":{"resource":approval_id}},
                  {"not_before":(now-timedelta(seconds=1)).isoformat(),"expires_at":(now+timedelta(minutes=10)).isoformat()},
                  human_control=True,approval_id=approval_id)
    digest=controller.approval_factory(session).get_approval(approval_id)['binding']['digest']
    added=replace(memo,id='memo-'+approval_id,approval_id=approval_id,approval_binding_digest=digest,
                  question='Question for '+approval_id)
    controller.store.save_memo(added)
    return added


def groups(controller,session):
    return {r.memo_id:r.group for r in controller.overview(session)}


def test_overview_classifies_without_presenting_or_binding(review):
    controller,session,app,memo,engine,transport=review
    for name in ('declined','returned','closed'):add_memo(review,name)
    other=replace(memo,id='memo-other',binding=replace(memo.binding,principal=Principal('someone-else','person','Other')))
    controller.store.save_memo(other)
    before=len(controller.store.evidence())
    rows=controller.overview(session)
    assert len(controller.store.evidence())==before and not any(m=='POST' for m,_ in transport.calls)
    assert {r.memo_id:(r.group,r.engine_status) for r in rows}=={
        'memo-1':('open','requested'),'memo-declined':('open','requested'),
        'memo-returned':('open','requested'),'memo-closed':('open','requested')}
    assert rows[0].question and rows[0].history==()

    p=opened(review).presentation;controller.acknowledge(session,p.id,['h-1'])
    controller.act(session,p.id,Verb.ACCEPT,operation_id=str(uuid.uuid4()))
    for name,verb in (('declined',Verb.DECLINE),('returned',Verb.RETURN)):
        q=controller.open(session,'memo-'+name).presentation;controller.acknowledge(session,q.id,['h-1'])
        controller.act(session,q.id,verb,operation_id=str(uuid.uuid4()),
                       reasons=('clarification_needed',) if verb is Verb.RETURN else ())
    engine.revoke('closed')
    rows={r.memo_id:r for r in controller.overview(session)}
    assert {k:r.group for k,r in rows.items()}=={'memo-1':'accepted','memo-declined':'declined',
        'memo-returned':'returned','memo-closed':'closed'}
    assert rows['memo-1'].approved_at
    assert rows['memo-1'].engine_status=='approved' and rows['memo-closed'].engine_status=='revoked'
    assert [h['verb'] for h in rows['memo-declined'].history]==['decline']
    # A revision after a return reopens the question on the new version.
    controller.store.save_memo(replace(controller.store.memo('memo-returned'),version=2))
    assert groups(controller,session)['memo-returned']=='open'


def test_overview_redacts_rows_the_pdp_does_not_allow(review):
    controller,session,app,memo,engine,transport=review
    controller.policy.transport.change=lambda d:d.update(effect='deny')
    before=len(transport.calls)
    [row]=controller.overview(session)
    assert (row.memo_id,row.group,row.reason,row.question,row.history)==('memo-1','unavailable','policy_denied',None,())
    assert len(transport.calls)==before


def test_overview_degrades_one_row_when_engine_is_unreachable(review):
    controller,session,app,memo,engine,transport=review
    original=transport.request
    def down(method,url,**kwargs):
        raise TransportError('down')
    transport.request=down
    [row]=controller.overview(session)
    assert (row.group,row.engine_status,row.reason)==('open',None,'engine_unavailable')
    assert row.question==memo.question
    transport.request=original


def test_overview_surfaces_unconfirmed_submission(review):
    controller,session,app,memo,engine,transport=review;p=opened(review).presentation
    controller.acknowledge(session,p.id,['h-1'])
    original=transport.request
    def lose_reply(method,url,**kwargs):
        result=original(method,url,**kwargs)
        if method=='POST':raise TransportError('lost response')
        return result
    transport.request=lose_reply
    controller.act(session,p.id,Verb.ACCEPT,operation_id='lost')
    assert groups(controller,session)=={'memo-1':'attention'}


def test_overview_requires_a_live_human_session(review):
    controller,session,*_=review
    with pytest.raises(ReviewError) as error:
        controller.overview(replace(session,expires_at=0))
    assert error.value.code=='session_expired'


def home(review):
    return call(review[2],'/',cookie=SESSION_COOKIE+'=fixture-session')


def test_home_lists_open_then_accepted_memos_and_keeps_the_manual_form(review):
    controller,session,app,memo,engine,transport=review
    r=home(review)
    assert r['status']==200 and 'Open for you (1)' in r['body'] and 'Accepted (' not in r['body']
    assert '<a href="/review?memo_id=memo-1"><strong>Approve the synthetic factory delivery?</strong></a>' in r['body']
    assert 'Approval status: requested' in r['body'] and 'Open a decision review' in r['body']
    assert 'private-brief-sentinel' not in r['body'] and 'private-packet-sentinel' not in r['body']
    assert len(controller.store.evidence())==0
    p=opened(review).presentation;controller.acknowledge(session,p.id,['h-1'])
    controller.act(session,p.id,Verb.ACCEPT,operation_id=str(uuid.uuid4()))
    r=home(review)
    assert 'Accepted (1)' in r['body'] and 'Open for you' not in r['body']
    assert 'Approval status: approved' in r['body'] and 'Entry recorded ' in r['body']
    assert '<li>Accept · version 1 · ' in r['body'] and 'confirmed' in r['body']


def test_home_escapes_memo_text_and_redacts_refused_rows(review):
    controller,session,app,memo,engine,transport=review
    hostile=add_memo(review,'hostile')
    controller.store.save_memo(replace(hostile,version=2,question='<img src=x onerror=alert(1)>'))
    r=home(review)
    assert '<img src=x' not in r['body'] and '&lt;img src=x onerror=alert(1)&gt;' in r['body']
    controller.policy.transport.change=lambda d:d.update(effect='deny')
    r=home(review)
    assert r['status']==200 and 'Not available (2)' in r['body']
    assert 'The permission service refused access to this memo.' in r['body']
    assert 'Approve the synthetic factory delivery?' not in r['body'] and 'onerror' not in r['body']


def test_home_without_session_shows_sign_in_only(review):
    controller,session,app,*_=review
    r=call(app,'/')
    assert 'Sign in with KeyCape' in r['body'] and 'Your decisions' not in r['body']
    assert controller.store.policy_observations()==[]


def test_home_keeps_the_memo_form_when_the_overview_fails(review,monkeypatch):
    controller,session,app,*_=review
    def broken(subject):raise sqlite3.OperationalError('disk')
    monkeypatch.setattr(controller.store,'memos_for',broken)
    r=home(review)
    assert r['status']==200 and 'could not be listed right now' in r['body'] and 'Open a decision review' in r['body']


def only(action,effect='deny'):
    return lambda d:d.update(effect=effect) if d['binding']['action']==action else None


def test_overview_asks_list_and_a_list_allow_never_opens_the_memo(review):
    controller,session,app,memo,engine,transport=review
    controller.policy.transport.change=only('read')
    r=home(review)
    assert 'Open for you (1)' in r['body'] and 'Approve the synthetic factory delivery?' in r['body']
    assert [json.loads(o['request'])['action'] for o in controller.store.policy_observations()]==['list']
    opened_=call(app,'/review',query='memo_id=memo-1',cookie=SESSION_COOKIE+'=fixture-session')
    assert opened_['status']==403 and 'private-brief-sentinel' not in opened_['body']
    assert controller.store.evidence()==[]


def test_list_row_discloses_no_brief_packet_highlight_or_note(review):
    controller,session,app,memo,engine,transport=review;p=opened(review).presentation
    controller.acknowledge(session,p.id,['h-1'])
    controller.act(session,p.id,Verb.DISCUSS,operation_id=str(uuid.uuid4()),note='private-note-sentinel')
    controller.policy.transport.change=only('read')
    [row]=controller.overview(session)
    assert row.group=='returned' and row.history[0]['verb']=='discuss' and 'note' not in row.history[0]
    body=home(review)['body']
    for sentinel in ('private-brief-sentinel','private-packet-sentinel','private-note-sentinel',
                     'Disposable test target','Full fixture terms'):
        assert sentinel not in body


def test_a_list_deny_redacts_even_when_read_would_allow(review):
    controller,session,*_=review
    controller.policy.transport.change=only('list')
    [row]=controller.overview(session)
    assert (row.group,row.question,row.reason)==('unavailable',None,'policy_denied')