admin tests
parent
9eb536e17c
commit
a41d563f5f
|
@ -3,7 +3,7 @@ from admin import app
|
||||||
from .keycloak_client import KeycloakClient
|
from .keycloak_client import KeycloakClient
|
||||||
from .moodle import Moodle
|
from .moodle import Moodle
|
||||||
from .nextcloud import Nextcloud
|
from .nextcloud import Nextcloud
|
||||||
from .nextcloud_exc import *
|
from .nextcloud_exc import ProviderItemExists
|
||||||
from .avatars import Avatars
|
from .avatars import Avatars
|
||||||
|
|
||||||
from .helpers import filter_roles_list, filter_roles_listofdicts
|
from .helpers import filter_roles_list, filter_roles_listofdicts
|
||||||
|
@ -89,7 +89,22 @@ class Admin():
|
||||||
'roles':[]}
|
'roles':[]}
|
||||||
log.warning(' SYSTEM READY TO HANDLE CONNECTIONS')
|
log.warning(' SYSTEM READY TO HANDLE CONNECTIONS')
|
||||||
# self.test_cohorts()
|
# self.test_cohorts()
|
||||||
self.delete_all_moodle_cohorts()
|
# self.delete_all_moodle_cohorts()
|
||||||
|
|
||||||
|
########## Testing: get user group
|
||||||
|
# cids=[c['id'] for c in self.moodle.get_cohorts()]
|
||||||
|
# relations=self.moodle.get_cohort_members(cids)
|
||||||
|
# for r in relations:
|
||||||
|
# print(self.get_internalgroup_from_moodlecohortid(r['cohortid'])['path'])
|
||||||
|
|
||||||
|
# def get_internalgroup_from_moodlecohortid(self,cohort_id):
|
||||||
|
# for g in self.internal['groups']:
|
||||||
|
# if not g['moodle']: continue
|
||||||
|
# if g['moodle_id'] == cohort_id: return g
|
||||||
|
# return ''
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
## This function should be moved to postup.py
|
## This function should be moved to postup.py
|
||||||
def default_setup(self):
|
def default_setup(self):
|
||||||
|
@ -104,7 +119,7 @@ class Admin():
|
||||||
self.keycloak.add_group('admin')
|
self.keycloak.add_group('admin')
|
||||||
## Add default admin user to group admin (for nextcloud, just in case we go there)
|
## Add default admin user to group admin (for nextcloud, just in case we go there)
|
||||||
admin_uid=self.keycloak_admin.get_user_id('admin')
|
admin_uid=self.keycloak_admin.get_user_id('admin')
|
||||||
self.keycloak_admin.group_user_add(uid,gid)
|
self.keycloak_admin.group_user_add(admin_uid,gid)
|
||||||
log.warning('KEYCLOAK: OK')
|
log.warning('KEYCLOAK: OK')
|
||||||
except:
|
except:
|
||||||
log.warning('KEYCLOAK: Seems to be there already')
|
log.warning('KEYCLOAK: Seems to be there already')
|
||||||
|
@ -227,32 +242,32 @@ class Admin():
|
||||||
|
|
||||||
|
|
||||||
### testing
|
### testing
|
||||||
def test_cohorts(self):
|
# def test_cohorts(self):
|
||||||
cohorts=self.moodle.get_cohorts()
|
# cohorts=self.moodle.get_cohorts()
|
||||||
|
|
||||||
testc=[c for c in cohorts if c['name'] in ['teacher','/teacher','student','/student']]
|
# testc=[c for c in cohorts if c['name'] in ['teacher','/teacher','student','/student']]
|
||||||
pprint(testc)
|
# pprint(testc)
|
||||||
|
|
||||||
groups=[]
|
# groups=[]
|
||||||
for u in self.internal['users']:
|
# for u in self.internal['users']:
|
||||||
groups=groups+u['keycloak_groups']
|
# groups=groups+u['keycloak_groups']
|
||||||
groups=list(dict.fromkeys(groups))
|
# groups=list(dict.fromkeys(groups))
|
||||||
pprint(groups)
|
# pprint(groups)
|
||||||
pprint([g for g in groups if 'teacher' in g['keycloak_groups']])
|
# pprint([g for g in groups if 'teacher' in g['keycloak_groups']])
|
||||||
exit(1)
|
# exit(1)
|
||||||
total=len(groups)
|
# total=len(groups)
|
||||||
i=0
|
# i=0
|
||||||
for g in groups:
|
# for g in groups:
|
||||||
parts=g.split('/')
|
# parts=g.split('/')
|
||||||
subpath=''
|
# subpath=''
|
||||||
for i in range(1,len(parts)):
|
# for i in range(1,len(parts)):
|
||||||
try:
|
# try:
|
||||||
log.warning(' MOODLE GROUPS: Adding group as cohort ('+str(i)+'/'+str(total)+'): '+subpath)
|
# log.warning(' MOODLE GROUPS: Adding group as cohort ('+str(i)+'/'+str(total)+'): '+subpath)
|
||||||
subpath=subpath+'/'+parts[i]
|
# subpath=subpath+'/'+parts[i]
|
||||||
self.moodle.add_system_cohort(subpath)
|
# self.moodle.add_system_cohort(subpath)
|
||||||
except:
|
# except:
|
||||||
log.error('probably exists')
|
# log.error('probably exists')
|
||||||
i=i+1
|
# i=i+1
|
||||||
|
|
||||||
### end testing
|
### end testing
|
||||||
|
|
||||||
|
@ -263,7 +278,7 @@ class Admin():
|
||||||
return True
|
return True
|
||||||
|
|
||||||
def get_moodle_users(self):
|
def get_moodle_users(self):
|
||||||
return [u for u in self.moodle.get_users_with_groups_and_roles() if u['username'] not in ['guest','ddadmin','admin'] and not u['username'].startswith('system')]
|
return [u for u in self.moodle.get_users_with_groups_and_roles() if u['username'] not in ['guest','ddadmin','admin'] and not u['username'].startswith('system_')]
|
||||||
|
|
||||||
## TOO SLOW. Not used.
|
## TOO SLOW. Not used.
|
||||||
# def get_moodle_users(self):
|
# def get_moodle_users(self):
|
||||||
|
@ -322,7 +337,6 @@ class Admin():
|
||||||
|
|
||||||
def _get_mix_users(self):
|
def _get_mix_users(self):
|
||||||
kusers=self.get_keycloak_users()
|
kusers=self.get_keycloak_users()
|
||||||
# pprint(kusers)
|
|
||||||
musers=self.get_moodle_users()
|
musers=self.get_moodle_users()
|
||||||
nusers=self.get_nextcloud_users()
|
nusers=self.get_nextcloud_users()
|
||||||
|
|
||||||
|
@ -367,6 +381,7 @@ class Admin():
|
||||||
del theuser['groups']
|
del theuser['groups']
|
||||||
users.append(theuser)
|
users.append(theuser)
|
||||||
|
|
||||||
|
# pprint([u['moodle_groups'] for u in users])
|
||||||
return users
|
return users
|
||||||
|
|
||||||
def get_roles(self):
|
def get_roles(self):
|
||||||
|
@ -420,10 +435,10 @@ class Admin():
|
||||||
|
|
||||||
moodle_exists=[g for g in mgroups if g['name'] == name]
|
moodle_exists=[g for g in mgroups if g['name'] == name]
|
||||||
if len(moodle_exists):
|
if len(moodle_exists):
|
||||||
thegroup['path']=''
|
thegroup['path']=moodle_exists[0]['name']
|
||||||
thegroup={**moodle_exists[0], **thegroup}
|
thegroup={**moodle_exists[0], **thegroup}
|
||||||
thegroup['moodle']=True
|
thegroup['moodle']=True
|
||||||
thegroup['moodle_id']=thegroup['id']
|
thegroup['moodle_id']=moodle_exists[0]['id']
|
||||||
else:
|
else:
|
||||||
thegroup['moodle']=False
|
thegroup['moodle']=False
|
||||||
|
|
||||||
|
@ -431,10 +446,10 @@ class Admin():
|
||||||
if len(nextcloud_exists):
|
if len(nextcloud_exists):
|
||||||
nextcloud={"id":nextcloud_exists[0],
|
nextcloud={"id":nextcloud_exists[0],
|
||||||
"name":nextcloud_exists[0],
|
"name":nextcloud_exists[0],
|
||||||
"path":''}
|
"path":nextcloud_exists[0]}
|
||||||
thegroup={**nextcloud, **thegroup}
|
thegroup={**nextcloud, **thegroup}
|
||||||
thegroup['nextcloud']=True
|
thegroup['nextcloud']=True
|
||||||
thegroup['nextcloud_id']=thegroup['id']
|
thegroup['nextcloud_id']=nextcloud_exists[0] ### is the path
|
||||||
else:
|
else:
|
||||||
thegroup['nextcloud']=False
|
thegroup['nextcloud']=False
|
||||||
|
|
||||||
|
@ -573,7 +588,6 @@ class Admin():
|
||||||
log.error(' MOODLE GROUPS: Group '+subpath+ ' probably already exists')
|
log.error(' MOODLE GROUPS: Group '+subpath+ ' probably already exists')
|
||||||
i=i+1
|
i=i+1
|
||||||
|
|
||||||
log.error('getting cohorts')
|
|
||||||
### Get all existing moodle cohorts
|
### Get all existing moodle cohorts
|
||||||
cohorts=self.moodle.get_cohorts()
|
cohorts=self.moodle.get_cohorts()
|
||||||
|
|
||||||
|
@ -582,23 +596,10 @@ class Admin():
|
||||||
if not u['moodle']:
|
if not u['moodle']:
|
||||||
log.error('Creating moodle user: '+u['username'])
|
log.error('Creating moodle user: '+u['username'])
|
||||||
user=self.moodle.create_user(u['email'],u['username'],'1Random 1String',u['first'],u['last'])[0]
|
user=self.moodle.create_user(u['email'],u['username'],'1Random 1String',u['first'],u['last'])[0]
|
||||||
# print(str(user))
|
# user_id=user['id']
|
||||||
user_id=user['id']
|
|
||||||
# [{'id': 5, 'username': 'xkrlzwd'}]
|
|
||||||
|
|
||||||
# for g in u['keycloak_groups']:
|
|
||||||
# log.info('Adding moodle user: '+u['username']+' to cohort '+g)
|
|
||||||
# cohort_id=[c['id'] for c in cohorts if c['name']==g][0]
|
|
||||||
|
|
||||||
# # print(user_id)
|
|
||||||
# # print(cohort_id)
|
|
||||||
# self.moodle.add_user_to_cohort(user_id,cohort_id)
|
|
||||||
|
|
||||||
## Update cohorts on existing
|
|
||||||
# self.resync_data()
|
|
||||||
# for u in self.internal['users']:
|
|
||||||
# if u['moodle']:
|
|
||||||
self.resync_data()
|
self.resync_data()
|
||||||
|
### Add user to their cohorts (groups)
|
||||||
for u in self.internal['users']:
|
for u in self.internal['users']:
|
||||||
total=len(u['keycloak_groups'])
|
total=len(u['keycloak_groups'])
|
||||||
index=0
|
index=0
|
||||||
|
@ -606,33 +607,25 @@ class Admin():
|
||||||
log.error('for user groups')
|
log.error('for user groups')
|
||||||
parts=g.split('/')
|
parts=g.split('/')
|
||||||
subpath=''
|
subpath=''
|
||||||
pprint(parts)
|
# pprint(parts)
|
||||||
for i in range(1,len(parts)):
|
for i in range(1,len(parts)):
|
||||||
if parts[i] in ['admin','manager','teacher','student']:
|
if parts[i] in ['admin','manager','teacher','student']:
|
||||||
subpath=parts[i]
|
subpath=parts[i]
|
||||||
else:
|
else:
|
||||||
subpath=subpath+'/'+parts[i]
|
subpath=subpath+'/'+parts[i]
|
||||||
# log.info('Adding moodle user: '+u['username']+' to cohort '+subpath)
|
|
||||||
try:
|
try:
|
||||||
cohort=[c for c in cohorts if c['name']==subpath][0]
|
cohort=[c for c in cohorts if c['name']==subpath][0]
|
||||||
except:
|
except:
|
||||||
pprint(subpath)
|
# pprint(subpath)
|
||||||
log.error(' MOODLE USER GROUPS: keycloak group does not exist as moodle cohort')
|
log.error(' MOODLE USER GROUPS: keycloak group '+subpath+' does not exist as moodle cohort. This should not happen. User '+u['username']+ ' not added.')
|
||||||
|
|
||||||
try:
|
try:
|
||||||
self.moodle.add_user_to_cohort(u['moodle_id'],cohort['id'])
|
self.moodle.add_user_to_cohort(u['moodle_id'],cohort['id'])
|
||||||
except:
|
except:
|
||||||
log.error(' MOODLE USER GROUPS: User '+u['username']+' already exists in cohort '+cohort['name'])
|
log.error(' MOODLE USER GROUPS: User '+u['username']+' already exists in cohort '+cohort['name'])
|
||||||
index=index+1
|
index=index+1
|
||||||
self.resync_data()
|
self.resync_data()
|
||||||
### MISING ASSING USER TO ROLE COHORT
|
|
||||||
|
|
||||||
# for g in u['keycloak_groups']:
|
|
||||||
# log.info('Adding moodle user: '+u['username']+' to cohort '+g)
|
|
||||||
# cohort_id=[c['id'] for c in cohorts if c['name']==g][0]
|
|
||||||
|
|
||||||
# print(u['moodle_id'])
|
|
||||||
# print('Adding to cohort id '+str(cohort_id)+' with name '+g)
|
|
||||||
# pprint(self.moodle.add_user_to_cohort(u['moodle_id'],cohort_id))
|
|
||||||
|
|
||||||
def delete_all_moodle_cohorts(self):
|
def delete_all_moodle_cohorts(self):
|
||||||
cohorts=self.moodle.get_cohorts()
|
cohorts=self.moodle.get_cohorts()
|
||||||
|
|
|
@ -96,6 +96,7 @@ class Moodle():
|
||||||
(headers,users)=self.moodle_pg.select_with_headers(q)
|
(headers,users)=self.moodle_pg.select_with_headers(q)
|
||||||
users_with_lists = [list(l[:-2])+([[]] if l[-2] == [None] else [list(set(l[-2]))]) + ([[]] if l[-1] == [None] else [list(set(l[-1]))]) for l in users]
|
users_with_lists = [list(l[:-2])+([[]] if l[-2] == [None] else [list(set(l[-2]))]) + ([[]] if l[-1] == [None] else [list(set(l[-1]))]) for l in users]
|
||||||
list_dict_users = [dict(zip(headers, r)) for r in users_with_lists]
|
list_dict_users = [dict(zip(headers, r)) for r in users_with_lists]
|
||||||
|
# pprint(list_dict_users)
|
||||||
return list_dict_users
|
return list_dict_users
|
||||||
|
|
||||||
## NOT USED. Too slow
|
## NOT USED. Too slow
|
||||||
|
@ -141,8 +142,9 @@ class Moodle():
|
||||||
user = self.call('core_cohort_add_cohort_members', members=members)
|
user = self.call('core_cohort_add_cohort_members', members=members)
|
||||||
return user
|
return user
|
||||||
|
|
||||||
def get_cohort_members(self, cohort_id):
|
def get_cohort_members(self, cohort_ids):
|
||||||
members = self.call('core_cohort_get_cohort_members', cohortids=[cohort_id])[0]['userids']
|
members = self.call('core_cohort_get_cohort_members', cohortids=cohort_ids)
|
||||||
|
#[0]['userids']
|
||||||
return members
|
return members
|
||||||
|
|
||||||
def delete_cohorts(self, cohortids):
|
def delete_cohorts(self, cohortids):
|
||||||
|
|
Loading…
Reference in New Issue