moparisthebest/wget
1
0
Fork 0
mirror of https://github.com/moparisthebest/wget synced 2024-07-03 16:38:41 -04:00
Code Issues Releases Wiki Activity

Refactor the Python based test suite

Browse Source 
This is a squashed commit of the following from parallel-wget:
ecd6977 Refactor mainly the test cases classes
d26c8eb Create package test for test case classes
507383d Move server classes to package server.protocol
195393b Create package conf where rules and hooks are put
42e482a Create package exc and move TestFailed to exc
82f44f3 Fix a typo in Test-Proto.py
31e5f33 From WgetTest.py move WgetFile to misc
422171d Create package misc, move ColourTerm.py to misc
This commit is contained in:
Zihang Chen 2014-07-24 16:46:50 +05:30 committed by Darshit Shah
parent b89cda0a77
commit 8b83306d54
54 changed files with 793 additions and 384 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

143
testenv/ChangeLog
View File

@ -1,3 +1,146 @@
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
* WgetTest.py: Split into test/base_test.py and test/http_test.py.
* Test-*.py: Optimize the imports according to changes of WgetTest.py
2014-03-13 Zihang Chen <chsc4698@gmail.com>
* server: (new package) package for the server classes
* server.http: (new package) package for HTTP server
* server.ftp: (new package) package for FTP server
* HTTPServer.py: Move to server/http/http_server.py. Also change the
CERTFILE to '../certs/wget-cert.pem'.
* FTPServer.py: Move to server/ftp/ftp_server.py.
* WgetTest.py: Optimize import respect to the server classes.
2014-03-13 Zihang Chen <chsc4698@gmail.com>
* conf: (new package) package for rule classes and hook methods
* WgetTest.py:
(CommonMethods.Authentication): Move to conf/authentication.py.
(CommonMethods.ExpectHeader): Move to conf/expect_header.py.
(CommonMethods.RejectHeader): Move to conf/reject_header.py.
(CommonMethods.Response): Move to conf/response.py.
(CommonMethods.SendHeader): Move to conf/send_header.py.
(CommonMethods.ServerFiles): Move to conf/server_files.py.
(CommonMethods.LocalFiles): Move to conf/local_files.py.
(CommonMethods.ServerConf): Move to conf/server_conf.py.
(CommonMethods.WgetCommands): Move to conf/wget_commands.py.
(CommonMethods.Urls): Move to conf/urls.py.
(CommonMethods.ExpectedRetcode): Move to conf/expected_retcode.py.
(CommonMethods.ExpectedFiles): Move to conf/expected_files.py.
(CommonMethods.FilesCrawled): Move to conf/files_crawled.py.
(CommonMethods.__check_downloaded_files): Rename to
_check_downloaded_files, so that the method is callable from outside the
class.
(CommomMethods.get_server_rules): Modify so that it utilizes the conf
package.
(HTTPTest): Add a method hook_call(configs, name) to reduce duplications
in pre_hook_call, call_test and post_hook_call utilizing the conf package.
* conf/hook_sample.py: (new file) sample for hooks
* conf/rule_sample.py: (new file) sample for rules
* REAMDE: Update sections about customizing rules and hooks.
2014-03-13 Zihang Chen <chsc4698@gmail.com>
* exc: (new package) package for miscellaneous exceptions
* WgetTest.py: Move TestFailed to exc/test_failed.py.
2014-03-13 Zihang Chen <chsc4698@gmail.com>
* Test-Proto.py: Fix a typo (line 71: server to servers).
2014-03-13 Zihang Chen <chsc4698@gmail.com>
* WgetTest.py: Move WgetFile to package misc.
* README: Modify documentation respect to WgetFile.
* Test-*.py: Optimize imports about WgetFile.
2014-03-13 Zihang Chen <chsc4698@gmail.com>
* misc: (new package) package for miscellaneous modules
* ColourTerm.py: Move to package misc and rename to colour_terminal.py,
add print_color functions to reduce the use of string literals like
"BLUE", "RED" etc.
* WgetTest.py:
(CommonMethods.Server_setup): Change invocation to printer to print_blue.
(CommonMethods.FilesCrawled): Change invocation to printer to print_red.
(HTTPTest.__init__): Change invocations to printer to print_red and
print_green respectively.
2014-01-02 Darshit Shah <darnir@gmail.com>
* Makefile.am: Add new Test--https.py to list of tests and EXTRA_DIST.
Also replace all tabs with spaces in file for conformity.

77
testenv/README
View File

@ -10,24 +10,34 @@ Run the './configure' command to generate the Makefile and then run 'make check'
to execute the Test Suite. Use the '-j n' option with 'make check' to execute
n tests simultaneously.
File List:
Structure:
================================================================================
* HTTPServer.py: This file contains a custom, programmatically configurable
HTTP Server for testing Wget. It runs an instance of Python's http.server
module.
* server: This package contains custom programmatically configurable servers
(both HTTP and FTP) for testing Wget. The HTTP server runs an instance of
Python's http.server module. The FTP server is to be implemented.
* WgetTest.py: This file contains various functions and global variables for
each instance of the server that is initiated. It includes functions to
start and stop the server, to initialze the test environment and to cleanup
after a test.
* test: This package contains the test case classes for HTTP and FTP. The
test case classes includes methods for initializing and cleaning up of the
test environment.
* Test-Proto.py: This is a prototype Test Case file. The file defines all
the acceptable elements and their uses. Typically, one must copy this file
and edit it for writing Test Cases.
* ColourTerm.py: A custom library for printing coloured output to the
terminal. Currently it only supports 4 colours in a *nix environment.
* exc: This package contains custom exception classes used in this test
suite.
* conf: This package contains the configuration classes for servers to be
configured with.
* misc: This package contains several helper modules used in this test
suite.
- colour_terminal.py: A custom module for printing coloured output to
the terminal. Currently it only supports 4 colours in a *nix
environment.
- wget_file.py: Module which contains WgetFile, which is a file data
container object.
Working:
================================================================================
@ -93,7 +103,8 @@ effort to get accustomed to.
All Test Files MUST begin with the following Three Lines:
#!/usr/bin/python3
from sys import exit
from WgetTest import {HTTPTest|FTPTest}, WgetFile
from WgetTest import {HTTPTest|FTPTest}
from misc.wget_file import WgetFile
It is recommended that a small description of the Test Case is provided next.
This would be very helpful to future contributors.
@ -101,7 +112,7 @@ Next, is the const variable, TEST_NAME that defines the name of the Test.
Each File in the Test must be represented as a WgetFile object. The WgetFile
Class has the following prototype:
WgetFile (String name, String contents, String timestamp, dict rules)
WgetFile (str name, str contents, str timestamp, dict rules)
None except name is a mandatory paramter, one may pass only those parameters
that are required by the File object.
@ -138,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:
================================================================================
@ -190,7 +202,8 @@ The test_options dictionary defines the commands to be used when the Test is
executed. The currently supported options are:
* Urls : A list of the filenames that Wget must attempt to
download. The complete URL will be created and passed to Wget automatically.
download. The complete URL will be created and passed to Wget
automatically. (alias URLs)
* WgetCommands : A string consisting of the various commandline switches
sent to Wget upon invokation. Any data placed between {{ }} in this string
will be replaced with the contents of self.<data> before being passed to
@ -207,7 +220,7 @@ hooks are usually used to run checks on the data, files downloaded, return code,
etc. The following hooks are currently supported:
* ExpectedRetcode : This is an integer value of the ReturnCode with which
Wget is expected to exit.
Wget is expected to exit. (alias ExpectedRetCode)
* ExpectedFiles : This is a list of WgetFile objects of the files that
must exist locally on disk in the Test directory.
* FilesCrawled : This requires a list of the Requests that the server is
@ -223,11 +236,31 @@ recommended method for writing new Test Case files is to copy Test-Proto.py and
modify it to ones needs.
In case you require any functionality that is not currently defined in List of
Rules defined above, you should add the required code in WgetTest.py.
In most cases, one requires a new Rule to be added for the Server to follow.
In such a case, create a new Class in WgetTest.py with the same name as the Rule
and define an __init__ () function to handle the data. A method must also be
defined in HTTPTest / FTPTest modules to handle the said Rule.
Rules defined above, you should implement a new class in the conf package. The
file name doesn't matter (though it's better to give it an appropriate name).
The new rule or hook class should be like this:
============================================
from conf import rule
@rule()
class MyNewRule:
def __init__(self, rule_arg):
self.rule_arg = rule_arg
# your rule initialization code goes here
============================================
from conf import hook
@hook()
class MyNewHook:
def __init__(self, hook_arg):
self.hook_arg = hook_arg
# your hook initialization code goes here
def __call__(self, test_obj):
# your hook code goes here
============================================
Once a new Test File is created, it must be added to the TESTS variable in
Makefile.am. This way the Test will be executed on running a 'make check'.

6
testenv/Test--https.py
View File

@ -1,6 +1,8 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile, HTTPS, HTTP
from test.http_test import HTTPTest
from test.base_test import HTTP, HTTPS
from misc.wget_file import WgetFile
"""
This test ensures that Wget can download files from HTTPS Servers
@ -45,7 +47,7 @@ err = HTTPTest (
pre_hook=pre_test,
test_params=test_options,
post_hook=post_test,
servers=Servers
protocols=Servers
).begin ()
exit (err)

3
testenv/Test--spider-r.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test executed Wget in Spider mode with recursive retrieval.

3
testenv/Test-Content-disposition-2.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures that Wget parses the Content-Disposition header

3
testenv/Test-Content-disposition.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures that Wget parses the Content-Disposition header

3
testenv/Test-Head.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures that Wget correctly handles responses to HEAD requests

3
testenv/Test-O.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures that Wget correctly handles the -O command for output

6
testenv/Test-Parallel-Proto.py
View File

@ -1,6 +1,8 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile, HTTP, HTTPS
from test.http_test import HTTPTest
from misc.constants import HTTP, HTTPS
from misc.wget_file import WgetFile
"""
This is a Prototype Test File for multiple servers.
@ -46,7 +48,7 @@ err = HTTPTest (
pre_hook=pre_test,
test_params=test_options,
post_hook=post_test,
servers=Servers
protocols=Servers
).begin ()
exit (err)

3
testenv/Test-Post.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
Simple test for HTTP POST Requests usiong the --method command

6
testenv/Test-Proto.py
View File

@ -1,6 +1,8 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile, HTTP, HTTPS
from test.http_test import HTTPTest
from misc.constants import HTTP, HTTPS
from misc.wget_file import WgetFile
"""
This is a Prototype Test File.
@ -67,7 +69,7 @@ err = HTTPTest (
pre_hook=pre_test,
test_params=test_options,
post_hook=post_test,
server=Servers
protocols=Servers
).begin ()
exit (err)

3
testenv/Test-auth-basic-fail.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures that Wget returns the correct exit code when Basic

3
testenv/Test-auth-basic.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures Wget's Basic Authorization Negotiation.

3
testenv/Test-auth-both.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures Wget's Basic Authorization Negotiation.

3
testenv/Test-auth-digest.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures Wget's Digest Authorization Negotiation.

3
testenv/Test-auth-no-challenge-url.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures Wget's Basic Authorization Negotiation, when credentials

3
testenv/Test-auth-no-challenge.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures Wget's Basic Authorization Negotiation, when the

3
testenv/Test-auth-retcode.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures that Wget returns the correct return code when sent

3
testenv/Test-auth-with-content-disposition.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures that Wget handles Content-Disposition correctly when

3
testenv/Test-c-full.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
Test Wget's response when the file requested already exists on disk with

3
testenv/Test-cookie-401.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures that Wget stores the cookie even in the event of a

3
testenv/Test-cookie-domain-mismatch.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures that Wget identifies bad servers trying to set cookies

3
testenv/Test-cookie-expires.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures that Wget handles Cookie expiry dates correctly.

3
testenv/Test-cookie.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from sys import exit
from WgetTest import HTTPTest, WgetFile
from test.http_test import HTTPTest
from misc.wget_file import WgetFile
"""
This test ensures that Wget's cookie jar support works correctly.

337
testenv/WgetTest.py
View File

@ -1,337 +0,0 @@
import os
import shutil
import shlex
import sys
import traceback
import HTTPServer
import re
import time
from subprocess import call
from ColourTerm import printer
from difflib import unified_diff
HTTP = "HTTP"
HTTPS = "HTTPS"
""" A Custom Exception raised by the Test Environment. """
class TestFailed (Exception):
def __init__ (self, error):
self.error = error
""" Class that defines methods common to both HTTP and FTP Tests. """
class CommonMethods:
TestFailed = TestFailed
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
def get_domain_addr (self, addr):
self.port = str (addr[1])
return addr[0] + ":" + str(addr[1]) + "/"
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")))
try:
retcode = call (params)
except FileNotFoundError as filenotfound:
raise TestFailed (
"The Wget Executable does not exist at the expected path")
return retcode
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 ('..')
try:
if os.getenv ("NO_CLEANUP") is None:
shutil.rmtree (testDir)
except Exception as ae:
print ("Unknown Exception while trying to remove Test Environment.")
def _exit_test (self):
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 __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
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 _replace_substring (self, string):
pattern = re.compile ('\{\{\w+\}\}')
match_obj = pattern.search (string)
if match_obj is not None:
rep = match_obj.group()
temp = getattr (self, rep.strip ('{}'))
string = string.replace (rep, temp)
return string
""" Test Rule Definitions """
""" This should really be taken out soon. All this extra stuff to ensure
re-use of old code is crap. Someone needs to re-write it. The new rework
branch is much better written, but integrating it requires effort.
All these classes should never exist. The whole server needs to modified.
"""
class Authentication:
def __init__ (self, auth_obj):
self.auth_type = auth_obj['Type']
self.auth_user = auth_obj['User']
self.auth_pass = auth_obj['Pass']
class ExpectHeader:
def __init__ (self, header_obj):
self.headers = header_obj
class RejectHeader:
def __init__ (self, header_obj):
self.headers = header_obj
class Response:
def __init__ (self, retcode):
self.response_code = retcode
class SendHeader:
def __init__ (self, header_obj):
self.headers = header_obj
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.
"""
server_rules = dict ()
for rule in file_obj.rules:
r_obj = getattr (self, rule) (file_obj.rules[rule])
server_rules[rule] = r_obj
return server_rules
""" Pre-Test Hook Function Calls """
def ServerFiles (self, server_files):
for i in range (0, self.servers):
file_list = dict ()
server_rules = dict ()
for file_obj in server_files[i]:
content = self._replace_substring (file_obj.content)
file_list[file_obj.name] = content
rule_obj = self.get_server_rules (file_obj)
server_rules[file_obj.name] = rule_obj
self.server_list[i].server_conf (file_list, server_rules)
def LocalFiles (self, local_files):
for file_obj in local_files:
file_handler = open (file_obj.name, "w")
file_handler.write (file_obj.content)
file_handler.close ()
def ServerConf (self, server_settings):
for i in range (0, self.servers):
self.server_list[i].server_sett (server_settings)
""" Test Option Function Calls """
def WgetCommands (self, command_list):
self.options = self._replace_substring (command_list)
def Urls (self, url_list):
self.urls = url_list
""" Post-Test Hook Function Calls """
def ExpectedRetcode (self, retcode):
if self.act_retcode != retcode:
pr = "Return codes do not match.\nExpected: " + str(retcode) + "\nActual: " + str(self.act_retcode)
raise TestFailed (pr)
def ExpectedFiles (self, exp_filesys):
self.__check_downloaded_files (exp_filesys)
def FilesCrawled (self, Request_Headers):
for i in range (0, self.servers):
headers = set(Request_Headers[i])
o_headers = self.Request_remaining[i]
header_diff = headers.symmetric_difference (o_headers)
if len(header_diff) is not 0:
printer ("RED", str (header_diff))
raise TestFailed ("Not all files were crawled correctly")
""" Class for HTTP Tests. """
class HTTPTest (CommonMethods):
# 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:
printer ("RED", "Error: " + tf.error)
self.tests_passed = False
except Exception as ae:
printer ("RED", "Unhandled Exception Caught.")
print ( ae.__str__ ())
traceback.print_exc ()
self.tests_passed = False
else:
printer ("GREEN", "Test Passed")
finally:
self._exit_test ()
def Server_setup (self, name, pre_hook, test_params, post_hook, servers):
self.name = name
self.server_types = servers
self.servers = len (servers)
printer ("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)
self.pre_hook_call (pre_hook)
self.call_test (test_params)
self.post_hook_call (post_hook)
def pre_hook_call (self, pre_hook):
for pre_hook_func in pre_hook:
try:
assert hasattr (self, pre_hook_func)
except AssertionError as ae:
self.stop_HTTP_Server ()
raise TestFailed ("Pre Test Function " + pre_hook_func + " not defined.")
getattr (self, pre_hook_func) (pre_hook[pre_hook_func])
def call_test (self, test_params):
for test_func in test_params:
try:
assert hasattr (self, test_func)
except AssertionError as ae:
self.stop_HTTP_Server ()
raise TestFailed ("Test Option " + test_func + " unknown.")
getattr (self, test_func) (test_params[test_func])
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):
for post_hook_func in post_hook:
try:
assert hasattr (self, post_hook_func)
except AssertionError as ae:
raise TestFailed ("Post Test Function " + post_hook_func + " not defined.")
getattr (self, post_hook_func) (post_hook[post_hook_func])
def init_HTTP_Server (self):
server = HTTPServer.HTTPd ()
server.start ()
return server
def init_HTTPS_Server (self):
server = HTTPServer.HTTPSd ()
server.start ()
return server
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 ()
""" WgetFile is a File Data Container object """
class WgetFile:
def __init__ (
self,
name,
content="Test Contents",
timestamp=None,
rules=dict()
):
self.name = name
self.content = content
self.timestamp = timestamp
self.rules = rules
# vim: set ts=4 sts=4 sw=4 tw=80 et :

47
testenv/conf/__init__.py Normal file
View File

@ -0,0 +1,47 @@
import os
# this file implements the mechanism of conf class auto-registration,
# don't modify this file if you have no idea what you're doing
def gen_hook():
hook_table = {}
class Wrapper:
"""
Decorator class which implements the conf class registration.
"""
def __init__(self, alias=None):
self.alias = alias
def __call__(self, cls):
# register the class object with the name of the class
hook_table[cls.__name__] = cls
if self.alias:
# also register the alias of the class
hook_table[self.alias] = cls
return cls
def find_hook(name):
if name in hook_table:
return hook_table[name]
else:
raise AttributeError
return Wrapper, find_hook
_register, find_conf = gen_hook()
hook = rule = _register
__all__ = ['hook', 'rule']
for module in os.listdir(os.path.dirname(__file__)):
# import every module under this package except __init__.py,
# so that the decorator `register` applies
# (nothing happens if the script is not loaded)
if module != '__init__.py' and module.endswith('.py'):
module_name = module[:-3]
mod = __import__('%s.%s' % (__name__, module_name),
globals(),
locals())
__all__.append(module_name)

9
testenv/conf/authentication.py Normal file
View File

@ -0,0 +1,9 @@
from conf import rule
@rule()
class Authentication:
def __init__ (self, auth_obj):
self.auth_type = auth_obj['Type']
self.auth_user = auth_obj['User']
self.auth_pass = auth_obj['Pass']

7
testenv/conf/expect_header.py Normal file
View File

@ -0,0 +1,7 @@
from conf import rule
@rule()
class ExpectHeader:
def __init__(self, header_obj):
self.headers = header_obj

42
testenv/conf/expected_files.py Normal file
View File

@ -0,0 +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, 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):
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.')

16
testenv/conf/expected_ret_code.py Normal file
View File

@ -0,0 +1,16 @@
from exc.test_failed import TestFailed
from conf import hook
@hook(alias='ExpectedRetcode')
class ExpectedRetCode:
def __init__(self, expected_ret_code):
self.expected_ret_code = expected_ret_code
def __call__(self, test_obj):
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)

18
testenv/conf/files_crawled.py Normal file
View File

@ -0,0 +1,18 @@
from misc.colour_terminal import print_red
from conf import hook
from exc.test_failed import TestFailed
@hook()
class FilesCrawled:
def __init__(self, request_headers):
self.request_headers = request_headers
def __call__(self, test_obj):
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.')

15
testenv/conf/hook_sample.py Normal file
View File

@ -0,0 +1,15 @@
from exc.test_failed import TestFailed
from conf import hook
# this file is a hook example
@hook(alias='SampleHookAlias')
class SampleHook:
def __init__(self, sample_hook_arg):
# do conf initialization here
self.arg = sample_hook_arg
def __call__(self, test_obj):
# implement hook here
# if you need the test case instance, refer to test_obj
pass

12
testenv/conf/local_files.py Normal file
View File

@ -0,0 +1,12 @@
from conf import hook
@hook()
class LocalFiles:
def __init__(self, local_files):
self.local_files = local_files
def __call__(self, _):
for f in self.local_files:
with open(f.name, 'w') as fp:
fp.write(f.content)

7
testenv/conf/reject_header.py Normal file
View File

@ -0,0 +1,7 @@
from conf import rule
@rule()
class RejectHeader:
def __init__ (self, header_obj):
self.headers = header_obj

7
testenv/conf/response.py Normal file
View File

@ -0,0 +1,7 @@
from conf import rule
@rule()
class Response:
def __init__(self, ret_code):
self.response_code = ret_code

10
testenv/conf/rule_sample.py Normal file
View File

@ -0,0 +1,10 @@
from conf import rule
@rule(alias='SampleRuleAlias')
class SampleRule:
def __init__(self, rule):
# do rule initialization here
# you may also need to implement a method the same name of this
# class in server/protocol/protocol_server.py to apply this rule.
self.rule = rule

7
testenv/conf/send_header.py Normal file
View File

@ -0,0 +1,7 @@
from conf import rule
@rule()
class SendHeader:
def __init__(self, header_obj):
self.headers = header_obj

11
testenv/conf/server_conf.py Normal file
View File

@ -0,0 +1,11 @@
from conf import hook
@hook()
class ServerConf:
def __init__(self, server_settings):
self.server_settings = server_settings
def __call__(self, test_obj):
for server in test_obj.servers:
server.server_sett(self.server_settings)

15
testenv/conf/server_files.py Normal file
View File

@ -0,0 +1,15 @@
from conf import hook
@hook()
class ServerFiles:
def __init__(self, server_files):
self.server_files = server_files
def __call__(self, test_obj):
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)

10
testenv/conf/urls.py Normal file
View File

@ -0,0 +1,10 @@
from conf import hook
@hook(alias='Urls')
class URLs:
def __init__(self, urls):
self.urls = urls
def __call__(self, test_obj):
test_obj.urls = self.urls

10
testenv/conf/wget_commands.py Normal file
View File

@ -0,0 +1,10 @@
from conf import hook
@hook()
class WgetCommands:
def __init__(self, commands):
self.commands = commands
def __call__(self, test_obj):
test_obj.wget_options = test_obj._replace_substring(self.commands)

0
testenv/exc/__init__.py Normal file
View File

7
testenv/exc/test_failed.py Normal file
View File

@ -0,0 +1,7 @@
class TestFailed(Exception):
""" A Custom Exception raised by the Test Environment. """
def __init__ (self, error):
self.error = error

0
testenv/misc/__init__.py Normal file
View File

8
testenv/ColourTerm.py → testenv/misc/colour_terminal.py
View File

@ -1,3 +1,4 @@
from functools import partial
import platform
from os import getenv
@ -20,4 +21,11 @@ def printer (color, string):
else:
print (string)
print_blue = partial(printer, 'BLUE')
print_red = partial(printer, 'RED')
print_green = partial(printer, 'GREEN')
print_purple = partial(printer, 'PURPLE')
print_yellow = partial(printer, 'YELLOW')
# vim: set ts=8 sw=3 tw=0 et :

16
testenv/misc/wget_file.py Normal file
View File

@ -0,0 +1,16 @@
class WgetFile:
""" WgetFile is a File Data Container object """
def __init__ (
self,
name,
content="Test Contents",
timestamp=None,
rules=None
):
self.name = name
self.content = content
self.timestamp = timestamp
self.rules = rules or {}

0
testenv/server/__init__.py Normal file
View File

0
testenv/server/ftp/__init__.py Normal file
View File

0
testenv/FTPServer.py → testenv/server/ftp/ftp_server.py
View File

0
testenv/server/http/__init__.py Normal file
View File

0
testenv/HTTPServer.py → testenv/server/http/http_server.py
View File

0
testenv/test/__init__.py Normal file
View File

226
testenv/test/base_test.py Normal file
View File

@ -0,0 +1,226 @@
import os
import shutil
import shlex
import traceback
import re
import time
from subprocess import call
from misc.colour_terminal import print_red, print_blue
from exc.test_failed import TestFailed
import conf
HTTP = "HTTP"
HTTPS = "HTTPS"
class BaseTest:
"""
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)
"""
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 = []
self.tests_passed = True
self.init_test_env()
self.ret_code = 0
def get_test_dir(self):
return self.name + '-test'
def init_test_env(self):
test_dir = self.get_test_dir()
try:
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)
return cmd_line
def __test_cleanup(self):
os.chdir('..')
try:
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()
def begin (self):
return 0 if self.tests_passed else 100
def call_test(self):
self.hook_call(self.test_params, 'Test Option')
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 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+\}\}')
match_obj = pattern.search (string)
if match_obj is not None:
rep = match_obj.group()
temp = getattr (self, rep.strip ('{}'))
string = string.replace (rep, temp)
return string
def instantiate_server_by(self, protocol):
"""
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

45
testenv/test/http_test.py Normal file
View File

@ -0,0 +1,45 @@
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(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.
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 instantiate_server_by(self, protocol):
server = {HTTP: HTTPd,
HTTPS: HTTPSd}[protocol]()
server.start()
return server
def request_remaining(self):
return [s.server_inst.get_req_headers()
for s in self.servers]
def stop_server(self):
for server in self.servers:
server.server_inst.shutdown()
# vim: set ts=4 sts=4 sw=4 tw=80 et :