Coverage for tests/unit_tests/common/test_proto_util.py: 100%

1""" 

2Tests of proto_util.py 

3""" 

4 

5import pytest 

6from tests.conftest import ApioRunner 

7from apio.common.proto.apio_testing_pb2 import MessageA, MessageB 

8from apio.common.proto_util import ( 

9 proto_from_json_dict, 

10 proto_to_json_dict, 

11 check_is_required, 

12 check_not_required, 

13) 

14 

15 

16def test_proto_from_dict_full(apio_runner: ApioRunner): 

17 """Test parsing of a proto from a json dict with all required and 

18 optional fields included.""" 

19 

20 with apio_runner.in_sandbox(): 

21 

22 json_dict1 = { 

23 "field-b1": { 

24 "field-a1": "aaa", 

25 "field-a2": "bbb", 

26 }, 

27 "field-b2": { 

28 "field-a1": "ccc", 

29 "field-a2": "ddd", 

30 }, 

31 } 

32 

33 # -- Parse proto from json dict and verify 

34 proto_msg = proto_from_json_dict( 

35 json_dict1, MessageB, "Fail to parse test proto" 

36 ) 

37 

38 assert proto_msg.field_b1.field_a1 == "aaa" 

39 assert proto_msg.field_b1.field_a2 == "bbb" 

40 assert proto_msg.field_b2.field_a1 == "ccc" 

41 assert proto_msg.field_b2.field_a2 == "ddd" 

42 

43 # -- Convert back to json dict and compare. 

44 json_dict2 = proto_to_json_dict(proto_msg) 

45 

46 assert json_dict2 == json_dict1 

47 

48 

49def test_proto_from_dict_minimal(apio_runner: ApioRunner): 

50 """Test parsing of a proto from a json dict with only required 

51 fields included.""" 

52 

53 with apio_runner.in_sandbox(): 

54 

55 json_dict1 = { 

56 "field-b1": { 

57 "field-a1": "aaa", 

58 }, 

59 } 

60 

61 # -- Parse proto from json dict and verify 

62 proto_msg = proto_from_json_dict( 

63 json_dict1, MessageB, "Fail to parse test proto" 

64 ) 

65 

66 assert proto_msg.field_b1.field_a1 == "aaa" 

67 

68 assert not proto_msg.HasField("field_b2") 

69 assert not proto_msg.field_b1.HasField("field_a2") 

70 

71 # -- Convert back to json dict and compare. 

72 json_dict2 = proto_to_json_dict(proto_msg) 

73 

74 assert json_dict2 == json_dict1 

75 

76 

77def test_proto_from_dict_missing_field(apio_runner: ApioRunner): 

78 """Test parsing of a proto from a json dict with a required 

79 field missing.""" 

80 

81 with apio_runner.in_sandbox(): 

82 

83 json_dict1 = { 

84 "field-b1": { 

85 # -- Required field 'field-a1' is missing. 

86 "field-a2": "aaa", 

87 }, 

88 } 

89 

90 with apio_runner.with_logger() as log: 

91 with pytest.raises(SystemExit) as e: 

92 _ = proto_from_json_dict( 

93 json_dict1, MessageB, "Fail to parse test proto" 

94 ) 

95 

96 assert e.value.code == 1 

97 assert "Missing required field 'field-b1.field-a1'" in log.out 

98 

99 

100def test_proto_from_dict_unknown_field(apio_runner: ApioRunner): 

101 """Test parsing of a proto from a json dict with a an unknown field.""" 

102 

103 with apio_runner.in_sandbox(): 

104 

105 json_dict1 = { 

106 "field-b1": {"field-a1": "aaa", "no-such-field": "bbb"}, 

107 } 

108 

109 with apio_runner.with_logger() as log: 

110 with pytest.raises(SystemExit) as e: 

111 _ = proto_from_json_dict( 

112 json_dict1, MessageB, "Fail to parse test proto" 

113 ) 

114 

115 print(log.out) 

116 

117 assert e.value.code == 1 

118 assert "Unknown field 'no-such-field'" in log.out 

119 

120 

121def test_check_is_required(apio_runner: ApioRunner): 

122 """Test test_check_is_required()""" 

123 

124 with apio_runner.in_sandbox(): 

125 

126 msg = MessageB(field_b1=MessageA(field_a1="aaa")) 

127 

128 # -- Check required fields only (success) 

129 check_is_required(msg, "field_b1") 

130 check_is_required(msg.field_b1, "field_a1") 

131 check_is_required(msg, "field_b1.field_a1") 

132 

133 # -- Check with an optional field (failure) 

134 with apio_runner.with_logger() as log: 

135 with pytest.raises(SystemExit) as e: 

136 check_is_required(msg, "field_b1", "field_b2") 

137 

138 assert e.value.code == 1 

139 assert ( 

140 "Field 'field_b2' of 'apio.common.proto.MessageB' " 

141 "is not required" in log.out 

142 ) 

143 

144 # -- Check with an optional nested field (failure) 

145 with apio_runner.with_logger() as log: 

146 with pytest.raises(SystemExit) as e: 

147 check_is_required(msg, "field_b1", "field_b1.field_a2") 

148 

149 assert e.value.code == 1 

150 assert ( 

151 "Field 'field_b1.field_a2' of 'apio.common.proto.MessageB' " 

152 "is not required" in log.out 

153 ) 

154 

155 # -- Check with an unknown field (failure) 

156 with apio_runner.with_logger() as log: 

157 with pytest.raises(SystemExit) as e: 

158 check_is_required(msg, "field_b1", "field_xyz") 

159 

160 assert e.value.code == 1 

161 assert ( 

162 "Field 'field_xyz' is not a field of protocol buffer " 

163 "message 'apio.common.proto.MessageB'" in log.out 

164 ) 

165 

166 # -- Check with an unknown nested field (failure) 

167 with apio_runner.with_logger() as log: 

168 with pytest.raises(SystemExit) as e: 

169 check_is_required(msg, "field_b1", "field_b1.field_xyz") 

170 

171 assert e.value.code == 1 

172 assert ( 

173 "Field 'field_b1.field_xyz' is not a field of protocol buffer " 

174 "message 'apio.common.proto.MessageB'" in log.out 

175 ) 

176 

177 

178def test_check_not_required(apio_runner: ApioRunner): 

179 """Test test_check_not_required()""" 

180 

181 with apio_runner.in_sandbox(): 

182 

183 msg = MessageB(field_b1=MessageA(field_a1="aaa")) 

184 

185 # -- Check optional fields only (success) 

186 check_not_required(msg, "field_b2") 

187 check_not_required(msg, "field_b2.field_a1") 

188 check_not_required(msg, "field_b2.field_a2") 

189 check_not_required(msg, "field_b1.field_a2") 

190 check_not_required(msg.field_b1, "field_a2") 

191 

192 # -- Check with required field (failure) 

193 with apio_runner.with_logger() as log: 

194 with pytest.raises(SystemExit) as e: 

195 check_not_required(msg, "field_b2", "field_b1") 

196 

197 assert e.value.code == 1 

198 assert ( 

199 "Field 'field_b1' of 'apio.common.proto.MessageB' " 

200 "is required" in log.out 

201 ) 

202 

203 # -- Check with an required nested field (failure) 

204 with apio_runner.with_logger() as log: 

205 with pytest.raises(SystemExit) as e: 

206 check_not_required(msg, "field_b2", "field_b1.field_a1") 

207 

208 assert e.value.code == 1 

209 assert ( 

210 "Field 'field_b1.field_a1' of 'apio.common.proto.MessageB' " 

211 "is required" in log.out 

212 ) 

213 

214 # -- Check with an unknown field (failure) 

215 with apio_runner.with_logger() as log: 

216 with pytest.raises(SystemExit) as e: 

217 check_is_required(msg, "field_b1", "field_xyz") 

218 

219 assert e.value.code == 1 

220 assert ( 

221 "Field 'field_xyz' is not a field of protocol buffer " 

222 "message 'apio.common.proto.MessageB'" in log.out 

223 ) 

224 

225 # -- Check with an unknown nested field (failure) 

226 with apio_runner.with_logger() as log: 

227 with pytest.raises(SystemExit) as e: 

228 check_is_required(msg, "field_b1", "field_b1.field_xyz") 

229 

230 assert e.value.code == 1 

231 assert ( 

232 "Field 'field_b1.field_xyz' is not a field of protocol buffer " 

233 "message 'apio.common.proto.MessageB'" in log.out 

234 )