vincent/aptly
1
0
Fork 0
mirror of https://github.com/aptly-dev/aptly.git synced 2026-06-12 06:30:35 +00:00
Code Issues Projects Releases Wiki Activity

Library for API system tests. #116

Browse Source
This commit is contained in:
Andrey Smirnov
2014-10-10 17:43:04 +04:00
parent 53f7fef4cf
commit 523d0d0945
3 changed files with 99 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
system/lib.py
+28 -16
View File
@@ -14,6 +14,7 @@ import string
import threading
import urllib
#import time
import pprint
import SocketServer
import SimpleHTTPServer
@@ -149,25 +150,28 @@ class BaseTest(object):
def run(self):
self.output = self.output_processor(self.run_cmd(self.runCmd, self.expectedCode))
def _start_process(self, command, stderr=subprocess.STDOUT, stdout=None):
if not hasattr(command, "__iter__"):
params = {
'files': os.path.join(os.path.dirname(inspect.getsourcefile(BaseTest)), "files"),
'udebs': os.path.join(os.path.dirname(inspect.getsourcefile(BaseTest)), "udebs"),
'testfiles': os.path.join(os.path.dirname(inspect.getsourcefile(self.__class__)), self.__class__.__name__),
}
if self.fixtureWebServer:
params['url'] = self.webServerUrl
command = string.Template(command).substitute(params)
command = shlex.split(command)
environ = os.environ.copy()
environ["LC_ALL"] = "C"
environ.update(self.environmentOverride)
return subprocess.Popen(command, stderr=stderr, stdout=stdout, env=environ)
def run_cmd(self, command, expected_code=0):
try:
#start = time.time()
if not hasattr(command, "__iter__"):
params = {
'files': os.path.join(os.path.dirname(inspect.getsourcefile(BaseTest)), "files"),
'udebs': os.path.join(os.path.dirname(inspect.getsourcefile(BaseTest)), "udebs"),
'testfiles': os.path.join(os.path.dirname(inspect.getsourcefile(self.__class__)), self.__class__.__name__),
}
if self.fixtureWebServer:
params['url'] = self.webServerUrl
command = string.Template(command).substitute(params)
command = shlex.split(command)
environ = os.environ.copy()
environ["LC_ALL"] = "C"
environ.update(self.environmentOverride)
proc = subprocess.Popen(command, stderr=subprocess.STDOUT, stdout=subprocess.PIPE, env=environ)
proc = self._start_process(command, stdout=subprocess.PIPE)
output, _ = proc.communicate()
#print "CMD %s: %.2f" % (" ".join(command), time.time()-start)
if proc.returncode != expected_code:
@@ -254,6 +258,10 @@ class BaseTest(object):
if os.path.exists(os.path.join(os.environ["HOME"], ".aptly", path)):
raise Exception("path %s exists" % (path, ))
def check_equal(self, a, b):
if a != b:
self.verify_match(a, b, match_prepare=pprint.pformat)
def verify_match(self, a, b, match_prepare=None):
if match_prepare is not None:
a = match_prepare(a)
@@ -287,3 +295,7 @@ class BaseTest(object):
def shutdown_webserver(self):
self.webserver.shutdown()
@classmethod
def shutdown_class(cls):
pass