#4120

Committed 07 Mar 2024 09:55PM UTC coverage: 90.392% (+0.3%) from 90.142%

Build # #4120

Build Type

push

circleci

Committed by

web-flow

Commit Message

Convert MethodViews to module functions for API requests (#3092)

Run Details

3229 of 3643 branches covered (88.64%)

Branch coverage included in aggregate %.

975 of 1085 new or added lines in 10 files covered. (89.86%)

2 existing lines in 1 file now uncovered.

5652 of 6182 relevant lines covered (91.43%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.26

/src/auslib/web/admin/views/required_signoffs.py

import json

import connexion
from flask import Response, jsonify, request
from sqlalchemy.sql.expression import null

from auslib.db import SignoffRequiredError
from auslib.global_state import dbo
from auslib.web.admin.views.base import handleGeneralExceptions, log, requirelogin, transactionHandler
from auslib.web.admin.views.problem import problem
from auslib.web.admin.views.scheduled_changes import (
    delete_scheduled_change,
    delete_signoffs_scheduled_change,
    get_all_scheduled_change_history,
    get_scheduled_change_history,
    get_scheduled_changes,
    post_enact_scheduled_change,
    post_scheduled_change,
    post_scheduled_change_history,
    post_scheduled_changes,
    post_signoffs_scheduled_change,
)
from auslib.web.common.history import get_input_dict


def get_required_signoffs(required_signoffs, where=None):
    rows = required_signoffs.select(where=where)
    return jsonify({"count": len(rows), "required_signoffs": [dict(rs) for rs in rows]})


def post_required_signoffs(required_signoffs, decisionFields, what, transaction, changed_by):
    where = {f: what.get(f) for f in decisionFields}
    if required_signoffs.select(where=where, transaction=transaction):
        raise SignoffRequiredError("Required Signoffs cannot be directly modified")
    else:
        try:
            required_signoffs.insert(changed_by=changed_by, transaction=transaction, **what)
            return Response(status=201, response=json.dumps({"new_data_version": 1}))
        except ValueError as e:
            log.warning("Bad input: %s", e.args)
            return problem(400, "Bad Request", str(e.args))


def delete_required_signoffs(*args, **kwargs):
    raise SignoffRequiredError("Required Signoffs cannot be directly deleted.")


def _get_filters_rs_history_api(table):
    history_table = table.history
    query = get_input_dict()
    where = [getattr(history_table, f) == query.get(f) for f in query]
    where.append(history_table.data_version != null())
    request = connexion.request
    if hasattr(history_table, "channel"):
        if request.args.get("channel"):
            where.append(history_table.channel == request.args.get("channel"))
    if hasattr(history_table, "product"):
        where.append(history_table.product != null())
        if request.args.get("product"):
            where.append(history_table.product == request.args.get("product"))
    if request.args.get("timestamp_from"):
        where.append(history_table.timestamp >= int(request.args.get("timestamp_from")))
    if request.args.get("timestamp_to"):
        where.append(history_table.timestamp <= int(request.args.get("timestamp_to")))
    return where


def get_rs_revisions(table, decisionFields, input_dict):
    if not table.select({f: input_dict.get(f) for f in decisionFields}):
        return problem(404, "Not Found", "Requested Required Signoff does not exist")

    try:
        page = int(connexion.request.args.get("page", 1))
        limit = int(connexion.request.args.get("limit", 100))
    except ValueError as msg:
        log.warning("Bad input: %s", msg)
        return problem(400, "Bad Request", str(msg))
    offset = limit * (page - 1)

    history_table = table.history
    where_count = [history_table.data_version != null()]
    for field in decisionFields:
        where_count.append(getattr(history_table, field) == input_dict.get(field))
    total_count = history_table.count(where=where_count)

    where = [getattr(history_table, f) == input_dict.get(f) for f in decisionFields]
    where.append(history_table.data_version != null())
    revisions = history_table.select(where=where, limit=limit, offset=offset, order_by=[history_table.timestamp.desc()])

    return jsonify(count=total_count, required_signoffs=revisions)


def get_all_rs_revisions(table):
    try:
        page = int(connexion.request.args.get("page", 1))
        limit = int(connexion.request.args.get("limit", 100))
    except ValueError as msg:
        log.warning("Bad input: %s", msg)
        return problem(400, "Bad Request", str(msg))
    offset = limit * (page - 1)

    where = _get_filters_rs_history_api(table)
    history_table = table.history
    total_count = history_table.count(where=where)
    revisions = history_table.select(where=where, limit=limit, offset=offset, order_by=[history_table.timestamp.desc()])

    return jsonify(count=total_count, required_signoffs=revisions)


def get_product_required_signoffs():
    where = {param: request.args[param] for param in ("product", "channel") if param in request.args}
    return get_required_signoffs(required_signoffs=dbo.productRequiredSignoffs, where=where)


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_product_required_signoffs(signoff, transaction, changed_by):
    what = {
        "product": signoff.get("product"),
        "channel": signoff.get("channel"),
        "role": signoff.get("role"),
        "signoffs_required": int(signoff.get("signoffs_required")),
    }
    return post_required_signoffs(
        required_signoffs=dbo.productRequiredSignoffs, decisionFields=["product", "channel", "role"], what=what, transaction=transaction, changed_by=changed_by
    )


@handleGeneralExceptions
def delete_product_required_signoffs():
    return delete_required_signoffs()


def get_product_rs_revisions():
    input_dict = {
        "product": connexion.request.args.get("product"),
        "role": connexion.request.args.get("role"),
        "channel": connexion.request.args.get("channel"),
    }
    return get_rs_revisions(dbo.productRequiredSignoffs, ["product", "channel", "role"], input_dict)


def get_all_product_rs_revisions():
    return get_all_rs_revisions(dbo.productRequiredSignoffs)


def get_product_rs_scheduled_changes():
    where = {f"base_{param}": request.args[param] for param in ("product", "channel") if param in request.args}
    return get_scheduled_changes(table=dbo.productRequiredSignoffs, where=where)


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_product_rs_scheduled_changes(sc_rs_product_body, transaction, changed_by):
    if sc_rs_product_body.get("when", None) is None:
        return problem(400, "Bad Request", "when cannot be set to null when scheduling a new change " "for a Product Required Signoff")
    change_type = sc_rs_product_body.get("change_type")

    what = {}
    for field in sc_rs_product_body:
        if change_type == "insert" and field == "data_version":
            continue
        what[field] = sc_rs_product_body[field]

    if change_type == "update":
        for field in ["signoffs_required", "data_version"]:
            if not what.get(field, None):
                return problem(400, "Bad Request", "Missing field", ext={"exception": "%s is missing" % field})
            else:
                what[field] = int(what[field])

    elif change_type == "insert":
        if not what.get("signoffs_required", None):
            return problem(400, "Bad Request", "Missing field", ext={"exception": "signoffs_required is missing"})
        else:
            what["signoffs_required"] = int(what["signoffs_required"])

    elif change_type == "delete":
        if not what.get("data_version", None):
            return problem(400, "Bad Request", "Missing field", ext={"exception": "data_version is missing"})
        else:
            what["data_version"] = int(what["data_version"])

    return post_scheduled_changes(
        sc_table=dbo.productRequiredSignoffs.scheduled_changes, what=what, transaction=transaction, changed_by=changed_by, change_type=change_type
    )


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_product_rs_scheduled_change(sc_id, sc_product_rs_body, transaction, changed_by):
    # TODO: modify UI and clients to stop sending 'change_type' in request body
    sc_table = dbo.productRequiredSignoffs.scheduled_changes
    sc_rs_product = sc_table.select(where={"sc_id": sc_id}, transaction=transaction, columns=["change_type"])
    if sc_rs_product:
        change_type = sc_rs_product[0]["change_type"]
    else:
        return problem(404, "Not Found", "Unknown sc_id", ext={"exception": "No scheduled change for product required signoff found for given sc_id"})
    what = {}
    for field in sc_product_rs_body:
        if (
            (change_type == "insert" and field not in ["when", "product", "channel", "role", "signoffs_required"])
            or (change_type == "update" and field not in ["when", "signoffs_required", "data_version"])
            or (change_type == "delete" and field not in ["when", "data_version"])
        ):
            continue

        what[field] = sc_product_rs_body[field]

    if change_type in ["update", "delete"] and not what.get("data_version", None):
        return problem(400, "Bad Request", "Missing field", ext={"exception": "data_version is missing"})

    if what.get("signoffs_required", None):
        what["signoffs_required"] = int(what["signoffs_required"])

    return post_scheduled_change(
        sc_table=sc_table,
        sc_id=sc_id,
        what=what,
        transaction=transaction,
        changed_by=changed_by,
        old_sc_data_version=sc_product_rs_body.get("sc_data_version", None),
    )


@requirelogin
@transactionHandler
@handleGeneralExceptions
def delete_product_rs_scheduled_change(sc_id, data_version, transaction, changed_by):
    return delete_scheduled_change(
        sc_table=dbo.productRequiredSignoffs.scheduled_changes, sc_id=sc_id, data_version=data_version, transaction=transaction, changed_by=changed_by
    )


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_product_rs_enact_scheduled_change(sc_id, transaction, changed_by):
    return post_enact_scheduled_change(sc_table=dbo.productRequiredSignoffs.scheduled_changes, sc_id=sc_id, transaction=transaction, changed_by=changed_by)


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_product_rs_signoffs_scheduled_change(sc_id, sc_post_signoffs_body, transaction, changed_by):
    return post_signoffs_scheduled_change(
        signoffs_table=dbo.productRequiredSignoffs.scheduled_changes.signoffs,
        sc_id=sc_id,
        what=sc_post_signoffs_body,
        transaction=transaction,
        changed_by=changed_by,
    )


@requirelogin
@transactionHandler
@handleGeneralExceptions
def delete_product_rs_signoffs_scheduled_change(sc_id, transaction, changed_by):
    return delete_signoffs_scheduled_change(
        signoffs_table=dbo.productRequiredSignoffs.scheduled_changes.signoffs, sc_id=sc_id, transaction=transaction, changed_by=changed_by
    )


def get_product_rs_scheduled_change_history(sc_id):
    return get_scheduled_change_history(sc_table=dbo.productRequiredSignoffs.scheduled_changes, sc_id=sc_id)


def get_all_product_rs_scheduled_change_history():
    return get_all_scheduled_change_history(sc_table=dbo.productRequiredSignoffs.scheduled_changes)


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_product_rs_scheduled_change_history(sc_id, transaction, changed_by):
    return post_scheduled_change_history(sc_table=dbo.productRequiredSignoffs.scheduled_changes, sc_id=sc_id, transaction=transaction, changed_by=changed_by)


def get_permissions_required_signoffs():
    where = {param: request.args[param] for param in ("product",) if param in request.args}
    return get_required_signoffs(required_signoffs=dbo.permissionsRequiredSignoffs, where=where)


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_permissions_required_signoffs(signoff, transaction, changed_by):
    what = {
        "product": signoff.get("product"),
        "role": signoff.get("role"),
        "signoffs_required": int(signoff.get("signoffs_required")),
    }
    return post_required_signoffs(
        required_signoffs=dbo.permissionsRequiredSignoffs, decisionFields=["product", "role"], what=what, transaction=transaction, changed_by=changed_by
    )


@handleGeneralExceptions
def delete_permissions_required_signoffs():
    return delete_required_signoffs()


def get_permissions_rs_revisions():
    input_dict = {"product": connexion.request.args.get("product"), "role": connexion.request.args.get("role")}
    return get_rs_revisions(dbo.permissionsRequiredSignoffs, ["product", "role"], input_dict)


def get_all_permissions_rs_revisions():
    return get_all_rs_revisions(dbo.permissionsRequiredSignoffs)


def get_permissions_rs_scheduled_changes():
    where = {f"base_{param}": request.args[param] for param in ("product",) if param in request.args}
    return get_scheduled_changes(table=dbo.permissionsRequiredSignoffs, where=where)


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_permissions_rs_scheduled_changes(sc_rs_permission_body, transaction, changed_by):
    if sc_rs_permission_body.get("when", None) is None:
        return problem(400, "Bad Request", "'when' cannot be set to null when scheduling a new change " "for a Permissions Required Signoff")
    change_type = sc_rs_permission_body.get("change_type")

    what = {}
    for field in sc_rs_permission_body:
        if change_type == "insert" and field == "data_version":
            continue
        what[field] = sc_rs_permission_body[field]

    if change_type == "update":
        for field in ["signoffs_required", "data_version"]:
            if not what.get(field, None):
                return problem(400, "Bad Request", "Missing field", ext={"exception": "%s is missing" % field})
            else:
                what[field] = int(what[field])

    elif change_type == "insert":
        if not what.get("signoffs_required", None):
            return problem(400, "Bad Request", "Missing field", ext={"exception": "signoffs_required is missing"})
        else:
            what["signoffs_required"] = int(what["signoffs_required"])

    elif change_type == "delete":
        if not what.get("data_version", None):
            return problem(400, "Bad Request", "Missing field", ext={"exception": "data_version is missing"})
        else:
            what["data_version"] = int(what["data_version"])

    return post_scheduled_changes(
        sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes, what=what, transaction=transaction, changed_by=changed_by, change_type=change_type
    )


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_permissions_rs_scheduled_change(sc_id, sc_permission_rs_body, transaction, changed_by):
    # TODO: modify UI and clients to stop sending 'change_type' in request body
    sc_table = dbo.permissionsRequiredSignoffs.scheduled_changes
    sc_rs_permission = sc_table.select(where={"sc_id": sc_id}, transaction=transaction, columns=["change_type"])
    if sc_rs_permission:
        change_type = sc_rs_permission[0]["change_type"]
    else:
        return problem(404, "Not Found", "Unknown sc_id", ext={"exception": "No scheduled change for permission required " "signoff found for given sc_id"})
    what = {}
    for field in sc_permission_rs_body:
        if (
            (change_type == "insert" and field not in ["when", "product", "role", "signoffs_required"])
            or (change_type == "update" and field not in ["when", "signoffs_required", "data_version"])
            or (change_type == "delete" and field not in ["when", "data_version"])
        ):
            continue
        what[field] = sc_permission_rs_body[field]

    if change_type in ["update", "delete"] and not what.get("data_version", None):
        return problem(400, "Bad Request", "Missing field", ext={"exception": "data_version is missing"})

    if what.get("signoffs_required", None):
        what["signoffs_required"] = int(what["signoffs_required"])

    return post_scheduled_change(
        sc_table=sc_table,
        sc_id=sc_id,
        what=what,
        transaction=transaction,
        changed_by=changed_by,
        old_sc_data_version=sc_permission_rs_body.get("sc_data_version", None),
    )


@requirelogin
@transactionHandler
@handleGeneralExceptions
def delete_permissions_rs_scheduled_change(sc_id, data_version, transaction, changed_by):
    return delete_scheduled_change(
        sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes, sc_id=sc_id, data_version=data_version, transaction=transaction, changed_by=changed_by
    )


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_permissions_rs_enact_scheduled_change(sc_id, transaction, changed_by):
    return post_enact_scheduled_change(sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes, sc_id=sc_id, transaction=transaction, changed_by=changed_by)


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_permissions_rs_signoffs_scheduled_change(sc_id, sc_post_signoffs_body, transaction, changed_by):
    return post_signoffs_scheduled_change(
        signoffs_table=dbo.permissionsRequiredSignoffs.scheduled_changes.signoffs,
        sc_id=sc_id,
        what=sc_post_signoffs_body,
        transaction=transaction,
        changed_by=changed_by,
    )


@requirelogin
@transactionHandler
@handleGeneralExceptions
def delete_permissions_rs_signoffs_scheduled_change(sc_id, transaction, changed_by):
    return delete_signoffs_scheduled_change(
        signoffs_table=dbo.permissionsRequiredSignoffs.scheduled_changes.signoffs, sc_id=sc_id, transaction=transaction, changed_by=changed_by
    )


def get_permissions_rs_scheduled_change_history(sc_id):
    return get_scheduled_change_history(sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes, sc_id=sc_id)


def get_all_permissions_rs_scheduled_change_history():
    return get_all_scheduled_change_history(sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes)


@requirelogin
@transactionHandler
@handleGeneralExceptions
def post_permissions_rs_scheduled_change_history(sc_id, transaction, changed_by):
    return post_scheduled_change_history(
        sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes, sc_id=sc_id, transaction=transaction, changed_by=changed_by
    )

1	import json	1✔
2
3	import connexion	1✔
4	from flask import Response, jsonify, request	1✔
5	from sqlalchemy.sql.expression import null	1✔
6
7	from auslib.db import SignoffRequiredError	1✔
8	from auslib.global_state import dbo	1✔
9	from auslib.web.admin.views.base import handleGeneralExceptions, log, requirelogin, transactionHandler	1✔
10	from auslib.web.admin.views.problem import problem	1✔
11	from auslib.web.admin.views.scheduled_changes import (	1✔
12	delete_scheduled_change,
13	delete_signoffs_scheduled_change,
14	get_all_scheduled_change_history,
15	get_scheduled_change_history,
16	get_scheduled_changes,
17	post_enact_scheduled_change,
18	post_scheduled_change,
19	post_scheduled_change_history,
20	post_scheduled_changes,
21	post_signoffs_scheduled_change,
22	)
23	from auslib.web.common.history import get_input_dict	1✔
24
25
26	def get_required_signoffs(required_signoffs, where=None):	1✔
27	rows = required_signoffs.select(where=where)	1✔
28	return jsonify({"count": len(rows), "required_signoffs": [dict(rs) for rs in rows]})	1✔
29
30
31	def post_required_signoffs(required_signoffs, decisionFields, what, transaction, changed_by):	1✔
32	where = {f: what.get(f) for f in decisionFields}	1✔
33	if required_signoffs.select(where=where, transaction=transaction):	1✔
34	raise SignoffRequiredError("Required Signoffs cannot be directly modified")	1✔
35	else:
36	try:	1✔
37	required_signoffs.insert(changed_by=changed_by, transaction=transaction, **what)	1✔
38	return Response(status=201, response=json.dumps({"new_data_version": 1}))	1✔
39	except ValueError as e:	1✔
40	log.warning("Bad input: %s", e.args)	1✔
41	return problem(400, "Bad Request", str(e.args))	1✔
42
43
44	def delete_required_signoffs(args, *kwargs):	1✔
45	raise SignoffRequiredError("Required Signoffs cannot be directly deleted.")	1✔
46
47
48	def _get_filters_rs_history_api(table):	1✔
49	history_table = table.history	1✔
50	query = get_input_dict()	1✔
51	where = [getattr(history_table, f) == query.get(f) for f in query]	1✔
52	where.append(history_table.data_version != null())	1✔
53	request = connexion.request	1✔
54	if hasattr(history_table, "channel"):	1✔
55	if request.args.get("channel"):	1!
NEW 56	where.append(history_table.channel == request.args.get("channel"))	×
57	if hasattr(history_table, "product"):	1!
58	where.append(history_table.product != null())	1✔
59	if request.args.get("product"):	1!
NEW 60	where.append(history_table.product == request.args.get("product"))	×
61	if request.args.get("timestamp_from"):	1✔
62	where.append(history_table.timestamp >= int(request.args.get("timestamp_from")))	1✔
63	if request.args.get("timestamp_to"):	1✔
64	where.append(history_table.timestamp <= int(request.args.get("timestamp_to")))	1✔
65	return where	1✔
66
67
68	def get_rs_revisions(table, decisionFields, input_dict):	1✔
69	if not table.select({f: input_dict.get(f) for f in decisionFields}):	1!
NEW 70	return problem(404, "Not Found", "Requested Required Signoff does not exist")	×
71
72	try:	1✔
73	page = int(connexion.request.args.get("page", 1))	1✔
74	limit = int(connexion.request.args.get("limit", 100))	1✔
NEW 75	except ValueError as msg:	×
NEW 76	log.warning("Bad input: %s", msg)	×
NEW 77	return problem(400, "Bad Request", str(msg))	×
78	offset = limit * (page - 1)	1✔
79
80	history_table = table.history	1✔
81	where_count = [history_table.data_version != null()]	1✔
82	for field in decisionFields:	1✔
83	where_count.append(getattr(history_table, field) == input_dict.get(field))	1✔
84	total_count = history_table.count(where=where_count)	1✔
85
86	where = [getattr(history_table, f) == input_dict.get(f) for f in decisionFields]	1✔
87	where.append(history_table.data_version != null())	1✔
88	revisions = history_table.select(where=where, limit=limit, offset=offset, order_by=[history_table.timestamp.desc()])	1✔
89
90	return jsonify(count=total_count, required_signoffs=revisions)	1✔
91
92
93	def get_all_rs_revisions(table):	1✔
94	try:	1✔
95	page = int(connexion.request.args.get("page", 1))	1✔
96	limit = int(connexion.request.args.get("limit", 100))	1✔
NEW 97	except ValueError as msg:	×
NEW 98	log.warning("Bad input: %s", msg)	×
NEW 99	return problem(400, "Bad Request", str(msg))	×
100	offset = limit * (page - 1)	1✔
101
102	where = _get_filters_rs_history_api(table)	1✔
103	history_table = table.history	1✔
104	total_count = history_table.count(where=where)	1✔
105	revisions = history_table.select(where=where, limit=limit, offset=offset, order_by=[history_table.timestamp.desc()])	1✔
106
107	return jsonify(count=total_count, required_signoffs=revisions)	1✔
108
109
110	def get_product_required_signoffs():	1✔
111	where = {param: request.args[param] for param in ("product", "channel") if param in request.args}	1✔
112	return get_required_signoffs(required_signoffs=dbo.productRequiredSignoffs, where=where)	1✔
113
114
115	@requirelogin	1✔
116	@transactionHandler	1✔
117	@handleGeneralExceptions	1✔
118	def post_product_required_signoffs(signoff, transaction, changed_by):	1✔
119	what = {	1✔
120	"product": signoff.get("product"),
121	"channel": signoff.get("channel"),
122	"role": signoff.get("role"),
123	"signoffs_required": int(signoff.get("signoffs_required")),
124	}
125	return post_required_signoffs(	1✔
126	required_signoffs=dbo.productRequiredSignoffs, decisionFields=["product", "channel", "role"], what=what, transaction=transaction, changed_by=changed_by
127	)
128
129
130	@handleGeneralExceptions	1✔
131	def delete_product_required_signoffs():	1✔
132	return delete_required_signoffs()	1✔
133
134
135	def get_product_rs_revisions():	1✔
136	input_dict = {	1✔
137	"product": connexion.request.args.get("product"),
138	"role": connexion.request.args.get("role"),
139	"channel": connexion.request.args.get("channel"),
140	}
141	return get_rs_revisions(dbo.productRequiredSignoffs, ["product", "channel", "role"], input_dict)	1✔
142
143
144	def get_all_product_rs_revisions():	1✔
145	return get_all_rs_revisions(dbo.productRequiredSignoffs)	1✔
146
147
148	def get_product_rs_scheduled_changes():	1✔
149	where = {f"base_{param}": request.args[param] for param in ("product", "channel") if param in request.args}	1✔
150	return get_scheduled_changes(table=dbo.productRequiredSignoffs, where=where)	1✔
151
152
153	@requirelogin	1✔
154	@transactionHandler	1✔
155	@handleGeneralExceptions	1✔
156	def post_product_rs_scheduled_changes(sc_rs_product_body, transaction, changed_by):	1✔
157	if sc_rs_product_body.get("when", None) is None:	1!
NEW 158	return problem(400, "Bad Request", "when cannot be set to null when scheduling a new change " "for a Product Required Signoff")	×
159	change_type = sc_rs_product_body.get("change_type")	1✔
160
161	what = {}	1✔
162	for field in sc_rs_product_body:	1✔
163	if change_type == "insert" and field == "data_version":	1!
NEW 164	continue	×
165	what[field] = sc_rs_product_body[field]	1✔
166
167	if change_type == "update":	1✔
168	for field in ["signoffs_required", "data_version"]:	1✔
169	if not what.get(field, None):	1!
NEW 170	return problem(400, "Bad Request", "Missing field", ext={"exception": "%s is missing" % field})	×
171	else:
172	what[field] = int(what[field])	1✔
173
174	elif change_type == "insert":	1✔
175	if not what.get("signoffs_required", None):	1!
NEW 176	return problem(400, "Bad Request", "Missing field", ext={"exception": "signoffs_required is missing"})	×
177	else:
178	what["signoffs_required"] = int(what["signoffs_required"])	1✔
179
180	elif change_type == "delete":	1!
181	if not what.get("data_version", None):	1!
NEW 182	return problem(400, "Bad Request", "Missing field", ext={"exception": "data_version is missing"})	×
183	else:
184	what["data_version"] = int(what["data_version"])	1✔
185
186	return post_scheduled_changes(	1✔
187	sc_table=dbo.productRequiredSignoffs.scheduled_changes, what=what, transaction=transaction, changed_by=changed_by, change_type=change_type
188	)
189
190
191	@requirelogin	1✔
192	@transactionHandler	1✔
193	@handleGeneralExceptions	1✔
194	def post_product_rs_scheduled_change(sc_id, sc_product_rs_body, transaction, changed_by):	1✔
195	# TODO: modify UI and clients to stop sending 'change_type' in request body
196	sc_table = dbo.productRequiredSignoffs.scheduled_changes	1✔
197	sc_rs_product = sc_table.select(where={"sc_id": sc_id}, transaction=transaction, columns=["change_type"])	1✔
198	if sc_rs_product:	1✔
199	change_type = sc_rs_product[0]["change_type"]	1✔
200	else:
201	return problem(404, "Not Found", "Unknown sc_id", ext={"exception": "No scheduled change for product required signoff found for given sc_id"})	1✔
202	what = {}	1✔
203	for field in sc_product_rs_body:	1✔
204	if (	1✔
205	(change_type == "insert" and field not in ["when", "product", "channel", "role", "signoffs_required"])
206	or (change_type == "update" and field not in ["when", "signoffs_required", "data_version"])
207	or (change_type == "delete" and field not in ["when", "data_version"])
208	):
209	continue	1✔
210
211	what[field] = sc_product_rs_body[field]	1✔
212
213	if change_type in ["update", "delete"] and not what.get("data_version", None):	1!
NEW 214	return problem(400, "Bad Request", "Missing field", ext={"exception": "data_version is missing"})	×
215
216	if what.get("signoffs_required", None):	1!
217	what["signoffs_required"] = int(what["signoffs_required"])	1✔
218
219	return post_scheduled_change(	1✔
220	sc_table=sc_table,
221	sc_id=sc_id,
222	what=what,
223	transaction=transaction,
224	changed_by=changed_by,
225	old_sc_data_version=sc_product_rs_body.get("sc_data_version", None),
226	)
227
228
229	@requirelogin	1✔
230	@transactionHandler	1✔
231	@handleGeneralExceptions	1✔
232	def delete_product_rs_scheduled_change(sc_id, data_version, transaction, changed_by):	1✔
233	return delete_scheduled_change(	1✔
234	sc_table=dbo.productRequiredSignoffs.scheduled_changes, sc_id=sc_id, data_version=data_version, transaction=transaction, changed_by=changed_by
235	)
236
237
238	@requirelogin	1✔
239	@transactionHandler	1✔
240	@handleGeneralExceptions	1✔
241	def post_product_rs_enact_scheduled_change(sc_id, transaction, changed_by):	1✔
242	return post_enact_scheduled_change(sc_table=dbo.productRequiredSignoffs.scheduled_changes, sc_id=sc_id, transaction=transaction, changed_by=changed_by)	1✔
243
244
245	@requirelogin	1✔
246	@transactionHandler	1✔
247	@handleGeneralExceptions	1✔
248	def post_product_rs_signoffs_scheduled_change(sc_id, sc_post_signoffs_body, transaction, changed_by):	1✔
249	return post_signoffs_scheduled_change(	1✔
250	signoffs_table=dbo.productRequiredSignoffs.scheduled_changes.signoffs,
251	sc_id=sc_id,
252	what=sc_post_signoffs_body,
253	transaction=transaction,
254	changed_by=changed_by,
255	)
256
257
258	@requirelogin	1✔
259	@transactionHandler	1✔
260	@handleGeneralExceptions	1✔
261	def delete_product_rs_signoffs_scheduled_change(sc_id, transaction, changed_by):	1✔
262	return delete_signoffs_scheduled_change(	1✔
263	signoffs_table=dbo.productRequiredSignoffs.scheduled_changes.signoffs, sc_id=sc_id, transaction=transaction, changed_by=changed_by
264	)
265
266
267	def get_product_rs_scheduled_change_history(sc_id):	1✔
268	return get_scheduled_change_history(sc_table=dbo.productRequiredSignoffs.scheduled_changes, sc_id=sc_id)	1✔
269
270
271	def get_all_product_rs_scheduled_change_history():	1✔
272	return get_all_scheduled_change_history(sc_table=dbo.productRequiredSignoffs.scheduled_changes)	1✔
273
274
275	@requirelogin	1✔
276	@transactionHandler	1✔
277	@handleGeneralExceptions	1✔
278	def post_product_rs_scheduled_change_history(sc_id, transaction, changed_by):	1✔
NEW 279	return post_scheduled_change_history(sc_table=dbo.productRequiredSignoffs.scheduled_changes, sc_id=sc_id, transaction=transaction, changed_by=changed_by)	×
280
281
282	def get_permissions_required_signoffs():	1✔
283	where = {param: request.args[param] for param in ("product",) if param in request.args}	1✔
284	return get_required_signoffs(required_signoffs=dbo.permissionsRequiredSignoffs, where=where)	1✔
285
286
287	@requirelogin	1✔
288	@transactionHandler	1✔
289	@handleGeneralExceptions	1✔
290	def post_permissions_required_signoffs(signoff, transaction, changed_by):	1✔
291	what = {	1✔
292	"product": signoff.get("product"),
293	"role": signoff.get("role"),
294	"signoffs_required": int(signoff.get("signoffs_required")),
295	}
296	return post_required_signoffs(	1✔
297	required_signoffs=dbo.permissionsRequiredSignoffs, decisionFields=["product", "role"], what=what, transaction=transaction, changed_by=changed_by
298	)
299
300
301	@handleGeneralExceptions	1✔
302	def delete_permissions_required_signoffs():	1✔
303	return delete_required_signoffs()	1✔
304
305
306	def get_permissions_rs_revisions():	1✔
307	input_dict = {"product": connexion.request.args.get("product"), "role": connexion.request.args.get("role")}	1✔
308	return get_rs_revisions(dbo.permissionsRequiredSignoffs, ["product", "role"], input_dict)	1✔
309
310
311	def get_all_permissions_rs_revisions():	1✔
312	return get_all_rs_revisions(dbo.permissionsRequiredSignoffs)	1✔
313
314
315	def get_permissions_rs_scheduled_changes():	1✔
316	where = {f"base_{param}": request.args[param] for param in ("product",) if param in request.args}	1✔
317	return get_scheduled_changes(table=dbo.permissionsRequiredSignoffs, where=where)	1✔
318
319
320	@requirelogin	1✔
321	@transactionHandler	1✔
322	@handleGeneralExceptions	1✔
323	def post_permissions_rs_scheduled_changes(sc_rs_permission_body, transaction, changed_by):	1✔
324	if sc_rs_permission_body.get("when", None) is None:	1!
NEW 325	return problem(400, "Bad Request", "'when' cannot be set to null when scheduling a new change " "for a Permissions Required Signoff")	×
326	change_type = sc_rs_permission_body.get("change_type")	1✔
327
328	what = {}	1✔
329	for field in sc_rs_permission_body:	1✔
330	if change_type == "insert" and field == "data_version":	1!
NEW 331	continue	×
332	what[field] = sc_rs_permission_body[field]	1✔
333
334	if change_type == "update":	1✔
335	for field in ["signoffs_required", "data_version"]:	1✔
336	if not what.get(field, None):	1!
NEW 337	return problem(400, "Bad Request", "Missing field", ext={"exception": "%s is missing" % field})	×
338	else:
339	what[field] = int(what[field])	1✔
340
341	elif change_type == "insert":	1✔
342	if not what.get("signoffs_required", None):	1!
NEW 343	return problem(400, "Bad Request", "Missing field", ext={"exception": "signoffs_required is missing"})	×
344	else:
345	what["signoffs_required"] = int(what["signoffs_required"])	1✔
346
347	elif change_type == "delete":	1!
348	if not what.get("data_version", None):	1!
NEW 349	return problem(400, "Bad Request", "Missing field", ext={"exception": "data_version is missing"})	×
350	else:
351	what["data_version"] = int(what["data_version"])	1✔
352
353	return post_scheduled_changes(	1✔
354	sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes, what=what, transaction=transaction, changed_by=changed_by, change_type=change_type
355	)
356
357
358	@requirelogin	1✔
359	@transactionHandler	1✔
360	@handleGeneralExceptions	1✔
361	def post_permissions_rs_scheduled_change(sc_id, sc_permission_rs_body, transaction, changed_by):	1✔
362	# TODO: modify UI and clients to stop sending 'change_type' in request body
363	sc_table = dbo.permissionsRequiredSignoffs.scheduled_changes	1✔
364	sc_rs_permission = sc_table.select(where={"sc_id": sc_id}, transaction=transaction, columns=["change_type"])	1✔
365	if sc_rs_permission:	1✔
366	change_type = sc_rs_permission[0]["change_type"]	1✔
367	else:
368	return problem(404, "Not Found", "Unknown sc_id", ext={"exception": "No scheduled change for permission required " "signoff found for given sc_id"})	1✔
369	what = {}	1✔
370	for field in sc_permission_rs_body:	1✔
371	if (	1✔
372	(change_type == "insert" and field not in ["when", "product", "role", "signoffs_required"])
373	or (change_type == "update" and field not in ["when", "signoffs_required", "data_version"])
374	or (change_type == "delete" and field not in ["when", "data_version"])
375	):
376	continue	1✔
377	what[field] = sc_permission_rs_body[field]	1✔
378
379	if change_type in ["update", "delete"] and not what.get("data_version", None):	1!
NEW 380	return problem(400, "Bad Request", "Missing field", ext={"exception": "data_version is missing"})	×
381
382	if what.get("signoffs_required", None):	1!
383	what["signoffs_required"] = int(what["signoffs_required"])	1✔
384
385	return post_scheduled_change(	1✔
386	sc_table=sc_table,
387	sc_id=sc_id,
388	what=what,
389	transaction=transaction,
390	changed_by=changed_by,
391	old_sc_data_version=sc_permission_rs_body.get("sc_data_version", None),
392	)
393
394
395	@requirelogin	1✔
396	@transactionHandler	1✔
397	@handleGeneralExceptions	1✔
398	def delete_permissions_rs_scheduled_change(sc_id, data_version, transaction, changed_by):	1✔
399	return delete_scheduled_change(	1✔
400	sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes, sc_id=sc_id, data_version=data_version, transaction=transaction, changed_by=changed_by
401	)
402
403
404	@requirelogin	1✔
405	@transactionHandler	1✔
406	@handleGeneralExceptions	1✔
407	def post_permissions_rs_enact_scheduled_change(sc_id, transaction, changed_by):	1✔
408	return post_enact_scheduled_change(sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes, sc_id=sc_id, transaction=transaction, changed_by=changed_by)	1✔
409
410
411	@requirelogin	1✔
412	@transactionHandler	1✔
413	@handleGeneralExceptions	1✔
414	def post_permissions_rs_signoffs_scheduled_change(sc_id, sc_post_signoffs_body, transaction, changed_by):	1✔
415	return post_signoffs_scheduled_change(	1✔
416	signoffs_table=dbo.permissionsRequiredSignoffs.scheduled_changes.signoffs,
417	sc_id=sc_id,
418	what=sc_post_signoffs_body,
419	transaction=transaction,
420	changed_by=changed_by,
421	)
422
423
424	@requirelogin	1✔
425	@transactionHandler	1✔
426	@handleGeneralExceptions	1✔
427	def delete_permissions_rs_signoffs_scheduled_change(sc_id, transaction, changed_by):	1✔
428	return delete_signoffs_scheduled_change(	1✔
429	signoffs_table=dbo.permissionsRequiredSignoffs.scheduled_changes.signoffs, sc_id=sc_id, transaction=transaction, changed_by=changed_by
430	)
431
432
433	def get_permissions_rs_scheduled_change_history(sc_id):	1✔
434	return get_scheduled_change_history(sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes, sc_id=sc_id)	1✔
435
436
437	def get_all_permissions_rs_scheduled_change_history():	1✔
438	return get_all_scheduled_change_history(sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes)	1✔
439
440
441	@requirelogin	1✔
442	@transactionHandler	1✔
443	@handleGeneralExceptions	1✔
444	def post_permissions_rs_scheduled_change_history(sc_id, transaction, changed_by):	1✔
NEW 445	return post_scheduled_change_history(	×
446	sc_table=dbo.permissionsRequiredSignoffs.scheduled_changes, sc_id=sc_id, transaction=transaction, changed_by=changed_by
447	)

mozilla-releng / balrog / #4120

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous