# -*- coding: UTF-8 -*-
# Copyright 2005, 2006 EIAO Consoritum
# This program is distributed under the terms of the GNU General
# Public License.
#
# This file is part of the European Internet Accessibility Observatory
# (EIAO)
#
# EIAO is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# EIAO is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with EIAO; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston,
# MA 02110-1301 USA
__author__ = 'Morten Goodwin Olsen'
__maintainer__ = 'Nils Ulltveit-Moe'
__version__ = "$Id$"
import RDF
import string
import doctest, unittest
import RDFgenerator
import RDFreaderwriter
from RDFreaderwriter import *
from RDFgeneratorerror import *
from sc import SystemConfiguration
from urlrep import *
"""Test suite for the RDF generator"""
#Doctests
result = []
sc = sc.SystemConfiguration()
RDFgenerator.urlrep = URLRepUtils3(sc)
#result.append(doctest.testmod(RDFgenerator))
#result.append(doctest.testmod(__import__('RDFreaderwriter')))
class TestSequenceFunctions(unittest.TestCase):
def setUp(self):
#RDFgenerator.urlrep = URLRepUtils3('eiaourlrep','eiaourlrep','eiaourlrep')
reload(RDFgenerator)
self.rg = RDFgenerator
self.rw = RDFreaderwriter('EIAO_test_rdfwriter')
self.sc = SystemConfiguration()
self.urlrep = URLRepUtils3(self.sc)
self.rg.setInstances(self.sc,self.urlrep)
#self.rw.db = rdfsql.rs_connect(self.sc.dbdatabase, self.sc.dbusername, self.sc.dbpassword, 'EIAO_test_rdfwriter')
self.rw.rdfmodel = 'EIAO_test_rdfwriter'
def tearDown(self):
self.rw.stop()
del self.rw.db
del self.sc
def testnomodulesset(self):
reload(RDFgenerator)
self.rg = RDFgenerator
fail = False
try:
self.rg.gettestrunrdf()
except InstanceNotPresentError:
fail = True
sc = SystemConfiguration()
#urlrep = HouseKeeping()
urlrep = URLRepUtils3(sc)
self.rg.setInstances(sc,urlrep)
fail = False
try:
self.rg.gettestrunrdf(1)
except InstanceNotPresentError:
fail = True
self.assert_(not fail)
def modelit(self, rdf):
model = RDF.Model()
parser = RDF.RDFXMLParser()
#Translating the RDF written into a model
parser.parse_string_into_model(model, rdf, base_uri="http://www.eiao.net/rdf/1.0#")
statements = model.find_statements(RDF.Statement(subject = None, predicate = None, object = None))
allstatements = [str(stat) for stat in statements]
return allstatements
def testemptyheader(self):
fail = False
try:
self.rg.getpagesurveyrdf('html
PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN"
"http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd"', 'Plone',{},'http://someurl.com')
except EmptyHeaderInformationError:
fail = True
self.assert_(fail)
def testillegaltestrun(self):
self.assert_(self.rg.gettestrunrdf(1))
fail = False
try:
self.rg.gettestrunrdf('NOT INTEGER')
except IllegalTestRunValueError:
fail = True
self.assert_(fail)
def testdbconnection(self):
self.assert_(self.rw.db)
def testrdfmodelpresent(self):
self.assert_(self.rw.rdfmodel)
def testwritecache(self):
self.rw.flushDB()
xml = self.rg.gettestrunrdf(1) + self.rg.getendtestrunrdf(1)
#self.rw.writeRDF(xml,performcache=True)
#self.rw.commit()
def testwritereadrdfsitesurvey(self):
xml = self.rg.getsitesurveyrdf('http://www.hole.kommune.no/',testrunnr = 1, sitesurvey = 'www.holekommune.no_survey_12', downloaddir='/var/local/cache/eiao/harvestman/storedfiles/')
#self.rw.writeRDF(xml)
def testwritereadrdfscenario(self):
#rdfsql.rs_flush(self.rw.db)
#self.rg.urlrep.setCurrentScenario('www.hole.kommune.no_survey_12_scenario_1')
#self.rg.urlrep.setCurrentSiteSurvey('www.hole.kommune.no_survey_12')
self.rg.pagedictionary['http://www.hia.no/'] = 'http://www.eiao.net_test_survey_12'
self.rg.pagedictionary['http://www.hia.no/english'] = 'http://www.eiao.net_test_survey_12'
self.rg.pagedictionary['http://www.hia.no/sonja'] = 'http://www.eiao.net_test_survey_12'
xml = self.rg.getscenariordf([('http://www.hia.no/',1,12,'http://www.hia.no/english',0,0),('http://www.hia.no/english',34,12,'http://www.hia.no/sonja',0,0)],self.urlrep.getCurrentScenario(),self.urlrep.getCurrentPageSurvey(),0.5)
#self.rw.writeRDF(xml)
for a in self.rw.asyncwriters:
a.join()
#a.join()
#Removing all whitespaces to remove difference in the XML
#for i in string.whitespace:
# xml = xml.replace(i,'')
#for i in string.whitespace:
# retrievedxml = retrievedxml.replace(i,'')
#xmllist = ['<'+ x for x in xml.split('<') if x != '']
#Checking for al elements in XMLlist, because XML might vary.
#for x in xmllist:
# self.assert_(x in retrievedxml)
def testwritepagesurvey(self):
rdfsql.rs_flush(self.rw.db)
#self.rg.urlrep.setCurrentPageSurvey('www.holekommune.no_pageSurvey_1')
(xml,currentsurvey) = self.rg.getpagesurveyrdf('html
PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN"
"http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd"', 'Plone',{'Expires':'Thu, 19 Nov 1981 ::00 GMT','Data':'Mon, 22 Aug 2005 ::31 GMT'},'http://www.someurl.com')
#self.rw.writeRDF(xml)
def testnorwegianrdf(self):
rdfsql.rs_flush(self.rw.db)
subject = 'http://www.eiao.net/rdf/testRun_1'
predicate = 'http://www.eiao.net/rdf/asserts'
object = 'someobjectøæå'
#self.rw.writeRDFTriple(subject, predicate, object, objectliteral = True)
def testwritenorwegianearl(self):
rdfsql.rs_flush(self.rw.db)
EARL = """
Web page (sampled automatically)
2005-12-3
"""
EARL = self.rw.rdfbegin + EARL + self.rw.rdfend
xml = self.rg.getearl('http://www.ellesea.no/æøå',EARL, preparseEARL=True,addnamespace=False)
def testwritesimpleearl(self):
EARL = """
Web page (sampled automatically)
2005-12-3
"""
xml = self.rg.getearl('http://www.ellesea.no/',EARL, preparseEARL=True)
self.rw.flushDB()
#self.rw.writeRDF(xml,performcache=True)
#self.rw.commit(async=True)
xml = self.rw.rdfbegin + xml + self.rw.rdfend
def testwritelargerdf(self):
self.rw.flushDB()
written =''
#Test run
xml = self.rg.gettestrunrdf(1) + self.rg.getendtestrunrdf(1)
self.rw.writeRDF(xml)
written += xml
#Site survey
xml = self.rg.getsitesurveyrdf('http://www.hole.kommune.no/',testrunnr = 1, sitesurvey = 'www.hole.kommune.no_survey_12', downloaddir='/var/local/cache/eiao/harvestman/storedfiles/')
self.rw.writeRDF(xml)
written += xml
#Scenarios
#self.rg.urlrep.setCurrentScenario('www.hole.kommune.no_survey_12_scenario_1')
#self.rg.urlrep.setCurrentSiteSurvey('www.hole.kommune.no_survey_12')
self.rg.pagedictionary['http://www.hia.no/'] = 'http://www.eiao.net_test_survey_12'
self.rg.pagedictionary['http://www.hia.no/english'] = 'http://www.eiao.net_test_survey_12'
self.rg.pagedictionary['http://www.hia.no/sonja'] = 'http://www.eiao.net_test_survey_12'
xml = self.rg.getscenariordf([('http://www.hia.no/',1,12,'http://www.hia.no/english',0,0),('http://www.hia.no/english',34,12,'http://www.hia.no/sonja',0,0)],self.urlrep.getCurrentScenario(),self.urlrep.getCurrentPageSurvey(),0.5)
self.rw.writeRDF(xml)
written += xml
#Page Survey
(xml,currentpagesurvey) = self.rg.getpagesurveyrdf('html
PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN"
"http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd"','Plone', {'Expires':'Thu, 19 Nov 1981 ::00 GMT','Data':'Mon, 22 Aug 2005 ::31 GMT'},'http://www.somesite.com')
self.rw.writeRDF(xml)
written += xml
def testwritelargeearl(self):
self.rw.flushDB()
f = open('earlexample.rdf')
EARL = ''.join(f.readlines())
f.close()
xmlwritten = self.rg.getearl('http://www.example.com/',EARL, preparseEARL=True)
def testwronglengthofrdf(self):
fail = False
try:
self.rg.getscenariordf([('http://www.hia.no',1,12,'http://www.hia.no/english',0),('http://www.hia.no/english',34,12,'http://www.hia.no/sonja',0)],self.urlrep.getCurrentScenario(),self.urlrep.getCurrentPageSurvey(),0.5)
except WrongLengthOfLogError:
fail = True
self.assert_(fail)
def testallrdf(self):
rdfsql.rs_flush(self.rw.db)
xml = self.rg.gettestrunrdf(1)
xml += self.rg.getsitesurveyrdf('http://www.hole.kommune.no/',testrunnr = 1, sitesurvey = 'www.holekommune.no_survey_12', downloaddir='/var/local/cache/eiao/harvestman/storedfiles/')
#self.rg.urlrep.setCurrentScenario('www.hole.kommune.no_survey_12_scenario_1')
#self.rg.urlrep.setCurrentSiteSurvey('www.hole.kommune.no_survey_12')
self.rg.pagedictionary['http://www.hia.no/'] = 'http://www.eiao.net_test_survey_12'
self.rg.pagedictionary['http://www.hia.no/english'] = 'http://www.eiao.net_test_survey_12'
self.rg.pagedictionary['http://www.hia.no/sonja'] = 'http://www.eiao.net_test_survey_12'
xml += self.rg.getscenariordf([('http://www.hia.no/',1,12,'http://www.hia.no/english',0,0),('http://www.hia.no/english',34,12,'http://www.hia.no/sonja',0,0)],self.urlrep.getCurrentScenario(),self.urlrep.getCurrentPageSurvey(),0.5)
self.rw.writeRDF(xml)
def testpagedictionary(self):
survey = 'http://www.eiao.net/rdf/1.0#somesurvey'
self.rg.addpagesurveytourlconnection(survey,'http://www.test.com')
self.assert_(self.rg.pagedictionary['http://www.test.com']==survey)
def testillegalrdf(self):
errorxml = """
%s
""" %('1', '2005-12-12')
fail = False
try:
self.rw.writeRDF(errorxml)
except IllegalRDFError:
fail = True
self.assert_(fail)
def testillegalearl(self):
errorEARL = """
Web page (sampled automatically)
2005-12-3
"""
#self.rg.urlrep.setCurrentPageSurvey('www.holekommune.no_pageSurvey_1')
fail = False
try:
xml = self.rg.getearl('http://www.ellesea.no/',errorEARL, preparseEARL = True)
except IllegalEARLError:
fail = True
#Note the difference between rdf:ID and rdf:about
correctEARL = """
Web page (sampled automatically)
2005-12-3
"""
#self.rg.urlrep.setCurrentPageSurvey('www.holekommune.no_pageSurvey_1')
xml = self.rg.getearl('http://www.ellesea.no/',correctEARL, preparseEARL = True)
self.assert_(xml)
def testalmostemptyrdf(self):
fail = False
try:
self.rw.writeRDF('\n')
except EmptyRDFError:
fail = True
self.assert_(fail)
def testemptyearl(self):
fail = False
try:
self.rg.getearl('http://example.com/','', preparseEARL=True)
except EmptyEARLError:
fail = True
self.assert_(fail)
def testnoassertionspresentinearl(self):
#self.rg.urlrep.setCurrentPageSurvey('www.holekommune.no_pageSurvey_1')
EARL = """
Web page (sampled automatically)
2005-12-3
"""
fail = False
try:
xml = self.rg.getearl('http://www.ellesea.no/',EARL, preparseEARL = True, assertions = [])
except NoAssertionsPresentEARLError:
fail = True
self.assert_(fail)
def testillegalcombination(self):
#self.rg.urlrep.setCurrentPageSurvey('www.holekommune.no_pageSurvey_1')
EARL = """
Web page (sampled automatically)
2005-12-3
"""
fail = False
try:
xml = self.rg.getearl('http://www.ellesea.no/',EARL, preparseEARL = False, assertions = [])
except IllegalCombinationError:
fail = True
self.assert_(fail)
if __name__ == "__main__":
result += unittest.main()