[Spambayes-checkins] spambayes/spambayes/test sb_test_support.py,
NONE, 1.1 test_programs.py, NONE, 1.1
Mark Hammond
mhammond at users.sourceforge.net
Tue Dec 2 19:34:19 EST 2003
Update of /cvsroot/spambayes/spambayes/spambayes/test
In directory sc8-pr-cvs1:/tmp/cvs-serv5358
Added Files:
sb_test_support.py test_programs.py
Log Message:
New sb_test_support file for utility functions used by test cases.
test_programs attempts to start and stop a number of programs - currently
sb_server.py, sb_server.py with a custom UI port, and the windows service.
--- NEW FILE: sb_test_support.py ---
# General utilities for the SpamBayes test suite
#
#
import sys, os
import unittest
def fix_sys_path():
# XXX - MarkH had the bright idea *after* writing this that we should
# ensure the CVS version of SpamBayes is *not* used to resolve SB imports.
# This would allow us to effectively test the distutils setup script, so
# any modules or files missing from the installed version raise errors.
"""Fix sys.path so that the core SpamBayes package,
*and* the SpamBayes scripts can be imported.
"""
this_dir = os.path.dirname(__file__)
try:
import spambayes.Version
except ImportError:
# Apparently SpamBayes is yet to be "setup.py install"
# We are in 'spambayes\spambayes\test' - 2 parents up should
# do it.
sb_dir = os.path.abspath(
os.path.join(this_dir, "..", ".."))
sys.path.insert(0, sb_dir)
import SpamBayes.Version
# Now do the same for the sb_* scripts
try:
import sb_server
except ImportError:
# Scripts are usually in "spambayes/scripts" (for an
# installed SpamBayes, they appear to be in
# os.path.join(sys.prefix(), "scripts"), which we may like to
# leverage - however, these test scripts are not currently
# installed).
script_dir = os.path.abspath(
os.path.join(this_dir, "..", "..", "scripts"))
sys.path.insert(0, script_dir)
import sb_server
# Entry point for all our 'simple' based test programs
def unittest_main(*args, **kwargs):
# I bet one day this will be more than this <wink>
unittest.main(*args, **kwargs)
--- NEW FILE: test_programs.py ---
import os
import sys
import errno
import unittest
import time
import sb_test_support
sb_test_support.fix_sys_path()
try:
True;False
except NameError: # 2.2 compat
True=(None is None);
False=not True
import sb_server
from spambayes.Options import options
default_shutdown_port = options["html_ui", "port"]
verbose = 0
class Spawner:
def __init__(self, test_case, spawn_args):
self.test_case = test_case
self.spawn_args = spawn_args
# If the command is a .py file, insert an executable.
if os.path.splitext(self.spawn_args[0])[1]=='.py':
self.spawn_args.insert(0, sys.executable)
self.pid = None
def _spawn(self, args):
return os.spawnv(os.P_NOWAIT, self.spawn_args[0], self.spawn_args)
def start(self):
raise NotImplementedError
def stop(self):
raise NotImplementedError
def is_running(self):
if self.pid is None:
return False
# damn - we could implement os.waitpid correctly
# using the win32api - but as at 2.3, you can't check
# if a pid is still running with os.waitpid
if sys.platform.startswith("win32"):
import win32process # sorry, ya gotta have win32all to run tests
import win32con
try:
rc = win32process.GetExitCodeProcess(self.pid)
result = rc==win32con.STILL_ACTIVE
except win32process.error:
result = False
else:
try:
os.waitpid(self.pid, os.WNOHANG)
result = True
except os.error, details:
if details.errno == errno.ECHILD:
result = False
# other exceptions invalid?
raise
# Wait a few seconds for the global mutex to catch up
for i in range(20):
time.sleep(0.25)
if result==is_any_sb_server_running():
break
# Check the platform agrees (could do xor, but even I wont be able
# to read it in a few weeks <wink>
if result:
self.test_case.failUnless(is_any_sb_server_running(),
"My server stopped, but global server mutex held")
else:
self.test_case.failUnless(not is_any_sb_server_running(),
"My server running, but no global server mutex held")
return result
class Spawner_sb_server(Spawner):
def __init__(self, test_case, args, shutdown_port = default_shutdown_port):
self.shutdown_port = shutdown_port
f = sb_server.__file__
if f.endswith(".pyc") or f.endswith(".pyo"):
f = f[:-1]
Spawner.__init__(self, test_case, [f]+args)
def start(self):
self.test_case.failUnless(not is_any_sb_server_running(),
"Should be no server running")
if verbose > 1:
print "Spawning", self.spawn_args
self.pid = self._spawn(self.spawn_args)
# wait for it to start - 5 secs, 0.25 per check
for i in range(20):
time.sleep(0.25)
if verbose > 1:
print "Waiting for start flags: running=%s, global_mutex=%s" \
% (self.is_running(), is_any_sb_server_running())
if self.is_running() and is_any_sb_server_running():
return
# gave up waiting.
self.test_case.fail("sb_server appeared to not start")
def stop(self):
# Copied from sb_server.stop()
# Shutdown as though through the web UI. This will save the DB, allow
# any open proxy connections to complete, etc.
from urllib import urlopen, urlencode
urlopen('http://localhost:%d/save' % self.shutdown_port,
urlencode({'how': 'Save & shutdown'})).read()
# wait for it to stop - 5 secs, 0.25 per check
for i in range(20):
time.sleep(0.25)
if not self.is_running() and not is_any_sb_server_running():
return
# gave up waiting.
self.test_case.fail("sb_server appeared to not stop")
def is_any_sb_server_running():
# reach into sb_server internals, as it is authoritative (sometimes <wink>)
try:
mutex = sb_server.open_platform_mutex()
sb_server.close_platform_mutex(mutex)
return False
except sb_server.AlreadyRunningException:
return True
class TestServer(unittest.TestCase):
def setUp(self):
self.failUnless(not is_any_sb_server_running(),
"Can't do sb_server tests while a server is running "\
"(platform mutex held)")
def tearDown(self):
# If we cause failure here, we mask the underlying error which left
# the server running - so just print the warning.
if is_any_sb_server_running():
print "WARNING:", self, "completed with the platform mutex held"
def _start_spawner(self, spawner):
self.failUnless(not spawner.is_running(),
"this spawneer can't be running")
spawner.start()
self.failUnless(spawner.is_running(),
"this spawner must be running after successful start")
self.failUnless(is_any_sb_server_running(),
"Platform mutex not held after starting")
def _stop_spawner(self, spawner):
self.failUnless(spawner.is_running(), "must be running to stop")
self.failUnless(is_any_sb_server_running(),
"Platform mutex must be held to stop")
spawner.stop()
self.failUnless(not spawner.is_running(), "didn't stop after stop")
self.failUnless(not is_any_sb_server_running(),
"Platform mutex still help after stop")
def test_sb_server_default(self):
# Should be using the default port from the options file.
from spambayes.Options import options
port = options["html_ui", "port"]
s = Spawner_sb_server(self, [])
self._start_spawner(s)
self._stop_spawner(s)
def test_sb_server_ui_port(self):
# Should be using the default port from the options file.
s = Spawner_sb_server(self, ["-u8899"], 8899)
self._start_spawner(s)
self._stop_spawner(s)
if sys.platform.startswith("win"):
import win32service # You need win32all to run the tests!
import win32serviceutil
import winerror
service_name = "pop3proxy"
class TestService(unittest.TestCase):
def setUp(self):
try:
win32serviceutil.QueryServiceStatus(service_name)
except win32service.error, details:
if details[0]==winerror.ERROR_SERVICE_DOES_NOT_EXIST:
self.was_installed = False
raise
else:
self.was_installed = True
self.failUnless(not is_any_sb_server_running(),
"Can't do service tests while a server is running "\
"(platform mutex held)")
def tearDown(self):
if is_any_sb_server_running():
print "WARNING:", self, "completed with the platform mutex held"
def _start_service(self):
win32serviceutil.StartService(service_name)
for i in range(10):
time.sleep(0.5)
status = win32serviceutil.QueryServiceStatus(service_name)
if status[1] == win32service.SERVICE_RUNNING:
break
if verbose > 1:
print "Service status is %d - still waiting" % status[1]
else:
self.fail("Gave up waiting for service to start")
def _stop_service(self):
# StopServiceWithDeps checks the status of each service as it
# stops it, which is exactly what we want here.
win32serviceutil.StopServiceWithDeps(service_name)
def test_simple_startstop(self):
self._start_service()
self._stop_service()
def test_remote_shutdown(self):
self._start_service()
# Should be using the default port from the options file.
from spambayes.Options import options
from urllib import urlopen, urlencode
port = options["html_ui", "port"]
urlopen('http://localhost:%d/save' % port,
urlencode({'how': 'Save & shutdown'})).read()
# wait for it to stop - 5 secs, 0.25 per check
for i in range(10):
time.sleep(0.5)
status = win32serviceutil.QueryServiceStatus(service_name)
if status[1] == win32service.SERVICE_STOPPED:
break
else:
self.fail("Gave up waiting for service to stop")
self.failUnless(not is_any_sb_server_running(),
"Should be no platform mutex held after stopping")
if __name__=='__main__':
sb_test_support.unittest_main()
More information about the Spambayes-checkins
mailing list