Skip to content
Snippets Groups Projects
Commit 0d9a63c7 authored by felmer's avatar felmer
Browse files

SP-1060, SWE-36: testing framework improved. test case screening finished.

SVN: 30224
parent c3c7d3c5
No related branches found
No related tags found
No related merge requests found
......@@ -9,20 +9,23 @@ import time
CI_BASE_URL = 'http://bs-ci01.ethz.ch:8090'
reuseRepository = False
suppressCleanUp = False
devMode = False
cmd = sys.argv[0]
if len(sys.argv) > 1:
firstArgument = sys.argv[1]
if firstArgument == '-r':
reuseRepository = True
if firstArgument == '-sr' or firstArgument == '-rs':
elif firstArgument == '-dr' or firstArgument == '-rd':
reuseRepository = True
suppressCleanUp = True
if firstArgument == '-s':
suppressCleanUp = True
if firstArgument == '-h':
print ("Usage: %s [-h|-r|-s|-rs]\n-h: prints this help\n-r: reuses artifact repository\n"
+ "-s: suppress clean up\n-sr: both options") % os.path.basename(cmd)
devMode = True
elif firstArgument == '-d':
devMode = True
elif firstArgument == '-h':
print ("Usage: %s [-h|-r|-d|-rd]\n-h: prints this help\n-r: reuses artifact repository\n"
+ "-d: developing mode\n-rd: both options") % os.path.basename(cmd)
exit(1)
else:
print "Unknown option: %s. Use option '-h' to see usage." % firstArgument
exit(1)
dirname = os.path.dirname(cmd)
......
import re
import os
import os.path
import shutil
......@@ -13,7 +14,42 @@ PLAYGROUND = 'targets/playground'
TEMPLATES = 'templates'
PSQL_EXE = 'psql'
class TestCase():
class TestCase(object):
"""
Abstract superclass of a test case.
Subclasses have to override execute() and optionally executeInDevMode().
The test case is run by invoking runTest().
Here is a skeleton of a test case:
#!/usr/bin/python
import settings
import systemtest.testcase
class TestCase(systemtest.testcase.TestCase):
def execute(self):
....
def executeInDevMode(self):
....
TestCase(settings, __file__).runTest()
There are two execution modes (controlled by command line option -d and -rd):
Normal mode:
1. Cleans up playground: Kills running servers and deletes playground folder of this test case.
2. Invokes execute() method.
3. Release resources: Shuts down running servers.
Developing mode:
1. Invokes executeInDevMode() method.
The developing mode allows to reuse already installed servers.
Servers might be restarted. This mode leads to fast development
of test code by doing incremental development. Working code
can be moved from executeInDevMode() to execute().
"""
def __init__(self, settings, filePath):
self.artifactRepository = settings.REPOSITORY
self.project = None
......@@ -21,22 +57,24 @@ class TestCase():
self.name = fileName[0:fileName.rfind('.')]
self.playgroundFolder = "%s/%s" % (PLAYGROUND, self.name)
self.numberOfFailures = 0
self.suppressCleanUp = settings.suppressCleanUp
self.devMode = settings.devMode
self.runningOpenbisInstances = []
def runTest(self):
"""
Runs this test case. This is a final method. It should not be overwritten.
"""
startTime = time.time()
print "\n/''''''''''''''''''' %s started at %s ''''''''''" % (self.name, time.strftime('%Y-%m-%d %H:%M:%S'))
print "\n/''''''''''''''''''' %s started at %s %s ''''''''''" % (self.name, time.strftime('%Y-%m-%d %H:%M:%S'),
'in DEV MODE' if self.devMode else '')
try:
if os.path.exists(self.playgroundFolder):
if not self.suppressCleanUp:
if not self.devMode:
if os.path.exists(self.playgroundFolder):
self._cleanUpPlayground()
os.makedirs(self.playgroundFolder)
else:
os.makedirs(self.playgroundFolder)
self.execute()
self.execute()
else:
self.executeInDevMode()
success = self.numberOfFailures == 0
except:
traceback.print_exc()
......@@ -44,6 +82,8 @@ class TestCase():
raise Exception("%s failed" % self.name)
finally:
duration = time.time() - startTime
if not self.devMode:
self.releaseResources()
if success:
print "\...........SUCCESS: %s executed in %d seconds .........." % (self.name, duration)
else:
......@@ -51,26 +91,56 @@ class TestCase():
def execute(self):
"""
Executes this test case. This is an abstract method which has to be overwritten in subclasses.
Executes this test case in normal mode.
This is an abstract method which has to be overwritten in subclasses.
"""
pass
def executeInDevMode(self):
"""
Executes this test case in developing mode.
This method can be overwritten in subclasses.
"""
pass
def releaseResources(self):
"""
Releases resources. It shuts down all running servers.
This method can be overwritten in subclasses.
Note, this method can be invoked in subclasses as follows:
super(type(self), self).releaseResources()
"""
self._shutdownSevers()
def assertPatternInLog(self, log, pattern):
if not re.search(pattern, log):
self.fail("Pattern doesn't match: %s" % pattern)
def assertEquals(self, itemName, expected, actual):
"""
Asserts that expected == actual. If not both will be printed and the test will be counted as failed.
Asserts that expected == actual. If not the test will be continued but counted as failed.
"""
if expected != actual:
self.numberOfFailures += 1
print "ERROR: %s\n expected: <%s>\n but was: <%s>" % (itemName, expected, actual)
self.fail("%s\n expected: <%s>\n but was: <%s>" % (itemName, expected, actual))
else:
print "%s as expected: <%s>" % (itemName, expected)
def fail(self, errorMessage):
"""
Prints specified error message and mark test case as failed.
"""
self.numberOfFailures += 1
print "ERROR: %s" % errorMessage
def installOpenbis(self, instanceName = 'openbis', technologies = []):
"""
Installs openBIS from the installer.
The instanceName specifies the subfolder in the playground folder where the instance will be installed.
The instanceName specifies the subfolder in the playground folder
where the instance will be installed.
In addition it is also part of the database names.
The technologies are an array of enabled technologies.
An instance of OpenbisController is returned.
......@@ -89,10 +159,20 @@ class TestCase():
util.writeProperties(consolePropertiesFile, consoleProperties)
util.executeCommand("%s/%s/run-console.sh" % (self.playgroundFolder, installerFileName),
"Couldn't install openBIS", consoleInput='admin\nadmin')
return OpenbisController(self.name, installPath, instanceName)
return OpenbisController(self, self.name, installPath, instanceName)
def createOpenbisController(self, instanceName = 'openbis'):
return OpenbisController(self.name, self._getOpenbisInstallPath(instanceName), instanceName, dropDatabases = False)
"""
Creates an openBIS controller object assuming that an openBIS instance for the specified name is installed.
"""
return OpenbisController(self, self.name, self._getOpenbisInstallPath(instanceName), instanceName, dropDatabases = False)
def installScreeningTestClient(self):
""" Installs the screening test client and returns an instance of ScreeningTestClient. """
zipFile = self.artifactRepository.getPathToArtifact(OPENBIS_STANDARD_TECHNOLOGIES_PROJECT, 'openBIS-screening-API')
installPath = "%s/screeningAPI" % self.playgroundFolder
util.unzip(zipFile, installPath)
return ScreeningTestClient(self, installPath)
def _getOpenbisInstallPath(self, instanceName):
return os.path.abspath("%s/%s" % (self.playgroundFolder, instanceName))
......@@ -107,7 +187,6 @@ class TestCase():
util.killProcess("%s/servers/openBIS-server/jetty/openbis.pid" % path)
util.deleteFolder(self.playgroundFolder)
def _getAndCreateFolder(self, folderPath):
"""
Creates a folder inside the playground. The argument is a relative path to the playground.
......@@ -117,11 +196,37 @@ class TestCase():
os.makedirs(path)
return path
def _shutdownSevers(self):
for instance in self.runningOpenbisInstances:
instance.allDown()
class ScreeningTestClient():
"""
Class representing the screeing test client.
"""
def __init__(self, testCase, installPath):
self.testCase = testCase
self.installPath = installPath
def run(self):
""" Runs the test client and returns the console output as a list of strings. """
output = util.executeCommand(['java', "-Djavax.net.ssl.trustStore=%s/openBIS.keystore" % self.installPath,
'-jar', "%s/openbis_screening_api.jar" % self.installPath, 'admin', 'admin',
'https://localhost:8443'], suppressStdOut = True)
with open("%s/log.txt" % self.installPath, 'w') as log:
for line in output:
log.write("%s\n" % line)
return output
class OpenbisController():
"""
Class to control AS and DSS of an installed openBIS instance.
"""
def __init__(self, testName, installPath, instanceName, dropDatabases = True):
def __init__(self, testCase, testName, installPath, instanceName, dropDatabases = True):
"""
Creates a new instance for specifies test case with specified test and instance name, installation path.
"""
self.testCase = testCase
self.testName = testName
self.instanceName = instanceName
self.installPath = installPath
......@@ -136,23 +241,70 @@ class OpenbisController():
self.databaseKind = "%s_%s" % (testName, instanceName)
self.asProperties['database.kind'] = self.databaseKind
self.asPropertiesModified = True
if dropDatabases:
util.dropDatabase(PSQL_EXE, "openbis_%s" % self.databaseKind)
self.dssServicePropertiesFile = "%s/servers/datastore_server/etc/service.properties" % installPath
self.dssProperties = util.readProperties(self.dssServicePropertiesFile)
self.dssProperties['path-info-db.databaseKind'] = self.databaseKind
self.dssProperties['imaging-database.kind'] = self.databaseKind
self.dssProperties['proteomics-database-kind'] = self.databaseKind
self.dssPropertiesModified = True
if dropDatabases:
util.dropDatabase(PSQL_EXE, "openbis_%s" % self.databaseKind)
util.dropDatabase(PSQL_EXE, "pathinfo_%s" % self.databaseKind)
util.dropDatabase(PSQL_EXE, "imaging_%s" % self.databaseKind)
util.dropDatabase(PSQL_EXE, "proteomics_%s" % self.databaseKind)
self._applyCorePlugins()
def assertFileExist(self, pathRelativeToInstallPath):
"""
Asserts that the specified path (relative to the installation path) exists.
"""
relativePath = "%s/%s" % (self.installPath, pathRelativeToInstallPath)
if os.path.exists(relativePath):
print "Path exists as expected: %s" % relativePath
else:
self.testCase.fail("Path doesn't exist: %s" % relativePath)
def assertEmptyFolder(self, pathRelativeToInstallPath):
"""
Asserts that the specified path (relative to the installation path) is an empty folder.
"""
relativePath = "%s/%s" % (self.installPath, pathRelativeToInstallPath)
if not os.path.isdir(relativePath):
self.testCase.fail("Doesn't exist or isn't a folder: %s" % relativePath)
files = os.listdir(relativePath)
if len(files) == 0:
print "Empty folder as expected: %s" % relativePath
else:
self.testCase.fail("%s isn't empty. It contains the following files:\n %s" % (relativePath, files))
def assertNumberOfDataSets(self, expectedNumberOfDataSets):
"""
Asserts that the specified number of data sets are in the store and in the database.
Some meta data of all data sets are returned.
"""
metaData = self.queryDatabase("openbis", "select e.code,data.code,t.code,location from data"
+ " join external_data as ed on ed.data_id = data.id"
+ " join data_set_types as t on data.dsty_id = t.id"
+ " join experiments as e on data.expe_id = e.id")
util.printResultSet(metaData)
self.testCase.assertEquals('number of data sets', expectedNumberOfDataSets, len(metaData));
for row in metaData:
self.assertFileExist("data/store/1/%s/original" % row[3])
return metaData
def createTestDatabase(self, databaseType):
"""
Creates a test database for the specified database type.
"""
database = "%s_%s" % (databaseType, self.databaseKind)
scriptPath = "%s/%s.sql" % (self.templatesFolder, database)
util.createDatabase(PSQL_EXE, database, scriptPath)
def queryDatabase(self, databaseType, queryStatement):
"""
Executes the specified SQL statement for the specified database type. Result set is returned
as a list of lists.
"""
database = "%s_%s" % (databaseType, self.databaseKind)
return util.queryDatabase(PSQL_EXE, database, queryStatement)
......@@ -162,6 +314,7 @@ class OpenbisController():
"""
self._saveAsPropertiesIfModified()
self._saveDssPropertiesIfModified()
self.testCase.runningOpenbisInstances.append(self)
util.executeCommand([self.bisUpScript], "Starting up openBIS AS '%s' failed." % self.instanceName)
util.executeCommand([self.dssUpScript], "Starting up openBIS DSS '%s' failed." % self.instanceName)
......@@ -171,6 +324,7 @@ class OpenbisController():
"""
self._saveAsPropertiesIfModified()
self._saveDssPropertiesIfModified()
self.testCase.runningOpenbisInstances.remove(self)
util.executeCommand([self.dssDownScript], "Shutting down openBIS DSS '%s' failed." % self.instanceName)
util.executeCommand([self.bisDownScript], "Shutting down openBIS AS '%s' failed." % self.instanceName)
......@@ -183,7 +337,7 @@ class OpenbisController():
"%s/servers/datastore_server/log/datastore_server_log.txt" % self.installPath,
timeOutInMinutes=5)
monitor.addNotificationCondition(util.RegexCondition('Incoming Data Monitor'))
monitor.waitUntilEvent(util.RegexCondition('Post registration'))
monitor.waitUntilEvent(util.RegexCondition('Post registration of (\\d*). of \\1 data sets'))
def _applyCorePlugins(self):
corePluginsFolder = "%s/servers/core-plugins" % self.installPath
......
import fcntl
import os
import os.path
import re
import time
import shutil
import subprocess
import threading
USER=os.environ['USER']
......@@ -54,7 +52,7 @@ def executeCommand(commandWithArguments, failingMessage = None, consoleInput = N
if exitValue != 0 and failingMessage != None:
print "---- FAILED %d: %s" % (exitValue, commandWithArguments)
raise Exception(failingMessage)
print "---- FINISHED: %s" % commandWithArguments
print "---- FINISHED: %s\n" % commandWithArguments
return lines
def killProcess(pidFile):
......@@ -71,7 +69,7 @@ def unzip(zipFile, destination):
"""
Unzips specified ZIP file at specified destination.
"""
executeCommand(['unzip', '-q', zipFile, '-d', destination], "Couldn't unzip %s at %s" % (zipFile, destination))
executeCommand(['unzip', '-q', '-o', zipFile, '-d', destination], "Couldn't unzip %s at %s" % (zipFile, destination))
def deleteFolder(folderPath):
"""
......@@ -104,7 +102,8 @@ def createDatabase(psqlExe, database, scriptPath = None):
def queryDatabase(psqlExe, database, queryStatement):
"""
Queries specified database by applying specified SQL statement.
Queries specified database by applying specified SQL statement and returns the result set as a list
where each row is a list, too.
"""
lines = executeCommand([psqlExe, '-U', 'postgres', '-tA', '-d', database, '-c', queryStatement],
"Couldn't execute query: %s" % queryStatement, suppressStdOut = True)
......@@ -112,6 +111,13 @@ def queryDatabase(psqlExe, database, queryStatement):
for line in lines:
result.append(line.split('|'))
return result
def printResultSet(resultSet):
"""
Prints the specified result set.
"""
for row in resultSet:
print row
class LogMonitor():
"""
......@@ -145,37 +151,42 @@ class LogMonitor():
self.conditions.append(condition)
startTime = self.timeProvider.time()
renderedStartTime = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(startTime))
self.printer.printMsg("Start monitoring %s log at %s" % (self.logName, renderedStartTime))
self.printer.printMsg(">>>>> Start monitoring %s log at %s >>>>>>>>>>>>>>>>>>>>" % (self.logName, renderedStartTime))
finalTime = startTime + self.timeOutInMinutes * 60
while True:
log = open(self.logFilePath, 'r')
try:
alreadyPrintedLines = set()
while True:
actualTime = self.timeProvider.time()
if actualTime > finalTime:
raise Exception("Time out after %d minutes for monitoring %s log."
% (self.timeOutInMinutes, self.logName))
line = log.readline()
if line == '':
break
match = re.match('(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}),\d{3} (.{6})(.*)', line)
if match == None:
continue
timestamp = match.group(1)
eventType = match.group(2)
message = match.group(3)
eventTime = time.mktime(time.strptime(timestamp, '%Y-%m-%d %H:%M:%S'))
if eventTime < startTime:
continue
for c in self.conditions:
if c.match(message):
self.printer.printMsg("%s log: %s" % (self.logName, line.strip()))
log = open(self.logFilePath, 'r')
while True:
actualTime = self.timeProvider.time()
if actualTime > finalTime:
raise Exception("Time out after %d minutes for monitoring %s log."
% (self.timeOutInMinutes, self.logName))
line = log.readline()
if line == '':
break
if condition.match(message):
return
if eventType.strip() == 'ERROR':
raise Exception("Error spotted in %s log." % self.logName)
log.seek(0, 1)
time.sleep(5)
match = re.match('(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}),\d{3} (.{6})(.*)', line)
if match == None:
continue
timestamp = match.group(1)
eventType = match.group(2)
message = match.group(3)
eventTime = time.mktime(time.strptime(timestamp, '%Y-%m-%d %H:%M:%S'))
if eventTime < startTime:
continue
for c in self.conditions:
if c.match(message) and not line in alreadyPrintedLines:
alreadyPrintedLines.add(line)
self.printer.printMsg(">> %s" % line.strip())
break
if condition.match(message):
return
if eventType.strip() == 'ERROR':
raise Exception("Error spotted in %s log." % self.logName)
log.seek(0, os.SEEK_CUR)
time.sleep(5)
finally:
self.printer.printMsg(">>>>> Finished monitoring %s log >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>\n" % self.logName)
class StartsWithCondition():
"""
......@@ -186,7 +197,7 @@ class StartsWithCondition():
def match(self, message):
return message.startswith(self.startsWithString)
class RegexCondition():
"""
A condition which matches if the message matches a specified regular expression.
......
......@@ -34,12 +34,14 @@ class UtilTest(TestCaseWithFiles):
monitor = self._createMonitor(logFile)
monitor.addNotificationCondition(util.RegexCondition('.*'))
monitor.waitUntilEvent(util.StartsWithCondition('Post registration'))
monitor.waitUntilEvent(util.RegexCondition('Post registration of (\\d*). of \\1 data sets'))
self.assertEqual(['Start monitoring TEST log at 2013-10-01 10:50:00',
'TEST log: 2013-10-01 10:50:00,025 WARN [qtp797130442-28] OPERATION',
'TEST log: 2013-10-01 10:50:20,559 INFO blabla',
'TEST log: 2013-10-01 10:50:30,559 INFO Post registration of 1. of 1 data sets'],
self.assertEqual(['>>>>> Start monitoring TEST log at 2013-10-01 10:50:00 >>>>>>>>>>>>>>>>>>>>',
'>> 2013-10-01 10:50:00,025 WARN [qtp797130442-28] OPERATION',
'>> 2013-10-01 10:50:20,559 INFO blabla',
'>> 2013-10-01 10:50:25,559 INFO Post registration of 1. of 2 data sets',
'>> 2013-10-01 10:50:30,559 INFO Post registration of 2. of 2 data sets',
'>>>>> Finished monitoring TEST log >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>\n'],
monitor.printer.recorder)
def test_LogMonitor_for_error_event(self):
......@@ -55,9 +57,11 @@ class UtilTest(TestCaseWithFiles):
self.fail('Exception expected')
except Exception as e:
self.assertEqual('Error spotted in TEST log.', str(e))
self.assertEqual(['Start monitoring TEST log at 2013-10-01 10:50:00',
'TEST log: 2013-10-01 10:50:00,025 WARN [qtp797130442-28] OPERATION',
'TEST log: 2013-10-01 10:50:20,559 ERROR test'], monitor.printer.recorder)
self.assertEqual(['>>>>> Start monitoring TEST log at 2013-10-01 10:50:00 >>>>>>>>>>>>>>>>>>>>',
'>> 2013-10-01 10:50:00,025 WARN [qtp797130442-28] OPERATION',
'>> 2013-10-01 10:50:20,559 ERROR test',
'>>>>> Finished monitoring TEST log >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>\n'],
monitor.printer.recorder)
def test_LogMonitor_timeout(self):
logFile = self._createExampleLog()
......@@ -68,13 +72,17 @@ class UtilTest(TestCaseWithFiles):
self.fail('Exception expected')
except Exception as e:
self.assertEqual('Time out after 1 minutes for monitoring TEST log.', str(e))
self.assertEqual(['>>>>> Start monitoring TEST log at 2013-10-01 10:50:00 >>>>>>>>>>>>>>>>>>>>',
'>>>>> Finished monitoring TEST log >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>\n'],
monitor.printer.recorder)
def _createExampleLog(self):
return self._createLogFromEvents(['2013-10-01 10:40:20,559 INFO blabla',
'2013-10-01 10:50:00,025 WARN [qtp797130442-28] OPERATION',
'ch.systemsx.cisd.common.exceptions.UserFailureException: Experiment',
'2013-10-01 10:50:20,559 INFO blabla',
'2013-10-01 10:50:30,559 INFO Post registration of 1. of 1 data sets',
'2013-10-01 10:50:25,559 INFO Post registration of 1. of 2 data sets',
'2013-10-01 10:50:30,559 INFO Post registration of 2. of 2 data sets',
'2013-10-01 10:50:40,559 INFO blabla',
'2013-10-01 10:50:50,559 INFO blabla',
'2013-10-01 10:51:30,559 INFO Too late'])
......
......@@ -3,24 +3,62 @@ import settings
import systemtest.testcase
class TestCase(systemtest.testcase.TestCase):
def execute(self):
"""
openbisController = self.installOpenbis(technologies = ['screening'])
openbisController.createTestDatabase('openbis')
openbisController.allUp()
openbisController.drop('data-incoming-analysis.zip', 'incoming-analysis')
self.assertEquals('spot sizes', [['24', '16']], self.getSpotSizes(openbisController))
self.assertSpotSizes(openbisController, [['24', '16']])
openbisController.drop('data-incoming-images-merged-channels.zip', 'incoming-images-merged-channels')
openbisController.drop('data-incoming-images-split-channels.zip', 'incoming-images-split-channels')
# """
openbisController.assertEmptyFolder('data/incoming-analysis');
openbisController.assertEmptyFolder('data/incoming-images-merged-channels');
openbisController.assertEmptyFolder('data/incoming-images-split-channels');
openbisController.assertNumberOfDataSets(3)
self.assertSpotSizes(openbisController, [['24', '16'], ['24', '16']])
self.assertFeatureVectorLabel(openbisController, 'HITRATE', 'Hit Rate')
self.assertFeatureVectorLabel(openbisController, 'CELLNUMBER', 'cellNumber')
client = self.installScreeningTestClient()
log = '\n'.join(client.run())
self.assertPatternInLog(log, "Experiments: \[/DEMO/DEMO_PROJECT/DEMO_EXPERIMENT \[20100623121102843-1\]\]")
self.assertPatternInLog(log, "Plates: \[/DEMO/PLATE1 \[20100624113752213-5\]")
self.assertPatternInLog(log, "Image datasets: \[[0-9]*-[0-9]* \(plate: /DEMO/PLATE3")
self.assertPatternInLog(log, "Feature vector datasets: \[[0-9]*-[0-9]* \(plate: /DEMO/PLATE2 \[20100624113756254-6\]")
self.assertPatternInLog(log, "Feature codes: \[CELLNUMBER, FEATRUE1, FEATRUE10, FEATRUE11, FEATRUE12, FEATRUE13, "
+ "FEATRUE14, FEATRUE15, FEATRUE16, FEATRUE2, FEATRUE3, FEATRUE4, FEATRUE5, FEATRUE6, "
+ "FEATRUE7, FEATRUE8, FEATRUE9, HITRATE, STD1, STD10, STD11, STD12, STD13, "
+ "STD14, STD15, STD16, STD2, STD3, STD4, STD5, STD6, STD7, STD8, STD9\]")
self.assertPatternInLog(log, "Loaded feature datasets: 1")
self.assertPatternInLog(log, "features labels: \[cellNumber, featrue1, featrue10, featrue11, featrue12, "
+ "featrue13, featrue14, featrue15, featrue16, featrue2, featrue3, featrue4, "
+ "featrue5, featrue6, featrue7, featrue8, featrue9, Hit Rate, std1, std10, "
+ "std11, std12, std13, std14, std15, std16, std2, std3, std4, std5, std6, "
+ "std7, std8, std9\]")
self.assertPatternInLog(log, "Features of the first dataset: datasetCode: [0-9]*-[0-9]*")
self.assertPatternInLog(log, "wellPosition: \[1, 2\], values: \[48.0, 0.0051865")
self.assertPatternInLog(log, "Image metadata: \[Dataset [0-9]*-[0-9]* \(plate: /DEMO/PLATE3 "
+ "\[20100624113759640-7\]\) has \[\[DAPI, GFP\]\] channels, 9 tiles\. "
+ "Image resolution: 720x468")
def executeInDevMode(self):
openbisController = self.createOpenbisController()
self.assertEquals('spot sizes', [['24', '16']], self.getSpotSizes(openbisController))
# openbisController.allDown()
# openbisController.allUp()
openbisController.drop('data-incoming-analysis.zip', 'incoming-analysis')
openbisController.assertNumberOfDataSets(3)
self.assertSpotSizes(openbisController, [['24', '16'], ['24', '16']])
self.assertFeatureVectorLabel(openbisController, 'HITRATE', 'Hit Rate')
self.assertFeatureVectorLabel(openbisController, 'CELLNUMBER', 'cellNumber')
def getSpotSizes(self, openbisController):
return openbisController.queryDatabase('imaging',
def assertSpotSizes(self, openbisController, expected):
actual = openbisController.queryDatabase('imaging',
'select spots_width,spots_height from containers order by spots_width')
self.assertEquals('spot sizes', expected, actual)
def assertFeatureVectorLabel(self, openbisController, featureCode, expectedFeatureLabel):
data = openbisController.queryDatabase('imaging',
"select distinct label from feature_defs where code = '%s'" % featureCode);
self.assertEquals("label of feature %s" % featureCode, [[expectedFeatureLabel]], data)
TestCase(settings, __file__).runTest()
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment