Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _validate_subscriber(self):
subscriber_type = self.properties[self.SUBSCRIBER].split(":", 1)[0]
if subscriber_type not in self.VALID_SUBSCRIBER_TYPES:
msg = (_("The subscriber type of must be one of: %s.")
% ", ".join(self.VALID_SUBSCRIBER_TYPES))
raise exception.StackValidationFailed(message=msg)
super(NetworkGateway, self).validate()
connections = self.properties[self.CONNECTIONS]
for connection in connections:
segmentation_type = connection[self.SEGMENTATION_TYPE]
segmentation_id = connection.get(self.SEGMENTATION_ID)
if segmentation_type == 'vlan' and segmentation_id is None:
msg = _("segmentation_id must be specified for using vlan")
raise exception.StackValidationFailed(message=msg)
if segmentation_type == 'flat' and segmentation_id:
msg = _(
"segmentation_id cannot be specified except 0 for "
"using flat")
raise exception.StackValidationFailed(message=msg)
if len(snippet) == 1:
fn_name, args = next(iter(snippet.items()))
Func = functions.get(fn_name)
if Func is not None:
try:
path = '.'.join([path, fn_name])
if (isinstance(Func, type) and
issubclass(Func, function.Macro)):
return Func(stack, fn_name, args,
functools.partial(recurse, path=path),
template)
else:
return Func(stack, fn_name, recurse(args, path))
except (ValueError, TypeError, KeyError) as e:
raise exception.StackValidationFailed(
path=path,
message=str(e))
return dict((k, recurse(v, mkpath(k)))
for k, v in snippet.items())
elif (not isinstance(snippet, str) and
isinstance(snippet, collections.Iterable)):
def mkpath(idx):
return ''.join([path, '[%d]' % idx])
return [recurse(v, mkpath(i)) for i, v in enumerate(snippet)]
else:
return snippet
function.resolve,
self.name).validate()
# validate if HTTPS_REDIRECT is true
self._validate_https_redirect()
# if a vip specifies and id, it can't specify version or type;
# otherwise version and type are required
for vip in self.properties[self.VIRTUAL_IPS]:
has_id = vip.get(self.VIRTUAL_IP_ID) is not None
has_version = vip.get(self.VIRTUAL_IP_IP_VERSION) is not None
has_type = vip.get(self.VIRTUAL_IP_TYPE) is not None
if has_id:
if (has_version or has_type):
message = _("Cannot specify type or version if VIP id is"
" specified.")
raise exception.StackValidationFailed(message=message)
elif not (has_version and has_type):
message = _("Must specify VIP type and version if no id "
"specified.")
raise exception.StackValidationFailed(message=message)
def validate_constraints(self, value, context=None, skipped=None):
if not skipped:
skipped = []
try:
for constraint in self.constraints:
if type(constraint) not in skipped:
constraint.validate(value, self, context)
except ValueError as ex:
raise exception.StackValidationFailed(message=str(ex))
def validate_nested_stack(self):
try:
name = "%s-%s" % (self.stack.name, self.name)
nested_stack = self._parse_nested_stack(
name,
self.child_template(),
self.child_params())
nested_stack.strict_validate = False
nested_stack.validate()
except AssertionError:
raise
except Exception as ex:
path = "%s<%s>" % (self.name, self.template_url)
raise exception.StackValidationFailed(
ex, path=[self.stack.t.RESOURCES, path])
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from heat.common import exception
from heat.common.i18n import _
from heat.engine import properties
from heat.engine import resource
from common.mixins import f5_common_resources
from common.mixins import F5BigIPMixin
from f5.utils.iapp_parser import IappParser
class IappFullTemplateValidationFailed(exception.StackValidationFailed):
pass
class F5SysiAppFullTemplate(F5BigIPMixin, resource.Resource):
'''Manages creation of an iApp® resource on the BIG-IP®.'''
PROPERTIES = (
BIGIP_SERVER,
PARTITION,
FULL_TEMPLATE
) = (
'bigip_server',
'partition',
'full_template'
)
def _validate_name(res_name):
if '/' in res_name:
message = _('Resource name may not contain "/"')
raise exception.StackValidationFailed(message=message)
def validate_deletion_policy(cls, policy):
path = rsrc_defn.DELETION_POLICY
if policy not in rsrc_defn.ResourceDefinition.DELETION_POLICIES:
msg = _('Invalid deletion policy "%s"') % policy
raise exception.StackValidationFailed(message=msg, path=path)
if policy == rsrc_defn.ResourceDefinition.SNAPSHOT:
if not callable(getattr(cls, 'handle_snapshot_delete', None)):
msg = _('"%s" deletion policy not supported') % policy
raise exception.StackValidationFailed(message=msg, path=path)
def raise_invalid_exception(zone_type, prp):
if self.properties.get(self.TYPE) == zone_type:
if not self.properties.get(prp):
msg = _('Property %(prp)s is required for zone type '
'%(zone_type)s') % {
"prp": prp,
"zone_type": zone_type
}
raise exception.StackValidationFailed(message=msg)