view PerfsDb/TestConfig.py @ 160:6995d5d12d88

filtering tests
author am@osimis.io
date Fri, 17 Aug 2018 16:17:57 +0200
parents df1f9946571c
children 27b3b0df5f90
line wrap: on
line source

import typing
import subprocess
import os
import shutil
from orthancRestApi import OrthancClient

from DbSize import DbSize
from DbType import DbType
from ConfigFileBuilder import ConfigFileBuilder
from DbServer import DbServer
from DbPopulator import DbPopulator
from Tests import *
from TestResult import TestResult

class TestConfig:

    def __init__(self,
        label: str,
        dbSize: DbSize,
        dbType: DbType=None,
        dbServer: DbServer=None
        ):

        self._dbSize = dbSize
        self._dbServer = dbServer
        self.label = label
        self._port = None
        self._orthancProcess = None
        self._repeatCount = 10

        if dbServer is not None:
            self._dbType = dbServer.dbType
            self._dbServer.setLabel(self.label)
            self._port = dbServer.port
        else:
            self._dbType = dbType
        
    def setRepeatCount(self, repeatCount: int):
        self._repeatCount = repeatCount

    def launchDbServer(self):
        if self._dbServer is not None:
            self._dbServer.launch()

    def launchOrthanc(self, orthancPath) -> bool:
        orthanc = OrthancClient("http://127.0.0.1:8042")
        
        print("Checking if Orthanc is already running")
        if orthanc.isAlive():
            print("Orthanc is already running")
            return False
        
        print("Launching Orthanc")
        self._orthancProcess = subprocess.Popen([
            os.path.join(orthancPath, "Orthanc"), 
            os.path.join(os.path.abspath(os.path.dirname(__file__)), "ConfigFiles", self.label + ".json"), 
        ])
       
        print("Waiting for Orthanc to start")
        orthanc.waitStarted(timeout=30)
        print("Orthanc has started")
        return True

    def stopOrthanc(self):
        if self._orthancProcess is not None:
            self._orthancProcess.terminate()
            self._orthancProcess.wait()

    def initializeDb(self):
        dbPopulator = DbPopulator(orthanc=OrthancClient("http://127.0.0.1:8042"), dbSize=self._dbSize)
        dbPopulator.populate()

    def runTests(self, tests: typing.List[Test]) -> typing.List[TestResult]:

        results = []
        for test in tests:
            test.setOrthancClient(OrthancClient("http://127.0.0.1:8042"))
            test.setRepeatCount(self._repeatCount)
            result = test.run()
            print(str(result))

            results.append(result)
        return results

    def clearDb(self):
        if self._dbServer is not None:
            self._dbServer.clear()
        
        # clear storage (in case of Sqlite DB, it will also clear the DB)
        shutil.rmtree(os.path.join(os.path.abspath(os.path.dirname(__file__)), "Storages/{name}".format(name=self.label)), ignore_errors=True)

    def generateOrthancConfigurationFile(self, pluginsPath: str):
        
        ConfigFileBuilder.generate(
            outputPath=os.path.join(os.path.abspath(os.path.dirname(__file__)), "ConfigFiles/{name}.json".format(name=self.label)), 
            pluginsPath=pluginsPath,
            storagePath=os.path.join(os.path.abspath(os.path.dirname(__file__)), "Storages/{name}".format(name=self.label)),
            dbType=self._dbType,
            dbSize=self._dbSize,
            port=self._port
            )