Commit e5dc4a70 authored by Julia Kaufhold's avatar Julia Kaufhold
Browse files

unit tests for B2Share client.

parent 8ba1c3c7
\ No newline at end of file
......@@ -4,11 +4,14 @@
################################################################################
# B2SAFE B2SHARE client Command Line Interface #
################################################################################
import sys
#sys.path.insert(0,'/usr/lib/python3.4/site-packages')
import logging.handlers
import requests
import argparse
import os
import pprint
import requests
from manifest import IRODSUtils
from configuration import Configuration
......@@ -247,4 +250,4 @@ if __name__ == "__main__":
parser_deleteDraft.set_defaults(func=deleteDraft)
args = parser.parse_args()
args.func(args)
\ No newline at end of file
args.func(args)
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
import requests
import jsonpatch
import requests
class B2shareClient():
......
......@@ -4,16 +4,18 @@
################################################################################
# Configuration Class #
################################################################################
import logging.handlers
import ConfigParser
import logging.handlers
import os
class Configuration():
"""
Get properties from filesystem
"""
defaultLogDir = os.getcwd() + os.sep + "logs"
defaultLogFileName = os.sep + "b2share_connection.log"
def __init__(self, conffile, debug, dryrun, logger):
self.conffile = conffile
self.debug = debug
self.dryrun = dryrun
......@@ -22,19 +24,41 @@ class Configuration():
self.log_level = {'INFO': logging.INFO, 'DEBUG': logging.DEBUG, \
'ERROR': logging.ERROR, 'WARNING': logging.WARNING, \
'CRITICAL': logging.CRITICAL}
self.config = ConfigParser.RawConfigParser()
# return configuration missing exception if expected config is not in the config file
def parseConf(self):
"""Parse the configuration file."""
self.config = ConfigParser.RawConfigParser()
#self.config = ConfigParser.RawConfigParser()
if not os.path.exists(self.conffile):
self.logger.error('missing configuration file %s:%s' % (self.conffile))
return # error config file not found
with open(self.conffile, "r") as confFile:
self.config.readfp(confFile)
try:
self.config.readfp(confFile)
except Exception as e:
self.configuration.logger.error(e)
return
logfilepath = self._getConfOption('Logging', 'log_file')
if not os.path.exists(logfilepath):
# if user did not specified a path for log file default to current dir + b2share_connection.log
# self.logger.error("no location for log file is specified, so try to default to <currend directory>/logs/b2share_connection.log")
try:
os.mkdir(self.defaultLogDir)
logfilepath = self.defaultLogDir + self.defaultLogFileName
except OSError:
# self.logger.error("no location for log file is specified, so try to default to <currend directory>/logs/b2share_connection.log")
logfilepath = os.getcwd()+ self.defaultLogFileName
#print("Creation of the directory %s failed, so defaulting to <currend directory>/b2share_connection.log" % defaultLogDir)
loglevel = self._getConfOption('Logging', 'log_level')
if self.debug:
loglevel = 'DEBUG'
self.logger.setLevel(self.log_level[loglevel])
rfh = logging.handlers.RotatingFileHandler(logfilepath, \
maxBytes=8388608, \
backupCount=9)
......@@ -53,8 +77,6 @@ class Configuration():
self.access_parameter = self._getConfOption('B2SHARE_HTTP_API', 'access_parameter')
self.get_community_schema_endpoint = self._getConfOption('B2SHARE_HTTP_API', 'get_community_schema_endpoint')
self.irods_zone_name = self._getConfOption('iRODS', 'zone_name')
self.irods_res = self._getConfOption('iRODS', 'resources')
self.irods_home_dir = self._getConfOption('iRODS', 'irods_home_dir')
self.irods_debug = self._getConfOption('iRODS', 'irods_debug', True)
......@@ -62,13 +84,18 @@ class Configuration():
"""
get the options from the configuration file
"""
if (self.config.has_option(section, option)):
opt = self.config.get(section, option)
if boolean:
if opt in ['True', 'true']: return True
else: return False
return opt
else:
self.logger.warning('missing parameter %s:%s' % (section,option))
return None
if self.config.has_section(section):
if (self.config.has_option(section, option)):
opt = self.config.get(section, option)
if boolean:
if opt in ['True', 'true']:
return True
else:
return False
return opt
else:
self.logger.error('missing parameter %s: in section %s' % (section, option))
else:
self.logger.error('missing section %s: with parameter %s' % (section, option))
return None
\ No newline at end of file
......@@ -2,10 +2,19 @@
[Logging]
# possible values: INFO, DEBUG, ERROR, WARNING, CRITICAL
log_level=DEBUG
log_file=b2safe_b2share.log
log_file=../log/b2share_connection.log
# section containing the sources for users and projects/groups'information
[B2SHARE]
scheme=https
address=trng-b2share.eudat.eu
path=/api/
[B2SHARE_HTTP_API]
# test instance of the B2SHARE_HTTP_API
host_name=https://trng-b2share.eudat.eu/
access_parameter=?access_token
list_communities_endpoint=api/communities/
get_community_schema_endpoint=/schemas/last
records_endpoint=api/records/
[iRODS]
#irods_home_dir=
# e.g. irods_home_dir=/MyZoneName/home/
# Boolean value: True | False
irods_debug=False
\ No newline at end of file
#please fill out at least the required fields with values as JSON strings in the line after the property name.
#e.g. for community - EUDAT, for open_access - true, for contributors - [{"contributor_name":"Hulk", "contributor_type": "Editor"}, {"contributor_name":"Banner", "contributor_type": "ContactPerson"}]
[required]
community=EUDAT
titles=[{"title":"Demo b2shareclient"}]
descriptions=[{"description":"A collection to demostrate the b2shareclient functionality", "description_type":"Abstract"}]
creators=[{"creator_name":"Kaufhold, Julia"}]
open_access=true
embargo_date=November 12, 2017
license={"license":"Public Domain Mark (PD)", "license_uri":"http://creativecommons.org/publicdomain/mark/1.0/"}
disciplines=["4.1.5 → Computer sciences → Data structures", "4.1.4 → Computer sciences → Algorithm"]
keywords=["Test", "Demo"]
contact_email=email@example.com
[optional]
contributors=[{"contributor_name":"Hulk", "contributor_type": "Editor"}, {"contributor_name":"Banner", "contributor_type": "ContactPerson"}]
resource_types=[{"resource_type":"Random text", "resource_type_general":"Text"}]
alternate_identifiers=[{"alternate_identifier":"/JULK_ZONE/home/irods/julia/collection_A", "alternate_identifier_type":"iRODS path"}]
version=V 1.0
publisher=Kaufhold, Julia. MPCDF
language=German [deu]
\ No newline at end of file
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import unittest
from testB2shareConnectionClient.configurationtest import ConfigurationTest
from testB2shareConnectionClient.b2shareclienttest import B2shareClientTest
def getSuite():
# first test the configurations class that loads the configurations for the B2shareClient, that is not working without
# than test the B2shareClient from the b2shareclientworker.py doing the requests to the B2Share
# as b2shareclient.py is only a wrapper calling the B2shareClient and connecting to the iRODS
configurationTests = unittest.TestLoader().loadTestsFromTestCase(ConfigurationTest)
b2shareClientTests = unittest.TestLoader().loadTestsFromTestCase(B2shareClientTest)
# TODO: at the end call the tests for the b2shareclient.py
return unittest.TestSuite([configurationTests, b2shareClientTests])
if __name__ == "__main__":
runner = unittest.TextTestRunner()
runner.run(getSuite())
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import unittest
from mock import MagicMock, patch
import json
import sys
import logging
test_logger = logging.getLogger('B2shareClientTest')
sys.path.append("../../cmd")
from b2shareclientworker import B2shareClient
class B2shareClientTest(unittest.TestCase):
def setUp(self):
# mock configuration
# TODO: change "confpath" to variable from call of the test or in the test suite
self.configurationMock = MagicMock(confpath = "/home/irods/B2SAFE-core/conf/b2safe_b2share.conf",
b2share_host_name = 'https://trng-b2share.eudat.eu/api',
list_communities_endpoint = '/communities',
access_parameter = '/?access_token=',
access_token = 'JFSUcYzTRpMdVCEyAk3mmtGcfpmH',
dryrun = False,
debug = True,
logger = test_logger)
self.b2shcl = B2shareClient(self.configurationMock)
self.draft_id_mock = "734c8cf3bb4f42cca0e9c4c3cf3cc86b"
self.community_id_mock = "e9b9792e-79fb-4b07-b6b4-b9c2bd06d095"
self.filePIDsString_mock = '[{"ePIC_PID": "http://hdl.handle.net/20.500.11946/5db21a88-e0d8-11e7-9eab-0050569e7581","key": "copytest.txt"},{"ePIC_PID": "http://hdl.handle.net/20.500.11946/619416e2-e0d8-11e7-8ca4-0050569e7581","key": "metatest2.txt"}]'
self.metadata_file_mock = ''
# TODO: change to variable from call of the test or in the test suite
filled_md_file_path = '/home/irods/B2SAFE-core/conf/b2share_metadata.json'
with open(filled_md_file_path, 'r') as metadata_file:
self.metadata_file_mock = metadata_file.read()
self.title_mock = "TestDraftTitle"
pass
def tearDown(self):
logging.shutdown()
pass
def testGetAllCommunities(self):
mock_get_patcher = patch('b2shareclient.requests.get')
mock_get = mock_get_patcher.start()
mock_get.return_value.json.return_value = json.loads('{"hits": {"hits": [{"created": "Wed, 21 Dec 2016 08:57:40 GMT", "description": "Aalto University", "id": "c4234f93-da96-4d2f-a2c8-fa83d0775212", "name": "Aalto"}]}}')
mock_get.return_value.status_code = 200
b2shcl = B2shareClient(self.configurationMock)
communities = b2shcl.getAllCommunities()
mock_get_patcher.stop()
self.assertTrue(bool(communities))
def testDeleteDraft(self):
mock_get_patcher = patch('b2shareclient.requests.delete')
mock_get = mock_get_patcher.start()
mock_get.return_value.status_code = 204
b2shcl = B2shareClient(self.configurationMock)
b2shcl.deleteDraft(self.draft_id_mock)
mock_get_patcher.stop()
self.assertTrue(True)
def testGetDraftByID(self):
mock_get_patcher = patch('b2shareclient.requests.get')
mock_get = mock_get_patcher.start()
mock_get.return_value.status_code = 200
mock_get.return_value.text = "response text mock"
b2shcl = B2shareClient(self.configurationMock)
draft = b2shcl.getDraftByID(self.draft_id_mock)
mock_get_patcher.stop()
self.assertTrue(bool(draft))
def testgGetCommunitySchema(self):
mock_get_patcher = patch('b2shareclient.requests.get')
mock_get = mock_get_patcher.start()
mock_get.return_value.status_code = 200
mock_get.return_value.text = "response text mock"
b2shcl = B2shareClient(self.configurationMock)
community_schema = b2shcl.getCommunitySchema(self.community_id_mock)
mock_get_patcher.stop()
self.assertTrue(bool(community_schema))
def testPublishRecord(self):
mock_get_patcher = patch('b2shareclient.requests.patch')
mock_get = mock_get_patcher.start()
mock_get.return_value.status_code = 200
mock_get.return_value.text = "response text mock"
b2shcl = B2shareClient(self.configurationMock)
b2shcl.publishRecord(self.draft_id_mock)
mock_get_patcher.stop()
self.assertTrue(True)
def testGetDraftMetadata(self):
mock_get_patcher = patch('b2shareclient.requests.get')
mock_get = mock_get_patcher.start()
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = json.loads('{"metadata":{"$schema": "https://trng-b2share.eudat.eu/api/communities/e9b9792e-79fb-4b07-b6b4-b9c2bd06d095/schemas/0#/draft_json_schema", "community": "e9b9792e-79fb-4b07-b6b4-b9c2bd06d095", "community_specific": {}, "external_pids": [{"ePIC_PID": "http://hdl.handle.net/20.500.11946/5db21a88-e0d8-11e7-9eab-0050569e7581","key": "copytest.txt"},{"ePIC_PID": "http://hdl.handle.net/20.500.11946/60fb7360-e0d8-11e7-a910-0050569e7581","key": "test3.txt"}],"open_access": true,"owners": [148],"publication_state": "draft","titles": [{"title": "testCollectionDraft1"}]}}')
mock_get.return_value.text = "response text mock"
b2shcl = B2shareClient(self.configurationMock)
draft_metadata = b2shcl.getDraftMetadata(self.draft_id_mock)
mock_get_patcher.stop()
self.assertTrue(bool(draft_metadata))
def testAddB2shareMetadata(self):
mock_patch_patcher = patch('b2shareclient.requests.patch')
mock_patch = mock_patch_patcher.start()
mock_patch.return_value.status_code = 200
self.b2shcl.addB2shareMetadata(self.draft_id_mock, self.metadata_file_mock)
mock_patch_patcher.stop()
self.assertTrue(bool(True))
def testCreateDraft(self):
mock_post_patcher = patch('b2shareclient.requests.post')
mock_post = mock_post_patcher.start()
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = json.loads('{"id":"734c8cf3bb4f42cca0e9c4c3cf3cc86b"}')
b2shcl = B2shareClient(self.configurationMock)
record_id = b2shcl.createDraft(self.community_id_mock, self.title_mock, self.filePIDsString_mock)
mock_post_patcher.stop()
self.assertTrue(bool(record_id))
#### test failure status_code
def testGetAllCommunities_failed(self):
mock_get_patcher = patch('scripts.tests.cmd.b2shareclient.requests.get')
mock_get = mock_get_patcher.start()
mock_get.return_value.json.return_value = json.loads('{}')
mock_get.return_value.status_code = 400
b2shcl = B2shareClient(self.configurationMock)
communities = b2shcl.getAllCommunities()
mock_get_patcher.stop()
self.assertFalse(bool(communities))
def testDeleteDraft_failed(self):
mock_get_patcher = patch('scripts.tests.cmd.b2shareclient.requests.delete')
mock_get = mock_get_patcher.start()
mock_get.return_value.status_code = 400
b2shcl = B2shareClient(self.configurationMock)
b2shcl.deleteDraft(self.draft_id_mock)
mock_get_patcher.stop()
self.assertTrue(True)
def testGetDraftByID_failed(self):
mock_get_patcher = patch('scripts.tests.cmd.b2shareclient.requests.get')
mock_get = mock_get_patcher.start()
mock_get.return_value.status_code = 400
mock_get.return_value.text = None
b2shcl = B2shareClient(self.configurationMock)
draft = b2shcl.getDraftByID(self.draft_id_mock)
mock_get_patcher.stop()
self.assertFalse(bool(draft))
def testgGetCommunitySchema_failed(self):
mock_get_patcher = patch('scripts.tests.cmd.b2shareclient.requests.get')
mock_get = mock_get_patcher.start()
mock_get.return_value.status_code = 400
mock_get.return_value.text = None
b2shcl = B2shareClient(self.configurationMock)
community_schema = b2shcl.getCommunitySchema(self.community_id_mock)
mock_get_patcher.stop()
self.assertFalse(bool(community_schema))
def testPublishRecord_failed(self):
mock_get_patcher = patch('scripts.tests.cmd.b2shareclient.requests.patch')
mock_get = mock_get_patcher.start()
mock_get.return_value.status_code = 400
mock_get.return_value.text = None
b2shcl = B2shareClient(self.configurationMock)
b2shcl.publishRecord(self.draft_id_mock)
mock_get_patcher.stop()
self.assertTrue(True)
def testGetDraftMetadata_failed(self):
mock_get_patcher = patch('scripts.tests.cmd.b2shareclient.requests.get')
mock_get = mock_get_patcher.start()
mock_get.return_value.status_code = 400
mock_get.return_value.json.return_value = json.loads('{}')
mock_get.return_value.text = None
b2shcl = B2shareClient(self.configurationMock)
draft_metadata = b2shcl.getDraftMetadata(self.draft_id_mock)
mock_get_patcher.stop()
self.assertFalse(bool(draft_metadata))
def testAddB2shareMetadata_failed(self):
mock_patch_patcher = patch('scripts.tests.cmd.b2shareclient.requests.patch')
mock_patch = mock_patch_patcher.start()
mock_patch.return_value.status_code = 400
self.b2shcl.addB2shareMetadata(self.draft_id_mock, self.metadata_file_mock)
mock_patch_patcher.stop()
self.assertTrue(bool(True))
def testCreateDraft_failed(self):
mock_post_patcher = patch('scripts.tests.cmd.b2shareclient.requests.post')
mock_post = mock_post_patcher.start()
mock_post.return_value.status_code = 400
mock_post.return_value.json.return_value = json.loads('{}')
b2shcl = B2shareClient(self.configurationMock)
record_id = b2shcl.createDraft(self.community_id_mock, self.title_mock, self.filePIDsString_mock)
mock_post_patcher.stop()
self.assertFalse(bool(record_id))
if __name__ == '__main__':
unittest.main()
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import unittest
import os
import logging
import sys
logger = logging.getLogger('ConfigurationTest')
sys.path.append("../../cmd")
from configuration import Configuration
class ConfigurationTest(unittest.TestCase):
def setUp(self):
# create tmp file with correct configs
self.conffile = os.getcwd() + os.sep + "ConfigurationTestTMPconfigFile.txt"
with open(self.conffile, 'w') as tmpConfFile:
tmpConfFile.write("[Logging]" + os.linesep + \
"log_file=../../log/b2share_connection.log" + os.linesep + \
"[B2SHARE_HTTP_API]" + os.linesep + \
"host_name=https://trng-b2share.eudat.eu/ " + os.linesep + \
"access_parameter=?access_token " + os.linesep + \
"list_communities_endpoint=api/communities/ " + os.linesep + \
"get_community_schema_endpoint=/schemas/last " + os.linesep + \
"records_endpoint=api/records/ " + os.linesep + \
"log_level=DEBUG " + os.linesep + \
"[iRODS] " + os.linesep + \
"irods_home_dir=/BasZone/home/ " + os.linesep + \
"irods_debug=False")
logger.setLevel('DEBUG')
defaultLogPath = Configuration.defaultLogDir
if os.path.exists(defaultLogPath):
configurationTestsLogfilepath = defaultLogPath + os.sep + "ConfigurationTest.log"
else:
configurationTestsLogfilepath = os.getcwd() + os.sep + "ConfigurationTest.log"
rfh = logging.handlers.RotatingFileHandler(configurationTestsLogfilepath, \
maxBytes=8388608, \
backupCount=9)
formatter = logging.Formatter('%(asctime)s %(levelname)s: '+ '[%(funcName)s] %(message)s')
rfh.setFormatter(formatter)
logger.addHandler(rfh)
self.configurationMock = Configuration(self.conffile, True, False, logger)
pass
def tearDown(self):
# delete tmp file with configs
if os.path.exists(self.conffile):
os.remove(self.conffile)
else:
print("The file does not exist")
# defaulting only if was not existing before so delete default logs after the test
# to save the logs for the test add existing log file path
logging.shutdown()
logfilepath = Configuration.defaultLogDir + Configuration.defaultLogFileName
if os.path.exists(logfilepath):
os.remove(logfilepath)
logfilepath = os.getcwd()+ Configuration.defaultLogFileName
if os.path.exists(logfilepath):
os.remove(logfilepath)
pass
def testParseConf(self):
# run parseConf to fill the configurationMock with mock options
self.configurationMock.parseConf()
# test _init_
self.assertEqual(self.conffile, self.configurationMock.conffile)
# test an option read from the config file
self.assertEqual("https://trng-b2share.eudat.eu/", self.configurationMock.b2share_host_name)
def test_getConfOption(self):
with open(self.configurationMock.conffile, "r") as confFile:
try:
self.configurationMock.config.readfp(confFile)
except Exception as e:
self.assertTrue(False, "Error during reading conf file for the test")
irodsDebugOption = self.configurationMock._getConfOption('iRODS', 'irods_debug', True)
self.assertNotEqual(None, irodsDebugOption, "Reading existing boolean option from configuration file.")
irodsHomeDirOption = self.configurationMock._getConfOption('iRODS', 'irods_home_dir')
self.assertNotEqual(None, irodsHomeDirOption, "Reading existing not boolean option from configuration file.")
def test_getConfOption_error(self):
try:
self.configurationMock._getConfOption("unknown", "option", True)
except Exception as e:
self.assertNotEqual(None, e, "Reading not existing boolean option from configuration file.")
try:
self.configurationMock._getConfOption("unknown", "option")
except Exception as ex:
self.assertNotEqual(None, ex, "Reading not existing not boolean option from configuration file.")
def testParseConfError(self):
# write error containing config file
with open(self.conffile, 'w') as tmpConfFile:
tmpConfFile.write("[Logging]" + os.linesep + \
"log_file=../../log/b2share_connection.log " + os.linesep + \
"[B2SHARE_HTTP_API" + os.linesep + \
"host_name=https://trng-b2share.eudat.eu/ " + os.linesep + \
"access_parameter=?access_token " + os.linesep + \
"list_communities_endpoint=api/communities/ " + os.linesep + \
"get_community_schema_endpoint=/schemas/last " + os.linesep + \
"records_endpoint=api/records/ " + os.linesep + \
"log_level=DEBUG " + os.linesep + \
"[iRODS] " + os.linesep + \
"irods_home_dir=/BasZone/home/ " + os.linesep + \
"irods_debug=False")
# run parseConf to fill the configurationMock with mock options with incorrect configuration file
self.configurationMock = Configuration(self.conffile, True, False, logger)
try:
self.configurationMock.parseConf()
self.assertTrue(False)
except Exception:
self.assertTrue(True)
if __name__ == "__main__":
unittest.main()
\ No newline at end of file
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment