Add a testflinger_common module with example usage

agent/charms/testflinger-agent-charm/src/charm.py

@@ -142,6 +142,7 @@ def update_repos(self):
  repo_path = f"{self._stored.agent_path}/testflinger"
  repo = Repo.clone_from(
  url=self._stored.testflinger_repo,
+ branch=self._stored.testflinger_branch,
  to_path=repo_path,
  no_checkout=True,
  depth=1,
@@ -151,6 +152,7 @@ def update_repos(self):
  f"origin/{self._stored.testflinger_branch}",
  "--",
  "agent",
+ "common",
  "device-connectors",
  )
  # Install the agent and device-connectors

agent/pyproject.toml

@@ -11,6 +11,7 @@ influxdb = "^5.3.2"
 pyyaml = "^6.0.1"
 requests = "^2.31.0"
 voluptuous = "^0.14.2"
+testflinger-common = { path = "../common" }
 
 [tool.poetry.dev-dependencies]
 pytest = "^8.1.2"

agent/testflinger_agent/agent.py

@@ -22,6 +22,7 @@
 from testflinger_agent.job import TestflingerJob
 from testflinger_agent.errors import TFServerError
 from testflinger_agent.config import ATTACHMENTS_DIR
+from testflinger_common.enums import JobState, TestPhase
 
 try:
  # attempt importing a tarfile filter, to check if filtering is supported
@@ -168,7 +169,11 @@ def unpack_attachments(self, job_data: dict, cwd: Path):
  # (so there is no interference with existing processes, especially
  # provisioning or firmware update, which are triggered when these
  # sections are not empty)
- for phase in ("provision", "firmware_update", "test"):
+ for phase in (
+ TestPhase.PROVISION,
+ TestPhase.FIRMWARE_UPDATE,
+ TestPhase.TEST,
+ ):
  phase_str = f"{phase}_data"
  try:
  phase_data = job_data[phase_str]
@@ -185,12 +190,12 @@ def unpack_attachments(self, job_data: dict, cwd: Path):
  def process_jobs(self):
  """Coordinate checking for new jobs and handling them if they exists"""
  TEST_PHASES = [
- "setup",
- "provision",
- "firmware_update",
- "test",
- "allocate",
- "reserve",
+ TestPhase.SETUP,
+ TestPhase.PROVISION,
+ TestPhase.FIRMWARE_UPDATE,
+ TestPhase.TEST,
+ TestPhase.ALLOCATE,
+ TestPhase.RESERVE,
  ]
 
  # First, see if we have any old results that we couldn't send last time
@@ -233,7 +238,10 @@ def process_jobs(self):
 
  for phase in TEST_PHASES:
  # First make sure the job hasn't been cancelled
- if self.client.check_job_state(job.job_id) == "cancelled":
+ if (
+ self.client.check_job_state(job.job_id)
+ == JobState.CANCELLED
+ ):
  logger.info("Job cancellation was requested, exiting.")
  break
  self.client.post_job_state(job.job_id, phase)
@@ -253,7 +261,7 @@ def process_jobs(self):
  logger.exception(e)
  finally:
  # Always run the cleanup, even if the job was cancelled
- job.run_test_phase("cleanup", rundir)
+ job.run_test_phase(TestPhase.CLEANUP, rundir)
  # clear job id
  self.client.post_agent_data({"job_id": ""})
 
@@ -266,7 +274,7 @@ def process_jobs(self):
  logger.exception(e)
  results_basedir = self.client.config.get("results_basedir")
  shutil.move(rundir, results_basedir)
- self.set_agent_state("waiting")
+ self.set_agent_state(JobState.WAITING)
 
  self.check_restart()
  if self.check_offline():

common/README.rst

@@ -0,0 +1,8 @@
+Testflinger Common
+==================
+
+The testflinger_common module may be used to store code that is useful to
+multiple projects so that it only needs to be defined once.
+
+There is no standalone script for testflinger_common as it is only intended
+to be imported and used by other projects.

common/pyproject.toml

@@ -0,0 +1,23 @@
+[tool.setuptools]
+packages = ["testflinger_common"]
+
+[tool.poetry]
+name = "testflinger-common"
+description = "Testflinger common modules"
+readme = "README.rst"
+version = "1.1.0"
+authors = []
+
+[tool.poetry.dependencies]
+python = "^3.8" # specify your Python version requirement here
+strenum = "^0.4.15"
+
+[tool.poetry.dev-dependencies]
+pytest = "^8.1.2"
+
+[build-system]
+requires = ["poetry-core>=1.0.0"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.black]
+line-length = 79

common/testflinger_common/enums.py

@@ -0,0 +1,45 @@
+# Copyright (C) 2024 Canonical
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+"""
+Job State and Test Phase Enums
+"""
+
+from strenum import StrEnum
+
+
+class JobState(StrEnum):
+ WAITING = "waiting"
+ SETUP = "setup"
+ PROVISION = "provision"
+ FIRMWARE_UPDATE = "firmware_update"
+ TEST = "test"
+ ALLOCATE = "allocate"
+ ALLOCATED = "allocated"
+ RESERVE = "reserve"
+ CLEANUP = "cleanup"
+ CANCELLED = "cancelled"
+ COMPLETED = "completed"
+
+
+class TestPhase(StrEnum):
+ SETUP = "setup"
+ PROVISION = "provision"
+ FIRMWARE_UPDATE = "firmware_update"
+ TEST = "test"
+ ALLOCATE = "allocate"
+ RESERVE = "reserve"
+ CLEANUP = "cleanup"
+