NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

test_scatterEventTracing.py (4385B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\script\sys\test\test_scatterEventTracing.py
      2 import mock
      3 import uthread
      4 import bluepy
      5 from carbon.common.script.sys.serviceConst import SERVICE_RUNNING
      6 from testsuites.testcases import ROLE_SERVER, ServerTestCase, init_prefs
      7 init_prefs(ROLE_SERVER)
      8 from carbon.common.script.sys import serviceManager
      9 from carbon.common.script.sys.service import Service
     10 EVENT = 'OnScatterEventTracingCoverage'
     11 FLAG = bluepy.STACKLESS_TRACING_ENABLED_KEY
     12 _SERVICE_MANAGER = None
     13 
     14 def _service_manager():
     15     global _SERVICE_MANAGER
     16     if _SERVICE_MANAGER is None:
     17         _SERVICE_MANAGER = serviceManager.ServiceManager()
     18     return _SERVICE_MANAGER
     19 
     20 
     21 class _Listener(object):
     22 
     23     def __init__(self):
     24         self.calls = []
     25 
     26     def OnScatterEventTracingCoverage(self, *args, **keywords):
     27         self.calls.append((args, keywords))
     28 
     29 
     30 class _Observer(_Listener):
     31     __notifyevents__ = [EVENT]
     32 
     33 
     34 class _ServiceListener(Service, _Listener):
     35     __guid__ = 'svc.scatterEventTracingCoverage'
     36     __notifyevents__ = [EVENT]
     37 
     38     def __init__(self):
     39         Service.__init__(self)
     40         _Listener.__init__(self)
     41         self.state = SERVICE_RUNNING
     42 
     43 
     44 class _ScatterTestCase(ServerTestCase):
     45     listener_class = None
     46 
     47     def setUp(self):
     48         super(_ScatterTestCase, self).setUp()
     49         self.serviceManager = _service_manager()
     50         self.listener = self.listener_class()
     51         self.serviceManager.RegisterNotify(self.listener)
     52 
     53     def tearDown(self):
     54         self.serviceManager.UnregisterNotify(self.listener)
     55         super(_ScatterTestCase, self).tearDown()
     56 
     57 
     58 class ObserverPathTests(_ScatterTestCase):
     59     listener_class = _Observer
     60 
     61     def _worker_keywords(self, keywords = None):
     62         with mock.patch.object(uthread, 'workerWithoutTheStars') as worker:
     63             self.serviceManager.ScatterEventWithoutTheStars(EVENT, (), keywords)
     64         self.assertEqual(1, worker.call_count, 'the observer should have been dispatched once')
     65         return worker.call_args[0][3]
     66 
     67     def test_declines_tracing_by_default(self):
     68         self.assertFalse(self._worker_keywords()[FLAG])
     69 
     70     def test_traces_when_the_sender_opts_in(self):
     71         self.assertTrue(self._worker_keywords({FLAG: True})[FLAG])
     72 
     73     def test_the_caller_keywords_are_not_mutated(self):
     74         keywords = {'alpha': 1}
     75         passed = self._worker_keywords(keywords)
     76         self.assertEqual({'alpha': 1}, keywords, "the caller's dict was modified in place")
     77         self.assertEqual(1, passed['alpha'], "the listener's own keywords were dropped")
     78 
     79     def test_the_listener_never_sees_the_flag(self):
     80         with mock.patch.object(bluepy, 'TaskletExt') as taskletExt:
     81             self.serviceManager.ScatterEventWithoutTheStars(EVENT, (7,), {'alpha': 1})
     82         self.assertFalse(taskletExt.call_args[1]['stackless_tracing_enabled'])
     83         applied = taskletExt.return_value.call_args
     84         self.assertEqual((7,), applied[0])
     85         self.assertEqual({'alpha': 1}, applied[1])
     86 
     87 
     88 class ServicePathTests(_ScatterTestCase):
     89     listener_class = _ServiceListener
     90 
     91     def _worker_call(self, keywords = None):
     92         with mock.patch.object(uthread, 'worker') as worker:
     93             self.serviceManager.ScatterEventWithoutTheStars(EVENT, (), keywords)
     94         self.assertEqual(1, worker.call_count, 'the service should have been dispatched once')
     95         return worker.call_args
     96 
     97     def test_declines_tracing_by_default(self):
     98         self.assertFalse(self._worker_call()[1]['stackless_tracing_enabled'])
     99 
    100     def test_traces_when_the_sender_opts_in(self):
    101         self.assertTrue(self._worker_call({FLAG: True})[1]['stackless_tracing_enabled'])
    102 
    103     def test_the_flag_stays_out_of_the_listener_keywords(self):
    104         listenerKeywords = self._worker_call({'alpha': 1})[0][6]
    105         self.assertNotIn(FLAG, listenerKeywords)
    106         self.assertEqual({'alpha': 1}, listenerKeywords)
    107 
    108     def test_the_listener_never_sees_the_flag(self):
    109         with mock.patch.object(bluepy, 'TaskletExt') as taskletExt:
    110             self.serviceManager.ScatterEventWithoutTheStars(EVENT, (7,), {'alpha': 1})
    111         self.assertFalse(taskletExt.call_args[1]['stackless_tracing_enabled'])
    112         self.serviceManager.MollycoddledUthread(*taskletExt.return_value.call_args[0])
    113         self.assertEqual([((7,), {'alpha': 1})], self.listener.calls)