first draft of the Xrn class, designed to replace sfa.util.namespace

diff --git a/sfa/util/namespace.py b/sfa/util/namespace.py
index 7e3c03c..961b3b3 100644 (file)
--- a/sfa/util/namespace.py
+++ b/sfa/util/namespace.py
@@ -1,3 +1,21 @@
+#
+#  WARNING - This file should be soon deprecated in favor of sfa.util.xrn
+#
+#
+#  WARNING - This file should be soon deprecated in favor of sfa.util.xrn
+#
+#
+#  WARNING - This file should be soon deprecated in favor of sfa.util.xrn
+#
+#
+#  WARNING - This file should be soon deprecated in favor of sfa.util.xrn
+#
+#
+#  WARNING - This file should be soon deprecated in favor of sfa.util.xrn
+#
+#
+#  WARNING - This file should be soon deprecated in favor of sfa.util.xrn
+#

 import re
 from sfa.util.faults import *
 URN_PREFIX = "urn:publicid:IDN"
 import re
 from sfa.util.faults import *
 URN_PREFIX = "urn:publicid:IDN"

diff --git a/sfa/util/xrn.py b/sfa/util/xrn.py
new file mode 100644 (file)
index 0000000..2230668
--- /dev/null
+++ b/sfa/util/xrn.py
@@ -0,0 +1,141 @@
+import re
+
+from sfa.util.faults import *
+from sfa.util.sfalogging import sfa_logger
+
+class Xrn:
+
+    ########## basic tools on HRNs
+    # split a HRN-like string into pieces
+    # this is like split('.') except for escaped (backslashed) dots
+    # e.g. hrn_split ('a\.b.c.d') -> [ 'a\.b','c','d']
+    @staticmethod
+    def hrn_split(hrn):
+        return [ x.replace('--sep--','\\.') for x in hrn.replace('\\.','--sep--').split('.') ]
+
+    # e.g. hrn_leaf ('a\.b.c.d') -> 'd'
+    @staticmethod
+    def hrn_leaf(hrn): return Xrn.hrn_split(hrn)[-1]
+
+    # e.g. hrn_path_list ('a\.b.c.d') -> ['a\.b', 'c']
+    @staticmethod
+    def hrn_path_list(hrn): return Xrn.hrn_split(hrn)[0:-1]
+    
+    # e.g. hrn_path ('a\.b.c.d') -> 'a\.b.c'
+    @staticmethod
+    def hrn_path(hrn): return '.'.join(Xrn.hrn_path_list(hrn))
+    
+    # e.g. escape ('a.b') -> 'a\.b'
+    @staticmethod
+    def escape(token): return re.sub(r'([^\\])\.', r'\1\.', token)
+    # e.g. unescape ('a\.b') -> 'a.b'
+    @staticmethod
+    def unescape(token): return token.replace('\\.','.')
+        
+    URN_PREFIX = "urn:publicid:IDN"
+
+    ########## basic tools on URNs
+    @staticmethod
+    def urn_full (urn):
+        if urn.startswith(Xrn.URN_PREFIX): return urn
+        else: return Xrn.URN_PREFIX+URN
+    @staticmethod
+    def urn_meaningful (urn):
+        if urn.startswith(Xrn.URN_PREFIX): return urn[len(Xrn.URN_PREFIX):]
+        else: return urn
+    @staticmethod
+    def urn_split (urn):
+        return Xrn.urn_meaningful(urn).split('+')
+
+    # provide either urn, or (hrn + type)
+    def __init__ (self, urn=None, hrn=None, type=None):
+        if urn: 
+            self.urn=urn
+            self.urn_to_hrn()
+        elif hrn and type: 
+            self.hrn=hrn
+            self.type=type
+            self.hrn_to_urn()
+        else:
+            raise SfaAPIError,"Xrn"
+
+    def get_urn(self): return self.urn
+    def get_hrn(self): return (self.hrn, self.type)
+
+    def get_leaf(self):
+        if not self.hrn: raise SfaAPIError, "Xrn"
+        if not hasattr(self,'leaf'): 
+            self.leaf=Xrn.hrn_split(self.hrn)[-1]
+        return self.leaf
+
+    def get_authority_hrn(self): 
+        if not self.hrn: raise SfaAPIError, "Xrn"
+        # self.authority keeps a list
+        if not hasattr(self,'authority'): 
+            self.authority=Xrn.hrn_path_list(self.hrn)
+        return '.'.join( self.authority )
+    
+    def get_authority_urn(self): 
+        if not self.hrn: raise SfaAPIError, "Xrn"
+        # self.authority keeps a list
+        if not hasattr(self,'authority'): 
+            self.authority=Xrn.hrn_path_list(self.hrn)
+        return ':'.join( [Xrn.unescape(x) for x in self.authority] )
+    
+    def urn_to_hrn(self):
+        """
+        compute tuple (hrn, type) from urn
+        """
+        
+        if not self.urn or not self.urn.startswith(Xrn.URN_PREFIX):
+            raise SfaAPIError, "Xrn"
+
+        parts = Xrn.urn_split(self.urn)
+        type=parts.pop(2)
+        # Remove the authority name (e.g. '.sa')
+        if type == 'authority': parts.pop()
+
+        # convert parts (list) into hrn (str) by doing the following
+        # 1. remove blank parts
+        # 2. escape dots inside parts
+        # 3. replace ':' with '.' inside parts
+        # 3. join parts using '.' 
+        hrn = '.'.join([Xrn.escape(part).replace(':','.') for part in parts if part]) 
+
+        self.hrn=str(hrn)
+        self.type=str(type)
+    
+    def hrn_to_urn(self):
+        """
+        compute urn from (hrn, type)
+        """
+
+        if not self.hrn or self.hrn.startswith(Xrn.URN_PREFIX):
+            raise SfaAPIError, "Xrn"
+
+        if self.type == 'authority':
+            self.authority = Xrn.hrn_split(self.hrn)
+            name = 'sa'   
+        else:
+            self.authority = Xrn.hrn_path_list(self.hrn)
+            name = Xrn.hrn_leaf(self.hrn)
+
+        authority_string = self.get_authority_urn()
+
+        if self.type == None:
+            urn = "+".join(['',authority_string,name])
+        else:
+            urn = "+".join(['',authority_string,self.type,name])
+        
+        self.urn = Xrn.URN_PREFIX + urn
+
+    def dump_string(self):
+        result="-------------------- XRN\n"
+        result += "URN=%s\n"%self.urn
+        result += "HRN=%s\n"%self.hrn
+        result += "TYPE=%s\n"%self.type
+        result += "LEAF=%s\n"%self.get_leaf()
+        result += "AUTH(hrn format)=%s\n"%self.get_authority_hrn()
+        result += "AUTH(urn format)=%s\n"%self.get_authority_urn()
+        return result
+        

diff --git a/tests/testAll.py b/tests/testAll.py
index ac3d47a..540a53c 100755 (executable)
--- a/tests/testAll.py
+++ b/tests/testAll.py
@@ -1,5 +1,6 @@
 #!/usr/bin/python
 #!/usr/bin/python

-from testNamespace import *
+from testXrn import *
+#from testNamespace import *

 # xxx broken-test
 #from testRights import *
 # xxx broken-test
 # xxx broken-test
 #from testRights import *
 # xxx broken-test

diff --git a/tests/testXrn.py b/tests/testXrn.py
new file mode 100755 (executable)
index 0000000..82707ac
--- /dev/null
+++ b/tests/testXrn.py
@@ -0,0 +1,65 @@
+#!/usr/bin/python
+import sys
+import unittest
+
+from sfa.util.faults import *
+from sfa.util.xrn import Xrn
+
+verbose=False
+
+class TestXrn(unittest.TestCase):
+
+    hrns=[ # hrn, type, expected_urn
+        ('ple.inria.baris','user', "urn:publicid:IDN+ple:inria+user+baris"),
+        ('emulab\.net.slice.jktest','slice', "urn:publicid:IDN+emulab.net:slice+slice+jktest"),
+        ('plc.princeton.tmack','user', "urn:publicid:IDN+plc:princeton+user+tmack"),
+        ('fake-pi1@onelab.eu','user', "urn:publicid:IDN+fake-pi1@onelab+user+eu"),
+        # not providing a type is currently not supported
+        ('fake-pi1@onelab.eu',None, None),
+        ]
+    
+    urns=[ # urn, expected_hrn, expected_type
+        ('urn:publicid:IDN+emulab:net+slice+jktest', "emulab.net.jktest", "slice"),
+        ('urn:publicid:IDN+emulab.net+slice+jktest', "emulab\\.net.jktest", "slice"),
+        ("urn:publicid:IDN+plc:princeton+user+tmack", "plc.princeton.tmack", "user"),
+        ]
+
+    def test_hrns(self):
+        for (h,t,exp_urn) in TestXrn.hrns:
+            print 'testing (',h,t,') expecting',exp_urn
+            if exp_urn:
+                xrn=Xrn(hrn=h,type=t)
+                if verbose: print xrn.dump_string()
+                urn=xrn.get_urn()
+                (h1,t1) = Xrn(urn=urn).get_hrn()
+                if h1!=h or t1!=t or urn!=exp_urn:
+                    print "hrn->urn->hrn : MISMATCH with in=(%s,%s) -- out=(%s,%s) -- urn=%s"%(h,t,h1,t1,urn)
+                self.assertEqual(h1,h)
+                self.assertEqual(t1,t)
+                self.assertEqual(urn,exp_urn)
+            else:
+                # could not figure how to use assertFails on object construction..
+                # with self.assertRaises(SfaAPIError):
+                #    Xrn(hrn=h,type=t).get_urn()
+                try:
+                    Xrn(hrn=h,type=t).get_urn()
+                    failure="Unexpectedly created Xrn object"
+                except SfaAPIError:
+                    failure=False
+                except Exception,e:
+                    failure="Xrn creation raised unexpected exception %r"%e
+                if failure: 
+                    print "hrn->urn->hrn - %s with HRN=%s TYPE=%s"%(failure,h,t)
+                    self.assertFalse(True)
+                    
+
+    def test_urns(self):
+        for (urn, exp_hrn, exp_type) in TestXrn.urns:
+            xrn=Xrn(urn=urn)
+            print 'testing urn',urn,'expecting (',exp_hrn,exp_type,')'
+            if verbose: print xrn.dump_string()
+            (h,t)=xrn.get_hrn()
+            urn1 = Xrn(hrn=h,type=t).get_urn()
+            if urn1!=urn:
+                print "urn->hrn->urn : MISMATCH with in=(%s) -- out=(%s) -- hrn=(%s,%s)"%(urn,urn1,h,t)
+            self.assertEqual(urn1,urn)