forked from indymedia/epicyon
Tests for delegation
parent
958e76ea89
commit
4f6e9eb87f
61
roles.py
61
roles.py
|
@ -61,49 +61,57 @@ def getRoles(baseDir: str,nickname: str,domain: str, \
|
|||
return actorJson['roles'][project]
|
||||
return None
|
||||
|
||||
def outboxDelegate(baseDir: str,messageJson: {},debug: bool) -> None:
|
||||
def outboxDelegate(baseDir: str,messageJson: {},debug: bool) -> bool:
|
||||
"""Handles receiving a delegation request
|
||||
"""
|
||||
if not messageJson.get('type'):
|
||||
return
|
||||
return False
|
||||
if not messageJson['type']=='Delegate':
|
||||
return
|
||||
return False
|
||||
if not messageJson.get('object'):
|
||||
return
|
||||
return False
|
||||
if not isinstance(messageJson['object'], dict):
|
||||
return
|
||||
return False
|
||||
if not messageJson['object'].get('type'):
|
||||
return
|
||||
return False
|
||||
if not messageJson['object']['type']=='Role':
|
||||
return
|
||||
return False
|
||||
if not messageJson['object'].get('object'):
|
||||
return
|
||||
return False
|
||||
if not messageJson['object'].get('actor'):
|
||||
return
|
||||
return False
|
||||
if not isinstance(messageJson['object']['object'], str):
|
||||
return
|
||||
return False
|
||||
if ';' not in messageJson['object']['object']:
|
||||
print('WARN: No ; separator between project and role')
|
||||
return
|
||||
if debug:
|
||||
print('DEBUG: delegate activity arrived in outbox')
|
||||
return False
|
||||
|
||||
delegatorNickname=getNicknameFromActor(messageJson['actor'])
|
||||
domain,port=getDomainFromActor(messageJson['actor'])
|
||||
project=messageJson['object']['object'].split(';')[0].strip()
|
||||
|
||||
# does the delegator have capability to delegate in this project?
|
||||
# instance delegators can delagate to other projects
|
||||
# than their own
|
||||
canDelegate=False
|
||||
delegatorRoles=getRoles(baseDir,delegatorNickname, \
|
||||
domain,project)
|
||||
domain,'instance')
|
||||
if delegatorRoles:
|
||||
if 'delegator' not in delegatorRoles:
|
||||
# instance delegators can delagate to other projects
|
||||
# than their own
|
||||
delegatorRoles=getRoles(baseDir,delegatorNickname, \
|
||||
domain,'instance')
|
||||
if 'delegator' in delegatorRoles:
|
||||
canDelegate=True
|
||||
|
||||
if canDelegate==False:
|
||||
canDelegate=True
|
||||
# non-instance delegators can only delegate within their project
|
||||
delegatorRoles=getRoles(baseDir,delegatorNickname, \
|
||||
domain,project)
|
||||
if delegatorRoles:
|
||||
if 'delegator' not in delegatorRoles:
|
||||
return
|
||||
|
||||
return False
|
||||
else:
|
||||
return False
|
||||
|
||||
if canDelegate==False:
|
||||
return False
|
||||
nickname=getNicknameFromActor(messageJson['object']['actor'])
|
||||
domainFull=domain
|
||||
if port:
|
||||
|
@ -115,10 +123,13 @@ def outboxDelegate(baseDir: str,messageJson: {},debug: bool) -> None:
|
|||
existingRoles=getRoles(baseDir,nickname,domain,project)
|
||||
if existingRoles:
|
||||
if role in existingRoles:
|
||||
print(nickname+'@'+domain+' is already assigned to the role '+role+' within the project '+project)
|
||||
return
|
||||
if debug:
|
||||
print(nickname+'@'+domain+' is already assigned to the role '+role+' within the project '+project)
|
||||
return False
|
||||
setRole(baseDir,nickname,domain,project,role)
|
||||
print(nickname+'@'+domain+' assigned to the role '+role+' within the project '+project)
|
||||
if debug:
|
||||
print(nickname+'@'+domain+' assigned to the role '+role+' within the project '+project)
|
||||
return True
|
||||
|
||||
def sendRoleViaServer(session,delegatorNickname: str,password: str,
|
||||
delegatorDomain: str,delegatorPort: int, \
|
||||
|
|
69
tests.py
69
tests.py
|
@ -45,6 +45,8 @@ from person import setPreferredNickname
|
|||
from person import setBio
|
||||
from person import setSkillLevel
|
||||
from roles import setRole
|
||||
from roles import getRoles
|
||||
from roles import outboxDelegate
|
||||
from auth import createBasicAuthHeader
|
||||
from auth import authorizeBasic
|
||||
from auth import storeBasicCredentials
|
||||
|
@ -935,6 +937,72 @@ def testCreatePerson():
|
|||
os.chdir(currDir)
|
||||
shutil.rmtree(baseDir)
|
||||
|
||||
def testDelegateRoles():
|
||||
print('testDelegateRoles')
|
||||
currDir=os.getcwd()
|
||||
nickname='test382'
|
||||
nicknameDelegated='test383'
|
||||
domain='badgerdomain.com'
|
||||
password='mypass'
|
||||
port=80
|
||||
httpPrefix='https'
|
||||
clientToServer=False
|
||||
useBlurhash=False
|
||||
baseDir=currDir+'/.tests_delegaterole'
|
||||
if os.path.isdir(baseDir):
|
||||
shutil.rmtree(baseDir)
|
||||
os.mkdir(baseDir)
|
||||
os.chdir(baseDir)
|
||||
|
||||
privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(baseDir,nickname,domain,port,httpPrefix,True,password)
|
||||
privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(baseDir,nicknameDelegated,domain,port,httpPrefix,True,'insecure')
|
||||
|
||||
httpPrefix='http'
|
||||
project='artechoke'
|
||||
role='delegator'
|
||||
newRoleJson = {
|
||||
'type': 'Delegate',
|
||||
'actor': httpPrefix+'://'+domain+'/users/'+nickname,
|
||||
'object': {
|
||||
'type': 'Role',
|
||||
'actor': httpPrefix+'://'+domain+'/users/'+nicknameDelegated,
|
||||
'object': project+';'+role,
|
||||
'to': [],
|
||||
'cc': []
|
||||
},
|
||||
'to': [],
|
||||
'cc': []
|
||||
}
|
||||
|
||||
assert outboxDelegate(baseDir,newRoleJson,False)
|
||||
# second time delegation has already happened so should return false
|
||||
assert outboxDelegate(baseDir,newRoleJson,False)==False
|
||||
|
||||
assert '"delegator"' in open(baseDir+'/accounts/'+nickname+'@'+domain+'.json').read()
|
||||
assert '"delegator"' in open(baseDir+'/accounts/'+nicknameDelegated+'@'+domain+'.json').read()
|
||||
|
||||
newRoleJson = {
|
||||
'type': 'Delegate',
|
||||
'actor': httpPrefix+'://'+domain+'/users/'+nicknameDelegated,
|
||||
'object': {
|
||||
'type': 'Role',
|
||||
'actor': httpPrefix+'://'+domain+'/users/'+nickname,
|
||||
'object': 'otherproject;otherrole',
|
||||
'to': [],
|
||||
'cc': []
|
||||
},
|
||||
'to': [],
|
||||
'cc': []
|
||||
}
|
||||
|
||||
# non-delegators cannot assign roles
|
||||
assert outboxDelegate(baseDir,newRoleJson,False)==False
|
||||
assert '"otherrole"' not in open(baseDir+'/accounts/'+nickname+'@'+domain+'.json').read()
|
||||
|
||||
assert False
|
||||
os.chdir(currDir)
|
||||
shutil.rmtree(baseDir)
|
||||
|
||||
def testAuthentication():
|
||||
print('testAuthentication')
|
||||
currDir=os.getcwd()
|
||||
|
@ -1242,4 +1310,5 @@ def runAllTests():
|
|||
testNoOfFollowersOnDomain()
|
||||
testFollows()
|
||||
testGroupFollowers()
|
||||
testDelegateRoles()
|
||||
print('Tests succeeded\n')
|
||||
|
|
Loading…
Reference in New Issue