Merge pull request #38 from simonsobs/dev

Add GraphQL mutation `loadExampleScript`

nextlinegraphql/plugins/ctrl/graphql/__init__.py

@@ -7,6 +7,7 @@
 MUTATE_RESET = (sub / 'Reset.gql').read_text()
 MUTATE_SEND_PDB_COMMAND = (sub / 'SendPdbCommand.gql').read_text()
 MUTATE_RUN_AND_CONTINUE = (sub / 'RunAndContinue.gql').read_text()
+MUTATE_LOAD_EXAMPLE_SCRIPT = (sub / 'LoadExampleScript.gql').read_text()
 
 sub = pwd / 'queries'
 QUERY_STATE = (sub / 'State.gql').read_text()

nextlinegraphql/plugins/ctrl/graphql/mutations/LoadExampleScript.gql

@@ -0,0 +1,3 @@
+mutation LoadExampleScript {
+  loadExampleScript
+}

nextlinegraphql/plugins/ctrl/schema/mutation.py

@@ -1,56 +1,75 @@
-from typing import TYPE_CHECKING, Optional
+from pathlib import Path
+from typing import Optional
 
 import strawberry
+from nextline import Nextline
 from strawberry.types import Info
 
-if TYPE_CHECKING:
-    from nextline import Nextline
+from nextlinegraphql.plugins.ctrl import example_script as example_script_module
+
+EXAMPLE_SCRIPT_PATH = Path(example_script_module.__file__).parent / 'script.py'
+assert EXAMPLE_SCRIPT_PATH.is_file()
+example_script = EXAMPLE_SCRIPT_PATH.read_text()
 
 
 async def mutate_exec(info: Info) -> bool:
-    nextline: Nextline = info.context["nextline"]
+    nextline = info.context['nextline']
+    assert isinstance(nextline, Nextline)
     await nextline.run()
     return True
 
 
 async def mutate_run_and_continue(info: Info) -> bool:
-    nextline: Nextline = info.context["nextline"]
+    nextline = info.context['nextline']
+    assert isinstance(nextline, Nextline)
     await nextline.run_and_continue()
     return True
 
 
 async def mutate_reset(info: Info, statement: Optional[str] = None) -> bool:
-    nextline: Nextline = info.context["nextline"]
+    nextline = info.context['nextline']
+    assert isinstance(nextline, Nextline)
     await nextline.reset(statement=statement)
     return True
 
 
 async def mutate_send_pdb_command(
     info: Info, command: str, prompt_no: int, trace_no: int
 ) -> bool:
-    nextline: Nextline = info.context["nextline"]
+    nextline = info.context['nextline']
+    assert isinstance(nextline, Nextline)
     await nextline.send_pdb_command(command, prompt_no, trace_no)
     return True
 
 
 async def mutate_interrupt(info: Info) -> bool:
-    nextline: Nextline = info.context["nextline"]
+    nextline = info.context['nextline']
+    assert isinstance(nextline, Nextline)
     await nextline.interrupt()
     return True
 
 
 async def mutate_terminate(info: Info) -> bool:
-    nextline: Nextline = info.context["nextline"]
+    nextline = info.context['nextline']
+    assert isinstance(nextline, Nextline)
     await nextline.terminate()
     return True
 
 
 async def mutate_kill(info: Info) -> bool:
-    nextline: Nextline = info.context["nextline"]
+    nextline = info.context['nextline']
+    assert isinstance(nextline, Nextline)
     await nextline.kill()
     return True
 
 
+async def mutate_load_example_script(info: Info) -> bool:
+    nextline = info.context['nextline']
+    assert isinstance(nextline, Nextline)
+    await nextline.reset(statement=example_script)
+    return True
+
+
 @strawberry.type
 class Mutation:
     exec: bool = strawberry.field(resolver=mutate_exec)
@@ -60,3 +79,4 @@ class Mutation:
     interrupt: bool = strawberry.field(resolver=mutate_interrupt)
     terminate: bool = strawberry.field(resolver=mutate_terminate)
     kill: bool = strawberry.field(resolver=mutate_kill)
+    load_example_script: bool = strawberry.field(resolver=mutate_load_example_script)

tests/plugins/ctrl/schema/mutations/test_load_example_script.py

@@ -0,0 +1,28 @@
+from pathlib import Path
+
+from nextline import Nextline
+from strawberry import Schema
+
+from nextlinegraphql.plugins.ctrl import example_script as example_script_module
+from nextlinegraphql.plugins.ctrl.graphql import MUTATE_LOAD_EXAMPLE_SCRIPT
+from nextlinegraphql.plugins.ctrl.schema import Mutation, Query, Subscription
+
+EXAMPLE_SCRIPT_PATH = Path(example_script_module.__file__).parent / 'script.py'
+example_script = EXAMPLE_SCRIPT_PATH.read_text()
+
+INITIAL_SCRIPT = '''
+import time
+time.sleep(0.001)
+'''.strip()
+
+
+async def test_query() -> None:
+    nextline = Nextline(INITIAL_SCRIPT)
+    async with nextline:
+        context = {'nextline': nextline}
+        schema = Schema(query=Query, mutation=Mutation, subscription=Subscription)
+        result = await schema.execute(MUTATE_LOAD_EXAMPLE_SCRIPT, context_value=context)
+        assert not result.errors
+        assert result.data
+        assert result.data['loadExampleScript'] is True
+        assert nextline.statement == example_script