Harden feature teardown cleanup

Author: pratyushjaiswal0806-dot

test/features/environment.py

@@ -133,12 +133,11 @@ def after_scenario(context, _):
             dbname = context.currentdb
             context.cli.expect_exact(f"{user}@{host}:{dbname}>", timeout=5)
         context.cli.sendcontrol("c")
-        user = context.conf["user"]
-        host = context.conf["host"]
-        dbname = context.currentdb
-        context.cli.expect_exact(f"{user}@{host}:{dbname}>", timeout=5)
         context.cli.sendcontrol("d")
-        context.cli.expect_exact(pexpect.EOF, timeout=5)
+        try:
+            context.cli.expect_exact(pexpect.EOF, timeout=5)
+        except pexpect.TIMEOUT:
+            context.cli.terminate(force=True)
 
     if os.path.exists(MY_CNF_BACKUP_PATH):
         shutil.move(MY_CNF_BACKUP_PATH, MY_CNF_PATH)

test/features/steps/iocommands.py

@@ -38,7 +38,6 @@ def step_edit_done_sql(context, query):
         wrappers.expect_exact(context, match, timeout=5)
     # Cleanup the command line.
     context.cli.sendcontrol("c")
-    wrappers.wait_prompt(context)
     # Cleanup the edited file.
     if context.editor_file_name and os.path.exists(context.editor_file_name):
         os.remove(context.editor_file_name)

test/pytests/test_feature_environment.py

@@ -17,48 +17,40 @@
 environment = cast(Any, environment_module)
 
 
-class PromptRaceCli:
+class TimeoutOnEofCli:
     def __init__(self) -> None:
-        self.interrupt_prompt_seen = False
-        self.closed = False
+        self.sent_controls: list[str] = []
+        self.terminated_force: bool | None = None
+
+    def sendcontrol(self, key: str) -> None:
+        self.sent_controls.append(key)
 
     def expect_exact(self, expected: object, timeout: int) -> None:
         if expected == pexpect.EOF:
-            if not self.closed:
-                raise pexpect.TIMEOUT('process still running')
-            return
-
-        if expected == 'root@127.0.0.1:mycli_behave_tests>':
-            self.interrupt_prompt_seen = True
-            return
+            raise pexpect.TIMEOUT('process still running')
 
         raise AssertionError(f'unexpected expectation: {expected!r}')
 
-    def sendcontrol(self, key: str) -> None:
-        if key == 'c':
-            return
-        if key == 'd':
-            self.closed = self.interrupt_prompt_seen
+    def terminate(self, force: bool = False) -> None:
+        self.terminated_force = force
 
 
-def test_after_scenario_waits_for_prompt_after_interrupt(monkeypatch, tmp_path) -> None:
+def test_after_scenario_terminates_when_teardown_eof_times_out(monkeypatch, tmp_path) -> None:
     log_file = tmp_path / 'mycli.test.log'
     log_file.write_text('')
     monkeypatch.setattr(environment, 'test_log_file', str(log_file))
     monkeypatch.setattr(environment, 'MY_CNF_BACKUP_PATH', str(tmp_path / '.my.cnf.backup'))
     monkeypatch.setattr(environment, 'MYLOGIN_CNF_BACKUP_PATH', str(tmp_path / '.mylogin.cnf.backup'))
     monkeypatch.setattr(environment, 'MYLOGIN_CNF_PATH', str(tmp_path / '.mylogin.cnf'))
 
+    cli = TimeoutOnEofCli()
     context = SimpleNamespace(
         atprompt=True,
-        cli=PromptRaceCli(),
-        conf={
-            'user': 'root',
-            'host': '127.0.0.1',
-            'dbname': 'mycli_behave_tests',
-        },
-        currentdb='mycli_behave_tests',
+        cli=cli,
         exit_sent=False,
     )
 
     environment.after_scenario(context, SimpleNamespace())
+
+    assert cli.sent_controls == ['c', 'd']
+    assert cli.terminated_force is True