File size: 2,784 Bytes
a2e06b3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | """S3 batch Lambda handler behaviour (assignPublicIp, app_config merge)."""
import importlib.util
import sys
from pathlib import Path
CDK_DIR = Path(__file__).resolve().parents[1]
LAMBDA_PATH = CDK_DIR / "config" / "lambda" / "lambda_function.py"
def _load_lambda_module():
spec = importlib.util.spec_from_file_location("batch_lambda", LAMBDA_PATH)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
sys.modules["batch_lambda"] = module
spec.loader.exec_module(module)
return module
def test_assign_public_ip_reads_env(monkeypatch):
_load_lambda_module()
monkeypatch.setenv("ECS_ASSIGN_PUBLIC_IP", "true")
reloaded = importlib.util.spec_from_file_location("batch_lambda2", LAMBDA_PATH)
module = importlib.util.module_from_spec(reloaded)
assert reloaded.loader is not None
reloaded.loader.exec_module(module)
assert module.ASSIGN_PUBLIC_IP is True
def test_build_environment_includes_app_config(monkeypatch):
mod = _load_lambda_module()
monkeypatch.setattr(
mod,
"_load_app_config_env",
lambda: {"RUN_AWS_FUNCTIONS": "True", "SHOW_COSTS": "False"},
)
monkeypatch.setattr(mod, "_load_default_env", lambda _bucket: {})
monkeypatch.setattr(
mod,
"s3",
type(
"S3",
(),
{
"get_object": staticmethod(
lambda Bucket, Key: {
"Body": type(
"Body",
(),
{
"read": staticmethod(
lambda: b"DIRECT_MODE_INPUT_FILE=doc.pdf\n"
)
},
)()
}
),
"exceptions": type("E", (), {"ClientError": Exception})(),
},
)(),
)
monkeypatch.setattr(
mod,
"ecs",
type(
"ECS",
(),
{"run_task": staticmethod(lambda **kwargs: {"tasks": [], "failures": []})},
)(),
)
event = {
"Records": [
{
"s3": {
"bucket": {"name": "output-bucket"},
"object": {"key": "input/config/job.env"},
}
}
]
}
mod.ENV_PREFIX = "input/config/"
mod.ENV_SUFFIX = ".env"
mod.CONTAINER_NAME = "app"
mod.CLUSTER = "cluster"
mod.TASK_DEF = "arn:aws:ecs:eu-west-2:123:task-definition/app:1"
mod.SUBNETS = ["subnet-1"]
mod.SECURITY_GROUPS = ["sg-1"]
result = mod.lambda_handler(event, None)
assert result["runs"]
assert result["runs"][0]["envCount"] >= 3
|