Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use create_result method in Slash
Best Python code snippet using slash
03_user_resource.py
Source:03_user_resource.py 
1#!/usr/share/ucs-test/runner /usr/bin/pytest-3 -l -v2## -*- coding: utf-8 -*-3## desc: test operations on user resource4## tags: [ucs_school_kelvin]5## exposure: dangerous6## packages: []7## bugs: []8from __future__ import unicode_literals9import logging10import random11import time12from multiprocessing import Pool13import pytest14import requests15from ucsschool.lib.models.group import SchoolClass16from ucsschool.lib.models.user import Staff as LibStaff, User as LibUser17from univention.testing.ucsschool.kelvin_api import (18 RESOURCE_URLS,19 api_call,20 create_remote_static,21 partial_update_remote_static,22)23from univention.testing.utils import wait_for_listener_replication, wait_for_s4connector_replication24try:25 from urlparse import urljoin # py226except ImportError:27 from urllib.parse import urljoin # py328logger = logging.getLogger("univention.testing.ucsschool")29def get_class_dn(class_name, school, lo):30 # copied from models.user as static version31 school_class = SchoolClass.cache(class_name, school)32 if school_class.get_relative_name() == school_class.name:33 if not school_class.exists(lo):34 class_name = "%s-%s" % (school, class_name)35 school_class = SchoolClass.cache(class_name, school)36 return school_class.dn37@pytest.fixture38def extract_class_dns(lo):39 def _func(attrs):40 school_class_objs = []41 for school, school_classes in attrs.get("school_classes", {}).items():42 school_class_objs.extend(SchoolClass.cache(sc, school) for sc in school_classes)43 return [get_class_dn(sc.name, sc.school, lo) for sc in school_class_objs]44 return _func45def test_list_resource_from_external(auth_header, lo):46 response = requests.get(RESOURCE_URLS["users"], headers=auth_header, params={"school": "DEMOSCHOOL"})47 res = response.json()48 assert isinstance(res, list), repr(res)49 assert isinstance(res[0], dict), repr(res)50 assert "name" in res[0], repr(res)51 assert "firstname" in res[0], repr(res)52 all_usernames_in_response = [user["name"] for user in res]53 for lib_user in LibUser.get_all(lo, "DEMOSCHOOL"):54 assert lib_user.name in all_usernames_in_response55def test_create_user_parallel_from_external_different_classes(56 auth_header,57 compare_import_user_and_resource,58 extract_class_dns,59 get_import_user,60 import_config,61 lo,62 make_user_attrs,63 schoolenv,64 setup_import_config,65 ucr,66):67 parallelism = 2068 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])69 ous = sorted([ou_name, ou_name2])70 logger.info("*** Using OUs %r and %r, parallelism=%d.", ous[0], ous[1], parallelism)71 attrs = [make_user_attrs(ous) for _i in range(parallelism)]72 for _attr in attrs:73 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(_attr))74 pool = Pool(processes=parallelism)75 job_args = [(auth_header, attr) for attr in attrs]76 t0 = time.time()77 map_async_result = pool.map_async(create_remote_static, job_args)78 results = map_async_result.get()79 t1 = time.time()80 logger.info("***** got %d results in %d seconds", len(results), t1 - t0)81 logger.debug("***** results=%r", results)82 errors = []83 for r in results:84 try:85 schoolenv.udm._cleanup.setdefault("users/user", []).append(r["dn"])86 except KeyError:87 # continue to collect user DNs, so we can cleanup as much as possible88 errors.append("Result without DN: {!r}.".format(r))89 assert not errors, " ".join(errors)90 wait_for_listener_replication()91 wait_for_s4connector_replication()92 for num, result in enumerate(results, start=1):93 logger.info("*** Checking result %d/%d (%r)...", num, parallelism, result["name"])94 user = get_import_user(result["dn"])95 compare_import_user_and_resource(user, result)96 logger.info("*** OK: LDAP <-> resource")97 # now compare with attrs98 for attr in attrs:99 if attr["name"] == user.name:100 break101 else:102 raise AssertionError("Could not find user with name {!r} in attrs.".format(user.name))103 import_user_cls = user.__class__104 user2 = import_user_cls(**attr)105 user2.disabled = "1" if attr["disabled"] else "0"106 user2.password = ""107 user2.roles = user.roles108 user2.school = ous[0]109 user2.schools = ous110 user2.ucsschool_roles = user.ucsschool_roles # not in attr111 # add mapped udm_properties not in attr112 mup = import_config["mapped_udm_properties"]113 user2.udm_properties.update(114 dict((prop, None) for prop in mup if prop not in user2.udm_properties)115 )116 compare_import_user_and_resource(user2, result, "ATTR")117 logger.info("*** OK: attr <-> resource")118def test_create_user_parallel_from_external_same_classes(119 auth_header,120 compare_import_user_and_resource,121 extract_class_dns,122 get_import_user,123 import_config,124 lo,125 make_user_attrs,126 schoolenv,127 setup_import_config,128 ucr,129):130 parallelism = 20131 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])132 ous = sorted([ou_name, ou_name2])133 ou1, ou2 = ous134 logger.info("*** Using OUs %r and %r, parallelism=%d.", ou1, ou2, parallelism)135 attrs = [make_user_attrs(ous) for _i in range(parallelism)]136 for _attr in attrs:137 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(_attr))138 # put everyone (except staff) into same classes139 everyone_classes = {}140 for attr in attrs:141 if attr["school_classes"]:142 everyone_classes = attr["school_classes"]143 # TODO: create bug report for this, or handle in API server:144 # work around school.lib failing when trying to create same class (and share) in two145 # processes146 group_dns = extract_class_dns(attr)147 for group_dn in group_dns:148 logger.debug("*** Creating group %r...", group_dn)149 LibUser.get_or_create_group_udm_object(group_dn, lo)150 break151 for attr in attrs:152 if not (len(attr["roles"]) == 1 and attr["roles"][0].endswith("/staff")):153 # don't set school_classes on staff154 attr["school_classes"] = everyone_classes155 pool = Pool(processes=parallelism)156 job_args = [(auth_header, attr) for attr in attrs]157 t0 = time.time()158 map_async_result = pool.map_async(create_remote_static, job_args)159 results = map_async_result.get()160 t1 = time.time()161 logger.info("***** got %d results in %d seconds", len(results), t1 - t0)162 logger.debug("***** results=%r", results)163 errors = []164 for r in results:165 try:166 schoolenv.udm._cleanup.setdefault("users/user", []).append(r["dn"])167 except KeyError:168 # continue to collect user DNs, so we can cleanup as much as possible169 errors.append("Result without DN: {!r}.".format(r))170 assert not errors, " ".join(errors)171 wait_for_listener_replication()172 wait_for_s4connector_replication()173 for num, result in enumerate(results, start=1):174 logger.info("*** Checking result %d/%d (%r)...", num, parallelism, result["name"])175 user = get_import_user(result["dn"])176 compare_import_user_and_resource(user, result)177 logger.info("*** OK: LDAP <-> resource")178 # now compare with attrs179 for attr in attrs:180 if attr["name"] == user.name:181 break182 else:183 raise AssertionError("Could not find user with name {!r} in attrs.".format(user.name))184 import_user_cls = user.__class__185 user2 = import_user_cls(**attr)186 user2.disabled = "1" if attr["disabled"] else "0"187 user2.password = ""188 user2.roles = user.roles189 user2.school = ou1190 user2.schools = ous191 user2.ucsschool_roles = user.ucsschool_roles # not in attr192 # add mapped udm_properties not in attr193 mup = import_config["mapped_udm_properties"]194 user2.udm_properties.update(195 dict((prop, None) for prop in mup if prop not in user2.udm_properties)196 )197 compare_import_user_and_resource(user2, result, "ATTR")198 logger.info("*** OK: attr <-> resource")199def test_partial_update_user_parallel_from_external_different_classes(200 auth_header,201 compare_import_user_and_resource,202 create_import_user,203 extract_class_dns,204 get_import_user,205 import_config,206 make_user_attrs,207 schoolenv,208 setup_import_config,209 ucr,210):211 parallelism = 20212 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])213 ous = sorted([ou_name, ou_name2])214 ou1, ou2 = ous215 logger.info("*** Using OUs %r and %r, parallelism=%d.", ou1, ou2, parallelism)216 # create users sequentially using Python interface217 jobs = []218 for _i in range(parallelism):219 create_attrs = make_user_attrs(ous, school=ou1, schools=ous) # overwrite URLs220 del create_attrs["roles"]221 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(create_attrs))222 user_obj = create_import_user(**create_attrs)223 logger.info("*** Created: %r", user_obj.to_dict())224 assert create_attrs["disabled"] == user_obj.disabled225 roles = tuple(user_obj.roles)226 if roles == ("pupil",):227 roles = ("student",)228 attrs_new = make_user_attrs(229 ous,230 partial=True,231 name=user_obj.name,232 roles=roles,233 source_uid=user_obj.source_uid,234 record_uid=user_obj.record_uid,235 disabled=create_attrs["disabled"], # TODO: changing 'disabled' in parallel fails236 )237 if isinstance(user_obj, LibStaff):238 create_attrs["school_classes"] = {}239 if "school_classes" in attrs_new:240 attrs_new["school_classes"] = {}241 logger.info("*** attrs_new=%r", attrs_new)242 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(attrs_new))243 jobs.append((create_attrs, attrs_new))244 # modify users in parallel using HTTP245 wait_for_listener_replication()246 wait_for_s4connector_replication()247 pool = Pool(processes=parallelism)248 t0 = time.time()249 map_async_result = pool.map_async(250 partial_update_remote_static, [(auth_header, job[0]["name"], job[1]) for job in jobs]251 )252 results = map_async_result.get()253 t1 = time.time()254 logger.info("***** got %d results in %d seconds", len(results), t1 - t0)255 logger.debug("***** results=%r", results)256 wait_for_listener_replication()257 wait_for_s4connector_replication()258 for num, result in enumerate(results, start=1):259 logger.info("*** Checking result %d/%d (%r)...", num, parallelism, result.get("name", "N/A"))260 logger.debug("***** result=%r", result)261 user = get_import_user(result["dn"]) # this will fail, when an error was reported262 compare_import_user_and_resource(user, result)263 logger.info("*** OK: LDAP <-> resource")264 # now compare with attrs265 for job in jobs:266 if job[0]["name"] == user.name:267 attr, new_attrs = job268 for k, v in new_attrs.items():269 if k == "school_classes" and not v:270 # special case `school_classes`: if newly empty but previously271 # non-empty -> use old value272 # see end of ImportUser.make_classes()273 # Bug #48045274 continue275 attr[k] = v276 break277 else:278 raise AssertionError("Could not find user with name {!r} in jobs.".format(user.name))279 import_user_cls = user.__class__280 logger.debug("***** initializing %s 'user2' from attr=%r", user.__class__.__name__, attr)281 user2 = import_user_cls(**attr)282 user2.disabled = "1" if attr["disabled"] else "0"283 logger.debug(284 "####### attr['disabled']=%r -> %r (%r)", attr["disabled"], user2.disabled, user2.name285 )286 user2.password = ""287 user2.roles = user.roles288 user2.school = user2.school.split("/")[-1] # URL 2 OU289 user2.schools = [ou.split("/")[-1] for ou in user2.schools] # URLs 2 OUs290 user2.ucsschool_roles = user.ucsschool_roles # not in attr291 # add mapped udm_properties not in attr292 mup = import_config["mapped_udm_properties"]293 user2.udm_properties.update(294 dict((prop, None) for prop in mup if prop not in user2.udm_properties)295 )296 compare_import_user_and_resource(user2, result, "ATTR")297 logger.info("*** OK: attr <-> resource")298def test_rename_single_user(299 auth_header,300 compare_import_user_and_resource,301 create_import_user,302 extract_class_dns,303 get_import_user,304 make_user_attrs,305 random_username,306 schoolenv,307 setup_import_config,308 ucr,309):310 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])311 ous = sorted([ou_name, ou_name2])312 ou1, ou2 = ous313 logger.info("*** Using OUs %r and %r.", ou1, ou2)314 name_old = random_username()315 logger.info("*** creating user with username %r", name_old)316 create_attrs = make_user_attrs(317 [ou1],318 school=ou1, # overwrite URLs319 schools=[ou1], # overwrite URLs320 partial=False,321 name=name_old,322 )323 del create_attrs["roles"]324 logger.info("*** create_attrs=%r", create_attrs)325 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(create_attrs))326 old_user_obj = create_import_user(**create_attrs)327 logger.info("*** API call (create) returned: %r", old_user_obj)328 name_new = random_username()329 logger.info("*** renaming user from %r to %r", name_old, name_new)330 schoolenv.udm._cleanup.setdefault("users/user", []).append(331 old_user_obj.dn.replace(name_old, name_new)332 )333 modify_attrs = {"name": name_new}334 logger.info("*** modify_attrs=%r", modify_attrs)335 resource_new = partial_update_remote_static((auth_header, name_old, modify_attrs))336 logger.info("*** API call (modify) returned: %r", resource_new)337 assert name_new == resource_new["name"]338 user = get_import_user(resource_new["dn"])339 assert name_new == user.name340 url = urljoin(RESOURCE_URLS["users"], name_new)341 assert resource_new["url"] == url342 resource_new2 = api_call("get", url, headers=auth_header)343 assert resource_new == resource_new2344 url = urljoin(RESOURCE_URLS["users"], name_old)345 response = requests.get(url, headers=auth_header)346 assert response.status_code == 404347 compare_import_user_and_resource(user, resource_new)348 logger.info("*** OK: LDAP <-> resource")349def test_create_user_without_name(350 auth_header,351 compare_import_user_and_resource,352 extract_class_dns,353 get_import_user,354 make_user_attrs,355 schoolenv,356 setup_import_config,357 ucr,358):359 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])360 ous = sorted([ou_name, ou_name2])361 ou1, ou2 = ous362 logger.info("*** Using OUs %r and %r.", ou1, ou2)363 attrs = make_user_attrs(ous)364 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(attrs))365 del attrs["name"]366 logger.debug("*** attrs=%r", attrs)367 expected_username = "test.{}.{}".format(attrs["firstname"][:2], attrs["lastname"][:3]).lower()368 result = create_remote_static((auth_header, attrs))369 assert result["name"] == expected_username370 user = get_import_user(result["dn"])371 assert user.name == expected_username372 compare_import_user_and_resource(user, result)373 logger.info("*** OK: LDAP <-> resource")374def test_move_teacher_one_school_only(375 auth_header,376 compare_import_user_and_resource,377 extract_class_dns,378 get_import_user,379 lo,380 make_user_attrs,381 random_username,382 schoolenv,383 ucr,384):385 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])386 ous = sorted([ou_name, ou_name2])387 ou1, ou2 = ous388 logger.info("*** Using OUs %r and %r.", ou1, ou2)389 logger.info("*** Going to move teacher from OU %r to %r ***", ou1, ou2)390 create_attrs = make_user_attrs([ou1], partial=False, roles=("teacher",))391 logger.info("*** create_attrs=%r", create_attrs)392 assert create_attrs["school"].strip("/").split("/")[-1] == ou1393 assert create_attrs["schools"] == [create_attrs["school"]]394 assert list(create_attrs["school_classes"].keys()) == [ou1]395 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(create_attrs))396 create_result = create_remote_static((auth_header, create_attrs))397 logger.debug("*** create_result=%r", create_result)398 assert create_result["name"] == create_attrs["name"]399 assert create_result["school"].strip("/").split("/")[-1] == ou1400 assert create_result["schools"] == [create_result["school"]]401 assert create_result["school_classes"] == create_attrs["school_classes"]402 user_old = get_import_user(create_result["dn"])403 logger.debug("*** user_old.school_classes=%r", user_old.school_classes)404 user_old_udm = user_old.get_udm_object(lo)405 old_groups = user_old_udm["groups"]406 logger.info("*** old_groups=%r", old_groups)407 for grp in old_groups:408 assert ou1 in grp409 assert ou2 not in grp410 modify_attrs = {411 "school": create_result["school"].replace(ou1, ou2),412 "schools": [create_result["school"].replace(ou1, ou2)],413 "school_classes": {ou2: sorted([random_username(4), random_username(4)])},414 }415 logger.info("*** modify_attrs=%r", modify_attrs)416 resource_new = partial_update_remote_static((auth_header, create_result["name"], modify_attrs))417 logger.info("*** API call (modify) returned: %r", resource_new)418 assert create_result["name"] == resource_new["name"]419 assert modify_attrs["school_classes"] == resource_new["school_classes"]420 user = get_import_user(resource_new["dn"])421 logger.debug("*** user.school_classes=%r", user.school_classes)422 assert create_result["name"] == user.name423 url = urljoin(RESOURCE_URLS["users"], create_result["name"])424 assert resource_new["url"] == url425 resource_new2 = api_call("get", url, headers=auth_header)426 assert resource_new == resource_new2427 compare_import_user_and_resource(user, resource_new)428 logger.info("*** OK: LDAP <-> resource")429 user_new_udm = user.get_udm_object(lo)430 new_groups = user_new_udm["groups"]431 logger.info("*** new_groups=%r", new_groups)432 for grp in new_groups:433 assert ou2 in grp434 assert ou1 not in grp435def test_move_teacher_remove_primary(436 auth_header,437 compare_import_user_and_resource,438 extract_class_dns,439 get_import_user,440 lo,441 make_user_attrs,442 random_username,443 schoolenv,444 ucr,445):446 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])447 ous = sorted([ou_name, ou_name2])448 ou1, ou2 = ous449 logger.info("*** Using OUs %r and %r.", ou1, ou2)450 logger.info(451 "*** Going to create teacher in OUs %r and %r, then remove it from primary (%r). ***",452 ou1,453 ou2,454 ou1,455 )456 create_attrs = make_user_attrs(ous, partial=False, roles=("teacher",))457 logger.info("*** create_attrs=%r", create_attrs)458 assert create_attrs["school"].strip("/").split("/")[-1] == ou1459 assert [460 create_attrs["schools"][0].strip("/").split("/")[-1],461 create_attrs["schools"][1].strip("/").split("/")[-1],462 ] == ous463 assert set(create_attrs["school_classes"].keys()) == {ou1, ou2}464 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(create_attrs))465 create_result = create_remote_static((auth_header, create_attrs))466 logger.debug("*** create_result=%r", create_result)467 assert create_result["name"] == create_attrs["name"]468 assert create_result["school"] == create_attrs["school"]469 assert set(create_result["schools"]) == set(create_attrs["schools"])470 assert create_result["school_classes"] == create_attrs["school_classes"]471 user_old = get_import_user(create_result["dn"])472 logger.debug("*** user_old.school_classes=%r", user_old.school_classes)473 user_old_udm = user_old.get_udm_object(lo)474 old_groups = user_old_udm["groups"]475 logger.info("*** old_groups=%r", old_groups)476 for grp_name in (477 "cn=lehrer-{0},cn=groups,ou={0},".format(ou1),478 "cn=lehrer-{0},cn=groups,ou={0},".format(ou2),479 "cn=Domain Users {0},cn=groups,ou={0},".format(ou1),480 "cn=Domain Users {0},cn=groups,ou={0},".format(ou2),481 ):482 assert any(dn.startswith(grp_name) for dn in old_groups)483 create_attrs_school_classes = dict(484 (ou, ["{}-{}".format(ou, k) for k in kls]) for ou, kls in create_attrs["school_classes"].items()485 )486 logger.info("*** user_old.school_classes =%r", user_old.school_classes)487 logger.info("*** create_attrs_school_classes=%r", create_attrs_school_classes)488 assert user_old.school_classes == create_attrs_school_classes489 modify_attrs = {490 "school": create_result["school"].replace(ou1, ou2),491 "schools": [create_result["school"].replace(ou1, ou2)],492 "school_classes": {ou2: sorted([random_username(4), random_username(4)])},493 }494 logger.info("*** modify_attrs=%r", modify_attrs)495 resource_new = partial_update_remote_static((auth_header, create_result["name"], modify_attrs))496 logger.info("*** API call (modify) returned: %r", resource_new)497 assert create_result["name"] == resource_new["name"]498 assert modify_attrs["school_classes"] == resource_new["school_classes"]499 logger.debug("*** zzz...")500 time.sleep(10)501 user = get_import_user(resource_new["dn"])502 logger.debug("*** user.school_classes=%r", user.school_classes)503 assert create_result["name"] == user.name504 url = urljoin(RESOURCE_URLS["users"], create_result["name"])505 assert resource_new["url"] == url506 resource_new2 = api_call("get", url, headers=auth_header)507 assert resource_new == resource_new2508 compare_import_user_and_resource(user, resource_new)509 logger.info("*** OK: LDAP <-> resource")510 user_new_udm = user.get_udm_object(lo)511 new_groups = user_new_udm["groups"]512 logger.info("*** new_groups=%r", new_groups)513 for grp in new_groups:514 assert ou2 in grp515 assert ou1 not in grp516def test_move_teacher_remove_primary_with_classes(517 auth_header,518 compare_import_user_and_resource,519 extract_class_dns,520 get_import_user,521 lo,522 make_user_attrs,523 schoolenv,524 ucr,525):526 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])527 ous = sorted([ou_name, ou_name2])528 ou1, ou2 = ous529 logger.info("*** Using OUs %r and %r.", ou1, ou2)530 logger.info(531 "*** Going to create teacher in OUs %r and %r, then remove it from primary (%r). ***",532 ou1,533 ou2,534 ou1,535 )536 create_attrs = make_user_attrs(ous, partial=False, roles=("teacher",))537 logger.info("*** create_attrs=%r", create_attrs)538 assert create_attrs["school"].strip("/").split("/")[-1] == ou1539 assert [540 create_attrs["schools"][0].strip("/").split("/")[-1],541 create_attrs["schools"][1].strip("/").split("/")[-1],542 ] == ous543 assert set(create_attrs["school_classes"].keys()) == {ou1, ou2}544 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(create_attrs))545 create_result = create_remote_static((auth_header, create_attrs))546 logger.debug("*** create_result=%r", create_result)547 assert create_result["name"] == create_attrs["name"]548 assert create_result["school"] == create_attrs["school"]549 assert set(create_result["schools"]) == set(create_attrs["schools"])550 assert create_result["school_classes"] == create_attrs["school_classes"]551 user_old = get_import_user(create_result["dn"])552 logger.debug("*** user_old.school_classes=%r", user_old.school_classes)553 user_old_udm = user_old.get_udm_object(lo)554 old_groups = user_old_udm["groups"]555 logger.info("*** old_groups=%r", old_groups)556 for grp_name in (557 "cn=lehrer-{0},cn=groups,ou={0},".format(ou1),558 "cn=lehrer-{0},cn=groups,ou={0},".format(ou2),559 "cn=Domain Users {0},cn=groups,ou={0},".format(ou1),560 "cn=Domain Users {0},cn=groups,ou={0},".format(ou2),561 ):562 assert any(dn.startswith(grp_name) for dn in old_groups)563 create_attrs_school_classes = dict(564 (ou, ["{}-{}".format(ou, k) for k in kls]) for ou, kls in create_attrs["school_classes"].items()565 )566 logger.info("*** user_old.school_classes =%r", user_old.school_classes)567 logger.info("*** create_attrs_school_classes=%r", create_attrs_school_classes)568 assert user_old.school_classes == create_attrs_school_classes569 modify_attrs = {570 "school": create_result["school"].replace(ou1, ou2),571 "schools": [create_result["school"].replace(ou1, ou2)],572 }573 logger.info("*** modify_attrs=%r", modify_attrs)574 logger.debug("*** zzz...")575 time.sleep(5)576 resource_new = partial_update_remote_static((auth_header, create_result["name"], modify_attrs))577 logger.info("*** API call (modify) returned: %r", resource_new)578 assert create_result["name"] == resource_new["name"]579 assert resource_new["school_classes"] == {ou2: create_attrs["school_classes"][ou2]}580 logger.debug("*** zzz...")581 time.sleep(5)582 user = get_import_user(resource_new["dn"])583 logger.debug("*** user.school_classes=%r", user.school_classes)584 assert user.school_classes == {585 ou2: ["{}-{}".format(ou2, k) for k in create_attrs["school_classes"][ou2]]586 }587 assert create_result["name"] == user.name588 url = urljoin(RESOURCE_URLS["users"], create_result["name"])589 assert resource_new["url"] == url590 resource_new2 = api_call("get", url, headers=auth_header)591 assert resource_new == resource_new2592 compare_import_user_and_resource(user, resource_new)593 logger.info("*** OK: LDAP <-> resource")594 user_new_udm = user.get_udm_object(lo)595 new_groups = user_new_udm["groups"]596 logger.info("*** new_groups=%r", new_groups)597 for grp in new_groups:598 assert ou2 in grp599 assert ou1 not in grp600def test_move_teacher_remove_primary_no_classes_in_new_school(601 auth_header,602 compare_import_user_and_resource,603 extract_class_dns,604 get_import_user,605 lo,606 make_user_attrs,607 schoolenv,608 setup_import_config,609 ucr,610):611 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])612 ous = sorted([ou_name, ou_name2])613 ou1, ou2 = ous614 logger.info("*** Using OUs %r and %r.", ou1, ou2)615 logger.info(616 "*** Going to create teacher in OUs %r and %r, then remove it from primary (%r). ***",617 ou1,618 ou2,619 ou1,620 )621 create_attrs = make_user_attrs(ous, partial=False, roles=("teacher",))622 del create_attrs["school_classes"][ou2]623 logger.info("*** create_attrs=%r", create_attrs)624 assert create_attrs["school"].strip("/").split("/")[-1] == ou1625 assert [626 create_attrs["schools"][0].strip("/").split("/")[-1],627 create_attrs["schools"][1].strip("/").split("/")[-1],628 ] == ous629 assert list(create_attrs["school_classes"].keys()) == [ou1]630 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(create_attrs))631 create_result = create_remote_static((auth_header, create_attrs))632 logger.debug("*** create_result=%r", create_result)633 assert create_result["name"] == create_attrs["name"]634 assert create_result["school"] == create_attrs["school"]635 assert set(create_result["schools"]) == set(create_attrs["schools"])636 assert create_result["school_classes"] == create_attrs["school_classes"]637 user_old = get_import_user(create_result["dn"])638 logger.debug("*** user_old.school_classes=%r", user_old.school_classes)639 user_old_udm = user_old.get_udm_object(lo)640 old_groups = user_old_udm["groups"]641 logger.info("*** old_groups=%r", old_groups)642 for grp_name in (643 "cn=lehrer-{0},cn=groups,ou={0},".format(ou1),644 "cn=lehrer-{0},cn=groups,ou={0},".format(ou1),645 "cn=Domain Users {0},cn=groups,ou={0},".format(ou1),646 "cn=Domain Users {0},cn=groups,ou={0},".format(ou2),647 ):648 assert any(dn.startswith(grp_name) for dn in old_groups)649 create_attrs_school_classes = dict(650 (ou, ["{}-{}".format(ou, k) for k in kls]) for ou, kls in create_attrs["school_classes"].items()651 )652 logger.info("*** user_old.school_classes =%r", user_old.school_classes)653 logger.info("*** create_attrs_school_classes=%r", create_attrs_school_classes)654 assert user_old.school_classes == create_attrs_school_classes655 modify_attrs = {656 "school": create_result["school"].replace(ou1, ou2),657 "schools": [create_result["school"].replace(ou1, ou2)],658 }659 logger.info("*** modify_attrs=%r", modify_attrs)660 logger.debug("*** zzz...")661 time.sleep(5)662 resource_new = partial_update_remote_static((auth_header, create_result["name"], modify_attrs))663 logger.info("*** API call (modify) returned: %r", resource_new)664 assert create_result["name"] == resource_new["name"]665 assert resource_new["school_classes"] == {}666 logger.debug("*** zzz...")667 time.sleep(5)668 user = get_import_user(resource_new["dn"])669 logger.debug("*** user.school_classes=%r", user.school_classes)670 assert user.school_classes == {}671 assert create_result["name"] == user.name672 url = urljoin(RESOURCE_URLS["users"], create_result["name"])673 assert resource_new["url"] == url674 resource_new2 = api_call("get", url, headers=auth_header)675 assert resource_new == resource_new2676 compare_import_user_and_resource(user, resource_new)677 logger.info("*** OK: LDAP <-> resource")678 user_new_udm = user.get_udm_object(lo)679 new_groups = user_new_udm["groups"]680 logger.info("*** new_groups=%r", new_groups)681 for grp in new_groups:682 assert ou2 in grp683 assert ou1 not in grp684def test_move_teacher_remove_primary_with_classes_and_rename(685 auth_header,686 compare_import_user_and_resource,687 extract_class_dns,688 get_import_user,689 lo,690 make_user_attrs,691 random_username,692 schoolenv,693 ucr,694):695 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])696 ous = sorted([ou_name, ou_name2])697 ou1, ou2 = ous698 logger.info("*** Using OUs %r and %r.", ou1, ou2)699 logger.info(700 "*** Going to create teacher in OUs %r and %r, then remove it from primary (%r) and rename "701 "it. ***",702 ou1,703 ou2,704 ou1,705 )706 create_attrs = make_user_attrs(ous, partial=False, roles=("teacher",))707 logger.info("*** create_attrs=%r", create_attrs)708 assert create_attrs["school"].strip("/").split("/")[-1] == ou1709 assert [710 create_attrs["schools"][0].strip("/").split("/")[-1],711 create_attrs["schools"][1].strip("/").split("/")[-1],712 ] == ous713 assert set(create_attrs["school_classes"].keys()) == {ou1, ou2}714 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(create_attrs))715 create_result = create_remote_static((auth_header, create_attrs))716 logger.debug("*** create_result=%r", create_result)717 assert create_result["name"] == create_attrs["name"]718 assert create_result["school"] == create_attrs["school"]719 assert set(create_result["schools"]) == set(create_attrs["schools"])720 assert create_result["school_classes"] == create_attrs["school_classes"]721 user_old = get_import_user(create_result["dn"])722 logger.debug("*** user_old.school_classes=%r", user_old.school_classes)723 user_old_udm = user_old.get_udm_object(lo)724 old_groups = user_old_udm["groups"]725 logger.info("*** old_groups=%r", old_groups)726 for grp_name in (727 "cn=lehrer-{0},cn=groups,ou={0},".format(ou1),728 "cn=lehrer-{0},cn=groups,ou={0},".format(ou2),729 "cn=Domain Users {0},cn=groups,ou={0},".format(ou1),730 "cn=Domain Users {0},cn=groups,ou={0},".format(ou2),731 ):732 assert any(dn.startswith(grp_name) for dn in old_groups)733 create_attrs_school_classes = dict(734 (ou, ["{}-{}".format(ou, k) for k in kls]) for ou, kls in create_attrs["school_classes"].items()735 )736 logger.info("*** user_old.school_classes =%r", user_old.school_classes)737 logger.info("*** create_attrs_school_classes=%r", create_attrs_school_classes)738 assert user_old.school_classes == create_attrs_school_classes739 modify_attrs = {740 "name": random_username(),741 "school": create_result["school"].replace(ou1, ou2),742 "schools": [create_result["school"].replace(ou1, ou2)],743 }744 assert user_old.name != modify_attrs["name"]745 logger.info("*** modify_attrs=%r", modify_attrs)746 logger.debug("*** zzz...")747 time.sleep(5)748 resource_new = partial_update_remote_static((auth_header, create_result["name"], modify_attrs))749 logger.info("*** API call (modify) returned: %r", resource_new)750 assert modify_attrs["name"] == resource_new["name"]751 assert resource_new["school_classes"] == {ou2: create_attrs["school_classes"][ou2]}752 logger.debug("*** zzz...")753 time.sleep(5)754 user = get_import_user(resource_new["dn"])755 logger.debug("*** user.school_classes=%r", user.school_classes)756 assert user.school_classes == {757 ou2: ["{}-{}".format(ou2, k) for k in create_attrs["school_classes"][ou2]]758 }759 assert modify_attrs["name"] == user.name760 url = urljoin(RESOURCE_URLS["users"], modify_attrs["name"])761 assert resource_new["url"] == url762 resource_new2 = api_call("get", url, headers=auth_header)763 assert resource_new == resource_new2764 compare_import_user_and_resource(user, resource_new)765 logger.info("*** OK: LDAP <-> resource")766 user_new_udm = user.get_udm_object(lo)767 new_groups = user_new_udm["groups"]768 logger.info("*** new_groups=%r", new_groups)769 for grp in new_groups:770 assert ou2 in grp771 assert ou1 not in grp772def test_modify_teacher_remove_all_classes(773 auth_header,774 compare_import_user_and_resource,775 extract_class_dns,776 get_import_user,777 lo,778 make_user_attrs,779 schoolenv,780 ucr,781):782 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])783 ou = random.choice([ou_name, ou_name2])784 logger.info("*** Going to create teacher in OU %r, then remove all its classes. ***", ou)785 create_attrs = make_user_attrs([ou], partial=False, roles=("teacher",))786 logger.info("*** create_attrs=%r", create_attrs)787 assert list(create_attrs["school_classes"].keys()) == [ou]788 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(create_attrs))789 create_result = create_remote_static((auth_header, create_attrs))790 logger.debug("*** create_result=%r", create_result)791 assert create_result["name"] == create_attrs["name"]792 assert create_result["school"] == create_attrs["school"]793 assert set(create_result["schools"]) == set(create_attrs["schools"])794 assert create_result["school_classes"] == create_attrs["school_classes"]795 user_old = get_import_user(create_result["dn"])796 logger.debug("*** user_old.school_classes=%r", user_old.school_classes)797 user_old_udm = user_old.get_udm_object(lo)798 old_groups = user_old_udm["groups"]799 logger.info("*** old_groups=%r", old_groups)800 for grp_name in (801 "cn=lehrer-{0},cn=groups,ou={0},".format(ou),802 "cn=Domain Users {0},cn=groups,ou={0},".format(ou),803 ):804 assert any(dn.startswith(grp_name) for dn in old_groups)805 create_attrs_school_classes = dict(806 (ou, ["{}-{}".format(ou, k) for k in kls]) for ou, kls in create_attrs["school_classes"].items()807 )808 logger.info("*** user_old.school_classes =%r", user_old.school_classes)809 logger.info("*** create_attrs_school_classes=%r", create_attrs_school_classes)810 assert user_old.school_classes == create_attrs_school_classes811 modify_attrs = {812 "school_classes": {},813 }814 logger.info("*** modify_attrs=%r", modify_attrs)815 logger.debug("*** zzz...")816 time.sleep(5)817 resource_new = partial_update_remote_static((auth_header, create_result["name"], modify_attrs))818 logger.info("*** API call (modify) returned: %r", resource_new)819 assert create_result["name"] == resource_new["name"]820 assert resource_new["school_classes"] == {}821 logger.debug("*** zzz...")822 time.sleep(5)823 user = get_import_user(resource_new["dn"])824 logger.debug("*** user.school_classes=%r", user.school_classes)825 assert user.school_classes == {}826 assert create_result["name"] == user.name827 url = urljoin(RESOURCE_URLS["users"], create_result["name"])828 assert resource_new["url"] == url829 resource_new2 = api_call("get", url, headers=auth_header)830 assert resource_new == resource_new2831 compare_import_user_and_resource(user, resource_new)832 logger.info("*** OK: LDAP <-> resource")833def test_modify_classes_2old_2new(834 auth_header,835 extract_class_dns,836 get_import_user,837 make_user_attrs,838 random_username,839 schoolenv,840 ucr,841):842 role = random.choice(("student", "teacher"))843 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])844 ou = random.choice([ou_name, ou_name2])845 logger.info("*** Going to create %s in OU %r. ***", role, ou)846 create_attrs = make_user_attrs([ou], partial=False, roles=(role,))847 logger.info("*** create_attrs=%r", create_attrs)848 assert list(create_attrs["school_classes"].keys()) == [ou]849 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(create_attrs))850 create_result = create_remote_static((auth_header, create_attrs))851 logger.debug("*** create_result=%r", create_result)852 assert create_result["name"] == create_attrs["name"]853 assert create_result["school"] == create_attrs["school"]854 assert set(create_result["schools"]) == set(create_attrs["schools"])855 assert create_result["school_classes"] == create_attrs["school_classes"]856 user_old = get_import_user(create_result["dn"])857 old_school_classes = user_old.school_classes858 logger.debug("*** old_school_classes=%r", old_school_classes)859 new_school_classes = dict(860 (ou, sorted([random_username(4), random_username(4)])) for ou in old_school_classes.keys()861 )862 logger.debug("*** new_school_classes=%r", new_school_classes)863 modify_attrs = {864 "school_classes": new_school_classes,865 }866 logger.info("*** modify_attrs=%r", modify_attrs)867 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(modify_attrs))868 resource_new = partial_update_remote_static((auth_header, create_result["name"], modify_attrs))869 logger.info("*** API call (modify) returned: %r", resource_new)870 assert create_result["name"] == resource_new["name"]871 assert resource_new["school_classes"] == new_school_classes872 user = get_import_user(resource_new["dn"])873 assert create_result["name"] == user.name874 logger.debug("*** user.school_classes=%r", user.school_classes)875 classes_without_ous = dict(876 (ou, [k.split("-", 1)[1] for k in kls]) for ou, kls in user.school_classes.items()877 )878 logger.debug("*** user.school_classes without ous=%r", classes_without_ous)879 assert classes_without_ous == new_school_classes880 logger.info("*** OK: 2 classes in old and 2 changed classes in new")881def test_modify_classes_0old_2new(882 auth_header,883 extract_class_dns,884 get_import_user,885 make_user_attrs,886 random_username,887 schoolenv,888 ucr,889):890 role = random.choice(("student", "teacher"))891 (ou_name, ou_dn), (ou_name2, ou_dn2) = schoolenv.create_multiple_ous(2, name_edudc=ucr["hostname"])892 ou = random.choice([ou_name, ou_name2])893 logger.info("*** Going to create %s in OU %r. ***", role, ou)894 create_attrs = make_user_attrs([ou], partial=False, roles=(role,), school_classes={})895 logger.info("*** create_attrs=%r", create_attrs)896 assert create_attrs["school_classes"] == {}897 create_result = create_remote_static((auth_header, create_attrs))898 logger.debug("*** create_result=%r", create_result)899 assert create_result["name"] == create_attrs["name"]900 assert create_result["school"] == create_attrs["school"]901 assert set(create_result["schools"]) == set(create_attrs["schools"])902 assert create_result["school_classes"] == {}903 user_old = get_import_user(create_result["dn"])904 old_school_classes = user_old.school_classes905 logger.debug("*** old_school_classes=%r", old_school_classes)906 assert old_school_classes == {}907 new_school_classes = dict(908 (ou, sorted([random_username(4), random_username(4)])) for ou in old_school_classes.keys()909 )910 logger.debug("*** new_school_classes=%r", new_school_classes)911 modify_attrs = {912 "school_classes": new_school_classes,913 }914 logger.info("*** modify_attrs=%r", modify_attrs)915 schoolenv.udm._cleanup.setdefault("groups/group", []).extend(extract_class_dns(modify_attrs))916 resource_new = partial_update_remote_static((auth_header, create_result["name"], modify_attrs))917 logger.info("*** API call (modify) returned: %r", resource_new)918 assert create_result["name"] == resource_new["name"]919 assert resource_new["school_classes"] == new_school_classes920 user = get_import_user(resource_new["dn"])921 assert create_result["name"] == user.name922 logger.debug("*** user.school_classes=%r", user.school_classes)923 classes_without_ous = dict(924 (ou, [k.split("-", 1)[1] for k in kls]) for ou, kls in user.school_classes.items()925 )926 logger.debug("*** user.school_classes without ous=%r", classes_without_ous)927 assert classes_without_ous == new_school_classes...
test_engines.py
Source:test_engines.py
1"""This unit test suite tests the application's engines."""2import os3from operator import itemgetter4import pytest5import app.engines6from app.messages import ErrorMessage, InfoMessage7@pytest.fixture8def default_args():9 """Contains the default engine test arguments"""10 template_name = 'flask-shell'11 template_folder = '~/templates'12 success_status = {'is_successful': True}13 failure_status = {'is_successful': False}14 dest_path = '~/Desktop'15 clone_name = 'test'16 dest = '~/Home'17 return dict(template_name=template_name, template_folder=template_folder,18 success_status=success_status, failure_status=failure_status,19 dest_path=dest_path, clone_name=clone_name, dest=dest)20@pytest.mark.engine21def test_create_existing_template_if_not_forced(mocker, default_args):22 """calls ErrorMessage with template_exists error_type"""23 template_name, template_folder, success_status = itemgetter(24 'template_folder', 'template_folder', 'success_status')(default_args)25 force_delete = False26 mocker.patch('os.mkdir', return_value=True)27 mocker.patch('app.engines.get_template', return_value=True)28 mocker.patch('app.engines.delete_template', return_value=success_status)29 create_result = app.engines.create_template(30 'src', template_name,31 {'type': 'file', 'execute':lambda src, dest: src},32 force_delete, template_folder)33 assert create_result['is_successful'] is False34 assert create_result['msg'] == ErrorMessage(35 'template_exists', template_name=template_name).get_message()36@pytest.mark.engine37def test_create_existing_template_with_force_flag(mocker, default_args):38 """calls delete_template"""39 template_name, template_folder, success_status = itemgetter(40 'template_folder', 'template_folder', 'success_status')(default_args)41 force_delete = True42 mocker.patch('os.mkdir', return_value=True)43 mocker.patch('app.engines.get_template', return_value=True)44 mocker.patch('app.engines.delete_template', return_value=success_status)45 app.engines.create_template(46 'src', template_name,47 {'type': 'file', 'execute':lambda src, dest: success_status},48 force_delete, template_folder)49 app.engines.delete_template.assert_called_once_with(50 template_name, template_folder)51@pytest.mark.engine52def test_create_if_delete_template_throws_error(mocker, default_args):53 """calls ErrorMessage with the delete_template error type"""54 template_name, template_folder, success_status, failure_status = \55 itemgetter('template_folder', 'template_folder',56 'success_status', 'failure_status')(default_args)57 force_delete = True58 mocker.patch('os.mkdir', return_value=True)59 mocker.patch('app.engines.get_template', return_value=True)60 mocker.patch('app.engines.delete_template', return_value=failure_status)61 create_result = app.engines.create_template(62 'src', template_name,63 {'type': 'file', 'execute':lambda src, dest: success_status},64 force_delete, template_folder)65 assert create_result['msg'] == ErrorMessage(66 'delete_template', template_name=template_name).get_message()67@pytest.mark.engine68def test_create_template_file_creates_a_directory(mocker, default_args):69 """A new directory is created with the template_path and file name"""70 template_name, template_folder, success_status = itemgetter(71 'template_folder', 'template_folder', 'success_status')(default_args)72 joined_path = '~/templates/flask-shell'73 mocker.patch('os.mkdir', return_value=True)74 mocker.patch('app.engines.get_template', return_value=True)75 mocker.patch('app.engines.delete_template', return_value=success_status)76 mocker.patch('os.path.join', return_value=joined_path)77 app.engines.create_template(78 'src', template_name,79 {'type': 'file', 'execute':lambda src, dest: success_status},80 True, template_folder)81 os.mkdir.assert_called_once_with(joined_path)82@pytest.mark.engine83def test_create_template_directory_wont_create_a_directory(84 mocker, default_args):85 """A new directory is not created with the template_path and file name"""86 template_name, template_folder, success_status, failure_status = \87 itemgetter('template_folder', 'template_folder', 'success_status',88 'failure_status')(default_args)89 mocker.patch('os.mkdir', return_value=True)90 mocker.patch('app.engines.get_template', return_value=True)91 mocker.patch('app.engines.delete_template', return_value=success_status)92 app.engines.create_template(93 'src', template_name,94 {'type': 'directory', 'execute':lambda src, dest: failure_status},95 True, template_folder)96 assert os.mkdir.assert_not_called97@pytest.mark.engine98def test_create_template_when_clone_function_errors(mocker, default_args):99 """calls ErrorMessage with create_template error type"""100 template_name, template_folder, success_status, failure_status = \101 itemgetter('template_folder', 'template_folder', 'success_status',102 'failure_status')(default_args)103 joined_path = '~/templates/flask-shell'104 mocker.patch('app.engines.get_template', return_value=True)105 mocker.patch('app.engines.delete_template', return_value=success_status)106 mocker.patch('os.path.join', return_value=joined_path)107 create_result = app.engines.create_template(108 'src', template_name,109 {'type': 'directory', 'execute':lambda src, dest: failure_status},110 True, template_folder)111 assert create_result['msg'] == ErrorMessage(112 'create_template', template_name=template_name).get_message()113@pytest.mark.engine114def test_create_template_with_no_errors(mocker, default_args):115 """calls InfoMessage with the template_created message type"""116 template_name, template_folder, success_status = itemgetter(117 'template_folder', 'template_folder', 'success_status')(default_args)118 joined_path = '~/templates/flask-shell'119 mocker.patch('os.mkdir', return_value=True)120 mocker.patch('app.engines.get_template', return_value=True)121 mocker.patch('app.engines.delete_template', return_value=success_status)122 mocker.patch('os.path.join', return_value=joined_path)123 create_result = app.engines.create_template(124 'src', template_name,125 {'type': 'file', 'execute':lambda src, dest: success_status},126 True, template_folder)127 assert create_result['is_successful'] is True128 assert create_result['msg'] == InfoMessage(129 'template_created', template_name=template_name).get_message()130@pytest.mark.engine131def test_clone_template_when_successful(mocker, default_args):132 """InfoMessage is called with the template_cloned message type"""133 template_name, template_folder, success_status, clone_name, dest = \134 itemgetter(135 'template_folder', 'template_folder', 'success_status',136 'clone_name', 'dest')(default_args)137 clone_name = 'test'138 dest = '~/Home'139 dest_path = os.path.join(dest, clone_name)140 mocker.patch('app.engines.get_template', return_value=True)141 create_result = app.engines.clone_template(142 dest, template_name, clone_name,143 {'type': 'file', 'execute':lambda src, dest: success_status},144 template_folder)145 assert create_result['is_successful'] is True146 assert create_result['msg'] == InfoMessage(147 'template_cloned', path=dest_path, template_name=template_name148 ).get_message()149@pytest.mark.engine150def test_clone_template_with_path_function_error(mocker, default_args):151 """calls ErrorMessage with the clone_template error type"""152 template_name, template_folder, failure_status, clone_name, dest = \153 itemgetter(154 'template_folder', 'template_folder', 'failure_status',155 'clone_name', 'dest')(default_args)156 joined_path = '~/templates/flask-shell'157 mocker.patch('os.mkdir', return_value=True)158 mocker.patch('app.engines.get_template', return_value=True)159 mocker.patch('os.path.join', return_value=joined_path)160 create_result = app.engines.clone_template(161 dest, template_name, clone_name,162 {'type': 'file', 'execute':lambda src, dest: failure_status},163 template_folder)164 assert create_result['is_successful'] is False165 assert create_result['msg'] == ErrorMessage(166 'clone_template', template_name=template_name).get_message()167@pytest.mark.engine168def test_clone_template_does_not_exist(mocker, default_args):169 """calls ErrorMessage with the template_missing error type"""170 template_name, template_folder, clone_name = itemgetter(171 'template_name', 'template_folder', 'clone_name')(default_args)172 mocker.patch('app.engines.get_template', return_value=False)173 create_result = app.engines.clone_template(174 'dest', template_name, clone_name,175 {'type': 'file', 'execute':lambda src, dest: src},176 template_folder)177 assert create_result['is_successful'] is False178 assert create_result['msg'] == ErrorMessage(179 'template_missing', template_name=template_name).get_message()180@pytest.mark.engine181def test_delete_template_when_successful(mocker, default_args):182 """calls InfoMessage with template_deleted message type"""183 template_name, template_folder, success_status = \184 itemgetter(185 'template_name', 'template_folder',186 'success_status')(default_args)187 mocker.patch('app.engines.get_template', return_value=True)188 mocker.patch('app.engines.delete_template', return_value=success_status)189 create_result = app.engines.remove_template(190 template_name, template_folder)191 assert create_result['is_successful'] is True192 assert create_result['msg'] == InfoMessage(193 'template_deleted', template_name=template_name).get_message()194@pytest.mark.engine195def test_delete_when_delete_template_throws_error(mocker, default_args):196 """calls ErrorMessage with delete_template error type"""197 template_name, template_folder, failure_status = \198 itemgetter(199 'template_name', 'template_folder',200 'failure_status')(default_args)201 mocker.patch('app.engines.get_template', return_value=True)202 mocker.patch('app.engines.delete_template', return_value=failure_status)203 create_result = app.engines.remove_template(204 template_name, template_folder)205 assert create_result['is_successful'] is False206 assert create_result['msg'] == ErrorMessage(207 'delete_template', template_name=template_name).get_message()208@pytest.mark.engine209def test_delete_when_template_not_found(mocker, default_args):210 """calls ErrorMessage with template_missing error type"""211 template_name, template_folder = \212 itemgetter('template_name', 'template_folder')(default_args)213 mocker.patch('app.engines.get_template', return_value=False)214 create_result = app.engines.remove_template(215 template_name, template_folder)216 assert create_result['is_successful'] is False217 assert create_result['msg'] == ErrorMessage(...
test_validate_product_manifest.py
Source:test_validate_product_manifest.py
1import os2import json3from elsepdm.tests.test_base_pdm import TestBasePdm4from elsepublic.elsepdm.dto.validate_product_manifest import (5 ValidateProductManifestParams,6 ValidateProductManifestResult,7)8from elsepdm.operations.manifest_validation.validate_product_manifest import ValidateProductManifestOperation9class TestValidateProductManifestOperation(TestBasePdm):10 def setUp(self):11 super().setUp()12 with open(os.path.join(self.fixtures_path, 'test_data.json')) as f:13 self.validate_manifest_dto = ValidateProductManifestParams(14 json_manifest=json.load(f))15 with open(os.path.join(self.fixtures_path, 'test_data_full.json')) as f:16 self.validate_full_manifest_dto = ValidateProductManifestParams(17 json_manifest=json.load(f))18 with open(os.path.join(self.fixtures_path, 'test_data_multiple_configurations.json')) as f:19 self.validate_multiple_configuration_manifest_dto = ValidateProductManifestParams(20 json_manifest=json.load(f))21 with open(os.path.join(self.fixtures_path, 'test_incorrect_data.json')) as f:22 self.validate_product_incorrect_manifest_dto = ValidateProductManifestParams(23 json_manifest=json.load(f))24 with open(os.path.join(self.fixtures_path, 'test_incorrect_multiple_configurations_data.json')) as f:25 self.validate_product_incorrect_multiple_configuration_manifest_dto = ValidateProductManifestParams(26 json_manifest=json.load(f))27 self.val_op = ValidateProductManifestOperation()28 def test_validate_product_manifest(self):29 """30 Test validate product manifest31 """32 create_result = self.val_op(data=self.validate_manifest_dto)33 self.assertIsInstance(create_result, ValidateProductManifestResult)34 self.assertTrue(create_result.is_valid)35 def test_validate_product_full_manifest(self):36 """37 Test validate product full manifest38 """39 create_result = self.val_op(self.validate_full_manifest_dto)40 self.assertIsInstance(create_result, ValidateProductManifestResult)41 self.assertTrue(create_result.is_valid)42 def test_validate_multiple_configuration_product_manifest(self):43 """44 Test validate multiple configuration product manifest45 """46 create_result = self.val_op(self.validate_multiple_configuration_manifest_dto)47 self.assertIsInstance(create_result, ValidateProductManifestResult)48 self.assertTrue(create_result.is_valid)49 def test_validate_product_incorrect_manifest(self):50 """51 Test validate product incorrect manifest52 """53 create_result = self.val_op(self.validate_product_incorrect_manifest_dto)54 self.assertIsInstance(create_result, ValidateProductManifestResult)55 self.assertFalse(create_result.is_valid)56 self.assertEqual(len(create_result.error_list), 6)57 def test_validate_product_incorrect_multiple_configurations_manifest(self):58 """59 Test validate product incorrect multiple configuration manifest60 """61 create_result = self.val_op(self.validate_product_incorrect_multiple_configuration_manifest_dto)62 self.assertIsInstance(create_result, ValidateProductManifestResult)63 self.assertFalse(create_result.is_valid)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Slash automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
