<commit_before>#!/usr/bin/python
# $Id:  $
"""
HTTP and SPARQL query test case support functions 

$Rev: $
"""

import os, os.path
import sys
import unittest
import logging
import httplib
import base64
import mimetypes
import urllib
import urlparse
import simplejson

if __name__ == "__main__":
    # For testing: 
    # add main library directory to python path if running stand-alone
    sys.path.append("..")

from MiscLib import TestUtils

logger = logging.getLogger('SparqlQueryTestCase')

# Originally copied from http://code.activestate.com/recipes/146306/:
def get_content_type(filename):
    return mimetypes.guess_type(filename)[0] or 'application/octet-stream'

# Originally copied from http://code.activestate.com/recipes/146306/:
def encode_multipart_formdata(fields, files):
    """
    fields is a sequence of (name, value) elements for regular form fields.
    files is a sequence of (name, filename, value, filetype) elements for data to be uploaded as files
    Return (content_type, body) ready for httplib.HTTP instance
    """
    BOUNDARY = '----------ThIs_Is_tHe_bouNdaRY_$'
    CRLF = '\r\n'
    L = []
    for (key, value) in fields:
        L.append('--' + BOUNDARY)
        L.append('Content-Disposition: form-data; name="%s"' % key)
        L.append('')
        L.append(value)
    for (key, filename, value, filetype) in files:
        L.append('--' + BOUNDARY)
        L.append('Content-Disposition: form-data; name="%s"; filename="%s"' % (key, filename))
        L.append('Content-Type: %s' % (filetype or get_content_type(filename)))
        L.append('')
        L.append(value)
    L.append('--' + BOUNDARY + '--')
    L.append('')
    body = CRLF.join(L)
    content_type = 'multipart/form-data; boundary=%s' % BOUNDARY
    return content_type, body

def bindingType(b):
    """
    Function returns the type of a variable binding.  Commonly 'uri' or 'literal'.
    """
    type = b['type']
    if type == "typed-literal" and b['datatype'] == "http://www.w3.org/2001/XMLSchema#string":
        type = 'literal' 
    return type

def findVarBindings(data, var):
    """
    Returns a list of (type,value) pairs to which the supplied variable is bound in the results
    """
    return [ (bindingType(b[var]),b[var]['value']) 
             for b in data['results']['bindings'] if var in b ]

def findBindingSets(data):
    """
    Returns a list of lists of (var:(type,value)) dictionaries from the supplied results
    """
    return [ dict([ (var,{'type':bindingType(bindset[var]), 'value':bindset[var]['value']} ) for var in bindset ]) 
             for bindset in data['results']['bindings'] ]

class SparqlQueryTestCase(unittest.TestCase):
    """
    Test simple query patterns against data in SPARQL endpoint
    
    Although this module appears as a test suite, its main intended use is as a class
    that can be subclassed in place of unittest.TestCase, providing additional methods
    for testing HTTP access and SPATRQL queries.
    """
    def setUp(self):
        # Default SPARQL endpoint details
        self._endpointhost = "localhost"
        self._endpointpath = "/sparqlquerytest"     # Really just a placeholder
        self._endpointuser = None
        self._endpointpass = None
        return

    def tearDown(self):
        return

    def setRequestEndPoint(self, endpointhost=None, endpointpath=None):
        if endpointhost or endpointpath:
            if endpointhost:
                self._endpointhost = endpointhost
                # Reset credentials if setting host
                self._endpointuser = None
                self._endpointpass = None
            if endpointpath: self._endpointpath = endpointpath
            logger.debug("setRequestEndPoint: endpointhost %s: " % self._endpointhost)
            logger.debug("setRequestEndPoint: endpointpath %s: " % self._endpointpath)
        return

    def setRequestUserPass(self, endpointuser=None, endpointpass=None):
        if endpointuser:
            self._endpointuser = endpointuser
            self._endpointpass = endpointpass
            logger.debug("setRequestEndPoint: endpointuser %s: " % self._endpointuser)
            logger.debug("setRequestEndPoint: endpointpass %s: " % self._endpointpass)
        else:
            self._endpointuser = None
            self._endpointpass = None
        return

    def getRequestPath(self, rel):
        rel = rel or ""
        return urlparse.urljoin(self._endpointpath,rel)

    def doRequest(self, command, resource, reqdata=None, reqheaders={}, expect_status=200, expect_reason="OK"):
        logger.debug(command+" http://"+self._endpointhost+self.getRequestPath(resource))
        if self._endpointuser:
            auth = base64.encodestring("%s:%s" % (self._endpointuser, self._endpointpass)).strip()
            reqheaders["Authorization"] = "Basic %s" % auth
        hc   = httplib.HTTPConnection(self._endpointhost)
        path = self.getRequestPath(resource)
        response     = None
        responsedata = None
        repeat       = 10
        while path and repeat > 0:
            repeat -= 1
            hc.request(command, path, reqdata, reqheaders)
            response = hc.getresponse()
            if response.status != 301: break
            path = response.getheader('Location', None)
            response.read()  # Seems to be needed to free up connection
        logger.debug("Status: %i %s" % (response.status, response.reason))
        if expect_status != "*": self.assertEqual(response.status, expect_status)
        if expect_reason != "*": self.assertEqual(response.reason, expect_reason)
        responsedata = response.read()
        hc.close()
        return (response, responsedata)

    def doHTTP_GET(self,
            endpointhost=None, endpointpath=None, resource=None,
            expect_status=200, expect_reason="OK",<commit_msg>Add test method getRequestUri, and use in doRequest<commit_after>#!/usr/bin/python
# $Id:  $
"""
HTTP and SPARQL query test case support functions 

$Rev: $
"""

import os, os.path
import sys
import unittest
import logging
import httplib
import base64
import mimetypes
import urllib
import urlparse
import simplejson

if __name__ == "__main__":
    # For testing: 
    # add main library directory to python path if running stand-alone
    sys.path.append("..")

from MiscLib import TestUtils

logger = logging.getLogger('SparqlQueryTestCase')

# Originally copied from http://code.activestate.com/recipes/146306/:
def get_content_type(filename):
    return mimetypes.guess_type(filename)[0] or 'application/octet-stream'

# Originally copied from http://code.activestate.com/recipes/146306/:
def encode_multipart_formdata(fields, files):
    """
    fields is a sequence of (name, value) elements for regular form fields.
    files is a sequence of (name, filename, value, filetype) elements for data to be uploaded as files
    Return (content_type, body) ready for httplib.HTTP instance
    """
    BOUNDARY = '----------ThIs_Is_tHe_bouNdaRY_$'
    CRLF = '\r\n'
    L = []
    for (key, value) in fields:
        L.append('--' + BOUNDARY)
        L.append('Content-Disposition: form-data; name="%s"' % key)
        L.append('')
        L.append(value)
    for (key, filename, value, filetype) in files:
        L.append('--' + BOUNDARY)
        L.append('Content-Disposition: form-data; name="%s"; filename="%s"' % (key, filename))
        L.append('Content-Type: %s' % (filetype or get_content_type(filename)))
        L.append('')
        L.append(value)
    L.append('--' + BOUNDARY + '--')
    L.append('')
    body = CRLF.join(L)
    content_type = 'multipart/form-data; boundary=%s' % BOUNDARY
    return content_type, body

def bindingType(b):
    """
    Function returns the type of a variable binding.  Commonly 'uri' or 'literal'.
    """
    type = b['type']
    if type == "typed-literal" and b['datatype'] == "http://www.w3.org/2001/XMLSchema#string":
        type = 'literal' 
    return type

def findVarBindings(data, var):
    """
    Returns a list of (type,value) pairs to which the supplied variable is bound in the results
    """
    return [ (bindingType(b[var]),b[var]['value']) 
             for b in data['results']['bindings'] if var in b ]

def findBindingSets(data):
    """
    Returns a list of lists of (var:(type,value)) dictionaries from the supplied results
    """
    return [ dict([ (var,{'type':bindingType(bindset[var]), 'value':bindset[var]['value']} ) for var in bindset ]) 
             for bindset in data['results']['bindings'] ]

class SparqlQueryTestCase(unittest.TestCase):
    """
    Test simple query patterns against data in SPARQL endpoint
    
    Although this module appears as a test suite, its main intended use is as a class
    that can be subclassed in place of unittest.TestCase, providing additional methods
    for testing HTTP access and SPATRQL queries.
    """
    def setUp(self):
        # Default SPARQL endpoint details
        self._endpointhost = "localhost"
        self._endpointpath = "/sparqlquerytest"     # Really just a placeholder
        self._endpointuser = None
        self._endpointpass = None
        return

    def tearDown(self):
        return

    def setRequestEndPoint(self, endpointhost=None, endpointpath=None):
        if endpointhost or endpointpath:
            if endpointhost:
                self._endpointhost = endpointhost
                # Reset credentials if setting host
                self._endpointuser = None
                self._endpointpass = None
            if endpointpath: self._endpointpath = endpointpath
            logger.debug("setRequestEndPoint: endpointhost %s: " % self._endpointhost)
            logger.debug("setRequestEndPoint: endpointpath %s: " % self._endpointpath)
        return

    def setRequestUserPass(self, endpointuser=None, endpointpass=None):
        if endpointuser:
            self._endpointuser = endpointuser
            self._endpointpass = endpointpass
            logger.debug("setRequestEndPoint: endpointuser %s: " % self._endpointuser)
            logger.debug("setRequestEndPoint: endpointpass %s: " % self._endpointpass)
        else:
            self._endpointuser = None
            self._endpointpass = None
        return

    def getRequestPath(self, rel):
        rel = rel or ""
        return urlparse.urljoin(self._endpointpath,rel)

    def getRequestUri(self, rel):
        return "http://"+self._endpointhost+self.getRequestPath(rel)

    def doRequest(self, command, resource, reqdata=None, reqheaders={}, expect_status=200, expect_reason="OK"):
        logger.debug(command+" "+self.getRequestUri(resource))
        if self._endpointuser:
            auth = base64.encodestring("%s:%s" % (self._endpointuser, self._endpointpass)).strip()
            reqheaders["Authorization"] = "Basic %s" % auth
        hc   = httplib.HTTPConnection(self._endpointhost)
        path = self.getRequestPath(resource)
        response     = None
        responsedata = None
        repeat       = 10
        while path and repeat > 0:
            repeat -= 1
            hc.request(command, path, reqdata, reqheaders)
            response = hc.getresponse()
            if response.status != 301: break
            path = response.getheader('Location', None)
            response.read()  # Seems to be needed to free up connection
        logger.debug("Status: %i %s" % (response.status, response.reason))
        if expect_status != "*": self.assertEqual(response.status, expect_status)
        if expect_reason != "*": self.assertEqual(response.reason, expect_reason)
        responsedata = response.read()
        hc.close()
        return (response, responsedata)

    def doHTTP_GET(self,
            endpointhost=None, endpointpath=None, resource=None,
            expect_status=200, expect_reason="OK",
{'fraction_non_alphanumeric': 0.11306553911205074, 'fraction_numerical': 0.005919661733615222, 'mean_word_length': 2.7964686998394863, 'pattern_counts': {'":': 8, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7040985', 'n_tokens_mistral': 3469, 'n_tokens_neox': 3365, 'n_words': 927}