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.