fix: notify and cancel windows

Signed-off-by: Cody Edwards <edwards@amazon.com>

Author: edwards-aws

pyproject.toml

@@ -143,7 +143,6 @@ addopts = [
     "--cov-report=html:build/coverage",
     "--cov-report=xml:build/coverage/coverage.xml",
     "--cov-report=term-missing",
-    "--numprocesses=auto",
     "--timeout=30"
 ]
 markers = [

scripts/windows_service_test.py

@@ -2,6 +2,7 @@
 
 import socket
 import logging
+import io
 from threading import Event
 from typing import Optional
 
@@ -15,6 +16,7 @@
 import shlex
 import win32con
 import win32api
+import pytest
 from getpass import getpass
 
 
@@ -54,32 +56,43 @@ def SvcDoRun(self):
         )
         code_location = os.environ["CODE_LOCATION"]
         pytest_args = os.environ.get("PYTEST_ARGS", None)
+        log_file_name = os.path.join(code_location, "test.log")
 
-        args = ["pytest", os.path.join(code_location, "test")]
+        args = [os.path.join(code_location, "test"), "-p", "no:xdist"]
 
         if pytest_args:
             args.extend(shlex.split(pytest_args, posix=False))
 
-        logging.basicConfig(
-            filename=os.path.join(code_location, "test.log"),
-            encoding="utf-8",
-            level=logging.INFO,
-            filemode="w",
-        )
-        process = subprocess.Popen(
-            args,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT,
-            text=True,
-            cwd=code_location,
-        )
-
-        while True:
-            output = process.stdout.readline()
-            if not output and process.poll() is not None:
-                break
-
-            logger.info(output.strip())
+        # logging.basicConfig(
+        #     filename=log_file_name,
+        #     encoding="utf-8",
+        #     level=logging.INFO,
+        #     filemode="w",
+        # )
+        # process = subprocess.Popen(
+        #     args,
+        #     stdout=subprocess.PIPE,
+        #     stderr=subprocess.STDOUT,
+        #     text=True,
+        #     cwd=code_location,
+        #     creationflags=subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_CONSOLE,
+        # )
+
+        with open(log_file_name, mode="w") as f:
+            sys.stdout = f
+            sys.stderr = f
+
+            ret = pytest.main(args)
+
+        # while True:
+        #     output = process.stdout.readline()
+        #     if not output and process.poll() is not None:
+        #         break
+
+        #     logger.info(output.strip())
+
+        # logger.info(sys.stdout.getvalue())
+        # logger.error(sys.stderr.getvalue())
 
         servicemanager.LogMsg(
             servicemanager.EVENTLOG_INFORMATION_TYPE,

src/openjd/sessions/_scripts/_windows/_signal_win_subprocess.py

@@ -55,11 +55,6 @@ def signal_process(pgid: int):
     if not kernel32.GenerateConsoleCtrlEvent(CTRL_BREAK_EVENT, pgid):
         raise ctypes.WinError()
 
-    if not kernel32.FreeConsole():
-        raise ctypes.WinError()
-    if not kernel32.AttachConsole(ATTACH_PARENT_PROCESS):
-        raise ctypes.WinError()
-
 
 if __name__ == "__main__":
     signal_process(int(sys.argv[1]))

src/openjd/sessions/_subprocess.py

@@ -11,8 +11,7 @@
 from queue import Queue, Empty
 from subprocess import DEVNULL, PIPE, STDOUT, Popen, list2cmdline, run
 from threading import Event, Thread
-from typing import Any
-from typing import Callable, Literal, Optional, Sequence, cast
+from typing import Callable, Literal, Optional, Sequence, cast, Any
 
 from ._linux._capabilities import try_use_cap_kill
 from ._linux._sudo import find_sudo_child_process_group_id
@@ -623,17 +622,22 @@ def _windows_notify_subprocess(self) -> None:
             str(WINDOWS_SIGNAL_SUBPROC_SCRIPT_PATH),
             str(self._process.pid),
         ]
-        result = run(
-            cmd,
-            stdout=PIPE,
-            stderr=STDOUT,
-            stdin=DEVNULL,
-            creationflags=CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW,
+        process = LoggingSubprocess(
+            logger=self._logger,
+            args=cmd,
+            encoding=self._encoding,
+            user=self._user,
+            os_env_vars=self._os_env_vars,
+            working_dir=self._working_dir,
+            creation_flags=CREATE_NO_WINDOW,
         )
-        if result.returncode != 0:
+
+        # Blocking call
+        process.run()
+
+        if process.exit_code != 0:
             self._logger.warning(
-                f"Failed to send signal 'CTRL_BREAK_EVENT' to subprocess {self._process.pid}: %s",
-                result.stdout.decode("utf-8"),
+                f"Failed to send signal 'CTRL_BREAK_EVENT' to subprocess {self._process.pid}",
                 extra=LogExtraInfo(
                     openjd_log_content=LogContent.PROCESS_CONTROL | LogContent.EXCEPTION_INFO
                 ),

src/openjd/sessions/_win32/_helpers.py

@@ -25,13 +25,17 @@
     # Constants
     LOGON32_LOGON_INTERACTIVE,
     LOGON32_PROVIDER_DEFAULT,
+    PI_NOUI,
+    PROFILEINFO,
     # Functions
     CloseHandle,
     CreateEnvironmentBlock,
     DestroyEnvironmentBlock,
     GetCurrentProcessId,
     LogonUserW,
     ProcessIdToSessionId,
+    LoadUserProfileW,
+    UnloadUserProfile,
 )
 
 
@@ -166,3 +170,33 @@ def environment_block_from_dict(env: dict[str, str]) -> c_wchar_p:
     env_block_str = null_delimited + "\0"
 
     return c_wchar_p(env_block_str)
+
+
+def load_user_profile(token: HANDLE, username: str) -> PROFILEINFO:
+    """
+    Load the user profile for the given logon token and user name
+
+    NOTE: The caller *MUST* call unload_user_profile when finished with the user profile
+
+    See: https://learn.microsoft.com/en-us/windows/win32/api/userenv/nf-userenv-loaduserprofilew
+    """
+    profile_info = PROFILEINFO()
+    profile_info.dwSize = sizeof(PROFILEINFO)
+    profile_info.lpUserName = username
+    profile_info.dwFlags = PI_NOUI
+    profile_info.lpProfilePath = None
+    
+    if not LoadUserProfileW(token, byref(profile_info)):
+        raise WinError()
+    
+    return profile_info
+
+
+def unload_user_profile(token: HANDLE, profile_info: PROFILEINFO) -> None:
+    """
+    Unload the user profile for the given token and profile.
+
+    See: https://learn.microsoft.com/en-us/windows/win32/api/userenv/nf-userenv-unloaduserprofile
+    """
+    if not UnloadUserProfile(token, profile_info.hProfile):
+        raise WinError()

test/openjd/sessions/conftest.py

@@ -21,6 +21,8 @@
     from openjd.sessions._win32._helpers import (  # type: ignore
         get_current_process_session_id,
         logon_user_context,
+        load_user_profile,
+        unload_user_profile,
     )
 
     TEST_RUNNING_IN_WINDOWS_SESSION_0 = 0 == get_current_process_session_id()
@@ -251,10 +253,10 @@ def windows_user() -> Generator[WindowsSessionUser, None, None]:
 
     if TEST_RUNNING_IN_WINDOWS_SESSION_0:
         try:
-            # Note: We don't load the user profile; it's currently not needed by our tests,
-            # and we're getting a mysterious crash when unloading it.
             with logon_user_context(user, password) as logon_token:
+                profile_info = load_user_profile(logon_token, user)
                 yield WindowsSessionUser(user, logon_token=logon_token)
+                unload_user_profile(logon_token, profile_info)
         except OSError as e:
             raise Exception(
                 f"Could not logon as {user}. Check the password that was provided in {WIN_PASS_ENV_VAR}."

test/openjd/sessions/test_runner_base.py

@@ -128,7 +128,7 @@ def test_basic_run(self, tmp_path: Path) -> None:
             logger=MagicMock(), session_working_directory=tmp_path, callback=callback
         ) as runner:
             # WHEN
-            runner._run([sys.executable, "-c", "import time; time.sleep(0.25)"])
+            runner._run([sys.executable.lower().replace("pythonservice.exe", "python.exe"), "-c", "import time; time.sleep(0.25)"])
 
             # THEN
             assert runner.state == ScriptRunnerState.RUNNING
@@ -173,7 +173,7 @@ def test_working_dir_is_cwd(
             logger=logger, session_working_directory=tmp_path, startup_directory=tmp_path
         ) as runner:
             # WHEN
-            runner._run([sys.executable, "-c", "import os; print(os.getcwd())"])
+            runner._run([sys.executable.lower().replace("pythonservice.exe", "python.exe"), "-c", "import os; print(os.getcwd())"])
             # Wait until the process exits.
             while runner.state == ScriptRunnerState.RUNNING:
                 time.sleep(0.1)
@@ -189,7 +189,7 @@ def test_failing_run(self, tmp_path: Path) -> None:
         # GIVEN
         with TerminatingRunner(logger=MagicMock(), session_working_directory=tmp_path) as runner:
             # WHEN
-            runner._run([sys.executable, "-c", "import sys; sys.exit(1)"])
+            runner._run([sys.executable.lower().replace("pythonservice.exe", "python.exe"), "-c", "import sys; sys.exit(1)"])
 
             # THEN
             while runner.state == ScriptRunnerState.RUNNING:
@@ -256,7 +256,7 @@ def test_run_with_env_vars(
             # WHEN
             runner._run(
                 [
-                    sys.executable,
+                    sys.executable.lower().replace("pythonservice.exe", "python.exe"),
                     "-c",
                     r"import os;print(*(f'{k} = {v}' for k,v in os.environ.items()), sep='\n')",
                 ]
@@ -297,8 +297,8 @@ def test_run_as_posix_user(
             # WHEN
             runner._run(
                 [
-                    # Note: Intentionally not `sys.executable`. Reasons:
-                    #  1) This is a cross-account command, and sys.executable may be in a user-specific venv
+                    # Note: Intentionally not `sys.executable.lower().replace("pythonservice.exe", "python.exe")`. Reasons:
+                    #  1) This is a cross-account command, and sys.executable.lower().replace("pythonservice.exe", "python.exe") may be in a user-specific venv
                     #  2) This test is, generally, intended to be run in a docker container where the system
                     #     python is the correct version that we want to run under.
                     "python",
@@ -348,8 +348,8 @@ def test_run_as_posix_user_with_env_vars(
             # WHEN
             runner._run(
                 [
-                    # Note: Intentionally not `sys.executable`. Reasons:
-                    #  1) This is a cross-account command, and sys.executable may be in a user-specific venv
+                    # Note: Intentionally not `sys.executable.lower().replace("pythonservice.exe", "python.exe")`. Reasons:
+                    #  1) This is a cross-account command, and sys.executable.lower().replace("pythonservice.exe", "python.exe") may be in a user-specific venv
                     #  2) This test is, generally, intended to be run in a docker container where the system
                     #     python is the correct version that we want to run under.
                     "python",
@@ -467,8 +467,8 @@ def test_run_as_windows_user_with_env_vars(
             # WHEN
             runner._run(
                 [
-                    # Note: Intentionally not `sys.executable`. Reasons:
-                    #  1) This is a cross-account command, and sys.executable may be in a user-specific venv
+                    # Note: Intentionally not `sys.executable.lower().replace("pythonservice.exe", "python.exe")`. Reasons:
+                    #  1) This is a cross-account command, and sys.executable.lower().replace("pythonservice.exe", "python.exe") may be in a user-specific venv
                     #  2) This test is, generally, intended to be run in a docker container where the system
                     #     python is the correct version that we want to run under.
                     "python",
@@ -518,8 +518,8 @@ def test_does_not_inherit_env_vars_posix(
             # WHEN
             runner._run(
                 [
-                    # Note: Intentionally not `sys.executable`. Reasons:
-                    #  1) This is a cross-account command, and sys.executable may be in a user-specific venv
+                    # Note: Intentionally not `sys.executable.lower().replace("pythonservice.exe", "python.exe")`. Reasons:
+                    #  1) This is a cross-account command, and sys.executable.lower().replace("pythonservice.exe", "python.exe") may be in a user-specific venv
                     #  2) This test is, generally, intended to be run in a docker container where the system
                     #     python is the correct version that we want to run under.
                     "python",
@@ -570,8 +570,8 @@ def test_does_not_inherit_env_vars_windows(
         ) as runner:
             # WHEN
             py_script = f"import os; v=os.environ.get('{var_name}'); print('NOT_PRESENT' if v is None else v)"
-            # Use the default 'python' rather than 'sys.executable' since we typically do not have access to
-            # sys.executable when running with impersonation since it's in a hatch environment for the local user.
+            # Use the default 'python' rather than 'sys.executable.lower().replace("pythonservice.exe", "python.exe")' since we typically do not have access to
+            # sys.executable.lower().replace("pythonservice.exe", "python.exe") when running with impersonation since it's in a hatch environment for the local user.
             runner._run(["python", "-c", py_script])
 
             # THEN
@@ -598,11 +598,11 @@ def test_cannot_run_twice(self, tmp_path: Path) -> None:
             logger=MagicMock(), session_working_directory=tmp_path, callback=callback
         ) as runner:
             # WHEN
-            runner._run([sys.executable, "-c", "print('hello')"])
+            runner._run([sys.executable.lower().replace("pythonservice.exe", "python.exe"), "-c", "print('hello')"])
 
             # THEN
             with pytest.raises(RuntimeError):
-                runner._run([sys.executable, "-c", "print('hello')"])
+                runner._run([sys.executable.lower().replace("pythonservice.exe", "python.exe"), "-c", "print('hello')"])
 
     @pytest.mark.usefixtures("message_queue", "queue_handler")
     def test_run_action(
@@ -623,7 +623,7 @@ def test_run_action(
         python_app_loc = (Path(__file__).parent / "support_files" / "app_20s_run.py").resolve()
         symtab = SymbolTable(
             source={
-                "Task.PythonInterpreter": sys.executable,
+                "Task.PythonInterpreter": sys.executable.lower().replace("pythonservice.exe", "python.exe"),
                 "Task.ScriptFile": str(python_app_loc),
             }
         )
@@ -683,7 +683,7 @@ def test_run_action_default_timeout(
         python_app_loc = (Path(__file__).parent / "support_files" / "app_20s_run.py").resolve()
         symtab = SymbolTable(
             source={
-                "Task.PythonInterpreter": sys.executable,
+                "Task.PythonInterpreter": sys.executable.lower().replace("pythonservice.exe", "python.exe"),
                 "Task.ScriptFile": str(python_app_loc),
             }
         )
@@ -755,7 +755,7 @@ def test_cancel_terminate(
             logger=logger, session_working_directory=tmp_path, callback=callback
         ) as runner:
             python_app_loc = (Path(__file__).parent / "support_files" / "app_20s_run.py").resolve()
-            runner._run([sys.executable, str(python_app_loc)])
+            runner._run([sys.executable.lower().replace("pythonservice.exe", "python.exe"), str(python_app_loc)])
 
             # WHEN
             runner.cancel()
@@ -789,7 +789,7 @@ def test_run_with_time_limit(
             python_app_loc = (Path(__file__).parent / "support_files" / "app_20s_run.py").resolve()
 
             # WHEN
-            runner._run([sys.executable, str(python_app_loc)], time_limit=timedelta(seconds=1))
+            runner._run([sys.executable.lower().replace("pythonservice.exe", "python.exe"), str(python_app_loc)], time_limit=timedelta(seconds=1))
 
             # THEN
             # Wait until the process exits. We'll be in CANCELING state between when the timeout is reached
@@ -818,7 +818,7 @@ def test_cancel_notify(
             python_app_loc = (
                 Path(__file__).parent / "support_files" / "app_20s_run_ignore_signal.py"
             ).resolve()
-            runner._run([sys.executable, str(python_app_loc)])
+            runner._run([sys.executable.lower().replace("pythonservice.exe", "python.exe"), str(python_app_loc)])
 
             # WHEN
             secs = 2 if not is_windows() else 5
@@ -971,7 +971,7 @@ def test_cancel_double_cancel_notify(
             python_app_loc = (
                 Path(__file__).parent / "support_files" / "app_20s_run_ignore_signal.py"
             ).resolve()
-            runner._run([sys.executable, str(python_app_loc)])
+            runner._run([sys.executable.lower().replace("pythonservice.exe", "python.exe"), str(python_app_loc)])
 
             # WHEN
             secs = 2 if not is_windows() else 5