"""Read-only WFP contract and cleanup tests; no filter mutations."""
import copy
import ctypes as C
import sys
import unittest
from pathlib import Path
from unittest.mock import Mock
sys.path.insert(0,str(Path(__file__).resolve().parent.parent/'apps/network-broker'))
from windows_wfp_probe import ReadOnlyWfp,verify_observed,Filter
PROVIDER='2a7bcbf2-e4f5-4e94-bd33-d5442a3bef50'
SUBLAYER='403a69b8-fc7c-460a-a247-72e0d72e6bf8'
def contract():
    expected=[];observed=[]
    for index in range(6):
        conditions=[{'field':'d78e1e87-8644-4ea5-9437-d809ecefc971','match':'equal','type':'FWP_BYTE_BLOB_TYPE','value':'01020304'}, {'field':'3971ef2b-623e-4f9a-8cb1-6e79b806b9a7','match':'equal','type':'FWP_UINT8','value':index}]
        expected.append({'layer':'layer-'+str(index),'action':'FWP_ACTION_BLOCK','clearActionRight':False,'conditions':conditions})
        observed.append({'provider':PROVIDER,'sublayer':SUBLAYER,'layer':'layer-'+str(index),'flags':1,'action':0x1001,'conditions':copy.deepcopy(conditions)})
    return {'filters':expected},observed
class WfpProbeTests(unittest.TestCase):
    def test_exact_contract_and_reordered_conditions(self):
        plan,items=contract();items.reverse()
        for item in items:item['conditions'].reverse()
        self.assertTrue(verify_observed(plan,items,PROVIDER,SUBLAYER)['verified'])
    def test_missing_duplicate_or_extra_filters(self):
        plan,items=contract()
        for bad in [items[:-1],items+items[:1],items[:3]+items[:1]]:
            self.assertFalse(verify_observed(plan,bad,PROVIDER,SUBLAYER)['verified'])
    def test_blob_type_match_layer_and_numeric_mismatch(self):
        plan,items=contract()
        for key,value in [('type','FWP_UINT32'),('value','deadbeef'),('match','not-equal')]:
            bad=copy.deepcopy(items);bad[0]['conditions'][0][key]=value
            self.assertEqual(verify_observed(plan,bad,PROVIDER,SUBLAYER)['reason'],'filter-condition-mismatch')
        bad=copy.deepcopy(items);bad[0]['layer']='other-layer';self.assertFalse(verify_observed(plan,bad,PROVIDER,SUBLAYER)['verified'])
    def test_disabled_nonpersistent_permit_softblock_or_wrong_owner(self):
        plan,items=contract()
        for key,value in [('flags',0),('flags',33),('flags',9),('action',0x1002),('provider','11111111-1111-1111-1111-111111111111'),('sublayer','22222222-2222-2222-2222-222222222222')]:
            bad=copy.deepcopy(items);bad[0][key]=value
            self.assertEqual(verify_observed(plan,bad,PROVIDER,SUBLAYER)['reason'],'filter-metadata-mismatch')
    def test_engine_denial_is_unverified_and_not_closed_or_enumerated(self):
        probe=ReadOnlyWfp.__new__(ReadOnlyWfp);probe.api=Mock();probe.api.FwpmEngineOpen0.return_value=5
        self.assertEqual(probe.inspect(contract()[0],PROVIDER,SUBLAYER),{'verified':False,'reason':'engine-unavailable','code':5})
        probe.api.FwpmFilterCreateEnumHandle0.assert_not_called();probe.api.FwpmEngineClose0.assert_not_called()
    def test_enum_handle_denial_closes_opened_engine(self):
        probe=ReadOnlyWfp.__new__(ReadOnlyWfp);probe.api=Mock()
        def opened(_name,_auth,_identity,_session,handle):handle._obj.value=123;return 0
        probe.api.FwpmEngineOpen0.side_effect=opened;probe.api.FwpmFilterCreateEnumHandle0.return_value=5
        self.assertEqual(probe.inspect(contract()[0],PROVIDER,SUBLAYER)['code'],5)
        probe.api.FwpmEngineClose0.assert_called_once();probe.api.FwpmFilterDestroyEnumHandle0.assert_not_called()
    def test_empty_enumeration_cleans_handle_and_engine(self):
        probe=ReadOnlyWfp.__new__(ReadOnlyWfp);probe.api=Mock()
        def opened(_name,_auth,_identity,_session,handle):handle._obj.value=123;return 0
        def enumeration(_engine,_template,handle):handle._obj.value=456;return 0
        def empty(_engine,_handle,_limit,_entries,count):count._obj.value=0;return 0
        probe.api.FwpmEngineOpen0.side_effect=opened;probe.api.FwpmFilterCreateEnumHandle0.side_effect=enumeration;probe.api.FwpmFilterEnum0.side_effect=empty
        self.assertEqual(probe.inspect(contract()[0],PROVIDER,SUBLAYER)['reason'],'filter-count-mismatch')
        probe.api.FwpmFilterDestroyEnumHandle0.assert_called_once();probe.api.FwpmEngineClose0.assert_called_once()
    def test_oversized_native_page_frees_memory_and_closes_handles(self):
        probe=ReadOnlyWfp.__new__(ReadOnlyWfp);probe.api=Mock()
        probe.api.FwpmEngineOpen0.side_effect=lambda *args:(setattr(args[-1]._obj,'value',123) or 0)
        probe.api.FwpmFilterCreateEnumHandle0.side_effect=lambda *args:(setattr(args[-1]._obj,'value',456) or 0)
        storage=(C.POINTER(Filter)*129)()
        def oversized(_engine,_handle,_limit,entries,count):
            C.cast(entries,C.POINTER(C.POINTER(C.POINTER(Filter))))[0]=C.cast(storage,C.POINTER(C.POINTER(Filter)))
            count._obj.value=129;return 0
        probe.api.FwpmFilterEnum0.side_effect=oversized
        self.assertEqual(probe.inspect(contract()[0],PROVIDER,SUBLAYER)['reason'],'enumeration-limit')
        probe.api.FwpmFreeMemory0.assert_called_once();probe.api.FwpmFilterDestroyEnumHandle0.assert_called_once();probe.api.FwpmEngineClose0.assert_called_once()
    def test_enumeration_error_cleans_handles(self):
        probe=ReadOnlyWfp.__new__(ReadOnlyWfp);probe.api=Mock()
        probe.api.FwpmEngineOpen0.side_effect=lambda *args:(setattr(args[-1]._obj,'value',123) or 0)
        probe.api.FwpmFilterCreateEnumHandle0.side_effect=lambda *args:(setattr(args[-1]._obj,'value',456) or 0)
        probe.api.FwpmFilterEnum0.return_value=5
        self.assertEqual(probe.inspect(contract()[0],PROVIDER,SUBLAYER)['reason'],'enumeration-failed')
        probe.api.FwpmFilterDestroyEnumHandle0.assert_called_once();probe.api.FwpmEngineClose0.assert_called_once()
if __name__=='__main__':unittest.main()