Coverage for /builds/BuildGrid/buildgrid/buildgrid/server/operations/filtering/sanitizer.py: 95.45%

1# Copyright (C) 2020 Bloomberg LP 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# <http://www.apache.org/licenses/LICENSE-2.0> 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15 

16import re 

17from abc import ABC 

18from datetime import datetime 

19from typing import Any, List 

20 

21import dateutil.parser 

22from dateutil.tz import tzutc 

23 

24import buildgrid.server.enums as enums 

25from buildgrid.server.exceptions import InvalidArgumentError 

26from buildgrid.server.operations.filtering import SortKey 

27 

28 

29class ValueSanitizer(ABC): 

30 """Base sanitizer class.""" 

31 

32 @property 

33 def valid_values(self) -> List[str]: 

34 """Return a list of valid values for the sanitizer. 

35 

36 This is only useful for sanitizers with a finite list of valid values. 

37 

38 Raises NotImplementedError for sanitizers which aren't based on an 

39 enum of valid values. 

40 

41 """ 

42 raise NotImplementedError() 

43 

44 # TODO probably make this generic? 

45 def sanitize(self, value_string: str) -> Any: 

46 """Method that takes an input string, validates that input string, 

47 and transforms it to a value of another type if necessary. 

48 

49 Raises InvalidArgumentError if the sanitization fails. Returns a 

50 value of an arbitrary type if it succeeds.""" 

51 raise NotImplementedError() 

52 

53 

54class RegexValueSanitizer(ValueSanitizer): 

55 """Sanitizer for regexable patterns.""" 

56 

57 def __init__(self, filter_name: str, regex_pattern: str) -> None: 

58 self.filter_name = filter_name 

59 self.regex = re.compile(f"^({regex_pattern})$") 

60 

61 def sanitize(self, value_string: str) -> str: 

62 if not self.regex.fullmatch(value_string): 

63 raise InvalidArgumentError(f"[{value_string}] is not a valid value for {self.filter_name}.") 

64 return value_string 

65 

66 

67class DatetimeValueSanitizer(ValueSanitizer): 

68 """Sanitizer for ISO 8601 datetime strings.""" 

69 

70 def __init__(self, filter_name: str) -> None: 

71 self.filter_name = filter_name 

72 

73 def sanitize(self, value_string: str) -> datetime: 

74 try: 

75 dt = dateutil.parser.isoparse(value_string) 

76 # Convert to UTC and remove timezone awareness 

77 if dt.tzinfo: 

78 dt = dt.astimezone(tz=tzutc()).replace(tzinfo=None) 

79 return dt 

80 except ValueError: 

81 raise InvalidArgumentError(f"[{value_string}] is not a valid value for {self.filter_name}.") 

82 

83 

84class OperationStageValueSanitizer(ValueSanitizer): 

85 """Sanitizer for the OperationStage type. 

86 

87 Matches valid OperationStage values and converts to the 

88 numeric representation of that stage.""" 

89 

90 def __init__(self, filter_name: str) -> None: 

91 self.filter_name = filter_name 

92 

93 @property 

94 def valid_values(self) -> List[str]: 

95 return [stage.name for stage in enums.OperationStage] 

96 

97 def sanitize(self, value_string: str) -> int: 

98 try: 

99 stage = value_string.upper() 

100 return enums.OperationStage[stage].value 

101 except KeyError: 

102 raise InvalidArgumentError(f"[{value_string}] is not a valid value for {self.filter_name}.") 

103 

104 

105class SortKeyValueSanitizer(ValueSanitizer): 

106 """Sanitizer for sort orders. 

107 

108 Produces a SortKey tuple, which specifies both a key name and a boolean 

109 indicating ascending/descending order.""" 

110 

111 def __init__(self, filter_name: str) -> None: 

112 self.filter_name = filter_name 

113 

114 def sanitize(self, value_string: str) -> SortKey: 

115 desc_key = "(desc)" 

116 asc_key = "(asc)" 

117 

118 key_name = value_string.lower().strip() 

119 if not key_name: 

120 raise InvalidArgumentError(f"Invalid sort key [{value_string}].") 

121 descending = False 

122 

123 if value_string.endswith(desc_key): 

124 descending = True 

125 key_name = key_name[: -len(desc_key)].strip() 

126 if not key_name: 

127 raise InvalidArgumentError(f"Invalid sort key [{value_string}].") 

128 

129 elif value_string.endswith(asc_key): 

130 key_name = key_name[: -len(asc_key)].strip() 

131 if not key_name: 

132 raise InvalidArgumentError(f"Invalid sort key [{value_string}].") 

133 

134 return SortKey(name=key_name, descending=descending)