Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use _check_logs method in localstack

Best Python code snippet using localstack_python

test_logs_command.py

Source:test_logs_command.py

...73        LOG.info("Invoking function %s", function_name)74        lambda_invoke_result = self.lambda_client.invoke(FunctionName=self._get_physical_id(function_name))75        LOG.info("Lambda invoke result %s", lambda_invoke_result)76        cmd_list = self.get_logs_command_list(self.stack_name, name=function_name)77        self._check_logs(cmd_list, [expected_log_output])78    @parameterized.expand([("ApiGwFunction",), ("SfnFunction",)])79    def test_tail(self, function_name: str):80        cmd_list = self.get_logs_command_list(self.stack_name, name=function_name, tail=True)81        tail_process = start_persistent_process(cmd_list)82        expected_log_output = f"Hello world from {function_name} function"83        LOG.info("Invoking function %s", function_name)84        lambda_invoke_result = self.lambda_client.invoke(FunctionName=self._get_physical_id(function_name))85        LOG.info("Lambda invoke result %s", lambda_invoke_result)86        def _check_logs(output: str, _: List[str]) -> bool:87            return expected_log_output in output88        try:89            read_until(tail_process, _check_logs, timeout=RETRY_COUNT * RETRY_SLEEP)90        finally:91            kill_process(tail_process)92    @parameterized.expand([("ApiGwFunction",), ("SfnFunction",)])93    def test_filter(self, function_name: str):94        log_filter = "this should be filtered"95        LOG.info("Invoking function %s", function_name)96        lambda_invoke_result = self.lambda_client.invoke(FunctionName=self._get_physical_id(function_name))97        LOG.info("Lambda invoke result %s", lambda_invoke_result)98        cmd_list = self.get_logs_command_list(self.stack_name, name=function_name, filter=log_filter)99        self._check_logs(cmd_list, [log_filter])100    @parameterized.expand(itertools.product(["HelloWorldServerlessApi"], ["hello", "world"]))101    def test_apigw_logs(self, apigw_name: str, path: str):102        apigw_url = f"{self._get_output_value(apigw_name)}{path}"103        apigw_result = requests.get(apigw_url)104        LOG.info("APIGW result %s", apigw_result)105        cmd_list = self.get_logs_command_list(self.stack_name, name=apigw_name, beta_features=True)106        self._check_logs(cmd_list, [f"HTTP Method: GET, Resource Path: /{path}"])107    @parameterized.expand([("MyStateMachine",)])108    def test_sfn_logs(self, state_machine_name: str):109        sfn_physical_id = self._get_physical_id(state_machine_name)110        sfn_invoke_result = self.sfn_client.start_execution(stateMachineArn=sfn_physical_id)111        execution_arn = sfn_invoke_result.get("executionArn", "")112        LOG.info("SFN invoke result %s", sfn_invoke_result)113        cmd_list = self.get_logs_command_list(self.stack_name, name=state_machine_name, beta_features=True)114        self._check_logs(cmd_list, execution_arn)115    @parameterized.expand(itertools.product(["HelloWorldServerlessApi"], ["hello"]))116    def test_end_to_end_apigw(self, apigw_name: str, path: str):117        apigw_url = f"{self._get_output_value(apigw_name)}{path}"118        apigw_result = requests.get(apigw_url)119        LOG.info("APIGW result %s", apigw_result)120        cmd_list = self.get_logs_command_list(self.stack_name, beta_features=True)121        self._check_logs(122            cmd_list, [f"HTTP Method: GET, Resource Path: /{path}", "Hello world from ApiGwFunction function"]123        )124    @parameterized.expand(itertools.product(["HelloWorldServerlessApi"], ["world"]))125    def test_end_to_end_sfn(self, apigw_name: str, path: str):126        apigw_url = f"{self._get_output_value(apigw_name)}{path}"127        apigw_result = requests.get(apigw_url)128        LOG.info("APIGW result %s", apigw_result)129        cmd_list = self.get_logs_command_list(self.stack_name, beta_features=True)130        self._check_logs(131            cmd_list,132            [133                f"HTTP Method: GET, Resource Path: /{path}",134                '"type": "TaskStateEntered"',135                "Hello world from ApiGwFunction function",136            ],137        )138    @parameterized.expand(itertools.product(["ApiGwFunction", "SfnFunction"], [None, "text", "json"]))139    def test_output(self, function_name: str, output: Optional[str]):140        expected_log_output = f"Hello world from {function_name} function"141        LOG.info("Invoking function %s", function_name)142        lambda_invoke_result = self.lambda_client.invoke(FunctionName=self._get_physical_id(function_name))143        LOG.info("Lambda invoke result %s", lambda_invoke_result)144        cmd_list = self.get_logs_command_list(self.stack_name, name=function_name, output=output, beta_features=True)145        self._check_logs(cmd_list, [expected_log_output], output=output)146    @parameterized.expand(147        itertools.product(148            ["ApiGwFunction", "SfnFunction"],149            [150                (None, None, True),151                (None, "1 minute", True),152                ("1 minute", None, True),153                ("now", None, False),154            ],155        )156    )157    def test_start_end(self, function_name: str, start_end_time_params: Tuple[Optional[str], Optional[str], bool]):158        (start_time, end_time, should_succeed) = start_end_time_params159        expected_log_output = f"Hello world from {function_name} function"160        LOG.info("Invoking function %s", function_name)161        lambda_invoke_result = self.lambda_client.invoke(FunctionName=self._get_physical_id(function_name))162        LOG.info("Lambda invoke result %s", lambda_invoke_result)163        cmd_list = self.get_logs_command_list(164            self.stack_name, name=function_name, start_time=start_time, end_time=end_time165        )166        if not should_succeed:167            with self.assertRaises(ValueError):168                self._check_logs(cmd_list, [expected_log_output], retries=2)169        else:170            self._check_logs(cmd_list, [expected_log_output])171    @parameterized.expand([("ApiGwFunction",), ("SfnFunction",)])172    def test_include_traces(self, function_name: str):173        expected_log_output = f"Hello world from {function_name} function"174        LOG.info("Invoking function %s", function_name)175        lambda_invoke_result = self.lambda_client.invoke(FunctionName=self._get_physical_id(function_name))176        LOG.info("Lambda invoke result %s", lambda_invoke_result)177        cmd_list = self.get_logs_command_list(178            self.stack_name, name=function_name, include_traces=True, beta_features=True179        )180        self._check_logs(cmd_list, ["New XRay Service Graph", "XRay Event [revision ", expected_log_output])181    def _check_logs(self, cmd_list: List, log_strings: List[str], output: str = "text", retries=RETRY_COUNT):182        for _ in range(retries):183            cmd_result = run_command(cmd_list)184            cmd_stdout = cmd_result.stdout.decode("utf-8")185            self.assertEqual(cmd_result.process.returncode, 0)186            log_string_found = True187            for log_string in log_strings:188                if output == "json":189                    if f'"message": "{log_string}\\n"' not in cmd_stdout:190                        log_string_found = False191                        break192                else:193                    if log_string not in cmd_stdout:194                        log_string_found = False195                        break...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.