def METHOD_NAME():
    dag = get_dag()
    with instance_for_test() as instance:
        manager = instance.compute_log_manager
        pre_execute_time = get_current_datetime_in_utc()
        # When tags are not set, run with current time
        job_def = make_dagster_job_from_airflow_dag(
            dag=dag,
        )
        result = job_def.execute_in_process(instance=instance)
        assert result.success
        for event in result.all_events:
            assert event.event_type_value != "STEP_FAILURE"
        capture_events = [
            event
            for event in result.all_events
            if event.event_type == DagsterEventType.LOGS_CAPTURED
        ]
        event = capture_events[0]
        assert event.logs_captured_data.step_keys == ["test_tags_dag__templated"]
        file_key = event.logs_captured_data.file_key
        post_execute_time = get_current_datetime_in_utc()
        compute_io_path = manager.get_local_path(result.run_id, file_key, ComputeIOType.STDOUT)
        assert os.path.exists(compute_io_path)
        stdout_file = open(compute_io_path, "r", encoding="utf8")
        file_contents = normalize_file_content(stdout_file.read())
        stdout_file.close()
        search_str = "command for dt "
        date_start = file_contents.find(search_str) + len(search_str)
        date_end = date_start + 10  # number of characters in YYYY-MM-DD
        date = file_contents[date_start:date_end]
        check_captured_logs(manager, result, date)
        pre_execute_time_fmt = pre_execute_time.strftime("%Y-%m-%d")
        post_execute_time_fmt = post_execute_time.strftime("%Y-%m-%d")
        assert date in [pre_execute_time_fmt, post_execute_time_fmt]