Coverage for middle_layer/allocate/domain_layer/entities/resource_specification_disposition.py: 90.70%

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/>. 

17import typing 

18from datetime import date, timedelta 

19 

20import sqlalchemy 

21from sqlalchemy import ForeignKey, Interval 

22from sqlalchemy.orm import Mapped, composite, mapped_column, relationship 

23 

24from allocate.domain_layer.entities.proprietary_period import ( 

25 ProprietaryPeriod, 

26 ProprietaryPeriodType, 

27) 

28from allocate.domain_layer.entities.scheduling_priority import SchedulingPriority 

29from common.domain_layer import JSON 

30from common.domain_layer.entities.base import Base 

31 

32if typing.TYPE_CHECKING: 

33 from allocate.domain_layer.entities.allocation_disposition import ( 

34 AllocationDisposition, 

35 ) 

36 from propose.domain_layer.entities.resource_specification import ( 

37 ResourceSpecification, 

38 ) 

39 

40 

41class ResourceSpecificationDisposition(Base): 

42 __tablename__ = "resource_specification_dispositions" 

43 resource_specification_disposition_id: Mapped[int] = mapped_column(primary_key=True) 

44 resource_specification_id: Mapped[int] = mapped_column( 

45 ForeignKey("resource_specifications.resource_specification_id", ondelete="CASCADE") 

46 ) 

47 resource_specification: Mapped["ResourceSpecification"] = relationship( 

48 back_populates="resource_specification_dispositions" 

49 ) 

50 allocated_amount: Mapped[float] = mapped_column() 

51 allocated_parameters: Mapped[dict[str, str]] = mapped_column(sqlalchemy.JSON) 

52 allocation_disposition_id: Mapped[int] = mapped_column( 

53 ForeignKey("allocation_dispositions.allocation_disposition_id", ondelete="cascade") 

54 ) 

55 allocation_disposition: Mapped["AllocationDisposition"] = relationship() 

56 

57 allocation_disposition_id = mapped_column( 

58 ForeignKey("allocation_dispositions.allocation_disposition_id", ondelete="CASCADE"), 

59 nullable=False, 

60 ) 

61 allocation_disposition = relationship("AllocationDisposition", back_populates="resource_specification_dispositions") 

62 allocation_request_id = mapped_column( 

63 ForeignKey("allocation_requests.allocation_request_id", ondelete="CASCADE"), 

64 nullable=False, 

65 ) 

66 allocation_request = relationship("AllocationRequest", back_populates="resource_specification_dispositions") 

67 

68 scheduling_priority_name: Mapped[str] = mapped_column(ForeignKey("scheduling_priorities.name")) 

69 scheduling_priority: Mapped[SchedulingPriority] = relationship(SchedulingPriority) 

70 scheduling_priority_locked: Mapped[bool] = mapped_column(nullable=False, default=False) 

71 proprietary_period_type: Mapped[ProprietaryPeriodType] = mapped_column( 

72 sqlalchemy.Enum(ProprietaryPeriodType, name="proprietary_period_type"), 

73 default=ProprietaryPeriodType.NO_PROPRIETARY_PERIOD, 

74 nullable=False, 

75 ) 

76 proprietary_period_duration: Mapped[timedelta] = mapped_column(Interval, nullable=True) 

77 proprietary_period_date: Mapped[date] = mapped_column(nullable=True) 

78 proprietary_period: Mapped[ProprietaryPeriod] = composite( 

79 ProprietaryPeriod, proprietary_period_type, proprietary_period_duration, proprietary_period_date 

80 ) 

81 

82 def __json__(self) -> JSON: 

83 overrides: JSON = { 

84 "allocated_parameters": self.allocated_parameters if isinstance(self.allocated_parameters, dict) else {} 

85 } 

86 return super().__json__() | overrides 

87 

88 def update_from_json(self, json: JSON): 

89 super().update_from_json(json) 

90 self.allocated_parameters = json["allocatedParameters"] 

91 

92 @classmethod 

93 def copy_from_resource_specification( 

94 cls, rs: "ResourceSpecification", ad: "AllocationDisposition" 

95 ) -> "ResourceSpecificationDisposition": 

96 """ 

97 Creates an ResourceSpecificationDisposition, copying fields from the supplied ResourceSpecification 

98 :param os: the resource specification to start from 

99 :param ad: the allocation disposition to use 

100 :return: a new ResourceSpecificationDisposition 

101 """ 

102 

103 return ResourceSpecificationDisposition( 

104 allocated_amount=rs.requested_amount, 

105 allocated_parameters=dict(rs.parameters), 

106 resource_specification=rs, 

107 allocation_disposition=ad, 

108 ) 

109 

110 def clone(self, ad: "AllocationDisposition"): 

111 return ResourceSpecificationDisposition( 

112 allocated_amount=self.allocated_amount, 

113 allocated_parameters=dict(self.allocated_parameters), 

114 resource_specification=self.resource_specification, 

115 allocation_disposition=ad, 

116 )