import asyncio
import time
from dataclasses import replace

import pytest

from hub_core.security.facts import AccountObservation, TenantObservation, PlatformFacts
from hub_core.security.identity import AccessFailure
from test_access_boundary import Owners


class Readers:
    def __init__(self):
        self.actor = Owners().actor
        self.account = AccountObservation(self.actor.issuer,self.actor.subject,'owner:platform',
            True,True,time.time(),'account:1',frozenset({'agent:root'}))
        self.tenant = TenantObservation('tenant:platform',True,time.time(),'tenant:1')
        self.calls = []

    async def read(self,**kwargs):
        self.calls.append(kwargs)
        return self.tenant if 'identifier' in kwargs else self.account

    def facts(self):
        return PlatformFacts(accounts=self,tenants=self,account_tenant='owner:platform',
                             mapping_evidence='review:platform-mapping')


def test_live_join_preserves_oldest_observation_and_detects_revocation():
    reader = Readers()
    facts = reader.facts()
    async def run():
        first = await facts.resolve(reader.actor,'/docs')
        assert first.checked_at == min(reader.account.checked_at,reader.tenant.checked_at)
        assert first.root_entitled and first.account_active and first.target_tenant_active
        assert first.producer_addresses == frozenset({'agent:root'})
        reader.account = replace(reader.account,root_entitled=False,evidence_id='account:2')
        second = await facts.resolve(reader.actor,'/docs')
        assert not second.root_entitled and second.evidence_id != first.evidence_id
        reader.account = replace(reader.account,account_active=False)
        reader.tenant = replace(reader.tenant,active=False)
        third = await facts.resolve(reader.actor,'/docs')
        assert not third.account_active and not third.actor_tenant_active and not third.target_tenant_active
    asyncio.run(run())
    assert len(reader.calls) == 6
    assert reader.calls[0] == dict(issuer=reader.actor.issuer,subject=reader.actor.subject,tenant='owner:platform')


@pytest.mark.parametrize('source,changes',[
    ('account',{'subject':'other'}),('account',{'issuer':'other'}),
    ('account',{'tenant':'tenant:platform'}),('tenant',{'identifier':'other'}),
    ('account',{'account_active':'true'}),('account',{'root_entitled':1}),
    ('tenant',{'active':None}),('account',{'checked_at':1}),
    ('tenant',{'checked_at':float('nan')}),('account',{'checked_at':float('inf')}),
    ('account',{'checked_at':time.time()+100}),('tenant',{'evidence_id':''}),
    ('account',{'producer_addresses':['agent:root']}),
])
def test_malformed_mismatched_or_stale_observation_refuses(source,changes):
    reader = Readers()
    setattr(reader,source,replace(getattr(reader,source),**changes))
    with pytest.raises(AccessFailure,match='owner_facts_unavailable_or_untrusted'):
        asyncio.run(reader.facts().resolve(reader.actor,'/docs'))


def test_owner_failure_has_no_cached_fallback_or_private_error():
    reader = Readers()
    facts = reader.facts()
    asyncio.run(facts.resolve(reader.actor,'/docs'))
    async def unavailable(**kwargs):
        raise RuntimeError('private owner details')
    reader.read = unavailable
    with pytest.raises(AccessFailure) as error:
        asyncio.run(facts.resolve(reader.actor,'/docs'))
    assert error.value.status == 503 and 'private' not in str(error.value)


@pytest.mark.parametrize('changes',[{'principal_type':'service'},{'tenant':'tenant:other'}])
def test_unadmitted_principal_never_calls_owners(changes):
    reader = Readers()
    with pytest.raises(AccessFailure) as error:
        asyncio.run(reader.facts().resolve(replace(reader.actor,**changes),'/docs'))
    assert error.value.status == 403 and not reader.calls


def test_slow_join_does_not_refresh_account_observation(monkeypatch):
    from types import SimpleNamespace
    from hub_core.security import facts as module
    reader = Readers()
    now = time.time()
    reader.account = replace(reader.account,checked_at=now-4.5)
    reader.tenant = replace(reader.tenant,checked_at=now)
    clock = SimpleNamespace(time=lambda: now)
    monkeypatch.setattr(module,'time',clock)
    async def read(**kwargs):
        if 'identifier' in kwargs:
            clock.time = lambda: now+1
            return reader.tenant
        return reader.account
    reader.read = read
    with pytest.raises(AccessFailure):
        asyncio.run(reader.facts().resolve(reader.actor,'/docs'))


@pytest.mark.parametrize('source,changes',[
    ('account',{'root_entitled':False}),('account',{'account_active':False}),
    ('tenant',{'active':False}),
])
def test_next_authorization_denies_withdrawn_owner_state(source,changes):
    reader = Readers()
    owners = Owners()
    controller = owners.controller()
    controller.facts = reader.facts()
    async def run():
        await controller.authorize('verified-root','hub.read','/docs','first','digest')
        setattr(reader,source,replace(getattr(reader,source),**changes))
        with pytest.raises(AccessFailure) as error:
            await controller.authorize('verified-root','hub.read','/docs','next','digest')
        assert error.value.status == 403
        assert len(owners.requests) == 1
    asyncio.run(run())


def test_hung_owner_read_is_bounded():
    reader = Readers()
    async def hung(**kwargs):
        await asyncio.Event().wait()
    reader.read = hung
    with pytest.raises(AccessFailure,match='owner_facts_unavailable_or_untrusted'):
        asyncio.run(reader.facts().resolve(reader.actor,'/docs'))