# Copyright (c) 2019-2023 Alexander Todorov <atodorov@MrSenko.com>
import os
from datetime import datetime
# change this to import cache in Python 3.9
from functools import lru_cache as cache
from . import TCMS
from .version import __version__
# pylint: disable=too-many-instance-attributes, too-many-public-methods
[docs]
class Backend:
"""
Facilitates RPC communications with the backend and implements
behavior described at:
http://kiwitcms.org/blog/atodorov/2018/11/05/test-runner-plugin-specification/
This class is intended to be used by Kiwi TCMS plugins implemented in
Python. The plugin will call::
backend = Backend()
backend.configure()
... parse test results ...
test_case_id, _ = backend.test_case_get_or_create(<description>)
backend.add_test_case_to_plan(test_case_id, backend.plan_id)
test_executions = backend.add_test_case_to_run(test_case_id,
backend.run_id)
for execution in test_executions:
backend.update_test_execution(execution['id'],
<status_id>,
<comment>)
:param prefix: Prefix which will be added to TestPlan.name and
TestRun.summary
.. versionadded:: 5.2
:type prefix: str
"""
_statuses = {}
name = "tcms_api.plugin_helpers.Backend"
version = __version__
@property
def prefix(self):
"""
Prefix may be overriden via ``TCMS_PREFIX`` environment variable.
.. versionadded:: 11.2
"""
return os.environ.get("TCMS_PREFIX", self._prefix)
def __init__(self, prefix="", verbose=False):
"""
:param prefix: Prefix which will be added to TestPlan.name and
TestRun.summary
.. versionadded:: 5.2
:type prefix: str
:param verbose: If ``True`` will print info about created records.
Defaults to ``False``
.. versionadded:: 11.3
:type verbose: bool
"""
self._prefix = prefix
self.verbose = verbose
self.rpc = None
self.run_id = None
self.plan_id = None
self.product_id = None
self.category_id = None
self.priority_id = None
self.confirmed_id = None
[docs]
def log_info(self, was_created, obj_prefix, obj_id):
if self.verbose:
verb = {
True: "added",
False: "reuse",
}[was_created]
print(f"{verb}: {obj_prefix}-{obj_id}")
[docs]
def get_statuses_by_weight(self, lookup_condition):
"""
Get a list of statuses based on lookup condition.
:param lookup_condition: ``tcms.testruns.models.TestExecutionStatus``
lookup condition
:type lookup_condition: dict
:rtype: list
"""
return self.rpc.TestExecutionStatus.filter(lookup_condition)
[docs]
def get_status_id_fallback(self, name):
"""
Get status based on weight if name not found
:param name: ``tcms.testruns.models.TestExecutionStatus`` name
:type name: str
:rtype: int
"""
lookup_condition = None
if name in ["PASSED", "WAIVED"]:
lookup_condition = "weight__gt"
elif name in ["FAILED", "ERROR"]:
lookup_condition = "weight__lt"
if not lookup_condition:
raise RuntimeError(
f"`lookup_condition`is not initialized when searching for {name}. "
"Please report a bug at https://github.com/kiwitcms/tcms-api/issues/new"
)
return self.get_statuses_by_weight({lookup_condition: 0})[0]["id"]
[docs]
def get_status_id(self, name):
"""
Get the PK of ``tcms.testruns.models.TestExecutionStatus``
matching the test execution status name or fallback based on
weight.
.. important::
Test runner plugins **must** call this method like so::
id = backend.get_status_id('FAILED')
:param name: ``tcms.testruns.models.TestExecutionStatus`` name
:type name: str
:rtype: int
"""
if name not in self._statuses:
try:
self._statuses[name] = self.rpc.TestExecutionStatus.filter(
{"name": name}
)[0]["id"]
except IndexError:
self._statuses[name] = self.get_status_id_fallback(name)
return self._statuses[name]
[docs]
def get_product_id(self, plan_id):
"""
Return a ``tcms.management.models.Product`` PK.
.. warning::
For internal use by `.configure()`!
:param plan_id: ``tcms.testplans.models.TestPlan`` PK
:type plan_id: int
:rtype: int
Order of precedence:
- `plan_id` is specified, then use TestPlan.product, otherwise
- use `$TCMS_PRODUCT` as Product.name if specified, otherwise
- use `$TRAVIS_REPO_SLUG` as Product.name if specified, otherwise
- use `$JOB_NAME` as Product.name if specified
If Product doesn't exist in the database it will be created with
the first ``tcms.management.models.Classification`` found!
"""
product_id = None
product_name = None
test_plan = self.rpc.TestPlan.filter({"pk": plan_id})
if test_plan:
product_id = test_plan[0]["product"]
product_name = test_plan[0]["product__name"]
else:
product_name = os.environ.get(
"TCMS_PRODUCT",
os.environ.get("TRAVIS_REPO_SLUG", os.environ.get("JOB_NAME")),
)
if not product_name:
raise RuntimeError(
"Product name not defined, "
"missing one of TCMS_PRODUCT, "
"TRAVIS_REPO_SLUG or JOB_NAME"
)
product = self.rpc.Product.filter({"name": product_name})
if not product:
class_id = self.rpc.Classification.filter({})[0]["id"]
product = [
self.rpc.Product.create(
{"name": product_name, "classification": class_id}
)
]
product_id = product[0]["id"]
return product_id, product_name
[docs]
def get_version_id(self, product_id):
"""
Return a ``tcms.management.models.Version`` (PK, name).
.. warning::
For internal use by `.configure()`!
:param product_id: ``tcms.management.models.Product`` PK
for which to look for Version
:type product_id: int
:return: (version_id, version_value)
:rtype: tuple(int, str)
Order of precedence:
- use `$TCMS_PRODUCT_VERSION` as Version.value if specified, or
- use `$TRAVIS_COMMIT` as Version.value if specified, otherwise
- use `$TRAVIS_PULL_REQUEST_SHA` as Version.value if specified,
otherwise
- use `$GIT_COMMIT` as Version.value if specified
If Version doesn't exist in the database it will be created with
the specified `product_id`!
"""
version_val = os.environ.get(
"TCMS_PRODUCT_VERSION",
os.environ.get(
"TRAVIS_COMMIT",
os.environ.get("TRAVIS_PULL_REQUEST_SHA", os.environ.get("GIT_COMMIT")),
),
)
if not version_val:
raise RuntimeError(
"Version value not defined, "
"missing one of TCMS_PRODUCT_VERSION, "
"TRAVIS_COMMIT, TRAVIS_PULL_REQUEST_SHA "
"or GIT_COMMIT"
)
version = self.rpc.Version.filter({"product": product_id, "value": version_val})
if not version:
version = [
self.rpc.Version.create({"product": product_id, "value": version_val})
]
return version[0]["id"], version_val
[docs]
def get_build_id(self, version_id):
"""
Return a ``tcms.management.models.Build`` (PK, name).
.. warning::
For internal use by `.configure()`!
:param version_id: ``tcms.management.models.Version`` PK
for which to look for Build
:type version_id: int
:return: (build_id, build_name)
:rtype: tuple(int, str)
Order of precedence:
- use `$TCMS_BUILD` as Build.name if specified, otherwise
- use `$TRAVIS_BUILD_NUMBER` as Build.name if specified, otherwise
- use `$BUILD_NUMBER` as Build.name if specified
If Build doesn't exist in the database it will be created with the
specified `version_id`!
"""
build_number = os.environ.get(
"TCMS_BUILD",
os.environ.get("TRAVIS_BUILD_NUMBER", os.environ.get("BUILD_NUMBER")),
)
if not build_number:
raise RuntimeError(
"Build number not defined, "
"missing one of TCMS_BUILD, "
"TRAVIS_BUILD_NUMBER or BUILD_NUMBER"
)
build = self.rpc.Build.filter({"name": build_number, "version": version_id})
if not build:
build = [
self.rpc.Build.create({"name": build_number, "version": version_id})
]
return build[0]["id"], build_number
[docs]
def get_plan_type_id(self):
"""
Return an **Integration** PlanType.
.. warning::
For internal use by `.configure()`!
:return: ``tcms.testplans.models.PlanType`` PK
:rtype: int
"""
plan_type = self.rpc.PlanType.filter({"name": "Integration"})
if not plan_type:
plan_type = [self.rpc.PlanType.create({"name": "Integration"})]
return plan_type[0]["id"]
[docs]
def external_plan_id(self): # pylint: disable=no-self-use
"""
Allows the user to specify `$TCMS_PLAN_ID` to point to an existing
TestPlan where new runs will be added!
.. warning::
Does not check if the specified TP exists!
:return: ``tcms.testplans.models.TestPlan`` PK or 0
:rtype: int
"""
return os.environ.get("TCMS_PLAN_ID", 0)
@property
@cache(maxsize=128)
def default_tester_id(self):
"""
Used internally and by default this is the user sending the API
request. Use `$TCMS_DEFAULT_TESTER_ID` to override!
Plugins may want to override this.
:return: User ID
:rtype: int
"""
return os.environ.get("TCMS_DEFAULT_TESTER_ID")
[docs]
def get_plan_id(self, run_id):
"""
If a TestRun with PK `run_id` exists then return the TestPlan to
which this TestRun is assigned, otherwise create new TestPlan with
Product and Version specified by environment variables.
.. warning::
For internal use by `.configure()`!
If ``TCMS_PARENT_PLAN`` environment variable is specified and a new
TestPlan is created then it will be created as a child TP.
.. versionadded:: 11.2
:param run_id: ``tcms.testruns.models.TestRun`` PK
:type run_id: int
:return: ``tcms.testplans.models.TestPlan`` PK
:rtype: int
"""
plan_id = self.external_plan_id()
if plan_id:
self.log_info(False, "TP", plan_id)
return plan_id
result = self.rpc.TestRun.filter({"pk": run_id})
if not result:
product_id, product_name = self.get_product_id(0)
version_id, version_name = self.get_version_id(product_id)
name = f"{self.prefix} Plan for {product_name} ({version_name})"
name = name[:255]
was_created = False
result = self.rpc.TestPlan.filter(
{"name": name, "product": product_id, "product_version": version_id}
)
if not result:
plan_type_id = self.get_plan_type_id()
args = {
"name": name,
"text": self.created_by_text,
"product": product_id,
"product_version": version_id,
"is_active": True,
"type": plan_type_id,
}
if self.default_tester_id:
args["author"] = self.default_tester_id
parent_plan_id = os.environ.get("TCMS_PARENT_PLAN")
if parent_plan_id:
args["parent"] = parent_plan_id
result = [self.rpc.TestPlan.create(args)]
was_created = True
# newly created TP
self.log_info(was_created, "TP", result[0]["id"])
return result[0]["id"]
# TP to which existing TR is assigned
self.log_info(False, "TP", result[0]["plan"])
return result[0]["plan"]
[docs]
def get_run_id(self):
"""
If `$TCMS_RUN_ID` is specified then assume the caller knows
what they are doing and try to add test results to that TestRun.
Otherwise will create a TestPlan and TestRun in which to record
the results!
.. warning::
For internal use by `.configure()`!
:return: ``tcms.testruns.models.TestRun`` PK
:rtype: int
"""
was_added = False
run_id = os.environ.get("TCMS_RUN_ID")
if not run_id:
product_id, product_name = self.get_product_id(0)
version_id, version_val = self.get_version_id(product_id)
build_id, build_number = self.get_build_id(version_id)
plan_id = self.get_plan_id(0)
# TR.manager is always the author of the TP, which is either
# another existing user (existing TP) or self.default_tester_id
# in case of newly created TP
manager_id = self.rpc.TestPlan.filter({"pk": plan_id})[0]["author"]
args = {
"summary": f"{self.prefix} Results for {product_name}, "
f"{version_val}, {build_number}",
"manager": manager_id,
"plan": plan_id,
"build": build_id,
"start_date": datetime.now().isoformat().replace("T", " ")[:19],
}
if self.default_tester_id:
args["default_tester"] = self.default_tester_id
else:
# b/c TestRun.create() always requires this argument
args["default_tester"] = manager_id
testrun = self.rpc.TestRun.create(args)
was_added = True
run_id = testrun["id"]
self.log_info(was_added, "TR", run_id)
return int(run_id)
[docs]
def finish_test_run(self):
"""
.. important::
Test runner plugins **may** call this method!
May be called at the end when there are no more test executions to
be sent to Kiwi TCMS. Default implementation will update
``TR.stop_date``.
:return: None
"""
self.rpc.TestRun.update(
self.run_id,
{
"stop_date": datetime.now().isoformat().replace("T", " ")[:19],
},
)
[docs]
def test_case_get_or_create(self, summary):
"""
Search for a TestCase with the specified `summary` and Product.
If it doesn't exist in the database it will be created!
.. important::
Test runner plugins **must** call this method!
:param summary: A TestCase summary
:type summary: str
:return: Serialized ``tcms.testcase.models.TestCase`` and boolean
flag to indicate if the TestCase has just been created!
:rtype: (dict, bool)
"""
summary = summary[:255]
created = False
test_case = self.rpc.TestCase.filter(
{
"summary": summary,
"category__product": self.product_id,
}
)
if not test_case:
test_case = [
self.rpc.TestCase.create(
{
"summary": summary,
"category": self.category_id,
"priority": self.priority_id,
"case_status": self.confirmed_id,
"notes": self.created_by_text,
"is_automated": True,
}
)
]
created = True
return test_case[0], created
[docs]
def add_test_case_to_plan(self, case_id, plan_id):
"""
Add a TestCase to a TestPlan if it is not already there!
.. important::
Test runner plugins **must** call this method!
:param case_id: ``tcms.testcases.models.TestCase`` PK
:type case_id: int
:param plan_id: ``tcms.testplans.models.TestPlan`` PK
:type plan_id: int
:return: None
"""
if not self.rpc.TestCase.filter({"pk": case_id, "plan": plan_id}):
self.rpc.TestPlan.add_case(plan_id, case_id)
[docs]
def add_test_case_to_run(self, case_id, run_id):
"""
Add a TestCase to a TestRun if it is not already there!
.. important::
Test runner plugins **must** call this method!
:param case_id: ``tcms.testcases.models.TestCase`` PK
:type case_id: int
:param run_id: ``tcms.testruns.models.TestRun`` PK
:type run_id: int
:return: List of serialized ``tcms.testruns.models.TestExecution``
objects
:rtype: list(dict)
"""
result = self.rpc.TestRun.add_case(run_id, case_id)
if not isinstance(result, list):
result = [result]
return result
# pylint: disable=too-many-arguments
[docs]
def update_test_execution(
self,
test_execution_id,
status_id,
comment=None,
start_date=None,
stop_date=None,
):
"""
Update TestExecution with a status and comment.
.. important::
Test runner plugins **must** call this method!
:param test_execution_id: ``tcms.testruns.models.TestExecution`` PK
:type test_execution_id: int
:param status_id: ``tcms.testruns.models.TestExecutionStatus`` PK,
for example the ID for PASSED, FAILED, etc.
:type status_id: int
:param comment: the string to add as a comment, defaults to None
:type comment: str
:param start_date: when execution began, default None
:type start_date: datetime
:param stop_date: when execution completed, default None
:type stop_date: datetime
:return: None
"""
args = {
"status": status_id,
"start_date": start_date,
"stop_date": stop_date,
}
if self.default_tester_id:
args["tested_by"] = self.default_tester_id
self.rpc.TestExecution.update(test_execution_id, args)
if comment:
self.add_comment(test_execution_id, comment)
@property
def created_by_text(self):
if not self.name:
raise NotImplementedError
if not self.version:
raise NotImplementedError
return f"Created by {self.name}, version {self.version}"