72 lines
3.3 KiB
Python
72 lines
3.3 KiB
Python
import ctypes as C
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'apps/network-broker'))
|
|
from windows_wfp_topology import TopologyObserver, CALLBACK
|
|
|
|
class ObserverTests(unittest.TestCase):
|
|
def observer(self, denied=False):
|
|
calls = []
|
|
class Api:
|
|
def FwpmEngineOpen0(self, *_args):
|
|
_args[-1]._obj.value = 9
|
|
return 0
|
|
def FwpmFilterSubscribeChanges0(self, _engine, _subscription, _callback, _context, handle):
|
|
calls.append('filter-subscribe')
|
|
handle._obj.value = 10
|
|
return 0
|
|
def FwpmSubLayerSubscribeChanges0(self, _engine, _subscription, _callback, _context, handle):
|
|
calls.append('sublayer-subscribe')
|
|
if denied:
|
|
return 5
|
|
handle._obj.value = 11
|
|
return 0
|
|
def FwpmFilterUnsubscribeChanges0(self, *_args): calls.append('filter-unsubscribe')
|
|
def FwpmSubLayerUnsubscribeChanges0(self, *_args): calls.append('sublayer-unsubscribe')
|
|
def FwpmEngineClose0(self, *_args): calls.append('close')
|
|
observer = TopologyObserver.__new__(TopologyObserver)
|
|
observer.api = Api()
|
|
observer.engine = C.c_void_p()
|
|
observer.handles = []
|
|
observer.changed = threading.Event()
|
|
observer.callback = CALLBACK(lambda *_args: observer.changed.set())
|
|
return observer, calls
|
|
|
|
def test_second_subscription_failure_releases_first_and_engine(self):
|
|
observer, calls = self.observer(True)
|
|
self.assertEqual(observer.start()['code'], 5)
|
|
self.assertTrue(observer.changed.is_set())
|
|
self.assertEqual(calls, ['filter-subscribe', 'sublayer-subscribe', 'filter-unsubscribe', 'close'])
|
|
self.assertEqual(observer.handles, [])
|
|
|
|
def test_actual_callback_invalidates_and_never_claims_full_visibility(self):
|
|
observer, calls = self.observer()
|
|
result = observer.start()
|
|
self.assertFalse(result['allObjectsReadable'])
|
|
observer.callback(None, None)
|
|
self.assertEqual(observer.snapshot([])['reason'], 'observer-unavailable')
|
|
observer.close()
|
|
self.assertEqual(calls[-3:], ['sublayer-unsubscribe', 'filter-unsubscribe', 'close'])
|
|
observer.close()
|
|
self.assertEqual(calls.count('close'), 1)
|
|
|
|
def test_authorized_readback_cannot_invent_visibility_and_change_is_sticky(self):
|
|
from types import SimpleNamespace
|
|
observer, _calls = self.observer()
|
|
rows=[{'key':str(index),'sublayer':'owned','provider':'provider'} for index in range(8)]
|
|
observer.revalidate=lambda _layers:{'complete':True,'allObjectsReadable':False,'filters':rows}
|
|
observer.reader=SimpleNamespace(inspect=lambda *_args:{'verified':True})
|
|
result=observer.authorized_snapshot({},'provider','owned',[])
|
|
self.assertFalse(result['allObjectsReadable'])
|
|
self.assertTrue(all(row['ownedVerified'] for row in result['filters']))
|
|
def raced(*_args):
|
|
observer.changed.set()
|
|
return {'verified':True}
|
|
observer.reader.inspect=raced
|
|
self.assertEqual(observer.authorized_snapshot({},'provider','owned',[])['reason'],'owned-readback-unavailable')
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|