From 387bd361217317865010568deac3125f246d76b0 Mon Sep 17 00:00:00 2001 From: mmikita95 Date: Thu, 15 Feb 2024 12:18:43 +0300 Subject: [PATCH] fix: introducing a fixture for proper teardown --- tests/test_app_runner.py | 329 ++++++++++++++++++++------------------- 1 file changed, 171 insertions(+), 158 deletions(-) diff --git a/tests/test_app_runner.py b/tests/test_app_runner.py index 51a39ef18..c396a57fb 100644 --- a/tests/test_app_runner.py +++ b/tests/test_app_runner.py @@ -6,9 +6,23 @@ import pytest from streamsync.ss_types import EventRequest, InitSessionRequest, InitSessionRequestPayload, StreamsyncEvent import asyncio +import contextlib from tests import test_app_dir + +@pytest.fixture +def setup_app_runner(): + @contextlib.contextmanager + def _manage_launch_args(app_dir, app_command): + ar = AppRunner(app_dir, app_command) + try: + yield ar + finally: + ar.shut_down() + return _manage_launch_args + + class TestAppRunner: numberinput_instance_path = [ @@ -28,161 +42,161 @@ class TestAppRunner: ] proposed_session_id = "c13a280fe17ec663047ec14de15cd93ad686fecf5f9a4dbf262d3a86de8cb577" - def test_init_wrong_path(self) -> None: - ar = AppRunner("./not_an_app", "run") - with pytest.raises(SystemExit) as wrapped_e: - ar.load() - assert wrapped_e.type == SystemExit - ar.shut_down() + @pytest.mark.usefixtures("setup_app_runner") + def test_init_wrong_path(self, setup_app_runner) -> None: + with setup_app_runner("./not_an_app", "run") as ar: + with pytest.raises(SystemExit) as wrapped_e: + ar.load() + assert wrapped_e.type == SystemExit def test_init_wrong_mode(self) -> None: with pytest.raises(ValueError): AppRunner(test_app_dir, "virus") @pytest.mark.asyncio - async def test_pre_session(self) -> None: - ar = AppRunner(test_app_dir, "run") - er = EventRequest( - type="event", - payload=StreamsyncEvent( + @pytest.mark.usefixtures("setup_app_runner") + async def test_pre_session(self, setup_app_runner) -> None: + with setup_app_runner(test_app_dir, "run") as ar: + er = EventRequest( + type="event", + payload=StreamsyncEvent( + type="virus", + instancePath=self.numberinput_instance_path, + payload={ + "virus": "yes" + } + ) + ) + ar.load() + r = await ar.dispatch_message(None, er) + assert r.status == "error" + + @pytest.mark.asyncio + @pytest.mark.usefixtures("setup_app_runner") + async def test_valid_session_invalid_event(self, setup_app_runner) -> None: + with setup_app_runner(test_app_dir, "run") as ar: + ar.load() + si = InitSessionRequest( + type="sessionInit", + payload=InitSessionRequestPayload( + cookies={}, + headers={}, + proposedSessionId=self.proposed_session_id + ) + ) + sres = await ar.dispatch_message(None, si) + assert sres.status == "ok" + assert sres.payload.model_dump().get("sessionId") == self.proposed_session_id + er = EventRequest(type="event", payload=StreamsyncEvent( type="virus", instancePath=self.numberinput_instance_path, payload={ "virus": "yes" } - ) - ) - ar.load() - r = await ar.dispatch_message(None, er) - assert r.status == "error" - ar.shut_down() + )) + rev = await ar.dispatch_message(None, er) + assert rev.status == "error" @pytest.mark.asyncio - async def test_valid_session_invalid_event(self) -> None: - ar = AppRunner(test_app_dir, "run") - ar.load() - si = InitSessionRequest( - type="sessionInit", - payload=InitSessionRequestPayload( - cookies={}, - headers={}, - proposedSessionId=self.proposed_session_id + @pytest.mark.usefixtures("setup_app_runner") + async def test_valid_event(self, setup_app_runner) -> None: + with setup_app_runner(test_app_dir, "run") as ar: + ar.load() + si = InitSessionRequest( + type="sessionInit", + payload=InitSessionRequestPayload( + cookies={}, + headers={}, + proposedSessionId=self.proposed_session_id + ) ) - ) - sres = await ar.dispatch_message(None, si) - assert sres.status == "ok" - assert sres.payload.model_dump().get("sessionId") == self.proposed_session_id - er = EventRequest(type="event", payload=StreamsyncEvent( - type="virus", - instancePath=self.numberinput_instance_path, - payload={ - "virus": "yes" - } - )) - rev = await ar.dispatch_message(None, er) - assert rev.status == "error" - ar.shut_down() + sres = await ar.dispatch_message(None, si) + assert sres.status == "ok" + assert sres.payload.model_dump().get("sessionId") == self.proposed_session_id + ev_req = EventRequest(type="event", payload=StreamsyncEvent( + type="ss-number-change", + instancePath=self.numberinput_instance_path, + payload="129673" + )) + ev_res = await ar.dispatch_message(self.proposed_session_id, ev_req) + assert ev_res.status == "ok" + assert ev_res.payload.result.get("ok") == True + assert ev_res.payload.mutations.get( + "+inspected_payload") == "129673.0" + assert ev_res.payload.mutations.get( + "+b.pet_count") == 129673 @pytest.mark.asyncio - async def test_valid_event(self) -> None: - ar = AppRunner(test_app_dir, "run") - ar.load() - si = InitSessionRequest( - type="sessionInit", - payload=InitSessionRequestPayload( - cookies={}, - headers={}, - proposedSessionId=self.proposed_session_id + @pytest.mark.usefixtures("setup_app_runner") + async def test_async_handler(self, setup_app_runner) -> None: + with setup_app_runner(test_app_dir, "run") as ar: + ar.load() + si = InitSessionRequest( + type="sessionInit", + payload=InitSessionRequestPayload( + cookies={}, + headers={}, + proposedSessionId=self.proposed_session_id + ) ) - ) - sres = await ar.dispatch_message(None, si) - assert sres.status == "ok" - assert sres.payload.model_dump().get("sessionId") == self.proposed_session_id - ev_req = EventRequest(type="event", payload=StreamsyncEvent( - type="ss-number-change", - instancePath=self.numberinput_instance_path, - payload="129673" - )) - ev_res = await ar.dispatch_message(self.proposed_session_id, ev_req) - assert ev_res.status == "ok" - assert ev_res.payload.result.get("ok") == True - assert ev_res.payload.mutations.get( - "+inspected_payload") == "129673.0" - assert ev_res.payload.mutations.get( - "+b.pet_count") == 129673 - ar.shut_down() + sres = await ar.dispatch_message(None, si) + assert sres.status == "ok" + assert sres.payload.model_dump().get("sessionId") == self.proposed_session_id - @pytest.mark.asyncio - async def test_async_handler(self) -> None: - ar = AppRunner(test_app_dir, "run") - ar.load() - si = InitSessionRequest( - type="sessionInit", - payload=InitSessionRequestPayload( - cookies={}, - headers={}, - proposedSessionId=self.proposed_session_id - ) - ) - sres = await ar.dispatch_message(None, si) - assert sres.status == "ok" - assert sres.payload.model_dump().get("sessionId") == self.proposed_session_id - - # Firing an event to bypass "initial" state mutations - ev_req = EventRequest(type="event", payload=StreamsyncEvent( - type="ss-number-change", - instancePath=self.numberinput_instance_path, - payload="129673" - )) - ev_res = await ar.dispatch_message(self.proposed_session_id, ev_req) - - ev_req = EventRequest(type="event", payload=StreamsyncEvent( - type="ss-click", - instancePath=self.async_handler_click_path - )) - ev_res = await ar.dispatch_message(self.proposed_session_id, ev_req) - assert ev_res.status == "ok" - assert ev_res.payload.result.get("ok") == True - assert "+counter" in ev_res.payload.mutations - ar.shut_down() + # Firing an event to bypass "initial" state mutations + ev_req = EventRequest(type="event", payload=StreamsyncEvent( + type="ss-number-change", + instancePath=self.numberinput_instance_path, + payload="129673" + )) + ev_res = await ar.dispatch_message(self.proposed_session_id, ev_req) + + ev_req = EventRequest(type="event", payload=StreamsyncEvent( + type="ss-click", + instancePath=self.async_handler_click_path + )) + ev_res = await ar.dispatch_message(self.proposed_session_id, ev_req) + assert ev_res.status == "ok" + assert ev_res.payload.result.get("ok") == True + assert "+counter" in ev_res.payload.mutations @pytest.mark.asyncio - async def test_bad_event_handler(self) -> None: - ar = AppRunner(test_app_dir, "run") - ar.load() - si = InitSessionRequest( - type="sessionInit", - payload=InitSessionRequestPayload( - cookies={}, - headers={}, - proposedSessionId=self.proposed_session_id + @pytest.mark.usefixtures("setup_app_runner") + async def test_bad_event_handler(self, setup_app_runner) -> None: + with setup_app_runner(test_app_dir, "run") as ar: + ar.load() + si = InitSessionRequest( + type="sessionInit", + payload=InitSessionRequestPayload( + cookies={}, + headers={}, + proposedSessionId=self.proposed_session_id + ) ) - ) - await ar.dispatch_message(None, si) - bad_button_instance_path = [ - {"componentId": "root", "instanceNumber": 0}, - {"componentId": "28a2212b-bc58-4398-8a72-2554e5296490", "instanceNumber": 0}, - {"componentId": "232d749a-5e0c-4802-bbe1-f8cae06db112", "instanceNumber": 0} - ] - ev_req = EventRequest(type="event", payload=StreamsyncEvent( - type="click", - instancePath=bad_button_instance_path, - payload={} - )) - ev_res = await ar.dispatch_message(self.proposed_session_id, ev_req) - print(repr(ev_res)) - assert ev_res.status == "ok" - assert ev_res.payload.result.get("ok") == False - ar.shut_down() - - def test_run_code_edit(self) -> None: - ar = AppRunner(test_app_dir, "run") - with pytest.raises(PermissionError): - ar.update_code(None, "exec(virus)") - with pytest.raises(PermissionError): - ar.save_code(None, "exec(virus)") - ar.shut_down() - + await ar.dispatch_message(None, si) + bad_button_instance_path = [ + {"componentId": "root", "instanceNumber": 0}, + {"componentId": "28a2212b-bc58-4398-8a72-2554e5296490", "instanceNumber": 0}, + {"componentId": "232d749a-5e0c-4802-bbe1-f8cae06db112", "instanceNumber": 0} + ] + ev_req = EventRequest(type="event", payload=StreamsyncEvent( + type="click", + instancePath=bad_button_instance_path, + payload={} + )) + ev_res = await ar.dispatch_message(self.proposed_session_id, ev_req) + print(repr(ev_res)) + assert ev_res.status == "ok" + assert ev_res.payload.result.get("ok") == False + + @pytest.mark.usefixtures("setup_app_runner") + def test_run_code_edit(self, setup_app_runner) -> None: + with setup_app_runner(test_app_dir, "run") as ar: + with pytest.raises(PermissionError): + ar.update_code(None, "exec(virus)") + with pytest.raises(PermissionError): + ar.save_code(None, "exec(virus)") + def run_loader_thread(self, app_runner: AppRunner) -> None: app_runner.update_code(None, "print('188542')") @@ -195,29 +209,28 @@ async def wait_for_code_update(self, app_runner: AppRunner) -> None: return True @pytest.mark.asyncio - async def test_code_update(self) -> None: - ar = AppRunner(test_app_dir, "edit") - ar.hook_to_running_event_loop() - ar.load() - wait_update_task = asyncio.create_task(self.wait_for_code_update(ar)) - loader_thread = threading.Thread(target=self.run_loader_thread, args=(ar,)) - loader_thread.start() - code_update_result = await wait_update_task - loader_thread.join() - - assert code_update_result == True - - si = InitSessionRequest( - type="sessionInit", - payload=InitSessionRequestPayload( - cookies={}, - headers={}, - proposedSessionId=self.proposed_session_id + @pytest.mark.usefixtures("setup_app_runner") + async def test_code_update(self, setup_app_runner) -> None: + with setup_app_runner(test_app_dir, "edit") as ar: + ar.hook_to_running_event_loop() + ar.load() + wait_update_task = asyncio.create_task(self.wait_for_code_update(ar)) + loader_thread = threading.Thread(target=self.run_loader_thread, args=(ar,)) + loader_thread.start() + code_update_result = await wait_update_task + loader_thread.join() + + assert code_update_result == True + + si = InitSessionRequest( + type="sessionInit", + payload=InitSessionRequestPayload( + cookies={}, + headers={}, + proposedSessionId=self.proposed_session_id + ) ) - ) - si_res = await ar.dispatch_message(None, si) - mail = list(si_res.payload.model_dump().get("mail")) - - assert mail[0].get("payload").get("message") == "188542\n" + si_res = await ar.dispatch_message(None, si) + mail = list(si_res.payload.model_dump().get("mail")) - ar.shut_down() \ No newline at end of file + assert mail[0].get("payload").get("message") == "188542\n"