ExternalVendorCode/tahoe-lafs
1
0
Fork 0
mirror of https://github.com/tahoe-lafs/tahoe-lafs.git synced 2025-01-31 08:25:35 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

factor some duplication out of the tests

Browse Source
This commit is contained in:
Jean-Paul Calderone 2023-06-20 07:45:51 -04:00
parent 357c9b003f
commit 02fba3b2b6
1 changed files with 43 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
src/allmydata/test/cli/test_run.py
View File

@ -174,24 +174,29 @@ class DaemonizeStopTests(SyncTestCase):
super().setUp() super().setUp()
def test_stop_on_stdin_close(self): def _make_daemon(self, extra_argv: list[str]) -> DaemonizeTheRealService:
""" """
We stop when stdin is closed. Create the daemonization service.
:param extra_argv: Extra arguments to pass between ``run`` and the
node path.
""" """
options = parse_options(["run", self.nodedir.path]) options = parse_options(["run"] + extra_argv + [self.nodedir.path])
options.stdout = StringIO() options.stdout = StringIO()
options.stderr = StringIO() options.stderr = StringIO()
options.stdin = StringIO() options.stdin = StringIO()
run_options = options.subOptions run_options = options.subOptions
return DaemonizeTheRealService(
with AlternateReactor(self.reactor):
service = DaemonizeTheRealService(
"client", "client",
self.nodedir.path, self.nodedir.path,
run_options, run_options,
) )
service.startService()
def _run_daemon(self) -> None:
"""
Simulate starting up the reactor so the daemon plugin can do its
stuff.
"""
# We happen to know that the service uses reactor.callWhenRunning # We happen to know that the service uses reactor.callWhenRunning
# to schedule all its work (though I couldn't tell you *why*). # to schedule all its work (though I couldn't tell you *why*).
# Make sure those scheduled calls happen. # Make sure those scheduled calls happen.
@ -200,11 +205,24 @@ class DaemonizeStopTests(SyncTestCase):
for f, a, k in waiting: for f, a, k in waiting:
f(*a, **k) f(*a, **k)
def _close_stdin(self) -> None:
"""
Simulate closing the daemon plugin's stdin.
"""
# there should be a single reader: our StandardIO process # there should be a single reader: our StandardIO process
# reader for stdin. Simulate it closing. # reader for stdin. Simulate it closing.
for r in self.reactor.getReaders(): for r in self.reactor.getReaders():
r.connectionLost(Failure(ConnectionDone())) r.connectionLost(Failure(ConnectionDone()))
def test_stop_on_stdin_close(self):
"""
We stop when stdin is closed.
"""
with AlternateReactor(self.reactor):
service = self._make_daemon([])
service.startService()
self._run_daemon()
self._close_stdin()
self.assertEqual(len(self.stop_calls), 1) self.assertEqual(len(self.stop_calls), 1)
def test_allow_stdin_close(self): def test_allow_stdin_close(self):
@ -212,33 +230,11 @@ class DaemonizeStopTests(SyncTestCase):
If --allow-stdin-close is specified then closing stdin doesn't If --allow-stdin-close is specified then closing stdin doesn't
stop the process stop the process
""" """
options = parse_options(["run", "--allow-stdin-close", self.nodedir.path])
options.stdout = StringIO()
options.stderr = StringIO()
options.stdin = StringIO()
run_options = options.subOptions
with AlternateReactor(self.reactor): with AlternateReactor(self.reactor):
service = DaemonizeTheRealService( service = self._make_daemon(["--allow-stdin-close"])
"client",
self.nodedir.path,
run_options,
)
service.startService() service.startService()
self._run_daemon()
# We happen to know that the service uses reactor.callWhenRunning self._close_stdin()
# to schedule all its work (though I couldn't tell you *why*).
# Make sure those scheduled calls happen.
waiting = self.reactor.whenRunningHooks[:]
del self.reactor.whenRunningHooks[:]
for f, a, k in waiting:
f(*a, **k)
# kind of cheating -- there are no readers, because we
# never instantiated a StandardIO in this case..
for r in self.reactor.getReaders():
r.connectionLost(Failure(ConnectionDone()))
self.assertEqual(self.stop_calls, []) self.assertEqual(self.stop_calls, [])