"""
**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
