Coverage for middle_layer/allocate/application_layer/rest_api/views/allocation_disposition.py: 92.31%
104 statements
« prev ^ index » next coverage.py v7.10.5, created at 2026-03-09 06:13 +0000
1# Copyright 2024 Associated Universities, Inc.
2#
3# This file is part of Telescope Time Allocation Tools (TTAT).
4#
5# TTAT is free software: you can redistribute it and/or modify
6# it under the terms of the GNU General Public License as published by
7# the Free Software Foundation, either version 3 of the License, or
8# any later version.
9#
10# TTAT is distributed in the hope that it will be useful,
11# but WITHOUT ANY WARRANTY; without even the implied warranty of
12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13# GNU General Public License for more details.
14#
15# You should have received a copy of the GNU General Public License
16# along with TTAT. If not, see <https://www.gnu.org/licenses/>.
17from http import HTTPStatus
19from pyramid.httpexceptions import HTTPBadRequest, HTTPUnauthorized
20from pyramid.request import Request
21from pyramid.response import Response
22from pyramid.view import view_config
24from allocate.domain_layer.entities.allocation_disposition import AllocationDisposition, AllocationDispositionState
25from allocate.domain_layer.entities.allocation_version import AllocationVersion
26from allocate.domain_layer.entities.proposal_disposition_group import ProposalDispositionGroup
27from allocate.domain_layer.entities.publication_destination import PublicationDestination
28from allocate.domain_layer.services.create_allocation_disposition_service import create_allocation_disposition
29from common.application_layer.rest_api import make_expected_params_message
30from common.application_layer.services.permissions_service import is_active_tac, is_author_of_completed_proposal
33def check_user_permissions(user_id: int, pdg, repo) -> tuple[bool, bool, bool]:
34 """
35 Check permissions for a user against a proposal disposition group.
37 :param user_id: The user ID to check
38 :param pdg: Proposal Disposition Group to check against
39 :param repo: Repository instance
40 :return: Tuple of (is_tta_member, is_active_tac_member, is_proposal_author)
41 """
42 # Check if user is a TTA member from identity.roles (this needs to be passed in)
43 is_tta_member = False # This will be passed from the request
45 # Check if user is an active TAC member for this solicitation
46 solicitation_id = pdg.solicitation_id
47 is_active_tac_member = is_active_tac(solicitation_id, user_id, repo)
49 # Check if user is an author of a completed proposal in this disposition group
50 is_proposal_author = is_author_of_completed_proposal(user_id, pdg.proposal_dispositions)
52 return is_tta_member, is_active_tac_member, is_proposal_author
55def get_allowed_publication_destinations(
56 is_tta_member: bool, is_active_tac_member: bool, is_proposal_author: bool
57) -> list:
58 """
59 Determine which publication destinations a user is allowed to see based on their roles.
61 :param is_tta_member: Whether the user is a TTA member
62 :param is_active_tac_member: Whether the user is an active TAC member
63 :param is_proposal_author: Whether the user is an author of a completed proposal
64 :return: List of allowed publication destinations
65 """
66 # For TTA members, include all published versions
67 if is_tta_member:
68 return [
69 None,
70 PublicationDestination.DISPOSITION,
71 PublicationDestination.APPROVAL,
72 PublicationDestination.CLOSEOUT,
73 ]
74 # For TAC members, include disposition and approval
75 elif is_active_tac_member:
76 return [
77 PublicationDestination.DISPOSITION,
78 PublicationDestination.APPROVAL,
79 ]
80 # For proposal authors, only include closeout
81 elif is_proposal_author:
82 return [PublicationDestination.CLOSEOUT]
83 # If none of the above, return an empty list (no access)
84 else:
85 return []
88def verify_publication_permissions(user_permissions, allocation_version):
89 """
90 Verify that a user has permission to access an allocation version based on its publication destination.
92 :param user_permissions: Tuple of (is_tta_member, is_active_tac_member, is_proposal_author)
93 :param allocation_version: The AllocationVersion to check access for
94 :raises HTTPUnauthorized: If the user doesn't have permission to access this allocation version
95 """
96 is_tta_member, is_active_tac_member, is_proposal_author = user_permissions
98 # Get allowed publication destinations based on user permissions
99 allowed_destinations = get_allowed_publication_destinations(is_tta_member, is_active_tac_member, is_proposal_author)
101 # Check if the allocation version's publication destination is in the allowed list
102 if allocation_version.published_destination not in allowed_destinations:
103 raise HTTPUnauthorized(
104 body=f"User does not have permission to access allocation version with "
105 f"publication destination {allocation_version.published_destination}"
106 )
109def filter_dispositions_for_proposal_author(published_ads: list, pdg, user_id: int) -> list:
110 """
111 Filter allocation dispositions to only include those for proposals authored by the given user.
113 :param published_ads: List of allocation dispositions
114 :param pdg: Proposal Disposition Group containing the proposals
115 :param user_id: User ID to check authorship against
116 :return: Filtered list of allocation dispositions
117 """
118 # First, collect all proposal IDs for which the user is an author
119 user_proposal_ids = set()
120 for pd in pdg.proposal_dispositions:
121 proposal = pd.proposal
122 # Using the existing function to check if user is author of this proposal
123 if proposal and proposal.state == "Completed" and is_author_of_completed_proposal(user_id, [pd]):
124 user_proposal_ids.add(proposal.proposal_id)
126 # Then filter allocation dispositions to only include those for the user's proposals
127 return [
128 ad
129 for ad in published_ads
130 if ad.proposal_disposition and ad.proposal_disposition.proposal.proposal_id in user_proposal_ids
131 ]
134@view_config(
135 route_name="allocation_disposition_list_by_allocation_version_id",
136 renderer="json",
137 permission="allocation_disposition_list",
138)
139def allocation_disposition_list_by_allocation_version_id(request: Request) -> Response:
140 """List all Allocation Dispositions for a given Allocation Version if a TTA member or active TAC member on the
141 associated solicitation. For proposal authors, only closeout dispositions are returned.
143 URL: allocation_versions/{{allocation_version_id}}/allocation_disposition
145 :param request: GET request
146 :return: Response with JSON-formatted list of all Allocation Dispositions on the given Version
147 or 404 response (HTTPNotFound) if no Allocation Version exists with the given id,
148 or 400 response (HTTPBadRequest) if allocation_version_id is not an integer,
149 or 401 response (HTTPUnauthorized) if the requesting user does not have permission
150 """
152 is_tta_member = "tta_member" in request.identity.roles
153 user_id = request.identity.user_id
155 allocation_version: AllocationVersion = request.repo.allocation_version_repo.by_id(
156 request.matchdict["allocation_version_id"]
157 )
159 # Get the proposal disposition group and its solicitation ID
160 pdg = allocation_version.proposal_disposition_group
161 solicitation_id = pdg.solicitation_id
163 # Check if user is an active TAC member for this solicitation
164 is_active_tac_member = is_active_tac(solicitation_id, user_id, request.repo)
166 # Check if user is an author of a completed proposal in this disposition group
167 is_proposal_author = is_author_of_completed_proposal(user_id, pdg.proposal_dispositions)
169 user_permissions = (is_tta_member, is_active_tac_member, is_proposal_author)
171 # User must be either TTA member, active TAC member, or author of a completed proposal
172 if not any(user_permissions):
173 raise HTTPUnauthorized(body=f"User {user_id} does not have permission to view these allocation dispositions.")
175 # Verify publication permissions
176 verify_publication_permissions(user_permissions, allocation_version)
178 allocation_dispositions: list[AllocationDisposition] = (
179 request.repo.allocation_disposition_repo.list_by_allocation_version_id(
180 request.matchdict["allocation_version_id"]
181 )
182 )
183 return Response(json_body=[allocation_disposition.__json__() for allocation_disposition in allocation_dispositions])
186@view_config(
187 route_name="allocation_disposition_list_by_proposal_disposition_group_id",
188 renderer="json",
189 permission="allocation_disposition_list",
190)
191def allocation_disposition_list_by_proposal_disposition_group_id(request: Request) -> Response:
192 """List all Allocation Dispositions for a given Proposal Disposition Group
194 URL: proposal_disposition_group/{{proposal_disposition_group_id}}/allocation_dispositions
196 :param request: GET request
197 :return: Response with JSON-formatted list of all Allocation Dispositions for the specified Proposal Disposition Group
198 or 404 response (HTTPNotFound) if no ProposalDispositionGroup exists with the given id,
199 or 400 response (HTTPBadRequest) if proposal_disposition_group_id is not an integer,
200 or 401 response (HTTPUnauthorized) if the requesting user does not have permission
201 """
203 is_tta_member = "tta_member" in request.identity.roles
204 user_id = request.identity.user_id
206 pdg: ProposalDispositionGroup = request.repo.proposal_disposition_group_repo.by_id(
207 request.matchdict["proposal_disposition_group_id"]
208 )
210 # Check if user is an active TAC member for this solicitation
211 is_active_tac_member = is_active_tac(pdg.solicitation_id, user_id, request.repo)
213 # Check if user is an author of any completed proposal in this disposition group
214 is_proposal_author = is_author_of_completed_proposal(user_id, pdg.proposal_dispositions)
216 user_permissions = (is_tta_member, is_active_tac_member, is_proposal_author)
218 # User must be either TTA member, active TAC member, or author of a completed proposal
219 if not any(user_permissions):
220 raise HTTPUnauthorized(body=f"User {user_id} does not have permission to view these allocation dispositions.")
222 # Get allowed publication destinations based on user permissions
223 published_destinations = get_allowed_publication_destinations(*user_permissions)
225 avs = request.repo.allocation_version_repo.list_by_group_id(pdg.proposal_disposition_group_id)
226 published_avs = []
227 for av in avs:
228 if av.published_destination in published_destinations:
229 published_avs.append(av)
231 published_ads = []
232 for av in published_avs:
233 published_ads.extend(
234 request.repo.allocation_disposition_repo.list_by_allocation_version_id(av.allocation_version_id)
235 )
237 # If author access, filter to only show dispositions for their own proposals
238 if is_proposal_author and not (is_tta_member or is_active_tac_member):
239 published_ads = filter_dispositions_for_proposal_author(published_ads, pdg, user_id)
241 return Response(json_body=[allocation_disposition.__json__() for allocation_disposition in published_ads])
244@view_config(
245 route_name="allocation_disposition_update",
246 renderer="json",
247 permission="allocation_disposition_update",
248)
249def allocation_disposition_update(request: Request) -> Response:
250 """
251 Update the allocation disposition with the properties found in this request.
252 :param request: PUT request with a list of Allocation Dispositions
253 :return: Response with the JSON for the updated allocation disposition
254 or 404 response (HTTPNotFound) if no Allocation Disposition exists with the given id,
255 or 400 response (HTTPBadRequest) if allocation_disposition_id is not an integer or the associated
256 Allocation Version is_readonly == True
257 or allocation disposition state is not valid,
258 or 401 response (HTTPUnauthorized) if the requesting user does not have permission
259 to update an Allocation Disposition
260 """
261 expected_params = [
262 "allocationDispositionId",
263 "state",
264 "schedulerInternalCommentsToTac",
265 ]
266 return_vals = []
267 for params in request.json_body:
268 if not all([expected in params for expected in expected_params]):
269 # JSON params do not contain all expected params
270 raise HTTPBadRequest(body=make_expected_params_message(expected_params, params.keys()))
272 # Locate the allocation disposition
273 ad: AllocationDisposition = request.lookup(params["allocationDispositionId"], AllocationDisposition)
275 # Check that the associated state of the allocation version to ensure this AD can be updated
276 if ad.allocation_version.is_read_only:
277 raise HTTPBadRequest(
278 body=f"The associated Allocation Version is in a read-only state preventing this "
279 f"Allocation Disposition from being updated"
280 )
282 dname: str = params["state"]
284 # attempt to find the AD state
285 try:
286 ad.state = AllocationDispositionState(dname.upper())
287 except ValueError:
288 raise HTTPBadRequest(body=f"Unable to set state on allocation disposition to unknown " f"state {dname}")
290 ad.scheduler_internal_comments_to_tac = params["schedulerInternalCommentsToTac"]
292 request.repo.allocation_disposition_repo.update(ad)
293 return_vals.append(ad)
295 return Response(status_code=HTTPStatus.OK, json_body=[ad.__json__() for ad in return_vals])
298@view_config(
299 route_name="allocation_disposition_restore",
300 renderer="json",
301 permission="allocation_disposition_update",
302)
303def allocation_disposition_restore(request: Request) -> Response:
304 """
305 Restore the allocation disposition using the original request
306 :param request: PUT request with the id of the Allocation Disposition to restore
307 :return: Response with the JSON for the updated allocation disposition
308 or 404 response (HTTPNotFound) if no Allocation Disposition exists with the given id,
309 or 400 response (HTTPBadRequest) if allocation_disposition_id is not an integer or the associated
310 Allocation Version is_readonly == True
311 or 401 response (HTTPUnauthorized) if the requesting user does not have permission
312 to restore an Allocation Disposition
313 """
315 current_ad: AllocationDisposition = request.lookup(
316 request.matchdict["allocation_disposition_id"], AllocationDisposition
317 )
319 if current_ad.allocation_version.is_read_only:
320 raise HTTPBadRequest(
321 body=f"The associated Allocation Version is in a read-only state preventing this "
322 f"Allocation Disposition from being restored"
323 )
325 current_ad.observation_specification_dispositions.clear() # Remove the existing OSDs
326 current_ad.allocated_science_targets.clear() # Remove the existing ASTs
328 # Restore
329 restored_ad = create_allocation_disposition(
330 current_ad.allocation_version, current_ad.allocation_request, current_ad.proposal_disposition
331 )
332 request.repo.allocation_disposition_repo.add(restored_ad)
333 request.repo.allocation_disposition_repo.delete(current_ad) # Delete the old AD
335 return Response(status_code=HTTPStatus.OK, json_body=restored_ad.__json__())