Source code for univention.testing.ucsschool.distribution

"""
**Class Distribution**\n

.. module:: distribution
    :platform: Unix

.. moduleauthor:: Ammar Najjar <najjar@univention.de>
"""
from __future__ import print_function

import os
import time

import six

import univention.testing.strings as uts
import univention.testing.ucr as ucr_test
import univention.testing.utils as utils
from univention.testing.umc import Client


[docs]class Distribution(object): """Contains the needed functionality for Materials distribution. By default the distribution is manual.\n :param school: name of the ou :type school: str :param connection: :type connection: UMC connection object :param ucr: :type ucr: UCR object :param name: name of distribution project to be added later :type name: str :param description: description of distribution project to be added later :type description: str :param sender: name of the creater user (teacher or admin) :type sender: str :param flavor: flavor of the acting user :type flavor: str ('teacher' or 'admin') :param distributeTime: time for automatic distribution :type distributeTime: str ('%I:%M') :param distributionDate: date for automatic distribution :type distributionDate: str ('%Y-%m-%d) :param collectionTime: time for automatic collection :type collectionTime: str ('%I:%M') :param collectionDate: date for automatic collection :type collectionDate: str ('%Y-%m-%d) :param distributeType: type of the distribution :type distributionType: str ('automatic' or 'manual') :param collectionTye: type of the collection :type collectionType: str ('automatic' or 'manual') :param files: names of material files for the distribution project :type files: list of str :param recipients: groups which are included in the distribution project :type recipients: list of group objects """ def __init__( self, school, connection=None, sender=None, flavor=None, ucr=None, description=None, name=None, distributeType="manual", distributeTime=None, distributeDate=None, collectType="manual", collectTime=None, collectDate=None, files=[], recipients=[], ): account = utils.UCSTestDomainAdminCredentials() admin = account.username passwd = account.bindpw self.school = school self.name = name if name else uts.random_string() self.description = description if description else uts.random_string() if distributeTime: self.distributeTime = distributeTime else: self.distributeTime = time.strftime("%I:%M") if distributeDate: self.distributeDate = distributeDate else: self.distributeDate = time.strftime("%Y-%m-%d") self.collectTime = collectTime if collectTime else time.strftime("%I:%M") self.collectDate = collectDate if collectDate else time.strftime("%Y-%m-%d") self.distributeType = distributeType self.collectType = collectType self.filename_encodings = files self.recipients = recipients self.ucr = ucr if ucr else ucr_test.UCSTestConfigRegistry() self.sender = sender if sender else admin self.flavor = flavor if flavor else "admin" if connection: self.client = connection else: self.client = Client(None, admin, passwd) self.distributed_version = 1 @property def files(self): return [file_name for file_name, encoding in self.filename_encodings] @property def files_encoded(self): if six.PY2: return [ file_name.decode(encoding).encode("UTF-8") for file_name, encoding in self.filename_encodings ] return [ file_name.decode(encoding) if isinstance(file_name, bytes) else file_name for file_name, encoding in self.filename_encodings ]
[docs] def query(self, filt="private", pattern=""): """Calles 'distribution/query' :param pattern: the pattern to use in the search :type pattern: str """ flavor = self.flavor param = {"filter": filt, "pattern": pattern} reqResult = self.client.umc_command("distribution/query", param, flavor).result result = [x["name"] for x in reqResult if reqResult is not None] return result
[docs] def get(self): """Calls 'distribute/get'""" name = [self.name] reqResult = self.client.umc_command("distribution/get", name, self.flavor).result return reqResult[0]
[docs] def idir(self, path): """Dir a specific path.\n :param path: wanted path :type path: str :return: list of file names """ files = [] for root, _, filenames in os.walk(path): for f in filenames: files.append(os.path.relpath(os.path.join(root, f), path)) return files
[docs] def genData(self, file_name, content_type, boundary, flavor, override_file_name=None): """Generates data in the form to be sent via http POST request.\n :param file_name: file name to be uploaded :type file_name: str :param content_type: type of the content of the file :type content_type: str ('text/plain',..) :param boundary: the boundary :type boundary: str (-------123091) :param flavor: flavor of the acting user :type flavor: str """ mime_file_name = override_file_name or os.path.basename(file_name) with open(file_name, "r") as f: data = r"""--{0} Content-Disposition: form-data; name="uploadedfile"; filename="{1}" Content-Type: {2} {3} --{0} Content-Disposition: form-data; name="flavor" {4} --{0} Content-Disposition: form-data; name="iframe" false --{0} Content-Disposition: form-data; name="uploadType" html5 --{0}-- """.format( boundary, mime_file_name, content_type, f.read(), flavor ) return data.replace("\n", "\r\n")
[docs] def uploadFile(self, file_name, content_type=None, override_file_name=None): """Uploads a file via http POST request.\n :param file_name: file name to be uploaded :type file_name: str :param content_type: type of the content of the file :type content_type: str ('text/plain',..) """ print("Uploading a file") content_type = content_type or "application/octet-stream" boundary = "---------------------------103454444410473823401882756" data = self.genData( file_name, content_type, boundary, self.flavor, override_file_name=override_file_name ) header_content = {"Content-Type": "multipart/form-data; boundary=%s" % (boundary,)} self.client.request("POST", "upload/distribution/upload", data, headers=header_content).result
[docs] def add(self): """Create files and upload them then add the project, calls: 'distribution/add' """ # creatng and uploading the files content_type = "text/plain" for filename in self.files: with open(filename, "w") as g: g.write("test_content") self.uploadFile(filename, content_type) print("Adding Project %s" % (self.name)) flavor = self.flavor recipients = [] for item in self.recipients: recipients.append(item.dn()) print("recipients=", recipients) param = [ { "object": { "collectDate": self.collectDate, "collectTime": self.collectTime, "collectType": self.collectType, "description": self.description, "distributeDate": self.distributeDate, "distributeTime": self.distributeTime, "distributeType": self.distributeType, "files": self.files_encoded, "name": self.name, "recipients": recipients, }, "options": None, } ] print("param=", param) reqResult = self.client.umc_command("distribution/add", param, flavor).result print("reqResult =", reqResult) if not reqResult[0]["success"]: utils.fail("Unable to add project (%r)" % (param,))
[docs] def check_add(self): """Calls 'distribution/query' and check the existance of the added project """ print("Checking %s addition" % (self.name,)) current = self.query(pattern=self.name) if not (self.name in current): utils.fail("Project %s was not added successfully" % (self.name,))
[docs] def put( self, description=None, distributeType=None, distributeTime=None, distributeDate=None, collectType=None, collectTime=None, collectDate=None, files=[], recipients=[], ): """Modifies the already existing project.\n :param description: description of the project to be added later :type description: str :param distributeTime: time for automatic distribution :type distributeTime: str ('%I:%M') :param distributionDate: date for automatic distribution :type distributionDate: str ('%Y-%m-%d) :param collectionTime: time for automatic collection :type collectionTime: str ('%I:%M') :param collectionDate: date for automatic collection :type collectionDate: str ('%Y-%m-%d) :param distributeType: type of the distribution :type distributionType: str ('automatic' or 'manual') :param collectionTye: type of the collection :type collectionType: str ('automatic' or 'manual') :param files: names of material files for the distribution project :type files: list of str :param recipients: groups which are included in the project :type recipients: list of group objects """ print("Editing Project %s" % (self.name)) description = description if description else self.description if distributeType: distributeType = distributeType else: distributeType = self.distributeType if distributeTime: distributeTime = distributeTime else: distributeTime = self.distributeTime if distributeDate: distributeDate = distributeDate else: distributeDate = self.distributeDate collectType = collectType if collectType else self.collectType collectTime = collectTime if collectTime else self.collectTime collectDate = collectDate if collectDate else self.collectDate files = files if files else self.files recipients = recipients if recipients else self.recipients new_recipients = [] for item in recipients: new_recipients.append(item.dn()) flavor = self.flavor param = [ { "object": { "collectDate": collectDate, "collectTime": collectTime, "collectType": collectType, "description": description, "distributeDate": distributeDate, "distributeTime": distributeTime, "distributeType": distributeType, "files": files, "name": self.name, "recipients": new_recipients, }, "options": None, } ] reqResult = self.client.umc_command("distribution/put", param, flavor).result print("reqResult =", reqResult) if not reqResult[0]["success"]: utils.fail("Unable to edit project with params =(%r)" % (param,)) else: self.description = description self.distributeType = distributeType self.distributeTime = distributeTime self.distributeDate = distributeDate self.collectType = collectType self.collectTime = collectTime self.collectDate = collectDate self.filename_encodings = [(x, "utf8") for x in files] self.recipients = recipients
[docs] def check_put(self, previousGetResult): """Calls 'distribution/get' and check the modified project :param previousGetResult: info from previous get :type previousGetResult: dict check changing sates for distribution and collection """ print("Checking %s modification" % (self.name,)) found = self.get() supposed = { "files": found["files"], "sender": found["sender"], "description": found["description"], "recipients": found["recipients"], "distributeType": found["distributeType"], "__type__": found["__type__"], "collectType": found["collectType"], "name": found["name"], "starttime": found["starttime"], "deadline": found["deadline"], } recips = [{"id": y.dn(), "label": y.name} for y in self.recipients] if self.distributeType != "automatic": sTime = None else: sTime = "%s %s" % (self.distributeDate, self.distributeTime) if self.collectType != "automatic": dTime = None else: dTime = "%s %s" % (self.collectDate, self.collectTime) current = { "files": self.files, "sender": self.sender, "description": self.description, "recipients": recips, "distributeType": self.distributeType, "__type__": "PROJECT", "collectType": self.collectType, "name": self.name, "starttime": sTime, "deadline": dTime, } print("supposed = ", supposed) print("current = ", current) assert supposed == current, "Project %s was not modified successfully,supposed!=current" % ( self.name, ) # check distribute check = "distribution" before_type = previousGetResult["distributeType"] after_type = found["distributeType"] before_time = previousGetResult["starttime"] after_time = found["starttime"] before_atJob = previousGetResult["atJobNumDistribute"] after_atJob = found["atJobNumDistribute"] fail_state = self.put_fail( before_type, after_type, before_time, after_time, before_atJob, after_atJob ) assert not fail_state, "Project %s was not modified successfully, %s: %s -> %s" % ( self.name, check, before_type, after_type, ) # check collect check = "collection" before_type = previousGetResult["collectType"] after_type = found["collectType"] before_time = previousGetResult["deadline"] after_time = found["deadline"] before_atJob = previousGetResult["atJobNumCollect"] after_atJob = found["atJobNumCollect"] fail_state = self.put_fail( before_type, after_type, before_time, after_time, before_atJob, after_atJob ) assert not fail_state, "Project %s was not modified successfully, %s: %s -> %s" % ( self.name, check, before_type, after_type, )
[docs] def put_fail(self, before_type, after_type, before_time, after_time, before_atJob, after_atJob): """Checks if the atjobs are in the expected formats :param before_type: type before using put command :type before_type: str :param after_type: type after using put command :type after_type: str :param before_atJob: atJobNum before using put command :type before_atJob: str or None :param after_atJob: atJobNum after using put command :type after_atJob: str or None :param before_time: time before using put command :type before_time: str :param after_time: time after using put command :type after_time: str """ fail_state = False # manual -> manual # atJobs == don't care if before_type == "manual" and after_type == "manual": pass # manual -> automatic # atJobs don't care -> int if before_type == "manual" and after_type == "automatic": fail_state = not (isinstance(after_atJob, int)) # automatic -> manual # atJobs int -> don't care if before_type == "automatic" and after_type == "manual": fail_state = not (isinstance(before_atJob, int)) # automatic -> automatic # atJobs int1 -> int2 & int1 < int2 if before_type == "automatic" and after_type == "automatic": fail1 = not (isinstance(before_atJob, int) and isinstance(after_atJob, int)) fail2 = not (before_time != after_time and (before_atJob < after_atJob)) fail_state = fail1 or fail2 return fail_state
[docs] def distribute(self): """Calls 'distribution/distribute'""" print("Distributing Project %s" % (self.name)) flavor = self.flavor reqResult = self.client.umc_command("distribution/distribute", [self.name], flavor).result assert reqResult[0]["success"], "Unable to distribute project (%r)" % (self.name,)
[docs] def check_distribute(self, users): """Checks if the distribution was successful by checking the file system.\n :param users: names of users to have the material distributed for :type users: list of str """ print("Checking %s distribution" % (self.name,)) for user in users: path = self.getUserFilesPath(user, "distribute") print("file_path=", path) existingFiles = self.idir(path) print("existingFiles=", existingFiles) files = self.files assert files == existingFiles, "Project files were not distributed for user %s:\n%r!=%r" % ( user, files, existingFiles, )
[docs] def collect(self): """Calls 'distribution/collect'""" print("Collecting Project %s" % (self.name)) flavor = self.flavor reqResult = self.client.umc_command("distribution/collect", [self.name], flavor).result assert reqResult[0]["success"], "Unable to collect project (%r)" % (self.name,)
[docs] def check_collect(self, users): """Checks if the collection was successful by checking the file system.\n :param users: names of users to have the material collected from :type users: list of str """ print("Checking %s collection" % (self.name,)) for user in users: path = self.getUserFilesPath(user, "collect", self.distributed_version) print("file_path=", path) existingFiles = self.idir(path) print("existingFiles=", existingFiles) files = self.files assert files == existingFiles, "Project files were not collected for user %s:\n%r!=%r" % ( user, files, existingFiles, )
[docs] def remove(self): """Calls 'distribution/remove'""" print("Removing Project %s" % (self.name)) flavor = self.flavor param = [{"object": self.name, "options": None}] reqResult = self.client.umc_command("distribution/remove", param, flavor).result assert not reqResult, "Unable to remove project (%r)" % (param,)
[docs] def check_remove(self): """Calls 'distribution/query' and check the existance of the removed project """ print("Checking %s removal" % (self.name,)) current = self.query(pattern=self.name) assert self.name not in current, "Project %s was not removed successfully" % (self.name,)
[docs] def checkFiles(self, files): """Calls 'distribution/checkfiles'""" print("Checking files Project %s" % (self.name)) flavor = self.flavor param = {"project": self.name, "filenames": files} reqResult = self.client.umc_command("distribution/checkfiles", param, flavor).result assert not reqResult, "Unable to chack files for project (%r)" % (param,)
[docs] def adopt(self, project_name): """Calls 'distribute/adopt'""" print("Adopting project", self.name) flavor = self.flavor reqResult = self.client.umc_command("distribution/adopt", [project_name], flavor).result assert not reqResult, "Failed to adopt project (%r)" % (project_name,)
[docs] def check_adopt(self, project_name): print("Checking adopting") q = self.query(pattern=project_name) assert project_name in q, "Project %s was not adopted successfully" % (project_name,)
[docs] def getUserFilesPath(self, user, purpose="distribute", version=1): """Gets the correct files path for a specific user depending on the value of the ucr variable ucsschool/import/roleshare.\n :param user: user name :type user: str :param purpose: either for distribution or collection :type purpose: str ('distribute' or 'collect') """ path = "" self.ucr.load() sender_dir_name = self.ucr.get( "ucsschool/datadistribution/datadir/sender", "Unterrichtsmaterial" ) project_dir_suffix = self.ucr.get( "ucsschool/datadistribution/datadir/sender/project/suffix", "-Ergebnisse" ) recipient_dir_name = self.ucr.get( "ucsschool/datadistribution/datadir/recipient", "Unterrichtsmaterial" ) if purpose == "distribute": path = "/home/{0}/schueler/{1}/{2}/{3}".format( self.school, user, recipient_dir_name, self.name ) elif purpose == "collect": path = "/home/{0}/lehrer/{1}/{2}/{3}{4}/{5}-{6:03d}".format( self.school, self.sender, sender_dir_name, self.name, project_dir_suffix, user, version, ) return path