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!!