fix: Handle KeyboardInterrupt that happens outside of the main test…

… loop inside the runner

Ref: #1325

docs/changelog.rst

@@ -4,6 +4,11 @@ Changelog
 `Unreleased`_ - TBD
 -------------------
 
+**Fixed**
+
+- Handle ``KeyboardInterrupt`` that happens outside of the main test loop inside the runner.
+  It makes interrupt handling consistent, independent at what point it happens. `#1325`_
+
 `3.11.1`_ - 2021-11-20
 ----------------------
 
@@ -2181,7 +2186,8 @@ Deprecated
 .. _0.3.0: https://github.com/schemathesis/schemathesis/compare/v0.2.0...v0.3.0
 .. _0.2.0: https://github.com/schemathesis/schemathesis/compare/v0.1.0...v0.2.0
 
-.. _#1328: https://github.com/schemathesis/schemathesis/issues/1228
+.. _#1328: https://github.com/schemathesis/schemathesis/issues/1328
+.. _#1325: https://github.com/schemathesis/schemathesis/issues/1325
 .. _#1290: https://github.com/schemathesis/schemathesis/issues/1290
 .. _#1287: https://github.com/schemathesis/schemathesis/issues/1287
 .. _#1280: https://github.com/schemathesis/schemathesis/issues/1280

src/schemathesis/runner/impl/core.py

@@ -91,8 +91,11 @@ def _finish() -> events.Finished:
             yield _finish()
             return
 
-        for event in self._execute(results, stop_event):
-            yield event
+        try:
+            for event in self._execute(results, stop_event):
+                yield event
+        except KeyboardInterrupt:
+            yield events.Interrupted()
 
         yield _finish()
 

test/conftest.py

@@ -149,6 +149,11 @@ def openapi3_schema_url(server, openapi_3_app):
     return f"http://127.0.0.1:{server['port']}/schema.yaml"
 
 
+@pytest.fixture()
+def openapi3_schema(openapi3_schema_url):
+    return oas_loaders.from_uri(openapi3_schema_url)
+
+
 @pytest.fixture
 def graphql_path():
     return "/graphql"

test/runner/test_runner.py

@@ -849,6 +849,31 @@ def test_graphql(graphql_url):
     assert finished.passed_count == 2
 
 
+@pytest.mark.operations("success")
+def test_interrupted_in_test(openapi3_schema):
+    # When an interrupt happens within a test body (check is called within a test body)
+    def check(response, case):
+        raise KeyboardInterrupt
+
+    *_, event, finished = from_schema(openapi3_schema, checks=(check,)).execute()
+    # Then the `Interrupted` event should be emitted
+    assert isinstance(event, events.Interrupted)
+
+
+@pytest.mark.operations("success")
+def test_interrupted_outside_test(mocker, openapi3_schema):
+    # See GH-1325
+    # When an interrupt happens outside of a test body
+    mocker.patch("schemathesis.runner.events.AfterExecution.from_result", side_effect=KeyboardInterrupt)
+
+    try:
+        *_, event, finished = from_schema(openapi3_schema).execute()
+        # Then the `Interrupted` event should be emitted
+        assert isinstance(event, events.Interrupted)
+    except KeyboardInterrupt:
+        pytest.fail("KeyboardInterrupt should be handled")
+
+
 @pytest.fixture(params=[1, 2], ids=["single-worker", "multi-worker"])
 def workers_num(request):
     return request.param