Commit 6db3e785 authored by Julia Kaufhold's avatar Julia Kaufhold
Browse files

unit tests for b2shareclient_cli.

parent e5dc4a70
#!/usr/bin/env python
# -*- coding: utf-8 -*-
################################################################################
# B2SAFE B2SHARE client Command Line Interface #
################################################################################
import argparse
import logging.handlers
import os
import pprint
import requests
from b2shareclient import B2shareClient
from configuration import Configuration
from manifest import IRODSUtils
logger = logging.getLogger('B2shareClientCLI')
def draft(args):
configuration = Configuration(args.confpath, args.debug, args.dryrun, logger)
configuration.parseConf()
logger.info("Start creating draft ...")
accessToken = getAccessTokenWithConfigs(configuration, args)
if accessToken is None:
logger.error("Drafting FAILED. No B2SHARE access token found in users meta data.")
return None
configuration.access_token = accessToken
b2shcl = B2shareClient(configuration)
filePIDsList = collectPIDsForCollection(args.collectionPath, configuration)
commID = args.communityID
if args.communityID is None:
commID = getCommunityIDByName(configuration, args.communityName)
recordId = None
if commID and filePIDsList:
recordId = b2shcl.createDraft(commID, args.title, filePIDsList)
if recordId is not None:
logger.info("Drafting for record "+recordId+" END.")
else:
logger.error("Drafting FAILED.")
return recordId
def getCommunityIDByName(configuration, community_name):
community_id = None
if not configuration:
return community_id
if not community_name:
return community_id
host = configuration.b2share_host_name
endpoint = configuration.list_communities_endpoint
acces_part = None
list_communities_url = None
if configuration.access_parameter and configuration.access_token:
acces_part = configuration.access_parameter + "=" + configuration.access_token
if acces_part and host and endpoint:
list_communities_url = host + endpoint + acces_part
if list_communities_url:
try:
#TODO: delete 'verify=False', if B2SHARE server has proper certificate
response = requests.get(url=list_communities_url, verify=False)
print(list_communities_url)
logger.debug("status code: " + str(response.status_code))
if (str(response.status_code) == "200"):
communities_list = response.json()["hits"]["hits"]
for community_object in communities_list:
name = community_object["name"]
if community_name == name:
community_id = community_object["id"]
else:
logger.error("NO community found: " + str(response.json()))
except requests.exceptions.RequestException as e:
logger.error(e)
return community_id
def getAllCommunities(args):
configuration = Configuration(args.confpath, args.debug, args.dryrun, logger)
configuration.parseConf()
logger.info("Start get all communities ...")
accessToken = getAccessTokenWithConfigs(configuration, args)
if accessToken is None:
logger.error("Drafting FAILED. No B2SHARE access token found in users meta data.")
return None
configuration.access_token = accessToken
b2shcl = B2shareClient(configuration)
communities = b2shcl.getAllCommunities()
if communities:
logger.info("All available communities: \n "+str(communities)+" END.")
print("List of communities and their id's: \n"+ pprint.pformat(communities))
else:
logger.error("get all communities FAILED.")
return communities
def collectPIDsForCollection(collectionPath, configuration):
PIDobjectsString = '['
irodsu = IRODSUtils(configuration.irods_home_dir, logger, configuration.irods_debug)
rc, res = irodsu.deepListDir(collectionPath)
if not res:
return None
filePathsMap = None
if res:
filePathsMap = collectFilePathsFromTree(res)
if not filePathsMap:
return None
for filePath in filePathsMap.keys():
filePID = irodsu.getMetadata(filePath, "PID")
if filePID :
# filePath[1:] deletes leading / in a path as requested in issue #112 on GitHub
PIDobject = '{"key":"'+filePath[1:]+'",'+' "ePIC_PID":"'+filePID[0] +'"}'
PIDobjectsString = PIDobjectsString + PIDobject +','
forLastElemIndex = len(PIDobjectsString) - 1 #delete last comma
PIDobjectsString = PIDobjectsString[:forLastElemIndex] + ']'
return PIDobjectsString
def collectFilePathsFromTree(filesTree):
filePaths = {}
for coll in filesTree:
for fp in filesTree[coll]['__files__']:
# loop over the files of the collection
filePaths[coll + os.sep + fp] = fp
if len(filesTree[coll]) > 1:
# there are also subdirs
del filesTree[coll]['__files__']
filemap = collectFilePathsFromTree(filesTree[coll])
# merge the map dictionaries
temp = filemap.copy()
temp.update(filePaths)
filePaths = temp
return filePaths
def addMetadata(args):
configuration = Configuration(args.confpath, args.debug, args.dryrun, logger)
configuration.parseConf()
logger.info("Adding metadata ...")
accessToken = getAccessTokenWithConfigs(configuration, args)
if accessToken is None:
logger.error("Adding metadata FAILED. No B2SHARE access token found in users meta data.")
return None
configuration.access_token = accessToken
irodsu = IRODSUtils(configuration.irods_home_dir, logger, configuration.irods_debug)
metadata_file = irodsu.getFile(args.metadata)
b2shcl = B2shareClient(configuration)
b2shcl.addB2shareMetadata(args.record_id, metadata_file)
logger.info("Added metadata")
def publish(args):
configuration = Configuration(args.confpath, args.debug, args.dryrun, logger)
configuration.parseConf()
logger.info("Publishing ...")
accessToken = getAccessTokenWithConfigs(configuration, args)
if not accessToken:
logger.error("Publishing FAILED. No B2SHARE access token found in users meta data.")
return None
configuration.access_token = accessToken
b2shcl = B2shareClient(configuration)
b2shcl.publishRecord(args.rec_id)
logger.info("Publishing END.")
#get access_token from users metadata in iRODS
def getAccessTokenWithConfigs(configuration, args):
irodsu = IRODSUtils(configuration.irods_home_dir, logger, configuration.irods_debug)
if irodsu:
users_metadata = irodsu.getMetadata(args.userName, "access_token", '-u')
if users_metadata:
return users_metadata[0]
else:
return None
else:
return None
def getCommunitySchema(args):
configuration = Configuration(args.confpath, args.debug, args.dryrun, logger)
configuration.parseConf()
logger.info("Get Community Schema ...")
accessToken = getAccessTokenWithConfigs(configuration, args)
if not accessToken:
logger.error("Get Community Schema FAILED. No B2SHARE access token found in users meta data.")
return None
configuration.access_token = accessToken
commID = args.commID
if commID is None:
commID = getCommunityIDByName(configuration, args.commName)
b2shcl = B2shareClient(configuration)
schema = b2shcl.getCommunitySchema(commID)
logger.info(str(schema))
logger.info("Get Community Schema END.")
return schema
def getDraftByID(args):
configuration = Configuration(args.confpath, args.debug, args.dryrun, logger)
configuration.parseConf()
logger.info("Get draft by ID ...")
accessToken = getAccessTokenWithConfigs(configuration, args)
if not accessToken:
logger.error("Get draft by ID FAILED. No B2SHARE access token found in users meta data.")
return None
configuration.access_token = accessToken
b2shcl = B2shareClient(configuration)
draft = b2shcl.getDraftByID(args.draft_id)
if draft:
logger.info("Request for a draft with id " + draft + " SUCCESSFUL.")
logger.info("Get draft by ID END.")
return draft
def deleteDraft(args):
configuration = Configuration(args.confpath, args.debug, args.dryrun, logger)
configuration.parseConf()
logger.info("DELETING DRAFT: " + args.draft_to_delete_id)
accessToken = getAccessTokenWithConfigs(configuration, args)
if not accessToken:
logger.error("DELETING DRAFT FAILED. No B2SHARE access token found in users meta data.")
return None
configuration.access_token = accessToken
b2shcl = B2shareClient(configuration)
b2shcl.deleteDraft(args.draft_to_delete_id)
logger.info("DELETING DRAFT END.")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='B2SAFE B2SHARE client')
parser.add_argument("--confpath", help="path to the configuration file")
parser.add_argument("-dbg", "--debug", action="store_true",
help="enable debug")
parser.add_argument("-d", "--dryrun", action="store_true",
help="run without performing any real change")
parser.add_argument("-u", "--userName", help="iRODS user name")
subparsers = parser.add_subparsers(help='sub-command help', dest='subcmd')
parser_draft = subparsers.add_parser('draft', help='create a draft record in B2Share')
comm_group = parser_draft.add_mutually_exclusive_group(required=True)
comm_group.add_argument("-c", "--communityName", help="B2Share community name")
comm_group.add_argument("-i", "--communityID", help="B2Share community id")
parser_draft.add_argument('-ti', '--title', help='title of the record')
parser_draft.add_argument('-cp', '--collectionPath', required=True, help='path to the collection in iRODS with files')
parser_draft.set_defaults(func=draft)
parser_meta = subparsers.add_parser('meta', help='add metadata to the draft')
parser_meta.add_argument('-id', '--record_id', required=True, help='the b2share id of the record')
parser_meta.add_argument('-md', '--metadata', required=True, help='path to the metadata JSON file of the record')
parser_meta.set_defaults(func=addMetadata)
parser_pub = subparsers.add_parser('pub', help='publish the draft')
parser_pub.add_argument('-pi', '--rec_id', required=True, help='the b2share id of the record')
parser_pub.set_defaults(func=publish)
parser_list_comm = subparsers.add_parser('listCommunities', help="list all communities with their names and id's")
parser_list_comm.set_defaults(func=getAllCommunities)
parser_commSchema = subparsers.add_parser('communitySchema', help='get community schema')
input_group = parser_commSchema.add_mutually_exclusive_group(required=True)
input_group.add_argument("-cn", "--commName", help="B2Share community name")
input_group.add_argument("-ci", "--commID", help="B2Share community id")
parser_commSchema.set_defaults(func=getCommunitySchema)
parser_getDraft = subparsers.add_parser('getDraft', help='get draft and if there write it to the log file')
parser_getDraft.add_argument('-di', '--draft_id', required=True, help='the b2share id of the record')
parser_getDraft.set_defaults(func=getDraftByID)
parser_deleteDraft = subparsers.add_parser('deleteDraft', help='delete the draft')
parser_deleteDraft.add_argument('-ddi', '--draft_to_delete_id', required=True, help='the b2share id of the record')
parser_deleteDraft.set_defaults(func=deleteDraft)
args = parser.parse_args()
args.func(args)
import unittest
from mock import MagicMock, patch
import json
import os
import sys
sys.path.append("../../cmd")
import b2shareclientCLI as b2shareclientCLI
import logging
test_logger = logging.getLogger('B2shareClientCLITest')
class argsMock():
title = "test"
collectionPath = ""
communityID = "EUDAT"
confpath = "/home/irods/B2SAFE-core/conf/b2share_connection.conf"
metadata = "/localpath/to/metadata"
record_id = "someID123"
rec_id = "someID123"
userName = "userName"
commID = "someCommunityID123"
commName = "Aalto"
draft_id = "someID123"
draft_to_delete_id = "someID123"
debug = True
dryrun = False
class B2shareClientCLITest(unittest.TestCase):
def setUp(self):
self.configurationMock = MagicMock(confpath = "/home/irods/B2SAFE-core/conf/b2share_connection.conf",
b2share_host_name = 'https://trng-b2share.eudat.eu/api',
list_communities_endpoint = '/communities',
access_parameter = '/?access_token=',
access_token = 'JFSUcYzTRpMdVCEyAk3mmtGcfpmH',
irods_home_dir = '',
irods_debug = True,
dryrun = False,
debug = True,
logger = test_logger)
self.collectionPathMock = "/local/path/to/B2SAFE/collection"
self.fileTreeMock = {'/JULK_ZONE/home/irods/julia/collection_A/subCollection_C/collection_A1': {'__files__': ['copytest.txt', 'test1.txt']}}
self.b2shcl = MagicMock()
pass
def tearDown(self):
pass
def testDraft(self):
with patch('configuration.Configuration.parseConf'):
with patch('b2shareclientCLI.getAccessTokenWithConfigs') as b2shareclientCLIMock:
with patch('b2shareclientCLI.collectPIDsForCollection') as b2shareclientCLIMock2:
with patch('b2shareclient.B2shareClient.createDraft') as b2shareclientMock:
b2shareclientCLIMock.return_value = "tocken_mock"
b2shareclientCLIMock2.return_value = [{"key":"BasZone/home/julia/testcollection/collection_A/metatest2.txt", "ePIC_PID":"20.500.11946/619416e2-e0d8-11e7-8ca4-0050569e7581"}]
b2shareclientMock.return_value = "ID_mock"
record_id = b2shareclientCLI.draft(argsMock)
self.assertEqual(record_id, "ID_mock", "testing creating draft ok")
def testGetCommunityIDByName(self):
mock_get_patcher = patch('b2shareclientCLI.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
community_id = b2shareclientCLI.getCommunityIDByName(self.configurationMock, "Aalto")
mock_get_patcher.stop()
self.assertIsNotNone(community_id)
def testGetAllCommunities(self):
with patch('configuration.Configuration.parseConf'):
with patch('b2shareclientCLI.getAccessTokenWithConfigs') as CLIMock:
with patch('b2shareclient.B2shareClient.getAllCommunities') as clientMock:
CLIMock.return_value = "tocken_mock"
clientMock.return_value = "communities_list"
communities = b2shareclientCLI.getAllCommunities(argsMock)
self.assertTrue(bool(communities))
@patch("configuration.Configuration")
def testAddMetadata(self, config):
with patch('b2shareclient.B2shareClient.addB2shareMetadata'):
with patch('b2shareclientCLI.getAccessTokenWithConfigs') as b2shareclientCLIMock:
with patch('manifest.IRODSUtils.getFile') as file_from_IRODSUtils_Mock:
config.irods_home_dir = "/username/home/"
config.irods_debug = True
config.parseConf.return_value = None
b2shareclientCLIMock.return_value = "tocken_mock"
file_from_IRODSUtils_Mock.return_value = "## coment \n\n [required] \n\n community \n EUDAT \n\n [optional]\n\n"
b2shareclientCLI.addMetadata(argsMock)
self.assertTrue(True)
def testPublish(self):
with patch('configuration.Configuration.parseConf'):
with patch('b2shareclientCLI.getAccessTokenWithConfigs') as CLIMock:
with patch('b2shareclient.B2shareClient.publishRecord'):
CLIMock.return_value = "tocken_mock"
b2shareclientCLI.publish(argsMock)
self.assertTrue(True)
def testGetAccessTokenWithConfigs(self):
with patch('manifest.IRODSUtils.getMetadata') as getMetadata_mock:
getMetadata_mock.return_value = ["tocken_mock"]
access_tocken = b2shareclientCLI.getAccessTokenWithConfigs(self.configurationMock, argsMock)
self.assertEqual(access_tocken, "tocken_mock", "access token is not as expected")
def testGetCommunitySchema(self):
with patch('configuration.Configuration.parseConf'):
with patch('b2shareclientCLI.getAccessTokenWithConfigs') as CLI_getTocken_Mock:
with patch('b2shareclientCLI.getCommunityIDByName') as CLI_getComID_Mock:
with patch('b2shareclient.B2shareClient.getCommunitySchema') as clientMock:
CLI_getTocken_Mock.return_value = "tocken_mock"
CLI_getComID_Mock.return_value = "comID"
clientMock.return_value = "community schema"
schema = b2shareclientCLI.getCommunitySchema(argsMock)
self.assertTrue(bool(schema))
argsMock.commID = None
schema = b2shareclientCLI.getCommunitySchema(argsMock)
self.assertTrue(bool(schema))
argsMock.commID = "someCommunityID123"
def testGetDraftByID(self):
with patch('configuration.Configuration.parseConf'):
with patch('b2shareclientCLI.getAccessTokenWithConfigs') as CLI_getTocken_Mock:
with patch('b2shareclient.B2shareClient.getDraftByID') as clientMock:
CLI_getTocken_Mock.return_value = "tocken_mock"
clientMock.return_value = "draft"
draft = b2shareclientCLI.getDraftByID(argsMock)
self.assertTrue(bool(draft))
def testDeleteDraft(self):
with patch('configuration.Configuration.parseConf'):
with patch('b2shareclientCLI.getAccessTokenWithConfigs') as CLI_getTocken_Mock:
with patch('b2shareclient.B2shareClient.deleteDraft'):
CLI_getTocken_Mock.return_value = "tocken_mock"
b2shareclientCLI.deleteDraft(argsMock)
self.assertTrue(True)
def testCollectPIDsForCollection(self):
with patch('manifest.IRODSUtils.deepListDir') as listDir_mock:
with patch('manifest.IRODSUtils.getMetadata') as metadata_mock:
with patch('b2shareclientCLI.collectFilePathsFromTree') as b2shareclientCLI_mock:
listDir_mock.return_value = ('', self.fileTreeMock)
metadata_mock.return_value = ["PID_mock"]
b2shareclientCLI_mock.return_value = {'/JULK_ZONE/home/irods/julia/collection_A/subCollection_C/EUDAT_manifest_METS.xml': 'EUDAT_manifest_METS.xml',
'/JULK_ZONE/home/irods/julia/collection_A/b2share_metadata.json': 'b2share_metadata.json'}
PIDobjectsString = b2shareclientCLI.collectPIDsForCollection(self.collectionPathMock, self.configurationMock)
expectedResultStr = '[{"key":"JULK_ZONE/home/irods/julia/collection_A/b2share_metadata.json", "ePIC_PID":"PID_mock"},{"key":"JULK_ZONE/home/irods/julia/collection_A/subCollection_C/EUDAT_manifest_METS.xml", "ePIC_PID":"PID_mock"}]'
self.assertEqual(expectedResultStr, PIDobjectsString)
def testCollectFilePathsFromTree(self):
filePaths = b2shareclientCLI.collectFilePathsFromTree(self.fileTreeMock)
expectedResult = {'/JULK_ZONE/home/irods/julia/collection_A/subCollection_C/collection_A1'+ os.sep +'copytest.txt': 'copytest.txt', '/JULK_ZONE/home/irods/julia/collection_A/subCollection_C/collection_A1'+ os.sep +'test1.txt': 'test1.txt'}
self.assertEqual(expectedResult, filePaths)
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