Move transform lookup routine along with actual functions.

Add unit test.

sippy/B2BTransforms.py

@@ -23,6 +23,8 @@
 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+import sys
+
 from sippy.SipRequest import SipRequest
 
 class HDR2Xattrs():
@@ -32,7 +34,7 @@ class HDR2Xattrs():
     def __init__(self, hdr_name:str):
         self.hdr_name = hdr_name
 
-    def __call__(req:SipRequest, cc:'CallController'):
+    def __call__(self, cc:'CallController', req:SipRequest):
         hfs = req.getHFs(self.hdr_name)
 
         if len(hfs) == 0:
@@ -52,3 +54,21 @@ def __call__(req:SipRequest, cc:'CallController'):
             cc.extra_attributes = extra_attributes
         else:
             cc.extra_attributes.extend(extra_attributes)
+
+class Nop():
+    def __init__(self, v:str): pass
+    def __call__(self, *a, **kwa): pass
+
+def getTransProc(value:str):
+    rparts = value.split('[', 1)
+    if not len(rparts) == 2 or not value.endswith(']'):
+        raise ValueError(f'getTransProc: `{value}` should be in the format `function[argument]`')
+    fname = rparts[0]
+    bts = sys.modules[__name__]
+    fclass = getattr(bts, fname)
+    farg = rparts[1][:-1]
+    return fclass(farg)
+
+if __name__ == '__main__':
+    for t in ('HDR2Xattrs[X-foo-hdr]',):
+        p = getTransProc(t)

sippy/MyConfigParser.py

@@ -33,7 +33,7 @@
     from configparser import RawConfigParser
     _boolean_states = RawConfigParser.BOOLEAN_STATES
 from sippy.SipConf import SipConf
-import sippy.B2BTransforms as bts
+from sippy.B2BTransforms import getTransProc
 
 SUPPORTED_OPTIONS = { \
  'acct_enable':       ('B', 'enable or disable Radius accounting'), \
@@ -220,13 +220,7 @@ def check_and_set(self, key, value, compat = True):
                 raise ValueError('sip_port should be in the range 1-65535')
             self['_sip_port'] = _value
         elif key == 'pre_auth_proc':
-            rparts = value.split('[', 1)
-            if not len(rparts) == 2 or not value.endswith(']'):
-                raise ValueError('pre_auth_proc should be in the format `function(argument)`')
-            fname = rparts[0]
-            fclass = getattr(bts, fname)
-            farg = rparts[1][:-1]
-            self['_pre_auth_proc'] = fclass(farg)
+            self['_pre_auth_proc'] = getTransProc(value)
         self[key] = value
 
     def options_help(self):

sippy/b2bua_radius.py

@@ -515,7 +515,7 @@ def recvRequest(self, req, sip_t):
             cc = CallController(remote_ip, source, req_source, req_target, self.global_config, pass_headers)
 
             if '_pre_auth_proc' in self.global_config:
-                self.global_config['_pre_auth_proc'](req, cc)
+                self.global_config['_pre_auth_proc'](cc, req)
 
             cc.challenge = challenge
             rval = cc.uaA.recvRequest(req, sip_t)

tests/test_B2BTransforms.py

@@ -0,0 +1,32 @@
+import unittest
+from sippy.B2BTransforms import getTransProc
+
+class FakeCC():
+    acctA = None
+    acctO = None
+class FakeRequest():
+    test: unittest.TestCase
+    def __init__(self, test): self.test = test
+    def getHFs(self, name):
+        got = name
+        want = 'X-foo-hdr'
+        self.test.assertEqual(want, got)
+        return tuple()
+class FakeEvent(): pass
+
+class TestB2BTransforms(unittest.TestCase):
+
+    def test_getTransProc(self):
+        transformations = [
+            ('HDR2Xattrs[X-foo-hdr]', (FakeCC(), FakeRequest(self))),
+            ('Nop[]', (None, None)),
+        ]
+
+        for t, args in transformations:
+            with self.subTest(t=t, args=args):
+                p = getTransProc(t)
+                self.assertIsNotNone(p, f"getTransProc({t}) returned None")
+                self.assertIsNone(p(*args), f"__call__({args}) returned not None")
+
+if __name__ == '__main__':
+    unittest.main()