Merge branch 'master' into release/5.3

pvapy/cli/adSimServer.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python
+
 '''
 Area Detector Simulation Server
 '''
@@ -178,7 +179,7 @@ def generateFrames(self):
             if self.maximum is not None:
                 mx = float(min(dtinfo.max, self.maximum))
             self.frames = np.random.uniform(mn, mx, size=(self.nf, self.ny, self.nx))
-            if datatype == 'float32':
+            if self.datatype == 'float32':
                 self.frames = np.float32(self.frames)
 
         print(f'Generated frame shape: {self.frames[0].shape}')

pvapy/cli/mirrorServer.py

@@ -1,8 +1,12 @@
 #!/usr/bin/env python
 
+'''
+PVA Mirror Server.
+'''
+
+import sys
 import argparse
 import time
-import numpy as np
 import pvaccess as pva
 
 __version__ = pva.__version__
@@ -17,11 +21,11 @@ def main():
     args, unparsed = parser.parse_known_args()
     if len(unparsed) > 0:
         print(f'Unrecognized argument(s): {" ".join(unparsed)}')
-        exit(1)
+        sys.exit(1)
 
     if args.channel_map is None:
         print('Channel map is not specified.')
-        exit(1)
+        sys.exit(1)
 
     # Cleanup channel map entries
     # They are expected to be of the form
@@ -39,8 +43,8 @@ def main():
         channelList.append(me[0])
         entryLength = len(me)
         if entryLength > 6:
-           print(f'Invalid channel map entry: {e}')
-           exit(1)
+            print(f'Invalid channel map entry: {e}')
+            sys.exit(1)
         if entryLength <= 2:
             # Add default source provider type
             me.append('pva')
@@ -56,17 +60,17 @@ def main():
         me[2] = providerTypeMap.get(me[2].lower())
         if me[2] is None:
             print(f'Invalid source provider type specified for entry: {e}')
-            exit(1)
+            sys.exit(1)
         me[3] = int(me[3])
         if me[3] < 0:
             print(f'Invalid source queue size specified for entry: {e}')
-            exit(1)
+            sys.exit(1)
         me[4] = int(me[4])
         if me[4] < 1:
             print(f'Invalid number of source monitors specified for entry: {e}')
-            exit(1)
+            sys.exit(1)
         mapEntries.append(me)
-        
+
     server = pva.PvaMirrorServer()
     server.start()
     startTime = time.time()
@@ -91,14 +95,14 @@ def main():
                 for c in channelList:
                     print(f'Channel {c} @ {now:.3f}: {server.getMirrorRecordCounters(c)}')
             time.sleep(sleepTime)
-        except KeyboardInterrupt as e:
+        except KeyboardInterrupt:
             print('Keyboard interrupt received, exiting.')
             break
     server.stop()
     print('\nFinal Channel Statistics:')
     now = time.time()
     for c in channelList:
         print(f'Channel {c} @ {now:.3f}: {server.getMirrorRecordCounters(c)}')
-    
+
 if __name__ == '__main__':
     main()

test/testAdSimServerCli.py

@@ -0,0 +1,15 @@
+'''
+Test AD Sim Server CLI.
+'''
+from unittest.mock import Mock
+import sys
+import pylint.lint
+
+from pvapy.cli import adSimServer
+
+def testLint(monkeypatch):
+    ''' Test for linting errors '''
+    monkeypatch.setattr(sys, 'exit', Mock())
+    pylint_opts = ['pvapy.cli.adSimServer', '--disable=all', '--enable=E,F', '--generated-members="pva.*"']
+    pylint.lint.Run(pylint_opts)
+    sys.exit.assert_called_once_with(0)

test/testHpcCollectorCli.py

@@ -1,9 +1,7 @@
 '''
-   Test HPC Collector CLI.
+Test HPC Collector CLI.
 '''
 from unittest.mock import Mock
-import tempfile
-import os
 import sys
 import pylint.lint
 

test/testHpcConsumerCli.py

@@ -1,9 +1,7 @@
 '''
-   Test HPC Consumer CLI.
+Test HPC Consumer CLI.
 '''
 from unittest.mock import Mock
-import tempfile
-import os
 import sys
 import pylint.lint
 

test/testMirrorServerCli.py

@@ -0,0 +1,15 @@
+'''
+Test Mirror Server CLI.
+'''
+from unittest.mock import Mock
+import sys
+import pylint.lint
+
+from pvapy.cli import mirrorServer
+
+def testLint(monkeypatch):
+    ''' Test for linting errors '''
+    monkeypatch.setattr(sys, 'exit', Mock())
+    pylint_opts = ['pvapy.cli.mirrorServer', '--disable=all', '--enable=E,F', '--generated-members="pva.*"']
+    pylint.lint.Run(pylint_opts)
+    sys.exit.assert_called_once_with(0)