import re
from itertools import repeat
from .common_schema import (
MAX_CHANNELS,
MAX_CHANNELS_PER_FSP,
MAX_CHANNELS_PER_STREAM,
MAX_STREAMS_PER_FSP,
)
"""
csp.validators module defines constants and functions for
validating CSP fields in schemas.
"""
MAX_NO_INTEGRATION_FACTOR = 10
[docs]
def validate_integration_factor(integration_factor: int) -> bool:
"""Checks if the integration_factor is valid.
:param integration_factor: Integration time for correlation products
:returns: True if integration_factor is valid
"""
return 1 <= integration_factor <= MAX_NO_INTEGRATION_FACTOR
def validate_output_host(
entry: list[int, str], start_channel_upper_bound: int
) -> bool:
# Ensure only two entries in each host table entry
if len(entry) != 2:
return False
# Validate channel ID
if entry[0] < 0 or entry[0] > start_channel_upper_bound:
return False
# Check IP address pattern lies between 0.0.0.0 -> 255.255.255.255
pattern = re.compile(r"^((25[0-5]|(2[0-4]|1\d|[1-9]|)\d)\.?\b){4}$")
if not pattern.fullmatch(entry[1]):
return False
return True
def validate_output_link_map(
entry: list[int], start_channel_upper_bound: int
) -> bool:
# Ensure only two entries in output_link_map entry
if len(entry) != 2:
return False
# Validate channel ID
if entry[0] < 0 or entry[0] > start_channel_upper_bound:
return False
# Validate link ID
if entry[1] != 1:
return False
return True
def validate_output_port_entry(
entry: list[int], start_channel_upper_bound: int
) -> bool:
# Ensure only two entries in output_port entry
if len(entry) != 2:
return False
# Validate start channel ID
if entry[0] < 0 or entry[0] > start_channel_upper_bound:
return False
# Validate port number
if entry[1] < 0 or entry[1] > 65535:
return False
return True
def validate_output_port(
mapping: list[list[int]], major_version: int, minor_version: int
) -> bool:
if (major_version, minor_version) >= (4, 0):
start_channel_upper_bound = MAX_CHANNELS
else:
start_channel_upper_bound = MAX_CHANNELS_PER_FSP - 1
# Validate must have at least 1 mapping
if len(mapping) == 0:
return False
if (major_version, minor_version) < (4, 1):
# Validate number of fsp streams
if len(mapping) > MAX_STREAMS_PER_FSP:
return False
# Validate each mapping is a valid entry
if False in list(
map(
validate_output_port_entry,
mapping,
repeat(start_channel_upper_bound),
)
):
return False
# AA0.5/AA1
# Validate start channels must be in increments of 20
# Validate channels must be in ascending order
expected_channel = 0
for index, value in enumerate(mapping):
if index == 0 and (major_version, minor_version) >= (4, 0):
expected_channel = value[0]
if value[0] != expected_channel:
return False
expected_channel = expected_channel + MAX_CHANNELS_PER_STREAM
return True