Copied from test_kvm.py

software/end-to-end-testing/test_beremiz.py

+import configparser
+import json
+import logging
+import time
+import unittest
+
+import slapos.client
+
+
+class EndToEndTestCase(unittest.TestCase):
+  @classmethod
+  def setUpClass(cls):
+    configp = configparser.ConfigParser()
+    configp.read('${slapos-client.cfg:output}')
+    args = type("empty_args", (), {})()
+    conf = slapos.client.ClientConfig(args, configp)
+    local = slapos.client.init(conf, logging.getLogger(__name__))
+    cls.slap = local['slap']
+    cls.supply = staticmethod(local['supply'])
+    cls._request = staticmethod(local['request'])
+    cls.product = staticmethod(local['product'])
+    cls._requested = {}
+
+  @classmethod
+  def tearDownClass(cls):
+    for args, kw in cls._requested.values():
+      kw['state']  = 'destroyed'
+      cls._request(*args, **kw)
+
+  @classmethod
+  def request(cls, *args, **kw):
+    instance_name = args[1]
+    cls._requested[instance_name] = (args, kw)
+    partition = cls._request(*args, **kw)
+    return cls.unwrapConnectionDict(partition.getConnectionParameterDict())
+
+  @staticmethod
+  def unwrapConnectionDict(connection_dict):
+    try:
+      connection_json = connection_dict['_']
+    except KeyError:
+      return connection_dict
+    return json.loads(connection_json)
+
+  @classmethod
+  def getInstanceInfos(cls, instance_name):
+    # adapated from cli/info
+    infos = cls.slap.registerOpenOrder().getInformation(instance_name)
+    class Infos:
+      def __init__(self, **kw):
+        self.__dict__.update(kw)
+    connection_dict = {
+      e['connection_key'] : e['connection_value']
+      for e in infos._connection_dict
+    }
+    return Infos(
+      software_url = infos._software_release_url,
+      software_type = infos._source_reference,
+      shared = bool(infos._root_slave),
+      requested_state = infos._requested_state,
+      parameter_dict = infos._parameter_dict,
+      connection_dict = cls.unwrapConnectionDict(connection_dict),
+      news = infos._news,
+    )
+
+  @classmethod
+  def getInstanceNews(cls, instance_name):
+    try:
+      news = cls.slap.registerOpenOrder().getInformation(instance_name)._news
+    except Exception:
+      return ()
+    return news['instance']
+
+  @classmethod
+  def getInstanceStatus(cls, instance_name):
+    # adapated from cli/info
+    status = 0b00
+    for e in cls.getInstanceNews(instance_name):
+      text = e.get('text', '')
+      if text.startswith('#access'):
+        status |= 0b01
+      elif text.startswith('#error'):
+        status |= 0b10
+      if status == 0b11:
+        break
+    return ('none', 'green', 'red', 'orange')[status]
+
+  @classmethod
+  def waitUntilGreen(cls, instance_name, timeout=80):
+    t0 = time.time()
+    while (status := cls.getInstanceStatus(instance_name)) != 'green':
+      msg = 'Instance %s status is still %s' % (instance_name, status)
+      print(msg)
+      if (time.time() - t0) > 60 * timeout:
+        raise TimeoutError(msg)
+      time.sleep(60)
+
+
+class KvmTest(EndToEndTestCase):
+  def test(self):
+    instance_name = time.strftime('e2e-test-kvm-%Y-%B-%d-%H:%M:%S')
+    # instance_name = 'e2e-kvm-test' # avoid timestamp to reuse instance
+    self.request(self.product.kvm, instance_name)
+    self.waitUntilGreen(instance_name)
+    connection_dict = self.request(self.product.kvm, instance_name)
+    self.assertIn('url', connection_dict)