|
1 | 1 | import enum |
2 | 2 | import json |
3 | 3 | import pickle |
| 4 | +import sys |
4 | 5 | import typing |
5 | 6 | import unittest |
6 | 7 | from datetime import date, datetime, time, timedelta |
@@ -3903,6 +3904,129 @@ class DataPoint(csp.Struct): |
3903 | 3904 | self.assertNotIn("_last_updated", json_data) |
3904 | 3905 | self.assertNotIn("_source", json_data["data"]) |
3905 | 3906 |
|
| 3907 | + def test_literal_types_validation(self): |
| 3908 | + """Test that Literal type annotations correctly validate input values in CSP Structs""" |
| 3909 | + |
| 3910 | + # Define a simple class with various Literal types |
| 3911 | + class StructWithLiterals(csp.Struct): |
| 3912 | + # String literals |
| 3913 | + color: Literal["red", "green", "blue"] |
| 3914 | + # Integer literals |
| 3915 | + size: Literal[1, 2, 3] |
| 3916 | + # Mixed type literals |
| 3917 | + status: Literal["on", "off", 0, 1, True, False] |
| 3918 | + # Optional literal with default |
| 3919 | + mode: Optional[Literal["fast", "slow"]] = "fast" |
| 3920 | + |
| 3921 | + # Test valid assignments |
| 3922 | + s1 = StructWithLiterals(color="red", size=2, status="on") |
| 3923 | + self.assertEqual(s1.color, "red") |
| 3924 | + self.assertEqual(s1.size, 2) |
| 3925 | + self.assertEqual(s1.status, "on") |
| 3926 | + self.assertEqual(s1.mode, "fast") # Default value |
| 3927 | + |
| 3928 | + s2 = StructWithLiterals.from_dict(dict(color="blue", size=1, status=True, mode="slow")) |
| 3929 | + s2_dump = s2.to_json() |
| 3930 | + s2_looped = TypeAdapter(StructWithLiterals).validate_json(s2_dump) |
| 3931 | + self.assertEqual(s2, s2_looped) |
| 3932 | + s2_dict = s2.to_dict() |
| 3933 | + s2_looped_dict = s2.from_dict(s2_dict) |
| 3934 | + self.assertEqual(s2_looped_dict, s2) |
| 3935 | + |
| 3936 | + # Invalid color, but from_dict still accepts |
| 3937 | + StructWithLiterals.from_dict(dict(color="yellow", size=1, status="on")) |
| 3938 | + |
| 3939 | + # Invalid size but from_dict still accepts |
| 3940 | + StructWithLiterals.from_dict(dict(color="red", size=4, status="on")) |
| 3941 | + |
| 3942 | + # Invalid status but from_dict still accepts |
| 3943 | + StructWithLiterals.from_dict(dict(color="red", size=1, status="standby")) |
| 3944 | + |
| 3945 | + # Invalid mode but from_dict still accepts |
| 3946 | + StructWithLiterals.from_dict(dict(color="red", size=1, mode=12)) |
| 3947 | + |
| 3948 | + # Invalid size and since the literals are all the same type |
| 3949 | + # If we give an incorrect type, we catch the error |
| 3950 | + with self.assertRaises(ValueError) as exc_info: |
| 3951 | + StructWithLiterals.from_dict(dict(color="red", size="adasd", mode=12)) |
| 3952 | + self.assertIn("Expected type <class 'int'> received <class 'str'>", str(exc_info.exception)) |
| 3953 | + |
| 3954 | + # Test valid values |
| 3955 | + result = TypeAdapter(StructWithLiterals).validate_python({"color": "green", "size": 3, "status": 0}) |
| 3956 | + self.assertEqual(result.color, "green") |
| 3957 | + self.assertEqual(result.size, 3) |
| 3958 | + self.assertEqual(result.status, 0) |
| 3959 | + |
| 3960 | + # Test invalid color with Pydantic validation |
| 3961 | + with self.assertRaises(ValidationError) as exc_info: |
| 3962 | + TypeAdapter(StructWithLiterals).validate_python({"color": "yellow", "size": 1, "status": "on"}) |
| 3963 | + self.assertIn("1 validation error for", str(exc_info.exception)) |
| 3964 | + self.assertIn("color", str(exc_info.exception)) |
| 3965 | + |
| 3966 | + # Test invalid size with Pydantic validation |
| 3967 | + with self.assertRaises(ValidationError) as exc_info: |
| 3968 | + TypeAdapter(StructWithLiterals).validate_python({"color": "red", "size": 4, "status": "on"}) |
| 3969 | + self.assertIn("1 validation error for", str(exc_info.exception)) |
| 3970 | + self.assertIn("size", str(exc_info.exception)) |
| 3971 | + |
| 3972 | + # Test invalid status with Pydantic validation |
| 3973 | + with self.assertRaises(ValidationError) as exc_info: |
| 3974 | + TypeAdapter(StructWithLiterals).validate_python({"color": "red", "size": 1, "status": "standby"}) |
| 3975 | + self.assertIn("1 validation error for", str(exc_info.exception)) |
| 3976 | + self.assertIn("status", str(exc_info.exception)) |
| 3977 | + |
| 3978 | + # Test invalid mode with Pydantic validation |
| 3979 | + with self.assertRaises(ValidationError) as exc_info: |
| 3980 | + TypeAdapter(StructWithLiterals).validate_python( |
| 3981 | + {"color": "red", "size": 1, "status": "on", "mode": "medium"} |
| 3982 | + ) |
| 3983 | + self.assertIn("1 validation error for", str(exc_info.exception)) |
| 3984 | + self.assertIn("mode", str(exc_info.exception)) |
| 3985 | + |
| 3986 | + def test_pipe_operator_types(self): |
| 3987 | + """Test using the pipe operator for union types in Python 3.10+""" |
| 3988 | + if sys.version_info >= (3, 10): # Only run on Python 3.10+ |
| 3989 | + # Define a class using various pipe operator combinations |
| 3990 | + class PipeTypesConfig(csp.Struct): |
| 3991 | + # Basic primitive types with pipe |
| 3992 | + id_field: str | int |
| 3993 | + # Pipe with None (similar to Optional) |
| 3994 | + description: str | None = None |
| 3995 | + # Multiple types with pipe |
| 3996 | + value: str | int | float | bool |
| 3997 | + # Container with pipe |
| 3998 | + tags: List[str] | Dict[str, str] | None = None |
| 3999 | + # Pipe with literal for comparison |
| 4000 | + status: Literal["active", "inactive"] | None = "active" |
| 4001 | + |
| 4002 | + # Test all valid types |
| 4003 | + valid_cases = [ |
| 4004 | + {"id_field": "string_id", "value": "string_value"}, |
| 4005 | + {"id_field": 42, "value": 123}, |
| 4006 | + {"id_field": "mixed", "value": 3.14}, |
| 4007 | + {"id_field": 999, "value": True}, |
| 4008 | + {"id_field": "with_desc", "value": 1, "description": "Description"}, |
| 4009 | + {"id_field": "with_dict", "value": 1, "tags": None}, |
| 4010 | + ] |
| 4011 | + |
| 4012 | + for case in valid_cases: |
| 4013 | + result = PipeTypesConfig.from_dict(case) |
| 4014 | + # use the other route to get back the result |
| 4015 | + result_to_dict_loop = TypeAdapter(PipeTypesConfig).validate_python(result.to_dict()) |
| 4016 | + self.assertEqual(result, result_to_dict_loop) |
| 4017 | + |
| 4018 | + # Test invalid values |
| 4019 | + invalid_cases = [ |
| 4020 | + {"id_field": 3.14, "value": 1}, # Float for id_field |
| 4021 | + {"id_field": None, "value": 1}, # None for required id_field |
| 4022 | + {"id_field": "test", "value": {}}, # Dict for value |
| 4023 | + {"id_field": "test", "value": None}, # None for required value |
| 4024 | + {"id_field": "test", "value": 1, "status": "unknown"}, # Invalid literal |
| 4025 | + ] |
| 4026 | + for case in invalid_cases: |
| 4027 | + with self.assertRaises(ValidationError): |
| 4028 | + TypeAdapter(PipeTypesConfig).validate_python(case) |
| 4029 | + |
3906 | 4030 |
|
3907 | 4031 | if __name__ == "__main__": |
3908 | 4032 | unittest.main() |
0 commit comments