.

slapos/test/promise/plugin/test_check_cpri_lock.py

@@ -42,45 +42,78 @@ class TestCheckCpriLock(TestPromisePluginMixin):
   def setUp(self):
     super(TestCheckCpriLock, self).setUp()
 
+    rf_info = \
+"""
+TRX SDR driver 2023-09-07, API v15/18
+PCIe CPRI /dev/sdr2@1:
+  Hardware ID: 0x4b12
+  DNA: [0x0048248a334a7054]
+  Serial: ''
+  FPGA revision: 2023-06-23  10:05:24
+  FPGA vccint: 0.98 V
+  FPGA vccaux: 1.76 V
+  FPGA vccbram: 0.98 V
+  FPGA temperature: 71.9 °C
+  Clock tune: 0.0 ppm
+  NUMA: 0
+  CPRI_option: '5' (x8) lock=no
+  DMA0: TX fifo: 66.67us  Usage=16/32768 (0%)
+  DMA0: RX fifo: 66.67us  Usage=16/32768 (0%)
+  DMA0 Underflows: 0
+  DMA0 Overflows: 0
+PCIe CPRI /dev/sdr3@1:
+  Hardware ID: 0x4b12
+  DNA: [0x0048248a334a7054]
+  Serial: ''
+  FPGA revision: 2023-06-23  10:05:24
+  FPGA vccint: 0.98 V
+  FPGA vccaux: 1.77 V
+  FPGA vccbram: 0.98 V
+  FPGA temperature: 71.7 °C
+  Clock tune: 0.0 ppm
+  NUMA: 0
+  CPRI_option: '5' (x8) lock=HW+SW rx/tx=46.606us
+    Port #0: T14=46.606us
+  DMA0: TX fifo: 66.67us  Usage=16/32768 (0%)
+  DMA0: RX fifo: 66.67us  Usage=16/32768 (0%)
+  DMA0 Underflows: 0
+  DMA0 Overflows: 0
+"""
+    data = {'message': 'rf', 'rf_info': rf_info}
+
+    self.amarisoft_rf_info_log = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'amarisoft_rf_info.json.log')
+    with open(self.amarisoft_rf_info_log, 'w') as f:
+      f.write(
+      """
+      {"time": "%s", "log_level": "INFO", "message": "RF info", "data": %s}
+      """ % ((datetime.now() - timedelta(seconds=5)).strftime("%Y-%m-%d %H:%M:%S")[:-3], data))
+
 
   def writePromise(self, **kw):
+    kw.update({'amarisoft-rf-info-log': self.amarisoft_rf_info_log})
     super(TestCheckCpriLock, self).writePromise(self.promise_name,
       "from %s import %s\nextra_config_dict = %r\n"
       % (RunPromise.__module__, RunPromise.__name__, kw))
 
-  def test_promise_success(self):
-    self.amarisoft_rf_info_log = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'amarisoft_rf_info.json.log')
-    if os.path.exists(self.amarisoft_rf_info_log):
-      os.remove(self.amarisoft_rf_info_log)
-    with open(self.amarisoft_rf_info_log, 'w+') as f:
-      f.write(
-      """
-      {"time": "%s", "log_level": "INFO", "message": "RF info", "data": {'rf_info': "CPRI: x16 HW SW"}}
-      """ % ((datetime.now() - timedelta(seconds=5)).strftime("%Y-%m-%d %H:%M:%S")[:-3],))
-      self.writePromise(**{
-        'amarisoft-rf-info-log': self.amarisoft_rf_info_log,
-#        'stats-period': 100,
-    })
+  def test_locked_ok(self):
+    self.writePromise(sdr_dev='3', sfp_port='1')
     self.configureLauncher()
     self.launcher.run()
 
+  def test_no_lock(self):
+    self.writePromise(sdr_dev='2', sfp_port='1')
+    self.configureLauncher()
+    with self.assertRaises(PromiseError):
+      self.launcher.run()
+    # XXX check output
 
-  def test_promise_fail(self):
-    self.amarisoft_rf_info_log = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'amarisoft_rf_info.json.log')
-    if os.path.exists(self.amarisoft_rf_info_log):
-      os.remove(self.amarisoft_rf_info_log)
-    with open(self.amarisoft_rf_info_log, 'w') as f:
-      f.write(
-      """
-      {"time": "%s", "log_level": "INFO", "message": "RF info", "data": {'rf_info': "CPRI: x16\\n"}}
-      """ % ((datetime.now() - timedelta(seconds=5)).strftime("%Y-%m-%d %H:%M:%S")[:-3],))
-    self.writePromise(**{
-        'amarisoft-rf-info-log': self.amarisoft_rf_info_log,
-#        'stats-period': 100,
-    })
+  def test_no_cpri(self):
+    self.writePromise(sdr_dev='1', sfp_port='0')
     self.configureLauncher()
     with self.assertRaises(PromiseError):
       self.launcher.run()
+    # XXX check output
+
 
 if __name__ == '__main__':
   unittest.main()