From 2b5a6c4057ba97b0d9f9c88b7145dc54ce7ea006 Mon Sep 17 00:00:00 2001 From: Alex Florenti Date: Fri, 16 Oct 2026 17:12:54 +0200 Subject: [PATCH] Fix session teardown lock inversion and bound POSIX shell cleanup --- app_shell/shell.py | 212 +++++++++++-- core/agent.py | 124 +++++--- tests/README.md | 42 +++ tests/test_shell_lifecycle.py | 575 ++++++++++++++++++++++++++++++++++ 4 files changed, 868 insertions(+), 85 deletions(-) create mode 100644 tests/README.md create mode 100644 tests/test_shell_lifecycle.py diff --git a/app_shell/shell.py b/app_shell/shell.py index 19b3617..d25235f 100644 --- a/app_shell/shell.py +++ b/app_shell/shell.py @@ -7,6 +7,7 @@ ''' import threading +import errno import os import sys import struct @@ -37,12 +38,120 @@ SHELL_TIMEOUT_RECOVERY = 20; #Seconds SHELL_VERSION = 1 +# Poll only while cleaning up a child. There is no idle periodic worker. +SHELL_PROCESS_TERM_GRACE = 0.5 +SHELL_PROCESS_KILL_GRACE = 0.5 +SHELL_PROCESS_REAP_INTERVAL = 1.0 + + +def _shell_process_exited(pid): + try: + waited, status = os.waitpid(pid, os.WNOHANG) + return waited == pid + except OSError as ex: + if ex.errno == errno.ECHILD: + return True + if ex.errno == errno.EINTR: + return False + raise + + +def _wait_shell_process(pid, grace): + deadline = time.monotonic() + grace + while True: + if _shell_process_exited(pid): + return True + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + time.sleep(min(0.05, remaining)) + + +def _stop_shell_foreground(pid, pio): + # Signal only the foreground job of this still-open, owned PTY. Root + # shells need explicit job cleanup too; terminal hangup alone is not + # sufficient in every launch context. Never signal our own group. + if pid < 1 or pio is None: + return + try: + group = os.tcgetpgrp(pio) + if group <= 1 or group == os.getpgrp(): + return + if os.getsid(group) != pid: + return + os.killpg(group, signal.SIGTERM) + deadline = time.monotonic() + SHELL_PROCESS_TERM_GRACE + while os.tcgetpgrp(pio) == group: + try: + os.killpg(group, 0) + except OSError as ex: + if ex.errno == errno.ESRCH: + return + raise + if time.monotonic() >= deadline: + # Recheck terminal ownership after the grace period, so a + # foreground change cannot make us signal a stale group. + if os.tcgetpgrp(pio) == group: + os.killpg(group, signal.SIGKILL) + return + time.sleep(0.05) + except OSError: + # No foreground group yet, job already gone or PTY already hung up. + # Continue the bounded cleanup of the directly owned shell process. + return + + +class ShellProcessReaper: + """One lazy worker per Agent, shared across shell app reloads. + + Only this registry owns deferred child PIDs. It never sends signals and + never uses a blocking waitpid. It exits as soon as the registry is empty. + """ + def __init__(self, agent_main): + self._agent_main = agent_main + self._condition = threading.Condition() + self._pending = set() + self._thread = None + + def add(self, pid): + with self._condition: + self._pending.add(pid) + if self._thread is None: + self._thread = threading.Thread(target=self._run, name="ShellProcessReaper") + self._thread.daemon = True + self._thread.start() + + def _run(self): + while True: + with self._condition: + if not self._pending: + self._thread = None + return + pending = list(self._pending) + for pid in pending: + try: + if _shell_process_exited(pid): + with self._condition: + self._pending.discard(pid) + except OSError: + # Keep ownership and retry; the initial deferral was logged. + pass + with self._condition: + if self._pending: + self._condition.wait(SHELL_PROCESS_REAP_INTERVAL) + + class Shell(): def __init__(self, agent_main): self._agent_main=agent_main self._list = {} self._list_semaphore = threading.Condition() + # App construction is serialized by Agent.get_app. Retain the registry + # on the Agent so an app reload cannot create one reaper per generation. + if not hasattr(agent_main, '_shell_process_reaper'): + agent_main._shell_process_reaper = ShellProcessReaper(agent_main) + self._process_reaper = agent_main._shell_process_reaper def destroy(self,bforce): if not bforce: @@ -515,6 +624,8 @@ def __init__(self, mgr, sid, col, row): self._reader=None self._writer=None self._libcrypt=None + self._cleanup_started=False + self._activity_registered=False def get_id(self): return self._id @@ -745,6 +856,9 @@ def open_session(self,u,p): except: os._exit(1) + # Take ownership before operations that can fail. + self._ppid = ppid + self._pio = pio fl = fcntl.fcntl(sys.stdin, fcntl.F_GETFL) fcntl.fcntl(pio, fcntl.F_SETFL, fl | os.O_NONBLOCK) fcntl.ioctl(pio, termios.TIOCSWINSZ, struct.pack("hhhh", self._rows, self._cols, 0, 0)) @@ -754,6 +868,7 @@ def open_session(self,u,p): self._writer = io.open(pio, 'wb', closefd=False) try: self._manager._cinfo.inc_activities_value("shellSession") + self._activity_registered=True except: None self._login_request = None @@ -761,47 +876,76 @@ def open_session(self,u,p): self.terminate() def _processIsAlive(self): - try: - if self._ppid>=0: - os.waitpid(self._ppid, os.WNOHANG) - os.kill(self._ppid, 0) # kill -0 tells us it's still alive - return True - else: + with self._semaphore: + if self._cleanup_started or self._ppid < 0: return False - except OSError: - return False - - def terminate(self): - self._login_request=None - try: - self._manager._cinfo.dec_activities_value("shellSession") - except: - None - self._bterm=True - if self._reader is not None: try: - self._reader.close() - except: - None - if self._writer is not None: + if _shell_process_exited(self._ppid): + self._ppid = -1 + return False + return True + except OSError: + # Let terminate retain ownership and retry cleanup on an OS error. + return True + + def terminate(self): + with self._semaphore: + if self._cleanup_started: + return + self._cleanup_started = True + self._login_request = None + self._bterm = True + pid, pio = self._ppid, self._pio + reader, writer = self._reader, self._writer + registered = self._activity_registered + self._ppid, self._pio = -1, None + self._reader, self._writer = None, None + self._activity_registered = False + _stop_shell_foreground(pid, pio) + if registered: try: - self._writer.close() + self._manager._cinfo.dec_activities_value("shellSession") except: - None - if self._ppid>=0: - if self._processIsAlive(): + self._manager._shlmain._agent_main.write_except(utils.get_exception()) + # io.open(..., closefd=False) does not close the PTY master. Close it + # explicitly too, releasing the terminal and hanging up foreground jobs. + for stream in (reader, writer): + if stream is not None: try: - os.kill(self._ppid, signal.SIGTERM) + stream.close() except: - None - time.sleep(0.5) - if self._processIsAlive(): + pass + if pio is not None: + try: + os.close(pio) + except OSError: + pass + if pid < 0: + return + reaped = False + try: + reaped = _shell_process_exited(pid) + if not reaped: try: - os.kill(self._ppid, signal.SIGKILL) - os.waitpid(self._ppid, 0) - except: - None - self._ppid=-1 + os.kill(pid, signal.SIGTERM) + except OSError as ex: + if ex.errno != errno.ESRCH: + raise + reaped = _wait_shell_process(pid, SHELL_PROCESS_TERM_GRACE) + if not reaped: + try: + os.kill(pid, signal.SIGKILL) + except OSError as ex: + if ex.errno != errno.ESRCH: + raise + reaped = _wait_shell_process(pid, SHELL_PROCESS_KILL_GRACE) + except OSError: + self._manager._shlmain._agent_main.write_except(utils.get_exception()) + finally: + if not reaped: + self._manager._shlmain._process_reaper.add(pid) + self._manager._shlmain._agent_main.write_err( + "AppShell:: process %s not reaped within cleanup grace; deferring collection" % pid) def is_terminate(self): if self._login_request is not None: diff --git a/core/agent.py b/core/agent.py index 4a62588..325fc05 100755 --- a/core/agent.py +++ b/core/agent.py @@ -196,6 +196,7 @@ def __init__(self,args): self._http_socket_pool=None self._agent_conn = None self._sessions={} + self._sessions_closing=set() self._sessions_update_status="" self._sessions_semaphore = threading.Condition() self._libs={} @@ -1982,36 +1983,48 @@ def _unload_apps(self): finally: self._libs_apps_semaphore.release() + def _unload_apps_if_idle(self): + # Same lock order as session publication: apps -> sessions. Release the + # sessions lock before calling app destroy methods, retaining the app + # registry lock so a new session cannot race with the idle unload. + with self._libs_apps_semaphore: + with self._sessions_semaphore: + if self._sessions: + return + self._unload_apps() + def _fire_close_conn_apps(self, idconn): - for k in self._apps: - md = self._apps[k] + # Snapshot app instances, but never run their callbacks under either + # the app registry lock or the global sessions lock. + with self._libs_apps_semaphore: + apps = list(self._apps.values()) + for md in apps: try: - func = None - try: - func = getattr(md, 'on_conn_close') - except AttributeError: - None + func = getattr(md, 'on_conn_close', None) if func is not None: func(idconn) except: - e = utils.get_exception() - self.write_except(e) + self.write_except(utils.get_exception()) def _close_all_sessions(self): - self._sessions_semaphore.acquire() - try: - for sid in self._sessions.keys(): + with self._sessions_semaphore: + sessions = [(sid, ses) for sid, ses in self._sessions.items() + if sid not in self._sessions_closing] + self._sessions_closing.update(sid for sid, ses in sessions) + for sid, ses in sessions: + try: try: - ses = self._sessions[sid] ses.close() + finally: self._fire_close_conn_apps(sid) - except: - ex = utils.get_exception() - self.write_err(utils.exception_to_string(ex)) - self._sessions={} - finally: - self._sessions_semaphore.release() - self._unload_apps() + except: + self.write_err(utils.exception_to_string(utils.get_exception())) + finally: + with self._sessions_semaphore: + if self._sessions.get(sid) is ses: + del self._sessions[sid] + self._sessions_closing.discard(sid) + self._unload_apps_if_idle() def open_session(self, msg): resp = {} @@ -2019,6 +2032,7 @@ def open_session(self, msg): conn_rcr = None appconn = None try: + iconntimeout = None if "connectionTimeout" in msg: try: iconntimeout=utils.Counter(int(msg["connectionTimeout"])-2) @@ -2046,26 +2060,32 @@ def open_session(self, msg): conn_rcr.set_max_attempt(int(msg["connRecoveryMaxAttempt"])) appconn = Connection(self, None, prop_conn, self.get_proxy_info()) sinfo=None - self._sessions_semaphore.acquire() - try: - while self._sessions_update_status=="INPROGRESS" and iconntimeout.is_elapsed(): - self._sessions_semaphore.wait(1) #WAIT FOR COMPLETE UPDATE - if self._sessions_update_status!="": - raise Exception("Open session failed: Update in progress.") - while True: - sid = generate_key(30) - if sid not in self._sessions: - sinfo=Session(self,appconn,sid,msg) - self._sessions[sid]=sinfo - resp["idSession"]=sid - resp["waitAccept"]=sinfo.get_wait_accept() - resp["passwordRequest"]=sinfo.get_password_request() - if conn_rcr is not None: - conn_rcr.set_msg_log("session (id: " + sinfo.get_idsession() + ", node: " + sinfo.get_host()+")") - appconn.set_recovery_conf(conn_rcr) - break - finally: - self._sessions_semaphore.release() + while True: + with self._sessions_semaphore: + while (self._sessions_update_status == "INPROGRESS" + and not iconntimeout.is_elapsed()): + self._sessions_semaphore.wait(0.1) + if self._sessions_update_status != "": + raise Exception("Open session failed: Update in progress.") + with self._libs_apps_semaphore: + with self._sessions_semaphore: + if self._sessions_update_status == "INPROGRESS": + continue + if self._sessions_update_status != "": + raise Exception("Open session failed: Update in progress.") + while True: + sid = generate_key(30) + if sid not in self._sessions: + sinfo=Session(self,appconn,sid,msg) + self._sessions[sid]=sinfo + resp["idSession"]=sid + resp["waitAccept"]=sinfo.get_wait_accept() + resp["passwordRequest"]=sinfo.get_password_request() + if conn_rcr is not None: + conn_rcr.set_msg_log("session (id: " + sinfo.get_idsession() + ", node: " + sinfo.get_host()+")") + appconn.set_recovery_conf(conn_rcr) + break + break except: ee = utils.get_exception() if appconn is not None: @@ -2077,19 +2097,21 @@ def open_session(self, msg): return resp def close_session(self, ses): - bcloseapps=False - self._sessions_semaphore.acquire() + sid = ses.get_idsession() + with self._sessions_semaphore: + if self._sessions.get(sid) is not ses or sid in self._sessions_closing: + return + # Keep closing sessions registered until callbacks finish. Updates + # and app unloading must not mistake in-flight cleanup for idle. + self._sessions_closing.add(sid) try: - sid = ses.get_idsession() - if sid in self._sessions: - del self._sessions[sid] - self._fire_close_conn_apps(sid) - if len(self._sessions)==0: - bcloseapps=True + self._fire_close_conn_apps(sid) finally: - self._sessions_semaphore.release() - if bcloseapps: - self._unload_apps() + with self._sessions_semaphore: + if self._sessions.get(sid) is ses: + del self._sessions[sid] + self._sessions_closing.discard(sid) + self._unload_apps_if_idle() def get_app_permission(self,cinfo,name): diff --git a/tests/README.md b/tests/README.md new file mode 100644 index 0000000..34ab0f7 --- /dev/null +++ b/tests/README.md @@ -0,0 +1,42 @@ +# Session and shell lifecycle regression tests + +Run from the repository root on a POSIX system with Python 3 (validated with Python 3.10.5): + +```sh +python3 -B tests/test_shell_lifecycle.py +``` + +The suite uses only the standard library. It compiles the actual class and +method bodies from `core/agent.py` and `app_shell/shell.py` using `ast`, replacing +network, native libraries and service bootstrap with small fakes. No agent is +installed or started, no configuration is read, and no DWService account or +network connection is needed. This allows testing cleanup and lock ordering +without compiling the native agent dependencies. + +The 33 tests cover callback lock inversion, concurrent open/close and idle app +unloading, update waits, duplicate and stale closes, bounded nonblocking reaping, +PTY descriptor ownership, partial initialization, signal guards, and a lazy +reaper shared across app reloads. The two registry stress tests exercise 650 +session lifecycles while checking callback uniqueness and/or heartbeat progress. +Injected OS failures never signal real PIDs. + +Five integration tests create real `/bin/sh` PTYs and child processes on macOS; +they are skipped elsewhere. They check immediate open/close (30 cycles), normal +foreground jobs, inherited ignored SIGTERM, foreground jobs ignoring both HUP +and TERM, and a direct child ignoring TERM. They assert process disappearance +before emergency cleanup and check descriptor/activity counts. Child shells use +a temporary HOME. Run in a disposable test environment: these tests create and +signal their own children, and use `ps` to observe them. The same suite can be run +as root there to exercise the service's privilege context. + +`DWAGENT_TEST_SOURCE` can point to a different repository checkout for negative +controls. These four tests fail on the unmodified base and pass with the fix: + +- `LifecycleTests.test_callback_releases_global_lock_and_remains_registered` +- `ProcessTests.test_bounded_wait_and_eventual_reap` +- `ProcessTests.test_pty_master_closed_after_one_termination` +- `ProcessTests.test_initialization_failure_closes_owned_fd` + +These are isolated lifecycle tests, not a full distribution/bootstrap test or +proof that every descendant (for example, a daemonized background job) is killed. +Windows terminal cleanup is not exercised by this suite. diff --git a/tests/test_shell_lifecycle.py b/tests/test_shell_lifecycle.py new file mode 100644 index 0000000..6ff9ad3 --- /dev/null +++ b/tests/test_shell_lifecycle.py @@ -0,0 +1,575 @@ +"""Regression tests using DWAgent method bodies from this checkout. + +No service, credentials, network, installed files, or unrelated processes are +touched. Actual PTYs use a temporary HOME. Fake OS faults never signal real PIDs. +""" +import ast +import errno +import fcntl +import io +import itertools +import json +import os +from pathlib import Path +import pty +import pwd +import signal +import struct +import subprocess +import sys +import tempfile +import termios +import threading +import time +from types import SimpleNamespace +import unittest +from unittest import mock + +ROOT=Path(__file__).resolve().parent +SOURCE=Path(os.environ.get('DWAGENT_TEST_SOURCE',ROOT.parent)) + + +class Counter: + def __init__(self,limit=0): self.limit=limit;self.reset() + def reset(self): self.start=time.monotonic() + def is_elapsed(self,seconds=None): return time.monotonic()-self.start >= (self.limit if seconds is None else seconds) + + +def method(source, cls, name, env): + tree=ast.parse(source.read_text()) + parent=next(n for n in tree.body if isinstance(n,ast.ClassDef) and n.name==cls) + node=next(n for n in parent.body if isinstance(n,ast.FunctionDef) and n.name==name) + local=dict(env) + exec(compile(ast.Module(body=[node],type_ignores=[]),str(source),'exec'),local) + return local[name] + + +def load(os_api=os): + utils=SimpleNamespace(Counter=Counter,get_exception=lambda:sys.exc_info()[1], + exception_to_string=str,path_exists=os.path.exists, + str_to_bytes=lambda s,enc='utf8':s.encode(enc), + bytes_to_str=lambda s,enc='utf8':s.decode(enc,errors='replace')) + platform=SimpleNamespace(is_windows=lambda:False,is_linux=lambda:False,is_mac=lambda:True, + generate_key=lambda n:os.urandom(n).hex()) + native=SimpleNamespace(get_instance=lambda:SimpleNamespace(get_utf8_lang=lambda:'en_US.UTF-8')) + env=dict(threading=threading,time=time,os=os_api,errno=errno,signal=signal,utils=utils, + io=io,sys=sys,struct=struct,json=json,fcntl=fcntl,pty=pty,termios=termios, + agent=platform,native=native,subprocess=subprocess, + SHELL_TIMEOUT_CONNECTION=40,SHELL_TIMEOUT_RECOVERY=20,SHELL_VERSION=1) + shell_path=SOURCE/'app_shell/shell.py' + tree=ast.parse(shell_path.read_text()) + nodes=[n for n in tree.body if isinstance(n,(ast.ClassDef,ast.FunctionDef)) + or isinstance(n,ast.Assign) and all(isinstance(t,ast.Name) and t.id.startswith('SHELL_') for t in n.targets)] + exec(compile(ast.Module(body=nodes,type_ignores=[]),str(shell_path),'exec'),env) + agent_path=SOURCE/'core/agent.py' + methods={name:method(agent_path,'Agent',name,env) for name in ( + '_fire_close_conn_apps','close_session','_close_all_sessions','get_active_sessions_status', + '_unload_apps','_unload_app')} + try: methods['_unload_apps_if_idle']=method(agent_path,'Agent','_unload_apps_if_idle',env) + except StopIteration: pass + class TestAgent: + def __init__(self): + self._sessions_semaphore=threading.Condition() + self._libs_apps_semaphore=threading.Condition() + self._sessions={};self._sessions_closing=set();self._sessions_update_status='' + self._apps={};self.errors=[];self.infos=[] + def write_except(self,error,*args): self.errors.append(str(error)) + def write_err(self,error): self.errors.append(str(error)) + def write_info(self,text): self.infos.append(text) + def get_config(self,key,default=None): return False if key=='recovery_session' else default + def get_proxy_info(self): return None + def _get_sys_info(self): return {} + def _get_agent_distr_id(self): return 65 + for name,fn in methods.items(): setattr(TestAgent,name,fn) + class TestSession: + def __init__(self,a,conn=None,sid='one',msg=None): + self._agent=a;self.sid=sid;self._activities={'shellSession':0} + def get_idsession(self): return self.sid + def get_wait_accept(self): return False + def get_password_request(self): return False + def get_last_activity_time(self): return time.time() + def get_init_time(self): return 0 + def get_access_type(self): return 'qa' + def get_user_name(self): return 'qa' + def get_ipaddress(self): return 'isolated' + def close(self): self._agent.close_session(self) + for name in ('inc_activities_value','dec_activities_value','get_activities'): + setattr(TestSession,name,method(agent_path,'Session',name,env)) + class Connection: + def __init__(self,*args): pass + def close(self): pass + counter=itertools.count() + open_env=dict(env,Session=TestSession,Connection=Connection,generate_key=lambda n:'new-%s'%next(counter)) + TestAgent.open_session=method(agent_path,'Agent','open_session',open_env) + return env,TestAgent,TestSession + + +def join(t,limit=2): + t.join(limit) + if t.is_alive(): raise AssertionError('thread did not finish: '+t.name) + + +def launch(fn,name='qa-worker'): + errors=[] + def run(): + try: fn() + except BaseException as e: errors.append(e) + t=threading.Thread(target=run,name=name);t.daemon=True;t.start() + return t,errors + + +OPEN=dict(connectionTimeout=8,connServer='isolated',connPort=0,connInstance='qa', + connVersion=1,connUser='qa',connPassword='not-a-credential') + + +class LifecycleTests(unittest.TestCase): + def setUp(self): + self.env,A,self.Session=load();self.a=A() + def add(self,sid='one'): + s=self.Session(self.a,sid=sid);self.a._sessions[sid]=s;return s + def app(self,callback=lambda sid:None,destroy=lambda force:True): + p=SimpleNamespace(on_conn_close=callback,destroy=destroy) + self.a._apps['test']=p;return p + def test_callback_releases_global_lock_and_remains_registered(self): + s=self.add();entered,release=threading.Event(),threading.Event() + self.app(lambda sid:(entered.set(),release.wait(3))) + t,errors=launch(lambda:self.a.close_session(s)) + try: + self.assertTrue(entered.wait(1)) + got=self.a._sessions_semaphore.acquire(timeout=.15) + self.assertTrue(got,'global sessions lock held during app callback') + if got: + try: self.assertIs(self.a._sessions['one'],s) + finally: self.a._sessions_semaphore.release() + status,errs=launch(self.a.get_active_sessions_status,'heartbeat') + join(status,.3);self.assertFalse(errs) + finally: release.set();join(t) + self.assertFalse(errors);self.assertFalse(self.a._sessions) + def test_lock_inversion_with_concurrent_shell_open(self): + s=self.add();manager_lock=threading.Condition() + manager_held,callback_entered=threading.Event(),threading.Event() + def cb(sid): + callback_entered.set() + with manager_lock: pass + self.app(cb) + def opener(): + with manager_lock: + manager_held.set() + if not callback_entered.wait(1): raise AssertionError('no callback') + s.inc_activities_value('shellSession') + opening,oe=launch(opener,'opening');self.assertTrue(manager_held.wait(1)) + closing,ce=launch(lambda:self.a.close_session(s),'closing') + join(opening,.8);join(closing,.8) + self.assertFalse(oe+ce) + def test_close_all_callbacks_and_session_close_release_global_lock(self): + sessions=[self.add(str(n)) for n in range(3)] + seen=[] + def check(): + got=self.a._sessions_semaphore.acquire(timeout=.1) + if got:self.a._sessions_semaphore.release() + seen.append(got) + def cb(sid): + t,err=launch(check);join(t);self.assertFalse(err) + self.app(cb) + self.a._close_all_sessions() + self.assertEqual(seen,[True]*3);self.assertFalse(self.a._sessions) + self.assertFalse(self.a._sessions_closing) + def test_duplicate_close_has_one_callback(self): + s=self.add();entered,release=threading.Event(),threading.Event();calls=[] + self.app(lambda sid:(calls.append(sid),entered.set(),release.wait(2))) + first,errors=launch(lambda:self.a.close_session(s));self.assertTrue(entered.wait(1)) + threads=[launch(lambda:self.a.close_session(s)) for _ in range(12)] + for t,e in threads:join(t);self.assertFalse(e) + release.set();join(first);self.assertFalse(errors);self.assertEqual(calls,['one']) + def test_stale_session_cannot_close_replacement(self): + old=self.Session(self.a,sid='one');fresh=self.add();calls=[] + self.app(lambda sid:calls.append(sid)) + self.a.close_session(old) + self.assertIs(self.a._sessions['one'],fresh);self.assertEqual(calls,[]) + def test_callback_can_open_new_session_without_unloading_it(self): + old=self.add();destroyed=[] + self.app(lambda sid:self.a.open_session(OPEN),lambda force:(destroyed.append(force),True)[1]) + self.a.close_session(old) + self.assertEqual(len(self.a._sessions),1);self.assertFalse(destroyed) + self.assertIn('test',self.a._apps) + def test_idle_unload_serializes_new_publication(self): + old=self.add();entered,release=threading.Event(),threading.Event() + self.app(destroy=lambda force:(entered.set(),release.wait(2),True)[2]) + closing,ce=launch(lambda:self.a.close_session(old));self.assertTrue(entered.wait(1)) + opening,oe=launch(lambda:self.a.open_session(OPEN)) + time.sleep(.05) + with self.a._sessions_semaphore:self.assertFalse(self.a._sessions) + self.assertTrue(opening.is_alive(),'new session published during old app unload') + # Heartbeat still works while destroy waits under the app registry lock. + status,se=launch(self.a.get_active_sessions_status);join(status,.2);self.assertFalse(se) + release.set();join(closing);join(opening);self.assertFalse(ce+oe) + self.assertEqual(len(self.a._sessions),1) + def test_all_close_does_not_unload_newly_opened_session(self): + self.add('old1');self.add('old2');calls=[];destroyed=[] + def callback(sid): + calls.append(sid) + if sid=='old1': self.a.open_session(OPEN) + self.app(callback,lambda force:(destroyed.append(force),True)[1]) + self.a._close_all_sessions() + self.assertEqual(calls,['old1','old2']);self.assertEqual(len(self.a._sessions),1) + self.assertFalse(destroyed) + def test_update_wait_does_not_hold_app_registry_lock(self): + self.a._sessions_update_status='INPROGRESS' + opening,oe=launch(lambda:self.a.open_session(OPEN),'open-during-update') + time.sleep(.05) + def update(): + self.a._unload_apps() + with self.a._sessions_semaphore: + self.a._sessions_update_status='' + self.a._sessions_semaphore.notify_all() + updater,ue=launch(update,'finish-update') + join(updater,.5);join(opening,.5) + self.assertFalse(ue+oe);self.assertEqual(len(self.a._sessions),1) + def test_expired_update_wait_rejects_without_publication(self): + self.a._sessions_update_status='INPROGRESS' + with self.assertRaisesRegex(Exception,'Update in progress'): + self.a.open_session(dict(OPEN,connectionTimeout=2)) + self.assertFalse(self.a._sessions) + def test_optional_connection_timeout_can_be_missing_or_invalid(self): + msg=dict(OPEN);msg.pop('connectionTimeout') + self.a.open_session(msg) + self.a.open_session(dict(OPEN,connectionTimeout='invalid')) + self.assertEqual(len(self.a._sessions),2) + def test_callback_exception_does_not_skip_other_apps_or_cleanup(self): + s=self.add();seen=[] + def bad(sid): raise RuntimeError('injected callback error') + self.app(bad) + self.a._apps['other']=SimpleNamespace(on_conn_close=lambda sid:seen.append(sid),destroy=lambda force:True) + self.a.close_session(s) + self.assertEqual(seen,['one']);self.assertFalse(self.a._sessions) + self.assertFalse(self.a._sessions_closing);self.assertTrue(self.a.errors) + def test_app_registry_snapshot_survives_callback_mutation(self): + s=self.add();seen=[] + def change(sid): + with self.a._libs_apps_semaphore:self.a._apps.pop('other',None) + self.app(change) + self.a._apps['other']=SimpleNamespace(on_conn_close=lambda sid:seen.append(sid),destroy=lambda force:True) + self.a.close_session(s);self.assertEqual(seen,['one']) + def test_stress_450_concurrent_opens_closes_and_heartbeat(self): + errors=[];stop=threading.Event();beats=[];calls=[];guard=threading.Lock() + # No app unload in this stress: a permanent anchor session models a + # concurrent active user. Idle/unload races are exercised separately. + self.add('anchor') + def cb(sid): + with guard:calls.append(sid) + self.app(cb) + def worker(): + for _ in range(75): + result=self.a.open_session(OPEN) + with self.a._sessions_semaphore:s=self.a._sessions.get(result['idSession']) + if s is not None:self.a.close_session(s) + def heartbeat(): + while not stop.is_set(): + self.a.get_active_sessions_status();beats.append(1);stop.wait(.002) + h,he=launch(heartbeat,'heartbeat-stress') + workers=[launch(worker,'lifecycle-%s'%i) for i in range(6)] + for t,e in workers:join(t,8);errors.extend(e) + stop.set();join(h);self.assertFalse(errors+he);self.assertTrue(beats) + self.assertEqual(len(calls),450);self.assertEqual(len(set(calls)),450) + self.assertEqual(set(self.a._sessions),{'anchor'});self.assertFalse(self.a._sessions_closing) + def test_stress_close_all_racing_with_new_sessions(self): + calls=[];lock=threading.Lock();stop=threading.Event() + self.app(lambda sid:(lock.acquire(),calls.append(sid),lock.release())) + # Registry-only stress; app reload/unload interlocks have dedicated tests. + self.a._unload_apps=lambda:None + def worker(): + for _ in range(100): + response=self.a.open_session(OPEN) + with self.a._sessions_semaphore:s=self.a._sessions.get(response['idSession']) + if s is not None:self.a.close_session(s) + def close_all(): + for _ in range(75):self.a._close_all_sessions();time.sleep(.001) + tasks=[launch(worker) for _ in range(2)]+[launch(close_all)] + for t,e in tasks:join(t,5);self.assertFalse(e) + self.a._close_all_sessions() + self.assertFalse(self.a._sessions);self.assertFalse(self.a._sessions_closing) + self.assertEqual(len(calls),200);self.assertEqual(len(set(calls)),200) + + +class OSProxy: + def __init__(self): + self.ready=threading.Event();self.release_blocking=threading.Event() + self.reaped=set();self.calls=[];self.signals=[];self.mutex=threading.Lock() + def __getattr__(self,k):return getattr(os,k) + def kill(self,pid,sig):self.signals.append((pid,sig)) + def waitpid(self,pid,options): + with self.mutex:self.calls.append((pid,options)) + if options==0:self.release_blocking.wait(5) + with self.mutex: + if pid in self.reaped:raise ChildProcessError(errno.ECHILD,'already collected') + if self.ready.is_set() or options==0: + self.reaped.add(pid);return pid,0 + return 0,0 + + +class ProcessTests(unittest.TestCase): + def terminal(self,api=None): + self.api=api or OSProxy();env,A,S=load(self.api);a=A();s=S(a) + app=env['Shell'](a) + manager=SimpleNamespace(_cinfo=s,_shlmain=app) + terminal=env['LinuxMac'](manager,1,80,24) + return terminal,env,a,s,app + def test_bounded_wait_and_eventual_reap(self): + t,env,a,s,app=self.terminal();t._ppid=424242 + t._activity_registered=True;s._activities['shellSession']=1 + started=time.monotonic();thread,errors=launch(t.terminate) + try: + join(thread,1.6) + self.assertLess(time.monotonic()-started,1.6) + self.assertFalse(errors) + self.assertTrue(all(options==os.WNOHANG for pid,options in self.api.calls)) + self.assertEqual(self.api.signals,[(424242,signal.SIGTERM),(424242,signal.SIGKILL)]) + self.assertIn(424242,app._process_reaper._pending) + self.assertEqual(s._activities['shellSession'],0) + finally: + self.api.ready.set();self.api.release_blocking.set();join(thread) + deadline=time.monotonic()+2 + while getattr(app,'_process_reaper',None) and app._process_reaper._thread is not None and time.monotonic()