mirror of
https://github.com/mirror/wget.git
synced 2025-01-27 04:40:41 +08:00
Refactor mainly the test cases classes
Files that were refactored are WgetTest.py, some conf scripts and some test case file. The purpose of the refactoring of the test cases classes is to provide a better interface for the incoming FTP test counterpart.
This commit is contained in:
Zihang Chen
Giuseppe Scrivano
parent
d26c8ebb16
commit
ecd69778bf
@ -1,3 +1,75 @@
|
||||
2014-03-13 Zihang Chen <chsc4698@gmail.com>
|
||||
|
||||
* base_test.py:
|
||||
(CommonMethods): Rename to BaseTest.
|
||||
(BaseTest): Implement __init__ method where the class-wide variables are
|
||||
initialized. Also variable names like `xxx_list` is renamed to its plural
|
||||
form, e.g. `server_list` => `servers`.
|
||||
(BaseTest.init_test_env): Remove name argument due to its unnecessarity.
|
||||
(BaseTest.get_test_dir): Because the path of the test directory is needed
|
||||
in multiple methods, this method is implemented.
|
||||
(BaseTest.get_domain_addr): Rewrite the return statement utilizing str
|
||||
formatting (which is more Pythonic).
|
||||
(BaseTest.get_cmd_line): Rename to gen_cmd_line. Change the variables with
|
||||
capitcal characters to lower ones. Also, the nested for loop is rewritten
|
||||
to a plain loop using the zip function.
|
||||
(BaseTest.__gen_local_filesys): Rename to gen_local_fs_snapshot. Move to
|
||||
ExpectedFiles in conf/expected_files.py and is marked as a static
|
||||
method. Refactor to a less verbose implementation.
|
||||
(BaseTest._check_downloaded_files): Rename to __call__ to agree with the
|
||||
invocation in test case classes. Move to ExpectedFiles in
|
||||
conf/expected_files.py.
|
||||
(BaseTest.get_server_rules): Refactor to a more Pythonic form utilizing
|
||||
dict.items() and is marked static.
|
||||
(BaseTest.stop_server): (new method) an abstract method which should stop
|
||||
the currently using servers.
|
||||
(BaseTest.instantiate_server_by): (new method) an abstract method which
|
||||
should instantiate a server instance according to the given argument.
|
||||
(BaseTest.__enter__): (new method) method which initialize the context
|
||||
manager
|
||||
(BaseTest.__exit__): (new method) method that finilize the context manager
|
||||
and deal with the exceptions during the execution of the with statement,
|
||||
subclasses can override this method for extensibility
|
||||
* http_test.py:
|
||||
(HTTPTest.__init__): Add call to super.__init__. Default values of
|
||||
pre_hook, test_params, post_hook are set to None to avoid a subtle bug of
|
||||
Python. Argument servers is renamed to protocols.
|
||||
(HTTPTest.Server_setup): Move to BaseTest and rename to server_setup.
|
||||
Calls to pre_hook_call, call_test, post_hook_call are removed.
|
||||
(HTTPTest.hook_call, pre_hook_call, call_test, post_hook_call): Move to
|
||||
BaseTest for that both HTTP test cases and FTP test cases may use these
|
||||
methods.
|
||||
(HTTPTest.init_HTTP_Server, init_HTTPS_Server): Merge and rename to
|
||||
instantiate_server_by to implement the abstract method in BaseTest.
|
||||
(HTTPTest.stop_HTTP_Server): Rename to stop_server to implement the
|
||||
abstract method in BaseTest. Also, pull out the part where remaining
|
||||
requests are gathered into a new method request_remaining.
|
||||
(BaseTest.act_retcode): Rename to ret_code because ExpectedRetCode is
|
||||
moved out from BaseTest, so the name act_retcode is actually a bit
|
||||
verbose.
|
||||
* conf/expected_ret_code.py:
|
||||
(ExpectedRetCode.__call__): Rewrite the str into a more readable form.
|
||||
* conf/files_crawled.py:
|
||||
(FilesCrawled.__call__): Refactor this method into a more Pythonic form
|
||||
utilizing the zip function.
|
||||
* conf/local_files.py:
|
||||
(LocalFiles__call__): Rewrite this method with the recommended with
|
||||
statement.
|
||||
* conf/server_conf.py:
|
||||
(ServerConf.__call__): Rewrite this method due to BaseTest.server_list is
|
||||
renamed to BaseTest.servers.
|
||||
* conf/server_files.py:
|
||||
(ServerFiles.__call__): Refactor the nested for loop into a plain one
|
||||
utilizing the zip function.
|
||||
* conf/urls.py:
|
||||
(URLs): Rename url_list to urls.
|
||||
* conf/wget_commands.py:
|
||||
(WgetCommands): Rename command_list to commands, rename test_obj.options
|
||||
to test_obj.wget_options.
|
||||
* Test--https.py, Test-Proto.py, Test-Parallel-Proto.py: Argument servers
|
||||
is changed to protocols due to change in the signature of
|
||||
HTTPTest.__init__.
|
||||
|
||||
2014-03-13 Zihang Chen <chsc4698@gmail.com>
|
||||
|
||||
* test: (new package) package for test case classes
|
||||
|
||||
@ -149,10 +149,11 @@ Both, the HTTPTest and FTPTest modules have the same prototype:
|
||||
pre_hook,
|
||||
test_options,
|
||||
post_hook,
|
||||
servers
|
||||
protocols
|
||||
}
|
||||
name expects the string name, and is usually passed the TEST_NAME variable,
|
||||
the three hooks, expect python dictionary objects and servers is an integer.
|
||||
name should be a string, and is usually passed to the TEST_NAME variable,
|
||||
the three hooks should be Python dict objects and protocols should be a list of
|
||||
protocols, like [HTTP, HTTPS].
|
||||
|
||||
Valid File Rules:
|
||||
================================================================================
|
||||
|
||||
@ -47,7 +47,7 @@ err = HTTPTest (
|
||||
pre_hook=pre_test,
|
||||
test_params=test_options,
|
||||
post_hook=post_test,
|
||||
servers=Servers
|
||||
protocols=Servers
|
||||
).begin ()
|
||||
|
||||
exit (err)
|
||||
|
||||
@ -48,7 +48,7 @@ err = HTTPTest (
|
||||
pre_hook=pre_test,
|
||||
test_params=test_options,
|
||||
post_hook=post_test,
|
||||
servers=Servers
|
||||
protocols=Servers
|
||||
).begin ()
|
||||
|
||||
exit (err)
|
||||
|
||||
@ -69,7 +69,7 @@ err = HTTPTest (
|
||||
pre_hook=pre_test,
|
||||
test_params=test_options,
|
||||
post_hook=post_test,
|
||||
servers=Servers
|
||||
protocols=Servers
|
||||
).begin ()
|
||||
|
||||
exit (err)
|
||||
|
||||
@ -1,10 +1,42 @@
|
||||
from difflib import unified_diff
|
||||
import os
|
||||
import sys
|
||||
from conf import hook
|
||||
from exc.test_failed import TestFailed
|
||||
|
||||
|
||||
@hook()
|
||||
class ExpectedFiles:
|
||||
def __init__(self, exp_filesys):
|
||||
self.exp_filesys = exp_filesys
|
||||
def __init__(self, expected_fs):
|
||||
self.expected_fs = expected_fs
|
||||
|
||||
@staticmethod
|
||||
def gen_local_fs_snapshot():
|
||||
snapshot = {}
|
||||
for parent, dirs, files in os.walk('.'):
|
||||
for name in files:
|
||||
f = {'content': ''}
|
||||
file_path = os.path.join(parent, name)
|
||||
with open(file_path) as fp:
|
||||
f['content'] = fp.read()
|
||||
snapshot[file_path[2:]] = f
|
||||
|
||||
return snapshot
|
||||
|
||||
def __call__(self, test_obj):
|
||||
test_obj._check_downloaded_files (self.exp_filesys)
|
||||
local_fs = self.gen_local_fs_snapshot()
|
||||
for file in self.expected_fs:
|
||||
if file.name in local_fs:
|
||||
local_file = local_fs.pop(file.name)
|
||||
if file.content != local_file['content']:
|
||||
for line in unified_diff(local_file['content'],
|
||||
file.content,
|
||||
fromfile='Actual',
|
||||
tofile='Expected'):
|
||||
print(line, file=sys.stderr)
|
||||
raise TestFailed('Contents of %s do not match.' % file.name)
|
||||
else:
|
||||
raise TestFailed('Expected file %s not found.' % file.name)
|
||||
if local_fs:
|
||||
print(local_fs)
|
||||
raise TestFailed('Extra files downloaded.')
|
||||
|
||||
@ -4,10 +4,13 @@ from conf import hook
|
||||
|
||||
@hook(alias='ExpectedRetcode')
|
||||
class ExpectedRetCode:
|
||||
def __init__(self, retcode):
|
||||
self.retcode = retcode
|
||||
def __init__(self, expected_ret_code):
|
||||
self.expected_ret_code = expected_ret_code
|
||||
|
||||
def __call__(self, test_obj):
|
||||
if test_obj.act_retcode != self.retcode:
|
||||
pr = "Return codes do not match.\nExpected: " + str(self.retcode) + "\nActual: " + str(test_obj.act_retcode)
|
||||
raise TestFailed (pr)
|
||||
if test_obj.ret_code != self.expected_ret_code:
|
||||
failure = "Return codes do not match.\n" \
|
||||
"Expected: %s\n" \
|
||||
"Actual: %s" % (self.expected_ret_code,
|
||||
test_obj.ret_code)
|
||||
raise TestFailed(failure)
|
||||
|
||||
@ -5,14 +5,14 @@ from exc.test_failed import TestFailed
|
||||
|
||||
@hook()
|
||||
class FilesCrawled:
|
||||
def __init__(self, Request_Headers):
|
||||
self.Request_Headers = Request_Headers
|
||||
def __init__(self, request_headers):
|
||||
self.request_headers = request_headers
|
||||
|
||||
def __call__(self, test_obj):
|
||||
for i in range (0, test_obj.servers):
|
||||
headers = set(self.Request_Headers[i])
|
||||
o_headers = test_obj.Request_remaining[i]
|
||||
header_diff = headers.symmetric_difference (o_headers)
|
||||
if len(header_diff) is not 0:
|
||||
print_red(header_diff)
|
||||
raise TestFailed ("Not all files were crawled correctly")
|
||||
for headers, remaining in zip(map(set, self.request_headers),
|
||||
test_obj.request_remaining()):
|
||||
diff = headers.symmetric_difference(remaining)
|
||||
|
||||
if diff:
|
||||
print_red(diff)
|
||||
raise TestFailed('Not all files were crawled correctly.')
|
||||
|
||||
@ -7,7 +7,6 @@ class LocalFiles:
|
||||
self.local_files = local_files
|
||||
|
||||
def __call__(self, _):
|
||||
for file_obj in self.local_files:
|
||||
file_handler = open (file_obj.name, "w")
|
||||
file_handler.write (file_obj.content)
|
||||
file_handler.close ()
|
||||
for f in self.local_files:
|
||||
with open(f.name, 'w') as fp:
|
||||
fp.write(f.content)
|
||||
|
||||
@ -7,5 +7,5 @@ class ServerConf:
|
||||
self.server_settings = server_settings
|
||||
|
||||
def __call__(self, test_obj):
|
||||
for i in range (0, test_obj.servers):
|
||||
test_obj.server_list[i].server_sett (self.server_settings)
|
||||
for server in test_obj.servers:
|
||||
server.server_sett(self.server_settings)
|
||||
|
||||
@ -7,12 +7,9 @@ class ServerFiles:
|
||||
self.server_files = server_files
|
||||
|
||||
def __call__(self, test_obj):
|
||||
for i in range (0, test_obj.servers):
|
||||
file_list = dict ()
|
||||
server_rules = dict ()
|
||||
for file_obj in self.server_files[i]:
|
||||
content = test_obj._replace_substring (file_obj.content)
|
||||
file_list[file_obj.name] = content
|
||||
rule_obj = test_obj.get_server_rules (file_obj)
|
||||
server_rules[file_obj.name] = rule_obj
|
||||
test_obj.server_list[i].server_conf (file_list, server_rules)
|
||||
for server, files in zip(test_obj.servers, self.server_files):
|
||||
rules = {f.name: test_obj.get_server_rules(f)
|
||||
for f in files}
|
||||
files = {f.name: test_obj._replace_substring(f.content)
|
||||
for f in files}
|
||||
server.server_conf(files, rules)
|
||||
|
||||
@ -3,8 +3,8 @@ from conf import hook
|
||||
|
||||
@hook(alias='Urls')
|
||||
class URLs:
|
||||
def __init__(self, url_list):
|
||||
self.url_list = url_list
|
||||
def __init__(self, urls):
|
||||
self.urls = urls
|
||||
|
||||
def __call__(self, test_obj):
|
||||
test_obj.urls = self.url_list
|
||||
test_obj.urls = self.urls
|
||||
|
||||
@ -3,8 +3,8 @@ from conf import hook
|
||||
|
||||
@hook()
|
||||
class WgetCommands:
|
||||
def __init__(self, command_list):
|
||||
self.command_list = command_list
|
||||
def __init__(self, commands):
|
||||
self.commands = commands
|
||||
|
||||
def __call__(self, test_obj):
|
||||
test_obj.options = test_obj._replace_substring (self.command_list)
|
||||
test_obj.wget_options = test_obj._replace_substring(self.commands)
|
||||
|
||||
@ -1,16 +1,11 @@
|
||||
import os
|
||||
import shutil
|
||||
import shlex
|
||||
import sys
|
||||
import traceback
|
||||
from server.http import http_server
|
||||
import re
|
||||
import time
|
||||
import shlex
|
||||
import shutil
|
||||
from subprocess import call
|
||||
from misc.colour_terminal import print_red, print_green, print_blue
|
||||
from difflib import unified_diff
|
||||
from misc.colour_terminal import print_red, print_blue
|
||||
from exc.test_failed import TestFailed
|
||||
import conf
|
||||
|
||||
@ -18,101 +13,159 @@ HTTP = "HTTP"
|
||||
HTTPS = "HTTPS"
|
||||
|
||||
|
||||
class CommonMethods:
|
||||
class BaseTest:
|
||||
|
||||
""" Class that defines methods common to both HTTP and FTP Tests. """
|
||||
"""
|
||||
Class that defines methods common to both HTTP and FTP Tests.
|
||||
Note that this is an abstract class, subclasses must implement
|
||||
* stop_server()
|
||||
* instantiate_server_by(protocol)
|
||||
"""
|
||||
|
||||
TestFailed = TestFailed
|
||||
def __init__(self, name, pre_hook, test_params, post_hook, protocols):
|
||||
"""
|
||||
Define the class-wide variables (or attributes).
|
||||
Attributes should not be defined outside __init__.
|
||||
"""
|
||||
self.name = name
|
||||
self.pre_configs = pre_hook or {} # if pre_hook == None, then
|
||||
# {} (an empty dict object) is
|
||||
# passed to self.pre_configs
|
||||
self.test_params = test_params or {}
|
||||
self.post_configs = post_hook or {}
|
||||
self.protocols = protocols
|
||||
|
||||
self.servers = []
|
||||
self.domains = []
|
||||
self.port = -1
|
||||
|
||||
self.wget_options = ''
|
||||
self.urls = []
|
||||
|
||||
def init_test_env (self, name):
|
||||
testDir = name + "-test"
|
||||
try:
|
||||
os.mkdir (testDir)
|
||||
except FileExistsError:
|
||||
shutil.rmtree (testDir)
|
||||
os.mkdir (testDir)
|
||||
os.chdir (testDir)
|
||||
self.tests_passed = True
|
||||
self.init_test_env()
|
||||
|
||||
def get_domain_addr (self, addr):
|
||||
self.port = str (addr[1])
|
||||
return addr[0] + ":" + str(addr[1]) + "/"
|
||||
self.ret_code = 0
|
||||
|
||||
def exec_wget (self, options, urls, domain_list):
|
||||
cmd_line = self.get_cmd_line (options, urls, domain_list)
|
||||
params = shlex.split (cmd_line)
|
||||
print (params)
|
||||
if os.getenv ("SERVER_WAIT"):
|
||||
time.sleep (float (os.getenv ("SERVER_WAIT")))
|
||||
def get_test_dir(self):
|
||||
return self.name + '-test'
|
||||
|
||||
def init_test_env(self):
|
||||
test_dir = self.get_test_dir()
|
||||
try:
|
||||
retcode = call (params)
|
||||
except FileNotFoundError as filenotfound:
|
||||
raise TestFailed (
|
||||
"The Wget Executable does not exist at the expected path")
|
||||
return retcode
|
||||
os.mkdir(test_dir)
|
||||
except FileExistsError:
|
||||
shutil.rmtree(test_dir)
|
||||
os.mkdir(test_dir)
|
||||
os.chdir(test_dir)
|
||||
|
||||
def get_domain_addr(self, addr):
|
||||
# TODO if there's a multiple number of ports, wouldn't it be
|
||||
# overridden to the port of the last invocation?
|
||||
self.port = str(addr[1])
|
||||
|
||||
return '%s:%s' % (addr[0], self.port)
|
||||
|
||||
def server_setup(self):
|
||||
print_blue("Running Test %s" % self.name)
|
||||
for protocol in self.protocols:
|
||||
instance = self.instantiate_server_by(protocol)
|
||||
self.servers.append(instance)
|
||||
|
||||
# servers instantiated by different protocols may differ in
|
||||
# ports and etc.
|
||||
# so we should record different domains respect to servers.
|
||||
domain = self.get_domain_addr(instance.server_address)
|
||||
self.domains.append(domain)
|
||||
|
||||
def exec_wget(self):
|
||||
cmd_line = self.gen_cmd_line()
|
||||
params = shlex.split(cmd_line)
|
||||
print(params)
|
||||
|
||||
if os.getenv("SERVER_WAIT"):
|
||||
time.sleep(float(os.getenv("SERVER_WAIT")))
|
||||
|
||||
try:
|
||||
ret_code = call(params)
|
||||
except FileNotFoundError:
|
||||
raise TestFailed("The Wget Executable does not exist at the "
|
||||
"expected path.")
|
||||
|
||||
return ret_code
|
||||
|
||||
def gen_cmd_line(self):
|
||||
test_path = os.path.abspath(".")
|
||||
wget_path = os.path.abspath(os.path.join(test_path,
|
||||
"..", '..', 'src', "wget"))
|
||||
|
||||
cmd_line = '%s %s ' % (wget_path, self.wget_options)
|
||||
for protocol, urls, domain in zip(self.protocols,
|
||||
self.urls,
|
||||
self.domains):
|
||||
# zip is function for iterating multiple lists at the same time.
|
||||
# e.g. for item1, item2 in zip([1, 5, 3],
|
||||
# ['a', 'e', 'c']):
|
||||
# print(item1, item2)
|
||||
# generates the following output:
|
||||
# 1 a
|
||||
# 5 e
|
||||
# 3 c
|
||||
protocol = protocol.lower()
|
||||
for url in urls:
|
||||
cmd_line += '%s://%s/%s ' % (protocol, domain, url)
|
||||
|
||||
print(cmd_line)
|
||||
|
||||
def get_cmd_line (self, options, urls, domain_list):
|
||||
TEST_PATH = os.path.abspath (".")
|
||||
WGET_PATH = os.path.join (TEST_PATH, "..", "..", "src", "wget")
|
||||
WGET_PATH = os.path.abspath (WGET_PATH)
|
||||
cmd_line = WGET_PATH + " " + options + " "
|
||||
for i in range (0, self.servers):
|
||||
for url in urls[i]:
|
||||
protocol = "http://" if self.server_types[i] is "HTTP" else "https://"
|
||||
cmd_line += protocol + domain_list[i] + url + " "
|
||||
# for url in urls:
|
||||
# cmd_line += domain_list[0] + url + " "
|
||||
print (cmd_line)
|
||||
return cmd_line
|
||||
|
||||
def __test_cleanup (self):
|
||||
testDir = self.name + "-test"
|
||||
os.chdir ('..')
|
||||
def __test_cleanup(self):
|
||||
os.chdir('..')
|
||||
try:
|
||||
if os.getenv ("NO_CLEANUP") is None:
|
||||
shutil.rmtree (testDir)
|
||||
except Exception as ae:
|
||||
if not os.getenv("NO_CLEANUP"):
|
||||
shutil.rmtree(self.get_test_dir())
|
||||
except:
|
||||
print ("Unknown Exception while trying to remove Test Environment.")
|
||||
|
||||
def _exit_test (self):
|
||||
self.__test_cleanup ()
|
||||
self.__test_cleanup()
|
||||
|
||||
def begin (self):
|
||||
return 0 if self.tests_passed else 100
|
||||
|
||||
""" Methods to check if the Test Case passes or not. """
|
||||
def call_test(self):
|
||||
self.hook_call(self.test_params, 'Test Option')
|
||||
|
||||
def __gen_local_filesys (self):
|
||||
file_sys = dict ()
|
||||
for parent, dirs, files in os.walk ('.'):
|
||||
for name in files:
|
||||
onefile = dict ()
|
||||
# Create the full path to file, removing the leading ./
|
||||
# Might not work on non-unix systems. Someone please test.
|
||||
filepath = os.path.join (parent, name)
|
||||
file_handle = open (filepath, 'r')
|
||||
file_content = file_handle.read ()
|
||||
onefile['content'] = file_content
|
||||
filepath = filepath[2:]
|
||||
file_sys[filepath] = onefile
|
||||
file_handle.close ()
|
||||
return file_sys
|
||||
try:
|
||||
self.ret_code = self.exec_wget()
|
||||
except TestFailed as e:
|
||||
raise e
|
||||
finally:
|
||||
self.stop_server()
|
||||
|
||||
def do_test(self):
|
||||
self.pre_hook_call()
|
||||
self.call_test()
|
||||
self.post_hook_call()
|
||||
|
||||
def _check_downloaded_files (self, exp_filesys):
|
||||
local_filesys = self.__gen_local_filesys ()
|
||||
for files in exp_filesys:
|
||||
if files.name in local_filesys:
|
||||
local_file = local_filesys.pop (files.name)
|
||||
if files.content != local_file ['content']:
|
||||
for line in unified_diff (local_file['content'], files.content, fromfile="Actual", tofile="Expected"):
|
||||
sys.stderr.write (line)
|
||||
raise TestFailed ("Contents of " + files.name + " do not match")
|
||||
else:
|
||||
raise TestFailed ("Expected file " + files.name + " not found")
|
||||
if local_filesys:
|
||||
print (local_filesys)
|
||||
raise TestFailed ("Extra files downloaded.")
|
||||
def hook_call(self, configs, name):
|
||||
for conf_name, conf_arg in configs.items():
|
||||
try:
|
||||
# conf.find_conf(conf_name) returns the required conf class,
|
||||
# then the class is instantiated with conf_arg, then the
|
||||
# conf instance is called with this test instance itself to
|
||||
# invoke the desired hook
|
||||
conf.find_conf(conf_name)(conf_arg)(self)
|
||||
except AttributeError:
|
||||
self.stop_server()
|
||||
raise TestFailed("%s %s not defined." %
|
||||
(name, conf_name))
|
||||
|
||||
def pre_hook_call(self):
|
||||
self.hook_call(self.pre_configs, 'Pre Test Function')
|
||||
|
||||
def post_hook_call(self):
|
||||
self.hook_call(self.post_configs, 'Post Test Function')
|
||||
|
||||
def _replace_substring (self, string):
|
||||
pattern = re.compile ('\{\{\w+\}\}')
|
||||
@ -123,14 +176,51 @@ class CommonMethods:
|
||||
string = string.replace (rep, temp)
|
||||
return string
|
||||
|
||||
|
||||
def get_server_rules (self, file_obj):
|
||||
""" The handling of expect header could be made much better when the
|
||||
options are parsed in a true and better fashion. For an example,
|
||||
see the commented portion in Test-basic-auth.py.
|
||||
def instantiate_server_by(self, protocol):
|
||||
"""
|
||||
server_rules = dict ()
|
||||
for rule in file_obj.rules:
|
||||
r_obj = conf.find_conf(rule)(file_obj.rules[rule])
|
||||
server_rules[rule] = r_obj
|
||||
Subclasses must override this method to actually instantiate servers
|
||||
for test cases.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def stop_server(self):
|
||||
"""
|
||||
Subclasses must implement this method in order to stop certain
|
||||
servers of different types.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@staticmethod
|
||||
def get_server_rules(file_obj):
|
||||
"""
|
||||
The handling of expect header could be made much better when the
|
||||
options are parsed in a true and better fashion. For an example,
|
||||
see the commented portion in Test-basic-auth.py.
|
||||
"""
|
||||
server_rules = {}
|
||||
for rule_name, rule in file_obj.rules.items():
|
||||
server_rules[rule_name] = conf.find_conf(rule_name)(rule)
|
||||
return server_rules
|
||||
|
||||
def __enter__(self):
|
||||
"""
|
||||
Initialization for with statement.
|
||||
"""
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
"""
|
||||
If the with statement got executed with no exception raised, then
|
||||
exc_type, exc_val, exc_tb are all None.
|
||||
"""
|
||||
if exc_val:
|
||||
self.tests_passed = False
|
||||
if exc_type is TestFailed:
|
||||
print_red('Error: %s.' % exc_val.error)
|
||||
else:
|
||||
print_red('Unhandled exception caught.')
|
||||
print(exc_val)
|
||||
traceback.print_tb(exc_tb)
|
||||
self.__test_cleanup()
|
||||
|
||||
return True
|
||||
|
||||
@ -1,107 +1,45 @@
|
||||
import traceback
|
||||
import conf
|
||||
from exc.test_failed import TestFailed
|
||||
from misc.colour_terminal import print_red, print_green, print_blue
|
||||
from server.http import http_server
|
||||
from test.base_test import CommonMethods, HTTP
|
||||
from misc.colour_terminal import print_green
|
||||
from server.http.http_server import HTTPd, HTTPSd
|
||||
from test.base_test import BaseTest, HTTP, HTTPS
|
||||
|
||||
|
||||
class HTTPTest (CommonMethods):
|
||||
class HTTPTest(BaseTest):
|
||||
|
||||
""" Class for HTTP Tests. """
|
||||
|
||||
# Temp Notes: It is expected that when pre-hook functions are executed, only an empty test-dir exists.
|
||||
# pre-hook functions are executed just prior to the call to Wget is made.
|
||||
# post-hook functions will be executed immediately after the call to Wget returns.
|
||||
# Temp Notes: It is expected that when pre-hook functions are executed, only an empty test-dir exists.
|
||||
# pre-hook functions are executed just prior to the call to Wget is made.
|
||||
# post-hook functions will be executed immediately after the call to Wget returns.
|
||||
|
||||
def __init__ (
|
||||
self,
|
||||
name="Unnamed Test",
|
||||
pre_hook=dict(),
|
||||
test_params=dict(),
|
||||
post_hook=dict(),
|
||||
servers=[HTTP]
|
||||
):
|
||||
try:
|
||||
self.Server_setup (name, pre_hook, test_params, post_hook, servers)
|
||||
except TestFailed as tf:
|
||||
print_red("Error: " + tf.error)
|
||||
self.tests_passed = False
|
||||
except Exception as ae:
|
||||
print_red("Unhandled Exception Caught.")
|
||||
print ( ae.__str__ ())
|
||||
traceback.print_exc ()
|
||||
self.tests_passed = False
|
||||
else:
|
||||
print_green("Test Passed")
|
||||
finally:
|
||||
self._exit_test ()
|
||||
def __init__(self,
|
||||
name="Unnamed Test",
|
||||
pre_hook=None,
|
||||
test_params=None,
|
||||
post_hook=None,
|
||||
protocols=(HTTP,)):
|
||||
super(HTTPTest, self).__init__(name,
|
||||
pre_hook,
|
||||
test_params,
|
||||
post_hook,
|
||||
protocols)
|
||||
with self:
|
||||
# if any exception occurs, self.__exit__ will be immediately called
|
||||
self.server_setup()
|
||||
self.do_test()
|
||||
print_green('Test Passed.')
|
||||
|
||||
def Server_setup (self, name, pre_hook, test_params, post_hook, servers):
|
||||
self.name = name
|
||||
self.server_types = servers
|
||||
self.servers = len (servers)
|
||||
print_blue("Running Test " + self.name)
|
||||
self.init_test_env (name)
|
||||
self.server_list = list()
|
||||
self.domain_list = list()
|
||||
for server_type in servers:
|
||||
server_inst = getattr (self, "init_" + server_type + "_Server") ()
|
||||
self.server_list.append (server_inst)
|
||||
domain = self.get_domain_addr (server_inst.server_address)
|
||||
self.domain_list.append (domain)
|
||||
#self.server = self.init_HTTP_Server ()
|
||||
#self.domain = self.get_domain_addr (self.server.server_address)
|
||||
def instantiate_server_by(self, protocol):
|
||||
server = {HTTP: HTTPd,
|
||||
HTTPS: HTTPSd}[protocol]()
|
||||
server.start()
|
||||
|
||||
self.pre_hook_call (pre_hook)
|
||||
self.call_test (test_params)
|
||||
self.post_hook_call (post_hook)
|
||||
|
||||
|
||||
def hook_call(self, configs, name):
|
||||
for conf_name, conf_arg in configs.items():
|
||||
try:
|
||||
# conf.find_conf(conf_name) returns the required conf class,
|
||||
# then the class is instantiated with conf_arg, then the
|
||||
# conf instance is called with this test instance itself to
|
||||
# invoke the desired hook
|
||||
conf.find_conf(conf_name)(conf_arg)(self)
|
||||
except AttributeError as e:
|
||||
print(e)
|
||||
self.stop_HTTP_Server()
|
||||
raise TestFailed("%s %s not defined." %
|
||||
(name, conf_name))
|
||||
|
||||
|
||||
def pre_hook_call (self, pre_hook):
|
||||
self.hook_call(pre_hook, 'Pre Test Function')
|
||||
|
||||
def call_test (self, test_params):
|
||||
self.hook_call(test_params, 'Test Option')
|
||||
|
||||
try:
|
||||
self.act_retcode = self.exec_wget (self.options, self.urls, self.domain_list)
|
||||
except TestFailed as tf:
|
||||
self.stop_HTTP_Server ()
|
||||
raise TestFailed (tf.__str__ ())
|
||||
self.stop_HTTP_Server ()
|
||||
|
||||
def post_hook_call (self, post_hook):
|
||||
self.hook_call(post_hook, 'Post Test Function')
|
||||
|
||||
def init_HTTP_Server (self):
|
||||
server = http_server.HTTPd ()
|
||||
server.start ()
|
||||
return server
|
||||
|
||||
def init_HTTPS_Server (self):
|
||||
server = http_server.HTTPSd ()
|
||||
server.start ()
|
||||
return server
|
||||
def request_remaining(self):
|
||||
return [s.server_inst.get_req_headers()
|
||||
for s in self.servers]
|
||||
|
||||
def stop_HTTP_Server (self):
|
||||
self.Request_remaining = list ()
|
||||
for server in self.server_list:
|
||||
server_req = server.server_inst.get_req_headers ()
|
||||
self.Request_remaining.append (server_req)
|
||||
server.server_inst.shutdown ()
|
||||
def stop_server(self):
|
||||
for server in self.servers:
|
||||
server.server_inst.shutdown()
|
||||
# vim: set ts=4 sts=4 sw=4 tw=80 et :
|
||||
|
||||
Loading…
Reference in New Issue
Block a user