source: TI12-security/trunk/NDG_XACML/ndg/xacml/test/test_context.py @ 6798

#!/usr/bin/env python
"""NDG XACML Context unit test package 

NERC DataGrid Project
"""
__author__ = "P J Kershaw"
__date__ = "26/03/10"
__copyright__ = "(C) 2010 Science and Technology Facilities Council"
__contact__ = "Philip.Kershaw@stfc.ac.uk"
__license__ = "BSD - see LICENSE file in top-level directory"
__contact__ = "Philip.Kershaw@stfc.ac.uk"
__revision__ = "$Id: $"
import unittest
from os import path
import logging
logging.basicConfig(level=logging.DEBUG)

from ndg.xacml.test import XACML_NDGTEST1_FILEPATH
from ndg.xacml.parsers.etree.factory import ReaderFactory
from ndg.xacml.core.context.pdpinterface import PDPInterface
from ndg.xacml.core.context.pdp import PDP
from ndg.xacml.core.context.handler import AbstractContextHandler
from ndg.xacml.core.attribute import Attribute
from ndg.xacml.core.attributevalue import AttributeValueFactory, AttributeValue
from ndg.xacml.core.context.request import Request
from ndg.xacml.core.context.response import Response
from ndg.xacml.core.context.result import Result, Decision
from ndg.xacml.core.context.subject import Subject
from ndg.xacml.core.context.resource import Resource
from ndg.xacml.core.context.action import Action

                
class TestContextHandler(AbstractContextHandler):
    """Test implementation of Context Handler"""
    
    def __init__(self):
        super(TestContextHandler, self).__init__()
        self.pip = None        
        
    def handlePEPRequest(self, myRequest):
        
        # Convert myRequest to XACML context request
        request = myRequest
        
        if self.pdp is None:
            raise TypeError('No "pdp" attribute set')
        
        response = self.pdp.evaluate(request)
        
        # Convert XACML context response to domain specific request
        myResponse = response
        
        return myResponse


class XACMLContextTestCase(unittest.TestCase):
    """Test PDP, PAP, PIP and Context handler"""
    
    def _createRequestCtx(self):
        request = Request()
        subject = Subject()
        
        attributeValueFactory = AttributeValueFactory()
        
        openidSubjectAttribute = Attribute()
        roleAttribute = Attribute()
        
        openidSubjectAttribute.attributeId = "urn:esg:openid"
        openidSubjectAttribute.dataType = \
                                    'http://www.w3.org/2001/XMLSchema#anyURI'

        StringAttributeValue = attributeValueFactory(
                                    'http://www.w3.org/2001/XMLSchema#anyURI')
        openidSubjectAttribute.attributeValues.append(StringAttributeValue())
    
        openidSubjectAttribute.attributeValues[-1].dataType = \
                                    'http://www.w3.org/2001/XMLSchema#anyURI'
        openidSubjectAttribute.attributeValues[-1].value = \
                                    'https://my.name.somewhere.ac.uk'
        
        subject.attributes.append(openidSubjectAttribute)

        roleAttribute.attributeId = "urn:ndg:security:authz:1.0:attr"
        roleAttribute.dataType = 'http://www.w3.org/2001/XMLSchema#string'
        roleAttribute.attributeValues.append(StringAttributeValue())
        roleAttribute.attributeValues[-1].dataType = \
                                    'http://www.w3.org/2001/XMLSchema#string'
        roleAttribute.attributeValues[-1].value = 'staff' 
        
        subject.attributes.append(roleAttribute)
                                  
        request.subjects.append(subject)
        
        resource = Resource()
        resourceAttribute = Attribute()
        resource.attributes.append(resourceAttribute)
        
        resourceAttribute.attributeId = \
                            "urn:oasis:names:tc:xacml:1.0:resource:resource-id"
                            
        resourceAttribute.dataType = "http://www.w3.org/2001/XMLSchema#anyURI"
        resourceAttribute.attributeValues.append(AttributeValue())
        resourceAttribute.attributeValues[-1].value = \
                                        'http://www.localhost/test_securedURI'

        request.resources.append(resource)
        
        request.action = Action()
        actionAttribute = Attribute()
        request.action.attributes.append(actionAttribute)
        
        actionAttribute.attributeId = \
                                "urn:oasis:names:tc:xacml:1.0:action:action-id"
        actionAttribute.dataType = "http://www.w3.org/2001/XMLSchema#string"
        actionAttribute.attributeValues.append(AttributeValue())
        actionAttribute.attributeValues[-1].value = 'read'
        
        return request
    
    def test01CreateRequest(self):
        requestCtx = self._createRequestCtx()
        self.assert_(requestCtx)
        
    def test02CreateResponse(self):
        response = Response()
        result = Result()
        response.results.append(result)
        result.decision = Decision()
        result.decision.value = Decision.NOT_APPLICABLE
        
    def test03AbstractCtxHandler(self):
        self.assertRaises(TypeError, AbstractContextHandler)
        
    def test04CreateCtxHandler(self):
        ctxHandler = TestContextHandler()
        
    def test04PDPInterface(self):
        self.assertRaises(TypeError, PDPInterface)
        
    def test05CreatePDP(self):
        pdp = PDP()
        self.assert_(pdp)
        
    def _createPDPfromPolicy(self):
        pdp = PDP.fromPolicySource(XACML_NDGTEST1_FILEPATH, ReaderFactory)
        return pdp
        
    def test06CreatePDPfromPolicy(self):
        pdp = self._createPDPfromPolicy()
        self.assert_(pdp)
        
    def test07EvaluatePDP(self):
        request = self._createRequestCtx()
        pdp = self._createPDPfromPolicy()
        response = pdp.evaluate(request)
        self.assert_(response)

        
if __name__ == "__main__":
    unittest.main()