Best Python code snippet using playwright-python
models.py
Source:models.py  
1# flake8: noqa2from django.core.servers.basehttp import FileWrapper3from django.http import HttpResponse, Http4044from envirocon.game.installed_games import InstalledGames, GameInterface5from funding_interventions import funding_points, funding_autoshocks6import os.path7fileroot = os.path.abspath(".") + "/game_many/files/"8class ConflictAssessment(GameInterface):9    def pages(self):10        return ('index', 'page2')11    def template(self, page_id=None, public_state=None):12        game_context = {'sampledata': "hello"}  # , documents:documents}13        if page_id == "country_narrative":14            return('file', servefile("ConflictAssessment_CountryNarrative.pdf",15                                     "Country_Narrative.pdf"))16        if page_id == "page2":17            return ('game_many/conflict_assessment.html', game_context)18        return ('game_many/conflict_assessment_intro.html', game_context)19    def variables(self, page_id=None):20        return ['conflict_assessment']21    def resources(self, game_state, onopen=False, onclosed=False):22        if onopen:23            return [{"page_id": 'country_narrative',24                     "type": 'file',25                     "title": 'Country Narrative.pdf',26                     }]27        else:28            return []29InstalledGames.register_game('conflict_assessment',30                             'Conflict Assessment',31                             ConflictAssessment())32class ExplainYourReportSelection(GameInterface):33    """34    Week 1, Explain Your Report Selection35    """36    def pages(self):37        return ('index', 'page2')38    def template(self, page_id=None, public_state=None):39        game_context = {'sampledata': "hello"}40        if page_id == "page2":41            return ('game_many/explain_your_report_selection.html',42                    game_context)43        return ('game_many/explain_your_report_selection_intro.html',44                game_context)45    def variables(self, page_id=None):46        return ['explain_your_report_selection']  # ,'additional_information']47InstalledGames.register_game('explain_your_report_selection',48                             'Explain Your Report Selection',49                             ExplainYourReportSelection())50class RecommendingInterventions(GameInterface):51    """52    Week 3, Recommending Interventions53    """54    def pages(self):55        return ('index', 'page2')56    def template(self, page_id=None, public_state=None):57        game_context = {'sampledata': "hello"}58        if page_id == "watching_brief":59            return(60                'file',61                servefile("RecommendingInterventions_FirstWatchingBrief.pdf",62                          "First_Watching_Brief.pdf"))63        if page_id == "page2":64            return ('game_many/recommending_interventions.html',65                    game_context)66        return ('game_many/recommending_interventions_intro.html',67                game_context)68    def variables(self, page_id=None):69        return ['recommending_interventions']70    def resources(self, game_state, onopen=False, onclosed=False):71        if onopen:72            return [{"page_id": 'watching_brief',73                     "type": 'file',74                     "title": 'First Watching Brief.pdf',75                     }]76        else:77            return []78InstalledGames.register_game('recommending_interventions',79                             'Recommending Interventions',80                             RecommendingInterventions())81class FundingInterventions(GameInterface):82    """83    Week 4, Funding Interventions84    """85    def pages(self):86        return ('index', 'page2')87    def template(self, page_id=None, public_state=None):88        game_context = {'sampledata': "hello"}89        if page_id == "page2":90            return ('game_many/funding_interventions.html', game_context)91        return ('game_many/funding_interventions_intro.html', game_context)92    def variables(self, page_id=None):93        return ['funding_interventions']94    def public_variables(self):95        return ['funding_interventions']96    def consequences(self, game_state):97        return DonorsConference().resources(game_state, onopen=True)98InstalledGames.register_game('funding_interventions',99                             'Funding Interventions',100                             FundingInterventions())101class TrackingYourProjects(GameInterface):102    """103    Week 4, Tracking Your Projects104    """105    def pages(self):106        return ('index', 'page2',)107    def template(self, page_id=None, public_state=None):108        funded = {}109        if public_state['resources_by_app'].has_key('tracking_your_projects'):110            if public_state['resources_by_app'][111                    'tracking_your_projects'].has_key('funded'):112                funded = public_state['resources_by_app'][113                    'tracking_your_projects']['funded']['value']114        game_context = {'funded': funded}115        if page_id == "page2":116            return ('game_many/tracking_your_projects.html', game_context)117        return ('game_many/tracking_your_projects_intro.html', game_context)118    def variables(self, page_id=None):119        return ['tracking_your_projects']120    def resources(self, game_state, onopen=False, onclosed=False):121        if onopen and game_state.has_key('funding_interventions'):122            funded = game_state['funding_interventions']123            return [{"page_id": 'funded',124                     "type": 'data',125                     'value': funded.keys()}]126    def autoshock(self, game_state):127        # to appear in Results Framework128        return funding_autoshocks('funding_interventions',129                                  game_state['funding_interventions'])130InstalledGames.register_game('tracking_your_projects',131                             'Tracking Your Projects',132                             TrackingYourProjects())133class ResultsFramework(GameInterface):134    """135    Week 5, Results Framework136    """137    def pages(self):138        return ('index', 'page2',)139    def template(self, page_id=None, public_state=None):140        game_context = {'sampledata': "hello"}141        if page_id == "overview":142            return('file', servefile("ResultsFramework_Overview.pdf",143                                     "Results_Framework_Overview.pdf"))144        if page_id == "matrices":145            return(146                'file',147                servefile("ResultsFramework_ClusterMatrices.xls",148                          "Results_Framework_Cluster_Matrices.xls",149                          "excel/ms-excel"))150        if page_id == "page2":151            return ('game_many/results_framework.html', game_context)152        return ('game_many/results_framework_intro.html', game_context)153    def variables(self, page_id=None):154        return ['results_framework']155    def resources(self, game_state, onopen=False, onclosed=False):156        if onopen:157            return [{"page_id": 'overview',158                     "type": 'file',159                     "title": 'Results Framework Overview.pdf',160                     },161                    {"page_id": 'matrices',162                     "type": 'file',163                     "title": 'Cluster Matrices.xls'}164                    ]165        else:166            return []167InstalledGames.register_game('results_framework',168                             'Results Framework',169                             ResultsFramework())170class DonorsConference(GameInterface):171    """172    Week 6, Donors Conference & More Interventions173    """174    def pages(self):175        return ('index', 'page2',)176    def template(self, page_id=None, public_state=None):177        points = 0178        ineligible = {}179        if public_state['resources_by_app'].has_key('donors_conference'):180            if public_state['resources_by_app']['donors_conference'].has_key(181                    'week4points'):182                points = public_state['resources_by_app'][183                    'donors_conference']['week4points']['value']184            if public_state['resources_by_app']['donors_conference'].has_key(185                    'ineligible'):186                ineligible = public_state['resources_by_app'][187                    'donors_conference']['ineligible']['value']188        #, 'wb2':wb2}189        game_context = {'week4points': points, 'ineligible': ineligible}190        if page_id == "summary":191            return('file',192                   servefile("DonorsConference_DonorsConferenceSummary.pdf",193                             "Donors_Conference_Summary.pdf"))194        if page_id == "watching_brief_1":195            return('file', servefile("DonorsConference_WatchingBrief1.pdf",196                                     "Second_Watching_Brief.pdf"))197        if page_id == "watching_brief_2":198            return('file', servefile("DonorsConference_WatchingBrief2.pdf",199                                     "Second_Watching_Brief.pdf"))200        if page_id == "watching_brief_3":201            return('file', servefile("DonorsConference_WatchingBrief3.pdf",202                                     "Second_Watching_Brief.pdf"))203        if page_id == "page2":204            return ('game_many/donors_conference.html', game_context)205        return ('game_many/donors_conference_intro.html', game_context)206    def variables(self, page_id=None):207        return ['donors_conference']208    def public_variables(self):209        return ['donors_conference']210    def resources(self, game_state, onopen=False, onclosed=False):211        if onopen and game_state.has_key('funding_interventions'):212            funded = game_state['funding_interventions']213            # certain interventions, if funded in week 4, are ineligible to be214            # funded in week 6215            ineligible = [intervention for intervention in funded216                          if intervention in ['timber-low',217                                              'nomadic-low',218                                              'agricultural-low',219                                              'desertification-low',220                                              'habitat-low',221                                              'water-low', ]]222            # teams who choose the 3-point option in any of the first 3223            # categories automatically get wb2224            wb2_choices = [intervention for intervention in funded225                           if intervention in ['water-supply-low',226                                               'sanitation-low',227                                               'waste-low']]228            autowb2 = False229            if wb2_choices != []:230                autowb2 = True231            unlock_wb2 = False232            points = funding_points(funded)233            if autowb2 or (points > 17 and points < 30):234                wb = {"page_id": 'watching_brief_2', "type":235                      'file', "title": 'Second Watching Brief.pdf'}236            elif points < 17:237                wb = {"page_id": 'watching_brief_1', "type":238                      'file', "title": 'Second Watching Brief.pdf'}239            else:  # 30 or more points240                wb = {"page_id": 'watching_brief_3', "type":241                      'file', "title": 'Second Watching Brief.pdf'}242                # yes, the third watching brief unlocks the WB2 map layers.243                # confusing but correct.244                unlock_wb2 = True245            return [{"page_id": 'summary',246                     "type": 'file',247                     "title": 'Donors Conference Summary.pdf',248                     },249                    wb,250                    {"page_id": 'week4points',251                        "type": 'data', 'value': points},252                    {"page_id": 'ineligible', "type":253                        'data', "value": ineligible},254                    {"page_id": 'wb2', "type": 'data', "value": unlock_wb2},255                    ]256        else:257            return []258    def consequences(self, game_state):259        return FinalPaper().resources(game_state, onopen=True)260    def autoshock(self, game_state):261        return funding_autoshocks('donors_conference',262                                  game_state['donors_conference'])263InstalledGames.register_game('donors_conference',264                             'Donors Conference',265                             DonorsConference())266class FinalPaper(GameInterface):267    """268    Week 8, Final Paper269    """270    def pages(self):271        return ('index', 'page2', 'page3')272    def template(self, page_id=None, public_state=None):273        game_context = {}274        if page_id == "watching_brief_1":275            return('file', servefile("FinalPaper_WatchingBrief1.pdf",276                                     "Third_Watching_Brief.pdf"))277        if page_id == "watching_brief_2":278            return('file', servefile("FinalPaper_WatchingBrief2.pdf",279                                     "Third_Watching_Brief.pdf"))280        if page_id == "watching_brief_3":281            return('file', servefile("FinalPaper_WatchingBrief3.pdf",282                                     "Third_Watching_Brief.pdf"))283        if page_id == "page2":284            return ('game_many/final_paper.html', game_context)285        if page_id == "page3":286            return ('game_many/final_paper2.html', game_context)287        return ('game_many/final_paper_intro.html', game_context)288    def variables(self, page_id=None):289        return ['final_paper']290    def public_variables(self, page_id=None):291        return ['final_paper']292    def resources(self, game_state, onopen=False, onclosed=False):293        if onopen and game_state.has_key('donors_conference'):294            # week 4 points carry over295            week4points = 0296            if game_state.has_key('funding_interventions'):297                week4points = funding_points(298                    game_state['funding_interventions'])299            week6points = funding_points(game_state['donors_conference'], 6)300            points = week4points + week6points301            # print points302            # teams who choose the 3-point option in any of the first 3303            # categories automatically get wb2304            funded = game_state['donors_conference']305            wb2_choices = [intervention for intervention in funded306                           if intervention in ['water-supply-low',307                                               'sanitation-low',308                                               'waste-low']]309            autowb2 = False310            if wb2_choices != []:311                autowb2 = True312            unlock_wb3 = False313            if autowb2 or (points > 32 and points < 59):314                wb = {"page_id": 'watching_brief_2', "type":315                      'file', "title": 'Third Watching Brief.pdf'}316            elif points < 33:317                wb = {"page_id": 'watching_brief_1', "type":318                      'file', "title": 'Third Watching Brief.pdf'}319            else:  # points > 58320                wb = {"page_id": 'watching_brief_3', "type":321                      'file', "title": 'Third Watching Brief.pdf'}322                unlock_wb3 = True323            return [wb,324                    {"page_id": 'wb3', "type": 'data', "value": unlock_wb3},325                    ]326        else:327            return []328InstalledGames.register_game('final_paper',329                             'Final Paper',330                             FinalPaper())331# helper function ##332def servefile(filename, name, mimetype="application/pdf"):333    # instead of serving the file directly, open it and pipe it over (for334    # security)335    path = InstalledGames.absolute_path("envirocon.game_many",336                                        "files/%s" % filename)337    # return 404 if the file DNE338    try:339        file_handle = open(path, "rb")340    except:341        raise Http404  # TODO this doesn't do what i expected342    response = HttpResponse(FileWrapper(file_handle), mimetype=mimetype)343    response['Content-Disposition'] = 'attachment; filename=%s' % name344    response['Content-Length'] = os.path.getsize(path)...Notifier_test.py
Source:Notifier_test.py  
1import pytest2from unittest.mock import Mock34import sys5# insert at 1, 0 is the script path (or '' in REPL)6if not '../event-notifier' in sys.path:7    sys.path.insert(1, '../event-notifier')8from EventNotifier.Notifier import Notifier91011@pytest.fixture(scope="class")  # scope="function" is default12def logger():13    import logging1415    l = logging.getLogger("multiprocessing")16    h = logging.StreamHandler()17    f = logging.Formatter("%(asctime)s: %(message)s")18    h.setFormatter(f)19    l.addHandler(h)20    l.setLevel(logging.ERROR)21    yield l222324@pytest.fixture(scope="class")  # scope="function" is default25def notifier(logger):26    return Notifier(["onCreate", "onOpen", "onModify", "onClose", "onDelete"], logger)272829@pytest.fixture(scope="class")  # scope="function" is default30def emptyNotifier(logger):31    return Notifier([], logger)323334@pytest.fixture(scope="class")  # scope="function" is default35def notifierWithoutLogger(logger):36    return Notifier(["onCreate", "onOpen", "onModify", "onDelete"])373839class TestNotifier:40    def test_initialised_OK(self, notifier):41        assert 5 == len(notifier.get_registered_events())424344    def test_initiallyEmpty_OK(self, emptyNotifier):45        assert 0 == len(emptyNotifier.get_registered_events())464748    def test_initialisedWithoutLogger_OK(self, notifierWithoutLogger):49        assert 4 == len(notifierWithoutLogger.get_registered_events())505152    def test_initialisedWithRepeatedEventNames_Raises(self, notifierWithoutLogger):53        with pytest.raises(KeyError):54            Notifier(["onCreate", "onOpen", "onModify", "onDelete", "onOpen"])555657    def test_supportedEventsAccessible_OK(self, notifierWithoutLogger):58        notifier = Notifier(["onCreate", "onOpen", "onModify", "onDelete", "onForget"])59        assert ["onCreate", "onOpen", "onModify", "onDelete", "onForget"] == notifier.get_registered_events()606162    def test_fireEvent_OK(self, notifierWithoutLogger):63        onOpenCallback = Mock()6465        notifierWithoutLogger.subscribe("onOpen", onOpenCallback)6667        onOpenCallback.assert_not_called()68        notifierWithoutLogger.raise_event("onOpen", 77)69        onOpenCallback.assert_called_once_with(77)707172    def test_fireEventWithNoneAsEventName_Raises(self, notifierWithoutLogger):73        onOpenCallback = Mock()7475        notifierWithoutLogger.subscribe("onOpen", onOpenCallback)7677        onOpenCallback.assert_not_called()78        with pytest.raises(KeyError):79            notifierWithoutLogger.raise_event(None, "event info")80        onOpenCallback.assert_not_called()818283    def test_fireEventWithNoneAsParameter_OK(self, notifierWithoutLogger):84        onOpenCallback = Mock()8586        notifierWithoutLogger.subscribe("onOpen", onOpenCallback)8788        onOpenCallback.assert_not_called()89        notifierWithoutLogger.raise_event("onOpen", None)90        onOpenCallback.assert_called_once_with(None)91        notifierWithoutLogger.raise_event("onOpen", None, None, None)92        onOpenCallback.assert_called_with(None, None, None)93        assert(2 == onOpenCallback.call_count)949596    def test_fireEventOnlyCorrectIsFired_OK(self, notifierWithoutLogger):97        onOpenCallback = Mock()98        onDeleteCallback = Mock()99100        notifierWithoutLogger.subscribe("onOpen", onOpenCallback)101102        onOpenCallback.assert_not_called()103        onDeleteCallback.assert_not_called()104105        notifierWithoutLogger.raise_event("onDelete", 258)106        notifierWithoutLogger.subscribe("onDelete", onDeleteCallback)107108        onOpenCallback.assert_not_called()109        onDeleteCallback.assert_not_called()110111        notifierWithoutLogger.raise_event("onOpen", 99)112        onOpenCallback.assert_called_once_with(99)113        onDeleteCallback.assert_not_called()114115        notifierWithoutLogger.raise_event("onDelete", "string as event")116        onOpenCallback.assert_called_once_with(99)117        onDeleteCallback.assert_called_once_with("string as event")118119120    def test_removedSubscriberIsNotFired_OK(self, notifierWithoutLogger):121        onCreateCallback = Mock()122        onModifyCallback = Mock()123124        notifierWithoutLogger.subscribe("onCreate", onCreateCallback)125        notifierWithoutLogger.subscribe("onModify", onModifyCallback)126127        onCreateCallback.assert_not_called()128        onModifyCallback.assert_not_called()129130        notifierWithoutLogger.raise_event("onCreate", "event: test_removedSubscriberIsNotFired_OK - onCreate")131        onCreateCallback.assert_called_once_with("event: test_removedSubscriberIsNotFired_OK - onCreate")132        onModifyCallback.assert_not_called()133134        notifierWithoutLogger.remove_subscribers_by_event_name("onCreate")135        notifierWithoutLogger.raise_event("onCreate", "event: second time")136137        onCreateCallback.assert_called_once_with("event: test_removedSubscriberIsNotFired_OK - onCreate")138        onModifyCallback.assert_not_called()139140        notifierWithoutLogger.raise_event("onModify", "onModify is still working...")141        onCreateCallback.assert_called_once_with("event: test_removedSubscriberIsNotFired_OK - onCreate")142        onModifyCallback.assert_called_once_with("onModify is still working...")143144145    def test_allRemovedNotFiredToo_OK(self, notifierWithoutLogger):146        onOpenCallback = Mock()147        onDeleteCallback = Mock()148149        notifierWithoutLogger.subscribe("onOpen", onOpenCallback)150        notifierWithoutLogger.subscribe("onDelete", onDeleteCallback)151152        onOpenCallback.assert_not_called()153        onDeleteCallback.assert_not_called()154155        notifierWithoutLogger.raise_event("onOpen", "event: test_removedSubscriberIsNotFired_OK - onOpen")156        onOpenCallback.assert_called_once_with("event: test_removedSubscriberIsNotFired_OK - onOpen")157        onDeleteCallback.assert_not_called()158159        notifierWithoutLogger.remove_all_subscribers()160161        notifierWithoutLogger.raise_event("onOpen", "event: second time")162163        onOpenCallback.assert_called_once_with("event: test_removedSubscriberIsNotFired_OK - onOpen")164        onDeleteCallback.assert_not_called()165166167    def test_repeatedlyAddingSameSubscriber_OK(self, notifierWithoutLogger):168        onCreateCallback = Mock()169        notifierWithoutLogger.subscribe("onCreate", onCreateCallback)170        notifierWithoutLogger.subscribe("onCreate", onCreateCallback)171        notifierWithoutLogger.subscribe("onCreate", onCreateCallback)172173        onCreateCallback.assert_not_called()174175        notifierWithoutLogger.raise_event("onCreate", "event: onCreate 2222")176        onCreateCallback.assert_called_once_with("event: onCreate 2222")177178179    #TODO: refactor this test to use parmetrised annotation180    def test_anyTypeAsEvent_OK(self):181        class Box:182            def __init__(self, name):183                self.name = name184185        a = Box("keyBoxA")186        b = Box("keyBoxB")187        c = Box("keyBoxB")  # intentionally named this way to look like b188189        notifier = Notifier(["onCreate", 5, 22.58, "onDelete", a, b])190        onCreateCallback = Mock()191        on5Callback      = Mock()192        onFloatCallback  = Mock()193        onBoxACallback   = Mock()194        onBoxBCallback   = Mock()195196        notifier.subscribe("onCreate", onCreateCallback)197        notifier.subscribe(5, on5Callback)198        notifier.subscribe(22.58, onFloatCallback)199        notifier.subscribe(a, onBoxACallback)200        notifier.subscribe(b, onBoxBCallback)201202        onCreateCallback.assert_not_called()203        on5Callback.assert_not_called()204        onFloatCallback.assert_not_called()205        onBoxACallback.assert_not_called()206        onBoxBCallback.assert_not_called()207208        notifier.raise_event("onCreate", "event: onCreate !!!!")209        onCreateCallback.assert_called_once_with("event: onCreate !!!!")210        on5Callback.assert_not_called()211        onFloatCallback.assert_not_called()212        onBoxACallback.assert_not_called()213        onBoxBCallback.assert_not_called()214215        with pytest.raises(KeyError):216            notifier.raise_event(6, "event: !!!!! 6 !!!!")   # unknown event - KeyError raised217        onCreateCallback.assert_called_once_with("event: onCreate !!!!")218        on5Callback.assert_not_called()219        onFloatCallback.assert_not_called()220        onBoxACallback.assert_not_called()221        onBoxBCallback.assert_not_called()222223        notifier.raise_event(5, "event: !!!!! 5 !!!!") # this one is registered - OK224        onCreateCallback.assert_called_once_with("event: onCreate !!!!")225        on5Callback.assert_called_once_with("event: !!!!! 5 !!!!")226        onFloatCallback.assert_not_called()227        onBoxACallback.assert_not_called()228        onBoxBCallback.assert_not_called()229230        with pytest.raises(KeyError):231            notifier.raise_event(22.59, "event: !!!!! 22.59 !!!!")  # unknown event - KeyError raised232        onCreateCallback.assert_called_once_with("event: onCreate !!!!")233        on5Callback.assert_called_once_with("event: !!!!! 5 !!!!")234        onFloatCallback.assert_not_called()235        onBoxACallback.assert_not_called()236        onBoxBCallback.assert_not_called()237238        notifier.raise_event(22.58, "event: !!!!! 22.58 !!!!")  # this one is registered - OK239        onCreateCallback.assert_called_once_with("event: onCreate !!!!")240        on5Callback.assert_called_once_with("event: !!!!! 5 !!!!")241        onFloatCallback.assert_called_once_with("event: !!!!! 22.58 !!!!")242        onBoxACallback.assert_not_called()243        onBoxBCallback.assert_not_called()244245        with pytest.raises(KeyError):246            notifier.raise_event(c, "event: Box c with name like it is b")  # unknown event (object is known, name is the same, but object is still different) - KeyError raised247        onCreateCallback.assert_called_once_with("event: onCreate !!!!")248        on5Callback.assert_called_once_with("event: !!!!! 5 !!!!")249        onFloatCallback.assert_called_once_with("event: !!!!! 22.58 !!!!")250        onBoxACallback.assert_not_called()251        onBoxBCallback.assert_not_called()252253        notifier.raise_event(b, "event: Box b")254        onCreateCallback.assert_called_once_with("event: onCreate !!!!")255        on5Callback.assert_called_once_with("event: !!!!! 5 !!!!")256        onFloatCallback.assert_called_once_with("event: !!!!! 22.58 !!!!")257        onBoxACallback.assert_not_called()258        onBoxBCallback.assert_called_once_with("event: Box b")259260261    def test_notifySubscribersWithMultipleParams_OK(self, notifier):262        onModifyCallback = Mock()263264        notifier.subscribe("onModify", onModifyCallback)265266        onModifyCallback.assert_not_called()267268        notifier.raise_event("onModify", "some text", 25, 16.99, named1 ="named param value", named2 = 25)269        onModifyCallback.assert_called_once_with("some text", 25, 16.99, named1 = "named param value", named2 = 25)270271272    def test_subscribeToAll_OK(self, notifier):273        onAnyCallback = Mock()274275        notifier.subscribe_to_all(onAnyCallback)276277        onAnyCallback.assert_not_called()278279        notifier.raise_event("onCreate", "event specific info here", event_type="onCreate")280        onAnyCallback.assert_called_once_with("event specific info here", event_type="onCreate")281282        notifier.raise_event("onOpen", "event specific info22 here", event_type="onOpen")283        onAnyCallback.assert_called_with("event specific info22 here", event_type="onOpen")284        assert (2 == onAnyCallback.call_count)285286        notifier.raise_event("onClose", "event specific info33 here", event_type="onClose")287        onAnyCallback.assert_called_with("event specific info33 here", event_type="onClose")288        assert (3 == onAnyCallback.call_count)289290291    def test_subscribeToAllIfWasAlreadyRegisteredToOne_OK(self, notifier):292        onAnyCallback = Mock()293294        notifier.subscribe("onCreate", onAnyCallback)295        notifier.subscribe_to_all(onAnyCallback)296297        onAnyCallback.assert_not_called()298299        notifier.raise_event("onCreate", "event specific info here", event_type="onCreate")300        onAnyCallback.assert_called_once_with("event specific info here", event_type="onCreate")301302        notifier.raise_event("onOpen", "event specific info22 here", event_type="onOpen")303        onAnyCallback.assert_called_with("event specific info22 here", event_type="onOpen")304        assert (2 == onAnyCallback.call_count)
...Examples_test.py
Source:Examples_test.py  
...3940    def test_remove_subscribers_by_event_name(self):41        from EventNotifier import Notifier42        class FileWatchDog():43            def onOpen(self, fileName, openMode):44                print(f"File {fileName} opened with {openMode} mode")4546            def onClose(self, fileName):47                print(f"File {fileName} closed")484950        def onOpenStandaloneMethod(fileName, openMode):51            print(f"StandaloneMethod: File {fileName} opened with {openMode} mode")5253        watchDog = FileWatchDog()5455        notifier = Notifier(["onCreate", "onOpen", "onModify", "onClose", "onDelete"])5657        notifier.subscribe("onOpen", watchDog.onOpen)58        notifier.subscribe("onOpen", onOpenStandaloneMethod)59        notifier.subscribe("onClose", watchDog.onClose)6061        print("\nAfter subscription:")62        notifier.raise_event("onOpen", openMode="w+", fileName="test_file.txt")63        notifier.raise_event("onClose", fileName="test_file.txt")6465        notifier.remove_subscribers_by_event_name("onOpen")6667        print("\nAfter removal of onOpen subscribers:")68        notifier.raise_event("onOpen", openMode="w+", fileName="test_file.txt")69        notifier.raise_event("onClose", fileName="test_file.txt")7071        notifier.remove_subscribers_by_event_name("onClose")7273        print("\nAfter removal of onClose subscribers:")74        notifier.raise_event("onOpen", openMode="w+", fileName="test_file.txt")75        notifier.raise_event("onClose", fileName="test_file.txt")76777879    def test_remove_all_subscribers(self):80        from EventNotifier import Notifier81        class FileWatchDog():82            def onOpen(self, fileName, openMode):83                print(f"File {fileName} opened with {openMode} mode")8485            def onClose(self, fileName):86                print(f"File {fileName} closed")878889        def onOpenStandaloneMethod(fileName, openMode):90            print(f"StandaloneMethod: File {fileName} opened with {openMode} mode")9192        watchDog = FileWatchDog()9394        notifier = Notifier(["onCreate", "onOpen", "onModify", "onClose", "onDelete"])9596        notifier.subscribe("onOpen", watchDog.onOpen)
...callback.py
Source:callback.py  
...6    @abstractmethod7    def onTrade(self, data: MarketData):8        '''onTrade'''9    @abstractmethod10    def onOpen(self, data: MarketData):11        '''onOpen'''12    @abstractmethod13    def onFill(self, data: MarketData):14        '''onFill'''15    @abstractmethod16    def onCancel(self, data: MarketData):17        '''onCancel'''18    @abstractmethod19    def onChange(self, data: MarketData):20        '''onChange'''21    @abstractmethod22    def onError(self, data: MarketData):23        '''onError'''24    def onExit(self):25        '''onExit'''26        pass27    def onAnalyze(self, data):28        '''onAnalyze'''29        pass30    def onHalt(self, data):31        '''onHalt'''32        pass33    def onContinue(self, data):34        '''onContinue'''35        pass36    def callback(self):37        return self38class NullCallback(Callback):39    def __init__(self):40        pass41    def onTrade(self, data: MarketData) -> None:42        pass43    def onReceived(self, data: MarketData) -> None:44        pass45    def onOpen(self, data: MarketData) -> None:46        pass47    def onFill(self, data: MarketData) -> None:48        pass49    def onCancel(self, data: MarketData) -> None:50        pass51    def onChange(self, data: MarketData) -> None:52        pass53    def onError(self, data: MarketData) -> None:54        pass55class Print(Callback):56    def __init__(self,57                 onTrade=True,58                 onReceived=True,59                 onOpen=True,60                 onFill=True,61                 onCancel=True,62                 onChange=True,63                 onError=True):64        if not onTrade:65            setattr(self, 'onTrade', False)66        if not onReceived:67            setattr(self, 'onReceived', False)68        if not onOpen:69            setattr(self, 'onOpen', False)70        if not onFill:71            setattr(self, 'onFilled', False)72        if not onCancel:73            setattr(self, 'onCancelled', False)74        if not onChange:75            setattr(self, 'onChange', False)76        if not onError:77            setattr(self, 'onError', False)78    def onTrade(self, data: MarketData) -> None:79        dlog.info(str(data))80    def onReceived(self, data: MarketData) -> None:81        dlog.info(str(data))82    def onOpen(self, data: MarketData) -> None:83        dlog.info(str(data))84    def onFill(self, data: MarketData) -> None:85        dlog.info(str(data))86    def onCancel(self, data: MarketData) -> None:87        dlog.info(str(data))88    def onChange(self, data: MarketData) -> None:89        dlog.info(str(data))90    def onError(self, data: MarketData) -> None:91        dlog.info(str(data))92    def onAnalyze(self, data) -> None:93        log.info('Analysis')94        pass95    def onHalt(self, data) -> None:96        log.info('Halt')...LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Get 100 minutes of automation test minutes FREE!!
