Coverage for ipam/forms/fields.py: 17%

1import json 

2 

3from django import forms 

4from django.core.exceptions import ValidationError 

5from django.utils.translation import gettext_lazy as _ 

6 

7from ipam.forms.widgets import PortMappingWidget 

8from ipam.utils import expand_port_mapping, group_port_mapping_rows 

9from ipam.validators import validate_port_mappings 

10 

11__all__ = ( 

12 'PortMappingField', 

13) 

14 

15 

16class PortMappingField(forms.Field): 

17 """ 

18 A form field for editing a service's port mappings. Presents one row per protocol (each with a 

19 comma/range list of ports) but cleans to the model's flat list of ``protocol/port`` strings, e.g. 

20 ``['tcp/80', 'tcp/443', 'udp/53']``. 

21 """ 

22 widget = PortMappingWidget 

23 

24 def prepare_value(self, value): 

25 # Group the flat ['tcp/80', 'tcp/443', 'udp/53'] list back into per-protocol rows for the widget. 

26 if value in (None, ''): 

27 return '[]' 

28 if isinstance(value, str): 

29 # An already-grouped JSON string (e.g. re-rendering a bound form) is passed through. A bare 

30 # 'protocol/port' string arrives when cloning a single-mapping object: the querystring 

31 # single-value collapse (normalize_querydict) yields a str rather than a list, so group it 

32 # like the list case instead of handing the widget unparseable JSON (which blanks the row). 

33 try: 

34 json.loads(value) 

35 except (TypeError, ValueError): 

36 return json.dumps(group_port_mapping_rows([value])) 

37 return value 

38 return json.dumps(group_port_mapping_rows(value)) 

39 

40 def to_python(self, value): 

41 if value in (None, ''): 

42 return [] 

43 # A list is assumed to already be the flat ['tcp/80', ...] form (e.g. set programmatically) 

44 if isinstance(value, list): 

45 mappings = value 

46 else: 

47 try: 

48 rows = json.loads(value) 

49 except (TypeError, ValueError): 

50 raise ValidationError(_("Invalid port mapping data.")) 

51 if not isinstance(rows, list): 

52 raise ValidationError(_("Invalid port mapping data.")) 

53 

54 mappings = [] 

55 for position, row in enumerate(rows, start=1): 

56 # The widget's JS always submits a list of {protocol, ports} objects, but the hidden 

57 # input is just POST data: a hand-crafted payload can put anything here, so validate the 

58 # shape rather than letting a non-dict row raise AttributeError (a 500) on .get() below. 

59 if not isinstance(row, dict): 

60 raise ValidationError(_("Invalid port mapping data.")) 

61 protocol = row.get('protocol') 

62 raw_ports = row.get('ports') 

63 # Likewise `protocol` is only ever a string, and `ports` either a string (the widget's 

64 # comma/range format) or a list of ports (set programmatically); anything else would reach 

65 # expand_port_mapping() and fail there on .strip(). 

66 if ( 

67 (protocol is not None and not isinstance(protocol, str)) 

68 or (raw_ports is not None and not isinstance(raw_ports, (str, list))) 

69 ): 

70 raise ValidationError(_("Invalid port mapping data.")) 

71 if isinstance(raw_ports, str): 

72 raw_ports = raw_ports.strip() 

73 # Ignore entirely-empty rows (e.g. the default blank row on an untouched form) 

74 if not protocol and not raw_ports: 

75 continue 

76 # Expand via the shared helper, which accepts either the widget's comma/range string or an 

77 # already-expanded list, rejects a blank protocol, and preserves a protocol-without-ports 

78 # row as a bare 'protocol/' token. Errors are re-raised with the row's position (among the 

79 # submitted rows — the widget omits entirely-blank ones), since it renders one row per 

80 # protocol and an unqualified "must specify a protocol" gives no clue which row to fix. 

81 # Errors from validate_port_mappings() below are deliberately left unqualified: each quotes 

82 # the offending mapping already, and a duplicate spans two rows. 

83 try: 

84 mappings.extend(expand_port_mapping(protocol, raw_ports)) 

85 except ValidationError as e: 

86 raise ValidationError([ 

87 _("Row {position}: {error}").format(position=position, error=message) 

88 for message in e.messages 

89 ]) 

90 

91 # Shared validation returns the canonical (normalized) list of protocol/port strings 

92 return validate_port_mappings(mappings) 

93 

94 def has_changed(self, initial, data): 

95 # Compare the parsed mappings rather than raw strings, so cosmetic differences (row/port 

96 # ordering, whitespace) don't register as a change. 

97 def normalize(value): 

98 try: 

99 return sorted(self.to_python(value)) 

100 except ValidationError: 

101 return None 

102 

103 return normalize(self.prepare_value(initial)) != normalize(data)