Category: blog

One commonly requested feature is being able to upload data from python. Given that all access is via a REST API, this is remarkably easy to do.

Eventually, we would like to add a proper upload / download library for Conical so that not only can people publish their test results from python, but they can also perform programmatic analysis on the data. That is on our book of work, but isn’t currently available.

In the meantime, we’ve put together the following script to allow uploading of data from your projects.

import requests

import enum
from datetime import datetime

class ConicalException(Exception):
    def __init__(self, message):
        self.message = message

class TestRunStatus(enum.Enum):
    unknown = 1
    exception = 2
    failed = 3
    passed = 4

class Product(object):
    def __init__(self, accessLayer, name, description):
        self.accessLayer = accessLayer
        self.name = name
        self.description = description

    def create_testrunset(self, testRunSetName, testRunSetDescription, testRunSetRefDate, testRunSetRunDate, tags = None):
        headers = self.accessLayer._makeHeaders()
        queryParameters = { "product": self.name, "name":testRunSetName, "description":testRunSetDescription, "refDate":testRunSetRefDate.strftime("%Y-%m-%d"), "runDate":testRunSetRunDate.strftime( "%Y-%m-%dT%H:%M:%S"), "tags":tags}
        response = requests.post( f"{self.accessLayer.url}/api/upload/CreateTestRunSet", headers = headers, params = queryParameters)
        if response:
            responseJson = response.json()
            trsID = responseJson["id"]
            trsName = responseJson["name"]
            trsDescription = responseJson["description"]
            trsRefDate = datetime.strptime( responseJson["refDate"], "%Y-%m-%dT%H:%M:%S")
            trsRunDate = datetime.strptime( responseJson["runDate"], "%Y-%m-%dT%H:%M:%S")

            retValue = TestRunSet(self.accessLayer, self.name, trsID, trsName, trsDescription, trsRefDate, trsRunDate)
            return retValue
        else:
            raise ConicalException("An exception occurred")

class TestRunSet(object):
    def __init__(self, accessLayer, productName, id, testRunSetName, testRunSetDescription, testRunSetRefDate, testRunSetRunDate):
        self.accessLayer = accessLayer
        self.productName = productName
        self.id = id
        self.name = testRunSetName
        self.description = testRunSetDescription
        self.testRunSetRefDate = testRunSetRefDate
        self.testRunSetRunDate = testRunSetRunDate

    def close(self):
        headers = self.accessLayer._makeHeaders()
        queryParameters = { "status": "standard"}
        response = requests.post( f"{self.accessLayer.url}/api/product/{self.productName}/TestRunSet/{self.id}/updateStatus", params = queryParameters, headers=headers)
        if not response:
            raise ConicalException("Unable to close open TRS")

    def create_testrun(self, testRunName, testRunDescription, testRunType, testRunStatus):
        headers = self.accessLayer._makeHeaders()
        queryParameters = { "product": self.productName, "testRunSetID": self.id, "name":testRunName, "description":testRunDescription, "testRunType":testRunType, "testStatus": testRunStatus.name}
        response = requests.post( f"{self.accessLayer.url}/api/upload/CreateTestRun", headers = headers, params = queryParameters)
        if response:
            responseJson = response.json()
            retValue = TestRun( self.accessLayer, self.productName, self.id, responseJson ["id"], testRunName, testRunDescription)
            return retValue
        else:
            raise ConicalException( "Unable to create test run")

class TestRun(object):
    def __init__(self, accessLayer, productName, trsID, id, name, description):
        self.accessLayer = accessLayer
        self.productName = productName
        self.trsID = trsID
        self.id = id
        self.name = name
        self.description = description

    def publish_results_text( self, resultsText):
        headers = self.accessLayer._makeHeaders()
        headers [ "Content-Type"] = "text/plain"
        queryParameters = { "product": self.productName, "testRunSetID":self.trsID, "testRunID":self.id, "resultType": "text"}
        response = requests.post( f"{self.accessLayer.url}/api/upload/publishTestRunResults", headers=headers, params = queryParameters, data = resultsText)
        if not response:
            raise ConicalException( "Unable to publish results text" )

    def publish_results_xml( self, resultsXml):
        headers = self.accessLayer._makeHeaders()
        headers [ "Content-Type"] = "text/plain"
        queryParameters = { "product": self.productName, "testRunSetID":self.trsID, "testRunID":self.id, "resultType": "xml"}
        response = requests.post( f"{self.accessLayer.url}/api/upload/publishTestRunResults", headers=headers, params = queryParameters, data = resultsXml)
        if not response:
            raise ConicalException( "Unable to publish results xml" )

    def publish_results_json( self, resultsJson):
        headers = self.accessLayer._makeHeaders()
        headers [ "Content-Type"] = "text/plain"
        queryParameters = { "product": self.productName, "testRunSetID":self.trsID, "testRunID":self.id, "resultType": "json"}
        response = requests.post( f"{self.accessLayer.url}/api/upload/publishTestRunResults", headers=headers, params = queryParameters, data = resultsJson)
        if not response:
            raise ConicalException( "Unable to publish results json" )

    def publish_results_csv( self, resultsCsv):
        headers = self.accessLayer._makeHeaders()
        headers [ "Content-Type"] = "text/plain"
        queryParameters = { "product": self.productName, "testRunSetID":self.trsID, "testRunID":self.id, "style": "csv"}
        response = requests.post( f"{self.accessLayer.url}/api/upload/publishTestRunXsvResults", headers=headers, params = queryParameters, data = resultsCsv)
        if not response:
            raise ConicalException( "Unable to publish results CSV" )

    def publish_results_tsv( self, resultsTsv):
        headers = self.accessLayer._makeHeaders()
        headers [ "Content-Type"] = "text/plain"
        queryParameters = { "product": self.productName, "testRunSetID":self.trsID, "testRunID":self.id, "style": "tsv"}
        response = requests.post( f"{self.accessLayer.url}/api/upload/publishTestRunXsvResults", headers=headers, params = queryParameters, data = resultsTsv)
        if not response:
            raise ConicalException( "Unable to publish results TSV" )

class ConicalAccessLayer(object):
    def __init__(self, url, accessToken = None):
        self.url = url
        self.accessToken = accessToken

    def _makeHeaders(self):
        headers = {}

        if self.accessToken != None:
            headers = { "Authorization": f"Bearer {self.accessToken}"}

        return headers

    def products(self):
        headers = self._makeHeaders()
        response = requests.get( f"{self.url}/api/products", headers = headers)
        
        productsArray = []
        for productJson in response.json():
            productO = Product(self, productJson["name"], productJson ["description"])
            productsArray.append(productO)

        return productsArray

    def get_product(self, productName):
        headers = self._makeHeaders()
        response = requests.get( f"{self.url}/api/product/{productName}", headers=headers)
        if response:
            productJson = response.json()
            p = Product( self, productJson["name"], productJson["description"])
            return p
        else:
            raise ConicalException(f"Unable to fetch '{productName}'")

Using the script is very simple. To do so, you’ll need to create an access token (unless you’ve configured the anonymous user to have write permissions – which we probably don’t recommend) and then:

token = "replace"
accessLayer = ConicalAccessLayer( "https://demo.conical.cloud", token)

dogfoodproduct = accessLayer.get_product( "dogfood-ui")
refDate = datetime(2022, 12, 23)
runDate = datetime(2022, 12, 23, 18, 23, 39)
trs = dogfoodproduct.create_testrunset( "TRS1", "descri", refDate, runDate)
print( f"Created TRS #{trs.id}")
tr = trs.create_testrun( "sample", "sample desc", "temp", TestRunStatus.passed)
print( f"Created TR #{tr.id}")
print( "Uploading test data" )
tr.publish_results_text( "Booooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooool")
tr.publish_results_xml( "<node><subNode1 /><subNode2>bob</subNode2></node>")
tr.publish_results_json( "{ \"bob\": 2.3, \"bib\": null}")
tr.publish_results_csv( "Col #1,Col #2,Col #3,Col #4,Col #5\n234,234,5,7,2\n234,23,41,5,15")
tr.publish_results_tsv( "Col #1\tCol #2\tCol #3\tCol #4\tCol #5\n238\t231\t5\t7\t4")
trs.close()
print( f"Closed trs")

If you have any comments / suggestions on how to improve python support, then please do get in touch with us, either via email, the contact form or in the comments below.

Note that we’re not python experts, so please be gentle with us!

Happy testing.