Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_vpc_trait_added_when_vpc_configured(self, lambda_app):
@lambda_app.lambda_function()
def foo(event, context):
pass
builder = ApplicationGraphBuilder()
config = self.create_config(lambda_app,
autogen_policy=True,
security_group_ids=['sg1', 'sg2'],
subnet_ids=['sn1', 'sn2'])
application = builder.build(config, stage_name='dev')
policy = application.resources[0].role.policy
assert policy == models.AutoGenIAMPolicy(
document=models.Placeholder.BUILD_STAGE,
traits=set([models.RoleTraits.VPC_NEEDED]),
)
def test_scheduled_event_models(self, scheduled_event_app):
config = self.create_config(scheduled_event_app,
app_name='scheduled-event',
autogen_policy=True)
builder = ApplicationGraphBuilder()
application = builder.build(config, stage_name='dev')
assert len(application.resources) == 1
event = application.resources[0]
assert isinstance(event, models.ScheduledEvent)
assert event.resource_name == 'foo-event'
assert event.rule_name == 'scheduled-event-dev-foo-event'
assert isinstance(event.lambda_function, models.LambdaFunction)
assert event.lambda_function.resource_name == 'foo'
context=LambdaErrorContext(
function_name='foo',
client_method_name='create_function',
deployment_size=1024 ** 2
)
)
deploy_error = ChaliceDeploymentError(lambda_error)
deploy_error_msg = str(deploy_error)
assert (
'Connection aborted. Timed out sending your app to Lambda.' in
deploy_error_msg
)
@attrs
class FooResource(models.Model):
name = attrib()
leaf = attrib()
def dependencies(self):
if not isinstance(self.leaf, list):
return [self.leaf]
return self.leaf
@attrs
class LeafResource(models.Model):
name = attrib()
@fixture
def lambda_app():
def create_function_resource(name):
return models.LambdaFunction(
resource_name=name,
function_name='appname-dev-%s' % name,
environment_variables={},
runtime='python2.7',
handler='app.app',
tags={},
timeout=60,
memory_size=128,
deployment_package=models.DeploymentPackage(filename='foo'),
role=models.PreCreatedIAMRole(role_arn='role:arn')
)
from chalice.deploy.newdeployer import ResultsRecorder
from chalice.deploy.newdeployer import DeploymentReporter
from chalice.deploy.swagger import SwaggerGenerator, TemplatedSwaggerGenerator
from chalice.deploy.planner import PlanStage, Variable
from chalice.deploy.planner import UnreferencedResourcePlanner, StringFormat
from chalice.deploy.newdeployer import Executor
from chalice.deploy.newdeployer import UnresolvedValueError
from chalice.deploy.models import APICall, StoreValue, RecordResourceValue
from chalice.deploy.models import RecordResourceVariable
from chalice.deploy.models import JPSearch, BuiltinFunction, Instruction
from chalice.policy import AppPolicyGenerator
from chalice.constants import LAMBDA_TRUST_POLICY
@attrs
class FooResource(models.Model):
name = attrib()
leaf = attrib()
def dependencies(self):
if not isinstance(self.leaf, list):
return [self.leaf]
return self.leaf
@attrs
class LeafResource(models.Model):
name = attrib()
@fixture
def lambda_app():
def test_can_plan_scheduled_event(self):
function = create_function_resource('function_name')
event = models.ScheduledEvent(
resource_name='bar',
rule_name='myrulename',
rule_description="my rule description",
schedule_expression='rate(5 minutes)',
lambda_function=function,
)
plan = self.determine_plan(event)
assert len(plan) == 4
self.assert_apicall_equals(
plan[0],
models.APICall(
method_name='get_or_create_rule_arn',
params={
'rule_name': 'myrulename',
'rule_description': 'my rule description',
'schedule_expression': 'rate(5 minutes)',
def test_scheduled_event_models(self, scheduled_event_app):
config = self.create_config(scheduled_event_app,
app_name='scheduled-event',
autogen_policy=True)
builder = ApplicationGraphBuilder()
application = builder.build(config, stage_name='dev')
assert len(application.resources) == 1
event = application.resources[0]
assert isinstance(event, models.ScheduledEvent)
assert event.resource_name == 'foo-event'
assert event.rule_name == 'scheduled-event-dev-foo-event'
assert isinstance(event.lambda_function, models.LambdaFunction)
assert event.lambda_function.resource_name == 'foo'
def test_can_generate_package(self):
generator = mock.Mock(spec=packager.LambdaDeploymentPackager)
generator.create_deployment_package.return_value = 'package.zip'
package = models.DeploymentPackage(models.Placeholder.BUILD_STAGE)
config = Config.create()
p = DeploymentPackager(generator)
p.handle(config, package)
assert package.filename == 'package.zip'
def _inject_into_function(self, config, lambda_function):
# type: (Config, Optional[models.LambdaFunction]) -> None
if lambda_function is None:
return
role = lambda_function.role
if role is None:
return
if (not self._policy_injected and
isinstance(role, models.ManagedIAMRole) and
isinstance(role.policy, models.AutoGenIAMPolicy) and
not isinstance(role.policy.document,
models.Placeholder)):
self._inject_policy(
role.policy.document,
POST_TO_WEBSOCKET_CONNECTION_POLICY.copy())
self._policy_injected = True
def _create_cloudwatchevent(self, resource):
# type: (models.CloudWatchEventBase) -> Sequence[InstructionMsg]
function_arn = Variable(
'%s_lambda_arn' % resource.lambda_function.resource_name
)
params = {'rule_name': resource.rule_name}
if isinstance(resource, models.ScheduledEvent):
resource = cast(models.ScheduledEvent, resource)
params['schedule_expression'] = resource.schedule_expression
if resource.rule_description is not None:
params['rule_description'] = resource.rule_description
else:
resource = cast(models.CloudWatchEvent, resource)
params['event_pattern'] = resource.event_pattern
plan = [
models.APICall(
method_name='get_or_create_rule_arn',
params=params,
output_var='rule-arn',
),
models.APICall(
method_name='connect_rule_to_lambda',
params={'rule_name': resource.rule_name,