diff --git a/lib/ansible/plugins/connection/aws_ssm.py b/lib/ansible/plugins/connection/aws_ssm.py
new file mode 100644
index 00000000000..39ac5527b6d
--- /dev/null
+++ b/lib/ansible/plugins/connection/aws_ssm.py
@@ -0,0 +1,548 @@
+# Based on the ssh connection plugin by Michael DeHaan
+#
+# Copyright: (c) 2018, Pat Sharkey <psharkey@cleo.com>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = """
+author:
+- Pat Sharkey (@psharkey) <psharkey@cleo.com>
+- HanumanthaRao MVL (@hanumantharaomvl) <hanumanth@flux7.com>
+- Gaurav Ashtikar (@gau1991 )<gaurav.ashtikar@flux7.com>
+connection: aws_ssm
+short_description: execute via AWS Systems Manager
+description:
+- This connection plugin allows ansible to execute tasks on an EC2 instance via the aws ssm CLI.
+version_added: "2.10"
+requirements:
+- The remote EC2 instance must be running the AWS Systems Manager Agent (SSM Agent).
+- The control machine must have the aws session manager plugin installed.
+- The remote EC2 linux instance must have the curl installed.
+options:
+  instance_id:
+    description: The EC2 instance ID.
+    vars:
+    - name: ansible_aws_ssm_instance_id
+  region:
+    description: The region the EC2 instance is located.
+    vars:
+    - name: ansible_aws_ssm_region
+    default: 'us-east-1'
+  bucket_name:
+    description: The name of the S3 bucket used for file transfers.
+    vars:
+    - name: ansible_aws_ssm_bucket_name
+  plugin:
+    description: This defines the location of the session-manager-plugin binary.
+    vars:
+    - name: ansible_aws_ssm_plugin
+    default: '/usr/local/bin/session-manager-plugin'
+  retries:
+    description: Number of attempts to connect.
+    default: 3
+    type: integer
+    vars:
+    - name: ansible_aws_ssm_retries
+  timeout:
+    description: Connection timeout seconds.
+    default: 60
+    type: integer
+    vars:
+    - name: ansible_aws_ssm_timeout
+"""
+
+EXAMPLES = r'''
+
+# Stop Spooler Process on Windows Instances
+- name: Stop Spooler Service on Windows Instances
+  vars:
+    ansible_connection: aws_ssm
+    ansible_shell_type: powershell
+    ansible_aws_ssm_bucket_name: nameofthebucket
+    ansible_aws_ssm_region: us-east-1
+  tasks:
+    - name: Stop spooler service
+      win_service:
+        name: spooler
+        state: stopped
+
+# Install a Nginx Package on Linux Instance
+- name: Install a Nginx Package
+  vars:
+    ansible_connection: aws_ssm
+    ansible_aws_ssm_bucket_name: nameofthebucket
+    ansible_aws_ssm_region: us-west-2
+  tasks:
+    - name: Install a Nginx Package
+      yum:
+        name: nginx
+        state: present
+
+# Create a directory in Windows Instances
+- name: Create a directory in Windows Instance
+  vars:
+    ansible_connection: aws_ssm
+    ansible_shell_type: powershell
+    ansible_aws_ssm_bucket_name: nameofthebucket
+    ansible_aws_ssm_region: us-east-1
+  tasks:
+    - name: Create a Directory
+      win_file:
+        path: C:\Windows\temp
+        state: directory
+
+# Making use of Dynamic Inventory Plugin
+# =======================================
+# aws_ec2.yml (Dynamic Inventory - Linux)
+# This will return the Instance IDs matching the filter
+#plugin: aws_ec2
+#regions:
+#    - us-east-1
+#hostnames:
+#    - instance-id
+#filters:
+#    tag:SSMTag: ssmlinux
+# -----------------------
+- name: install aws-cli
+  hosts: all
+  gather_facts: false
+  vars:
+    ansible_connection: aws_ssm
+    ansible_aws_ssm_bucket_name: nameofthebucket
+    ansible_aws_ssm_region: us-east-1
+  tasks:
+  - name: aws-cli
+    raw: yum install -y awscli
+    tags: aws-cli
+# Execution: ansible-playbook linux.yaml -i aws_ec2.yml
+# The playbook tasks will get executed on the instance ids returned from the dynamic inventory plugin using ssm connection.
+# =====================================================
+# aws_ec2.yml (Dynamic Inventory - Windows)
+#plugin: aws_ec2
+#regions:
+#    - us-east-1
+#hostnames:
+#    - instance-id
+#filters:
+#    tag:SSMTag: ssmwindows
+# -----------------------
+- name: Create a dir.
+  hosts: all
+  gather_facts: false
+  vars:
+    ansible_connection: aws_ssm
+    ansible_shell_type: powershell
+    ansible_aws_ssm_bucket_name: nameofthebucket
+    ansible_aws_ssm_region: us-east-1
+  tasks:
+    - name: Create the directory
+      win_file:
+        path: C:\Temp\SSM_Testing5
+        state: directory
+# Execution:  ansible-playbook win_file.yaml -i aws_ec2.yml
+# The playbook tasks will get executed on the instance ids returned from the dynamic inventory plugin using ssm connection.
+'''
+
+import os
+import boto3
+import getpass
+import json
+import os
+import pty
+import random
+import re
+import select
+import string
+import subprocess
+import time
+
+from functools import wraps
+from ansible import constants as C
+from ansible.errors import AnsibleConnectionFailure, AnsibleError, AnsibleFileNotFound
+from ansible.module_utils.six import PY3
+from ansible.module_utils.six.moves import xrange
+from ansible.module_utils._text import to_bytes, to_native, to_text
+from ansible.plugins.connection import ConnectionBase
+from ansible.plugins.shell.powershell import _common_args
+from ansible.utils.display import Display
+
+display = Display()
+
+
+def _ssm_retry(func):
+    """
+    Decorator to retry in the case of a connection failure
+    Will retry if:
+    * an exception is caught
+    Will not retry if
+    * remaining_tries is <2
+    * retries limit reached
+    """
+    @wraps(func)
+    def wrapped(self, *args, **kwargs):
+        remaining_tries = int(self.get_option('retries')) + 1
+        cmd_summary = "%s..." % args[0]
+        for attempt in range(remaining_tries):
+            cmd = args[0]
+
+            try:
+                return_tuple = func(self, *args, **kwargs)
+                display.vvv(return_tuple, host=self.host)
+                break
+
+            except (AnsibleConnectionFailure, Exception) as e:
+                if attempt == remaining_tries - 1:
+                    raise
+                else:
+                    pause = 2 ** attempt - 1
+                    if pause > 30:
+                        pause = 30
+
+                    if isinstance(e, AnsibleConnectionFailure):
+                        msg = "ssm_retry: attempt: %d, cmd (%s), pausing for %d seconds" % (attempt, cmd_summary, pause)
+                    else:
+                        msg = "ssm_retry: attempt: %d, caught exception(%s) from cmd (%s), pausing for %d seconds" % (attempt, e, cmd_summary, pause)
+
+                    display.vv(msg, host=self.host)
+
+                    time.sleep(pause)
+
+                    # Do not attempt to reuse the existing session on retries
+                    self.close()
+
+                    continue
+
+        return return_tuple
+    return wrapped
+
+
+def chunks(lst, n):
+    """Yield successive n-sized chunks from lst."""
+    for i in range(0, len(lst), n):
+        yield lst[i:i + n]
+
+
+class Connection(ConnectionBase):
+    ''' AWS SSM based connections '''
+
+    transport = 'aws_ssm'
+    allow_executable = False
+    allow_extras = True
+    has_pipelining = False
+    is_windows = False
+    _client = None
+    _session = None
+    _stdout = None
+    _session_id = ''
+    _timeout = False
+    MARK_LENGTH = 26
+
+    def __init__(self, *args, **kwargs):
+
+        super(Connection, self).__init__(*args, **kwargs)
+        self.host = self._play_context.remote_addr
+
+        if getattr(self._shell, "SHELL_FAMILY", '') == 'powershell':
+            self.delegate = None
+            self.has_native_async = True
+            self.always_pipeline_modules = True
+            self.module_implementation_preferences = ('.ps1', '.exe', '')
+            self.protocol = None
+            self.shell_id = None
+            self._shell_type = 'powershell'
+            self.is_windows = True
+
+    def _connect(self):
+        ''' connect to the host via ssm '''
+
+        self._play_context.remote_user = getpass.getuser()
+
+        if not self._session_id:
+            self.start_session()
+        return self
+
+    def start_session(self):
+        ''' start ssm session '''
+
+        if self.get_option('instance_id') is None:
+            self.instance_id = self.host
+        else:
+            self.instance_id = self.get_option('instance_id')
+
+        display.vvv(u"ESTABLISH SSM CONNECTION TO: {0}".format(self.instance_id), host=self.host)
+
+        executable = self.get_option('plugin')
+        if not os.path.exists(to_bytes(executable, errors='surrogate_or_strict')):
+            raise AnsibleError("failed to find the executable specified %s."
+                               " Please verify if the executable exists and re-try." % executable)
+
+        profile_name = ''
+        region_name = self.get_option('region')
+        ssm_parameters = dict()
+
+        client = boto3.client('ssm', region_name=region_name)
+        self._client = client
+        response = client.start_session(Target=self.instance_id, Parameters=ssm_parameters)
+        self._session_id = response['SessionId']
+
+        cmd = [
+            executable,
+            json.dumps(response),
+            region_name,
+            "StartSession",
+            profile_name,
+            json.dumps({"Target": self.instance_id}),
+            client.meta.endpoint_url
+        ]
+
+        display.vvvv(u"SSM COMMAND: {0}".format(to_text(cmd)), host=self.host)
+
+        stdout_r, stdout_w = pty.openpty()
+        session = subprocess.Popen(
+            cmd,
+            stdin=subprocess.PIPE,
+            stdout=stdout_w,
+            stderr=subprocess.PIPE,
+            close_fds=True,
+            bufsize=0,
+        )
+
+        os.close(stdout_w)
+        self._stdout = os.fdopen(stdout_r, 'rb', 0)
+        self._session = session
+        self._poll_stdout = select.poll()
+        self._poll_stdout.register(self._stdout, select.POLLIN)
+
+        # Disable command echo and prompt.
+        self._prepare_terminal()
+
+        display.vvv(u"SSM CONNECTION ID: {0}".format(self._session_id), host=self.host)
+
+        return session
+
+    @_ssm_retry
+    def exec_command(self, cmd, in_data=None, sudoable=True):
+        ''' run a command on the ssm host '''
+
+        super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable)
+
+        display.vvv(u"EXEC {0}".format(to_text(cmd)), host=self.host)
+
+        session = self._session
+
+        mark_begin = "".join([random.choice(string.ascii_letters) for i in xrange(self.MARK_LENGTH)])
+        if self.is_windows:
+            mark_start = mark_begin + " $LASTEXITCODE"
+        else:
+            mark_start = mark_begin
+        mark_end = "".join([random.choice(string.ascii_letters) for i in xrange(self.MARK_LENGTH)])
+
+        # Wrap command in markers accordingly for the shell used
+        cmd = self._wrap_command(cmd, sudoable, mark_start, mark_end)
+
+        self._flush_stderr(session)
+
+        for chunk in chunks(cmd, 1024):
+            session.stdin.write(to_bytes(chunk, errors='surrogate_or_strict'))
+
+        # Read stdout between the markers
+        stdout = ''
+        win_line = ''
+        begin = False
+        stop_time = int(round(time.time())) + self.get_option('timeout')
+        while session.poll() is None:
+            remaining = stop_time - int(round(time.time()))
+            if remaining < 1:
+                self._timeout = True
+                display.vvvv(u"EXEC timeout stdout: {0}".format(to_text(stdout)), host=self.host)
+                raise AnsibleConnectionFailure("SSM exec_command timeout on host: %s"
+                                               % self.instance_id)
+            if self._poll_stdout.poll(1000):
+                line = self._filter_ansi(self._stdout.readline())
+                display.vvvv(u"EXEC stdout line: {0}".format(to_text(line)), host=self.host)
+            else:
+                display.vvvv(u"EXEC remaining: {0}".format(remaining), host=self.host)
+                continue
+
+            if not begin and self.is_windows:
+                win_line = win_line + line
+                line = win_line
+
+            if mark_start in line:
+                begin = True
+                if not line.startswith(mark_start):
+                    stdout = ''
+                continue
+            if begin:
+                if mark_end in line:
+                    display.vvvv(u"POST_PROCESS: {0}".format(to_text(stdout)), host=self.host)
+                    returncode, stdout = self._post_process(stdout, mark_begin)
+                    break
+                else:
+                    stdout = stdout + line
+
+        stderr = self._flush_stderr(session)
+
+        return (returncode, stdout, stderr)
+
+    def _prepare_terminal(self):
+        ''' perform any one-time terminal settings '''
+
+        if not self.is_windows:
+            cmd = "stty -echo\n" + "PS1=''\n"
+            cmd = to_bytes(cmd, errors='surrogate_or_strict')
+            self._session.stdin.write(cmd)
+
+    def _wrap_command(self, cmd, sudoable, mark_start, mark_end):
+        ''' wrap command so stdout and status can be extracted '''
+
+        if self.is_windows:
+            if not cmd.startswith(" ".join(_common_args) + " -EncodedCommand"):
+                cmd = self._shell._encode_script(cmd, preserve_rc=True)
+            cmd = cmd + "; echo " + mark_start + "\necho " + mark_end + "\n"
+        else:
+            if sudoable:
+                cmd = "sudo " + cmd
+            cmd = "echo " + mark_start + "\n" + cmd + "\necho $'\\n'$?\n" + "echo " + mark_end + "\n"
+
+        display.vvvv(u"_wrap_command: '{0}'".format(to_text(cmd)), host=self.host)
+        return cmd
+
+    def _post_process(self, stdout, mark_begin):
+        ''' extract command status and strip unwanted lines '''
+
+        if self.is_windows:
+            # Value of $LASTEXITCODE will be the line after the mark
+            trailer = stdout[stdout.rfind(mark_begin):]
+            last_exit_code = trailer.splitlines()[1]
+            if last_exit_code.isdigit:
+                returncode = int(last_exit_code)
+            else:
+                returncode = -1
+            # output to keep will be before the mark
+            stdout = stdout[:stdout.rfind(mark_begin)]
+
+            # If it looks like JSON remove any newlines
+            if stdout.startswith('{'):
+                stdout = stdout.replace('\n', '')
+
+            return (returncode, stdout)
+        else:
+            # Get command return code
+            returncode = int(stdout.splitlines()[-2])
+
+            # Throw away ending lines
+            for x in range(0, 3):
+                stdout = stdout[:stdout.rfind('\n')]
+
+            return (returncode, stdout)
+
+    def _filter_ansi(self, line):
+        ''' remove any ANSI terminal control codes '''
+        line = to_text(line)
+
+        if self.is_windows:
+            osc_filter = re.compile(r'\x1b\][^\x07]*\x07')
+            line = osc_filter.sub('', line)
+            ansi_filter = re.compile(r'(\x9B|\x1B\[)[0-?]*[ -/]*[@-~]')
+            line = ansi_filter.sub('', line)
+
+            # Replace or strip sequence (at terminal width)
+            line = line.replace('\r\r\n', '\n')
+            if len(line) == 201:
+                line = line[:-1]
+
+        return line
+
+    def _flush_stderr(self, subprocess):
+        ''' read and return stderr with minimal blocking '''
+
+        poll_stderr = select.poll()
+        poll_stderr.register(subprocess.stderr, select.POLLIN)
+        stderr = ''
+
+        while subprocess.poll() is None:
+            if poll_stderr.poll(1):
+                line = subprocess.stderr.readline()
+                display.vvvv(u"stderr line: {0}".format(to_text(line)), host=self.host)
+                stderr = stderr + line
+            else:
+                break
+
+        return stderr
+
+    def _get_url(self, client_method, bucket_name, out_path, http_method):
+        ''' Generate URL for get_object / put_object '''
+        client = boto3.client('s3')
+        return client.generate_presigned_url(client_method, Params={'Bucket': bucket_name, 'Key': out_path}, ExpiresIn=3600, HttpMethod=http_method)
+
+    @_ssm_retry
+    def _file_transport_command(self, in_path, out_path, ssm_action):
+        ''' transfer a file from using an intermediate S3 bucket '''
+
+        s3_path = out_path.replace('\\', '/')
+        bucket_url = 's3://%s/%s' % (self.get_option('bucket_name'), s3_path)
+
+        if self.is_windows:
+            put_command = "Invoke-WebRequest -Method PUT -InFile '%s' -Uri '%s' -UseBasicParsing" % (
+                in_path, self._get_url('put_object', self.get_option('bucket_name'), s3_path, 'PUT'))
+            get_command = "Invoke-WebRequest '%s' -OutFile '%s'" % (
+                self._get_url('get_object', self.get_option('bucket_name'), s3_path, 'GET'), out_path)
+        else:
+            put_command = "curl --request PUT --upload-file '%s' '%s'" % (
+                in_path, self._get_url('put_object', self.get_option('bucket_name'), s3_path, 'PUT'))
+            get_command = "curl '%s' -o '%s'" % (
+                self._get_url('get_object', self.get_option('bucket_name'), s3_path, 'GET'), out_path)
+
+        client = boto3.client('s3')
+        if ssm_action == 'get':
+            (returncode, stdout, stderr) = self.exec_command(put_command, in_data=None, sudoable=False)
+            with open(to_bytes(out_path, errors='surrogate_or_strict'), 'wb') as data:
+                client.download_fileobj(self.get_option('bucket_name'), s3_path, data)
+        else:
+            with open(to_bytes(in_path, errors='surrogate_or_strict'), 'rb') as data:
+                client.upload_fileobj(data, self.get_option('bucket_name'), s3_path)
+            (returncode, stdout, stderr) = self.exec_command(get_command, in_data=None, sudoable=False)
+
+        # Check the return code
+        if returncode == 0:
+            return (returncode, stdout, stderr)
+        else:
+            raise AnsibleError("failed to transfer file to %s %s:\n%s\n%s" %
+                               (to_native(in_path), to_native(out_path), to_native(stdout), to_native(stderr)))
+
+    def put_file(self, in_path, out_path):
+        ''' transfer a file from local to remote '''
+
+        super(Connection, self).put_file(in_path, out_path)
+
+        display.vvv(u"PUT {0} TO {1}".format(in_path, out_path), host=self.host)
+        if not os.path.exists(to_bytes(in_path, errors='surrogate_or_strict')):
+            raise AnsibleFileNotFound("file or module does not exist: {0}".format(to_native(in_path)))
+
+        return self._file_transport_command(in_path, out_path, 'put')
+
+    def fetch_file(self, in_path, out_path):
+        ''' fetch a file from remote to local '''
+
+        super(Connection, self).fetch_file(in_path, out_path)
+
+        display.vvv(u"FETCH {0} TO {1}".format(in_path, out_path), host=self.host)
+        return self._file_transport_command(in_path, out_path, 'get')
+
+    def close(self):
+        ''' terminate the connection '''
+        if self._session_id:
+
+            display.vvv(u"CLOSING SSM CONNECTION TO: {0}".format(self.instance_id), host=self.host)
+            if self._timeout:
+                self._session.terminate()
+            else:
+                cmd = b"\nexit\n"
+                self._session.communicate(cmd)
+
+            display.vvvv(u"TERMINATE SSM SESSION: {0}".format(self._session_id), host=self.host)
+            self._client.terminate_session(SessionId=self._session_id)
+            self._session_id = ''
diff --git a/test/integration/targets/connection_aws_ssm/aliases b/test/integration/targets/connection_aws_ssm/aliases
new file mode 100644
index 00000000000..96a01dcfbe2
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aliases
@@ -0,0 +1,6 @@
+cloud/aws
+destructive
+shippable/aws/group1
+non_local
+needs/root
+needs/target/connection
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup.yml b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup.yml
new file mode 100644
index 00000000000..7cd735b9a1c
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup.yml
@@ -0,0 +1,3 @@
+- hosts: localhost
+  roles:
+    - role: aws_ssm_integration_test_setup
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/README.md b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/README.md
new file mode 100644
index 00000000000..bc12a83e1d7
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/README.md
@@ -0,0 +1,43 @@
+# AWS SSM Integration Test Setup
+
+## aws_ssm_integration_test_setup_teardown
+
+An Ansible role was created to perform integration test across aws_ssm connection plugin. The role performs the following actions.
+
+- Create AWS Resources in user specified region.
+- Perform integration Test across aws_ssm connection plugin.
+- TearDown/Remove AWS Resources that are created for testing plugin.
+
+### Prerequisites
+
+- Make sure the machine used for testing already has Ansible repo with ssm connection plugin.
+- AWS CLI/IAM-Role configured to the machine which has permissions to spin-up AWS resources.
+
+### Variables referred in Ansible Role
+
+The following table provide details about variables referred within Ansible Role.
+
+| Variable Name | Details |
+| ------ | ------ |
+| aws_region | Name of AWS-region |
+| iam_role_name | Name of IAM Role which will be attached to newly-created EC2-Instance |
+| iam_policy_name | Name of IAM Policy which will be attached to the IAM role referred above |
+| instance_type | Instance type user for creating EC2-Instance |
+| instance_id | AWS EC2 Instance-Id (This gets populated by role) |
+| bucket_name | Name of S3 buckted used by SSM (This gets populated by role) |
+
+### Example Playbook
+
+A sample example to demonstrate the usage of role within Ansible-playbook.(Make sure the respective variables are passed as parameters.)
+
+```yaml
+   - hosts: localhost
+     roles:
+       - aws_ssm_integration_test_setup_teardown
+```
+
+#### Author's Information
+
+Krishna Nand Choudhary (krishnanandchoudhary)
+Nikhil Araga (araganik)
+Gaurav Ashtikar (gau1991)
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/defaults/main.yml b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/defaults/main.yml
new file mode 100644
index 00000000000..d6e025594fa
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/defaults/main.yml
@@ -0,0 +1,4 @@
+---
+instance_type: t2.micro
+linux_ami_name: amzn-ami-hvm-2018.03.0.20190611-x86_64-ebs
+windows_ami_name: Windows_Server-2019-English-Full-Base-2019.11.13
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/files/ec2-trust-policy.json b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/files/ec2-trust-policy.json
new file mode 100644
index 00000000000..63d22eaecd8
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/files/ec2-trust-policy.json
@@ -0,0 +1,13 @@
+{
+	"Version": "2008-10-17",
+	"Statement": [
+	  {
+		"Sid": "",
+		"Effect": "Allow",
+		"Principal": {
+		  "Service": "ec2.amazonaws.com"
+		},
+		"Action": "sts:AssumeRole"
+	  }
+	]
+  }
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/tasks/debian.yml b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/tasks/debian.yml
new file mode 100644
index 00000000000..b75f3ec30a3
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/tasks/debian.yml
@@ -0,0 +1,25 @@
+---
+- name: Download SSM plugin
+  get_url:
+    url: https://s3.amazonaws.com/session-manager-downloads/plugin/latest/ubuntu_64bit/session-manager-plugin.deb
+    dest: /tmp/session-manager-plugin.deb
+    mode: 0440
+  tags: setup_infra
+
+# We are not install deb package here, as deb package has systemd service which fails during the installation
+# on containerized env. As we will only session-manager-plugin executable, we are extracting and copying deb file.
+- name: Extract SSM plugin Deb File
+  shell: ar x session-manager-plugin.deb
+  args:
+    chdir: /tmp
+  tags: setup_infra
+
+- name: Extract SSM Plugin Control File
+  shell: tar -zxvf data.tar.gz -C /
+  args:
+    chdir: /tmp
+  tags: setup_infra
+
+- name: Check the SSM Plugin
+  shell: /usr/local/sessionmanagerplugin/bin/session-manager-plugin --version
+  tags: setup_infra
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/tasks/main.yml b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/tasks/main.yml
new file mode 100644
index 00000000000..dae7e27747a
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/tasks/main.yml
@@ -0,0 +1,156 @@
+---
+## Task file for setup/teardown AWS resources for aws_ssm integration testing
+- block:
+    - name: set up aws connection info
+      set_fact:
+        aws_connection_info: &aws_connection_info
+          aws_access_key: "{{aws_access_key}}"
+          aws_secret_key: "{{aws_secret_key}}"
+          security_token: "{{security_token}}"
+          region: "{{aws_region}}"
+      no_log: yes
+
+    - name: AMI Lookup
+      ec2_ami_info:
+        owners: 'amazon'
+        filters:
+          name: '{{ item }}'
+        <<: *aws_connection_info  
+      register: ec2_amis
+      loop:
+        - '{{ linux_ami_name }}'
+        - '{{ windows_ami_name }}'
+
+    - name: Set facts with latest AMIs
+      vars:
+        latest_linux_ami: '{{ ec2_amis.results[0].images | sort(attribute="creation_date") | last }}'
+        latest_windows_ami: '{{ ec2_amis.results[1].images | sort(attribute="creation_date") | last }}'
+      set_fact:
+        linux_ami_id: '{{ latest_linux_ami.image_id }}'
+        windows_ami_id: '{{ latest_windows_ami.image_id }}'
+
+    - name: Install Session Manager Plugin for Debian/Ubuntu
+      include_tasks: debian.yml
+      when: ansible_distribution == "Ubuntu" or ansible_distribution == "Debian"
+      register: install_plugin_debian
+
+    - name: Install Session Manager Plugin for RedHat/Amazon
+      include_tasks: redhat.yml
+      when: ansible_distribution == "CentOS" or ansible_distribution == "RedHat" or ansible_distribution == "Amazon"
+      register: install_plugin_redhat
+
+    - name: Fail if the plugin was not installed
+      fail:
+        msg: The distribution does not contain the required Session Manager Plugin
+      when:
+        - install_plugin_debian is skipped
+        - install_plugin_redhat is skipped
+
+    - name: Install Boto3
+      pip:
+        name: boto3
+
+    - name: Install Boto
+      pip:
+        name: boto
+
+    - name: Ensure IAM instance role exists
+      iam_role:
+        name: "ansible-test-{{resource_prefix}}-aws-ssm-role"
+        assume_role_policy_document: "{{ lookup('file','ec2-trust-policy.json') }}"
+        state: present
+        create_instance_profile: yes
+        managed_policy:
+        - AmazonEC2RoleforSSM
+        <<: *aws_connection_info
+      register: role_output
+
+    - name: Create S3 bucket
+      s3_bucket:
+        name: "{{resource_prefix}}-aws-ssm-s3"
+        <<: *aws_connection_info
+      register: s3_output
+
+    - name: Wait for IAM Role getting created
+      pause:
+        seconds: 10
+
+    - name: Create Linux EC2 instance
+      ec2:
+        instance_type: "{{instance_type}}"
+        image: "{{linux_ami_id}}"
+        wait: "yes"
+        count: 1
+        instance_profile_name: "{{role_output.iam_role.role_name}}"
+        instance_tags:
+          Name: "{{resource_prefix}}-integration-test-aws-ssm-linux"
+        user_data: |
+                    #!/bin/sh
+                    sudo systemctl start amazon-ssm-agent
+        state: present
+        <<: *aws_connection_info
+      register: linux_output
+
+    - name: Create Windows EC2 instance
+      ec2:
+        instance_type: "{{instance_type}}"
+        image: "{{windows_ami_id}}"
+        wait: "yes"
+        count: 1
+        instance_profile_name: "{{role_output.iam_role.role_name}}"
+        instance_tags:
+          Name: "{{resource_prefix}}-integration-test-aws-ssm-windows"
+        user_data: |
+                    <powershell>
+                    Invoke-WebRequest -Uri "https://amazon-ssm-us-east-1.s3.amazonaws.com/latest/windows_amd64/AmazonSSMAgentSetup.exe" -OutFile "C:\AmazonSSMAgentSetup.exe"
+                    Start-Process -FilePath C:\AmazonSSMAgentSetup.exe -ArgumentList "/S","/v","/qn" -Wait
+                    Restart-Service AmazonSSMAgent
+                    </powershell>
+        state: present
+        <<: *aws_connection_info
+      register: windows_output
+
+    - name: Wait for EC2 to be available
+      wait_for_connection:
+        delay: 300
+
+    - name: Create Inventory file for Linux host
+      template:
+        dest: "{{playbook_dir}}/inventory-linux.aws_ssm"
+        src: inventory-linux.aws_ssm.j2
+
+    - name: Create Inventory file for Windows host
+      template:
+        dest: "{{playbook_dir}}/inventory-windows.aws_ssm"
+        src: inventory-windows.aws_ssm.j2
+
+    - name: Create AWS Keys Environement
+      template:
+        dest: "{{playbook_dir}}/aws-env-vars.sh"
+        src: aws-env-vars.j2
+      no_log: yes
+
+  always:
+    - name: Create EC2 Linux vars_to_delete.yml
+      template:
+        dest: "{{playbook_dir}}/ec2_linux_vars_to_delete.yml"
+        src: ec2_linux_vars_to_delete.yml.j2
+      ignore_errors: yes
+
+    - name: Create EC2 Windows vars_to_delete.yml
+      template:
+        dest: "{{playbook_dir}}/ec2_windows_vars_to_delete.yml"
+        src: ec2_windows_vars_to_delete.yml.j2
+      ignore_errors: yes
+
+    - name: Create S3 vars_to_delete.yml
+      template:
+        dest: "{{playbook_dir}}/s3_vars_to_delete.yml"
+        src: s3_vars_to_delete.yml.j2
+      ignore_errors: yes
+
+    - name: Create IAM Role vars_to_delete.yml
+      template:
+        dest: "{{playbook_dir}}/iam_role_vars_to_delete.yml"
+        src: iam_role_vars_to_delete.yml.j2
+      ignore_errors: yes
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/tasks/redhat.yml b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/tasks/redhat.yml
new file mode 100644
index 00000000000..6bf73a02b71
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/tasks/redhat.yml
@@ -0,0 +1,13 @@
+---
+- name: Download SSM plugin
+  get_url:
+    url: https://s3.amazonaws.com/session-manager-downloads/plugin/latest/linux_64bit/session-manager-plugin.rpm
+    dest: /tmp/session-manager-plugin.rpm
+    mode: 0440
+  tags: setup_infra
+
+- name: Install SSM Plugin
+  yum:
+    name: /tmp/session-manager-plugin.rpm
+    state: present
+  tags: setup_infra
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/aws-env-vars.j2 b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/aws-env-vars.j2
new file mode 100644
index 00000000000..1e3821ad847
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/aws-env-vars.j2
@@ -0,0 +1,4 @@
+export AWS_ACCESS_KEY_ID={{aws_access_key}}
+export AWS_SECRET_ACCESS_KEY={{aws_secret_key}}
+export AWS_SECURITY_TOKEN={{security_token}}
+export AWS_REGION={{aws_region}}
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/ec2_linux_vars_to_delete.yml.j2 b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/ec2_linux_vars_to_delete.yml.j2
new file mode 100644
index 00000000000..8af1e3b514a
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/ec2_linux_vars_to_delete.yml.j2
@@ -0,0 +1,2 @@
+---
+linux_instance_id: {{linux_output.instance_ids[0]}}
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/ec2_windows_vars_to_delete.yml.j2 b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/ec2_windows_vars_to_delete.yml.j2
new file mode 100644
index 00000000000..d216f37225b
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/ec2_windows_vars_to_delete.yml.j2
@@ -0,0 +1,2 @@
+---
+windows_instance_id: {{windows_output.instance_ids[0]}}
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/iam_role_vars_to_delete.yml.j2 b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/iam_role_vars_to_delete.yml.j2
new file mode 100644
index 00000000000..0d87d3ed6f7
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/iam_role_vars_to_delete.yml.j2
@@ -0,0 +1,2 @@
+---
+iam_role_name: {{role_output.iam_role.role_name}}
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/inventory-linux.aws_ssm.j2 b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/inventory-linux.aws_ssm.j2
new file mode 100644
index 00000000000..7e97e5f830f
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/inventory-linux.aws_ssm.j2
@@ -0,0 +1,12 @@
+[aws_ssm]
+{{linux_output.instance_ids[0]}} ansible_aws_ssm_instance_id={{linux_output.instance_ids[0]}} ansible_aws_ssm_region={{aws_region}}
+
+[aws_ssm:vars]
+ansible_connection=aws_ssm
+ansible_aws_ssm_bucket_name={{s3_output.name}}
+ansible_aws_ssm_plugin=/usr/local/sessionmanagerplugin/bin/session-manager-plugin
+ansible_python_interpreter=/usr/bin/env python
+
+# support tests that target testhost
+[testhost:children]
+aws_ssm
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/inventory-windows.aws_ssm.j2 b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/inventory-windows.aws_ssm.j2
new file mode 100644
index 00000000000..0b6a28c8a93
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/inventory-windows.aws_ssm.j2
@@ -0,0 +1,12 @@
+[aws_ssm]
+{{windows_output.instance_ids[0]}} ansible_aws_ssm_instance_id={{windows_output.instance_ids[0]}} ansible_aws_ssm_region={{aws_region}}
+
+[aws_ssm:vars]
+ansible_shell_type=powershell
+ansible_connection=aws_ssm
+ansible_aws_ssm_bucket_name={{s3_output.name}}
+ansible_aws_ssm_plugin=/usr/local/sessionmanagerplugin/bin/session-manager-plugin
+
+# support tests that target testhost
+[testhost:children]
+aws_ssm
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/s3_vars_to_delete.yml.j2 b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/s3_vars_to_delete.yml.j2
new file mode 100644
index 00000000000..3839fb3c6ea
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_setup/templates/s3_vars_to_delete.yml.j2
@@ -0,0 +1,2 @@
+---
+bucket_name: {{s3_output.name}}
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_teardown.yml b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_teardown.yml
new file mode 100644
index 00000000000..13c62c1f90c
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_teardown.yml
@@ -0,0 +1,3 @@
+- hosts: localhost
+  roles:
+    - role: aws_ssm_integration_test_teardown
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_teardown/README.md b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_teardown/README.md
new file mode 100644
index 00000000000..bc12a83e1d7
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_teardown/README.md
@@ -0,0 +1,43 @@
+# AWS SSM Integration Test Setup
+
+## aws_ssm_integration_test_setup_teardown
+
+An Ansible role was created to perform integration test across aws_ssm connection plugin. The role performs the following actions.
+
+- Create AWS Resources in user specified region.
+- Perform integration Test across aws_ssm connection plugin.
+- TearDown/Remove AWS Resources that are created for testing plugin.
+
+### Prerequisites
+
+- Make sure the machine used for testing already has Ansible repo with ssm connection plugin.
+- AWS CLI/IAM-Role configured to the machine which has permissions to spin-up AWS resources.
+
+### Variables referred in Ansible Role
+
+The following table provide details about variables referred within Ansible Role.
+
+| Variable Name | Details |
+| ------ | ------ |
+| aws_region | Name of AWS-region |
+| iam_role_name | Name of IAM Role which will be attached to newly-created EC2-Instance |
+| iam_policy_name | Name of IAM Policy which will be attached to the IAM role referred above |
+| instance_type | Instance type user for creating EC2-Instance |
+| instance_id | AWS EC2 Instance-Id (This gets populated by role) |
+| bucket_name | Name of S3 buckted used by SSM (This gets populated by role) |
+
+### Example Playbook
+
+A sample example to demonstrate the usage of role within Ansible-playbook.(Make sure the respective variables are passed as parameters.)
+
+```yaml
+   - hosts: localhost
+     roles:
+       - aws_ssm_integration_test_setup_teardown
+```
+
+#### Author's Information
+
+Krishna Nand Choudhary (krishnanandchoudhary)
+Nikhil Araga (araganik)
+Gaurav Ashtikar (gau1991)
diff --git a/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_teardown/tasks/main.yml b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_teardown/tasks/main.yml
new file mode 100644
index 00000000000..7993733bbd8
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/aws_ssm_integration_test_teardown/tasks/main.yml
@@ -0,0 +1,85 @@
+---
+- name: Set up AWS connection info
+  set_fact:
+    aws_connection_info: &aws_connection_info
+      aws_access_key: "{{aws_access_key}}"
+      aws_secret_key: "{{aws_secret_key}}"
+      region: "{{aws_region}}"
+      security_token: "{{security_token}}"
+  no_log: true
+
+- name: Check if ec2_linux_vars_to_delete.yml is present
+  stat:
+    path: "{{playbook_dir}}/ec2_linux_vars_to_delete.yml"
+  register: ec2_linux_vars_file
+
+- name: Include variable file to delete EC2 Linux infra
+  include_vars: "{{playbook_dir}}/ec2_linux_vars_to_delete.yml"
+  when: ec2_linux_vars_file.stat.exists == true
+
+- name: Check if ec2_windows_vars_to_delete.yml is present
+  stat:
+    path: "{{playbook_dir}}/ec2_windows_vars_to_delete.yml"
+  register: ec2_windows_vars_file
+
+- name: Include variable file to delete EC2 Windows infra
+  include_vars: "{{playbook_dir}}/ec2_windows_vars_to_delete.yml"
+  when: ec2_windows_vars_file.stat.exists == true
+
+- name: Check if s3_vars_to_delete.yml is present
+  stat:
+    path: "{{playbook_dir}}/s3_vars_to_delete.yml"
+  register: s3_vars_file
+
+- name: Include variable file to delete S3 Infra infra
+  include_vars: "{{playbook_dir}}/s3_vars_to_delete.yml"
+  when: s3_vars_file.stat.exists == true
+
+- name: Check if iam_role_vars_to_delete.yml is present
+  stat:
+    path: "{{playbook_dir}}/iam_role_vars_to_delete.yml"
+  register: iam_role_vars_file
+
+- name: Include variable file to delete IAM Role infra
+  include_vars: "{{playbook_dir}}/iam_role_vars_to_delete.yml"
+  when: iam_role_vars_file.stat.exists == true
+
+- name: Terminate Windows EC2 instances that were previously launched
+  ec2:
+    instance_ids:
+      - "{{windows_instance_id}}"
+    state: absent
+    <<: *aws_connection_info
+  ignore_errors: yes
+  when: ec2_windows_vars_file.stat.exists == true
+
+- name: Terminate Linux EC2 instances that were previously launched
+  ec2:
+    instance_ids:
+      - "{{linux_instance_id}}"
+    state: absent
+    <<: *aws_connection_info
+  ignore_errors: yes
+  when: ec2_linux_vars_file.stat.exists == true
+
+- name: Delete S3 bucket
+  aws_s3:
+    bucket: "{{bucket_name}}"
+    mode: delete
+    <<: *aws_connection_info
+  ignore_errors: yes
+  when: s3_vars_file.stat.exists == true
+
+- name: Delete IAM role
+  iam_role:
+    name: "{{iam_role_name}}"
+    state: absent
+    <<: *aws_connection_info
+  ignore_errors: yes
+  when: iam_role_vars_file.stat.exists == true
+
+- name: Delete AWS keys environement
+  file:
+    path: "{{playbook_dir}}/aws-env-vars.sh"
+    state: absent
+  ignore_errors: yes
diff --git a/test/integration/targets/connection_aws_ssm/inventory.aws_ssm.template b/test/integration/targets/connection_aws_ssm/inventory.aws_ssm.template
new file mode 100644
index 00000000000..afbee1aeeca
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/inventory.aws_ssm.template
@@ -0,0 +1,10 @@
+[aws_ssm]
+@NAME ansible_aws_ssm_instance_id=@HOST ansible_aws_ssm_region=@AWS_REGION
+
+[aws_ssm:vars]
+ansible_connection=aws_ssm
+ansible_aws_ssm_bucket_name=@S3_BUCKET
+
+# support tests that target testhost
+[testhost:children]
+aws_ssm
diff --git a/test/integration/targets/connection_aws_ssm/runme.sh b/test/integration/targets/connection_aws_ssm/runme.sh
new file mode 100755
index 00000000000..1d9b38733de
--- /dev/null
+++ b/test/integration/targets/connection_aws_ssm/runme.sh
@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+
+set -eux
+
+CMD_ARGS=("$@")
+
+# Destroy Environment
+cleanup() {
+
+    cd ../connection_aws_ssm
+
+    ansible-playbook -c local aws_ssm_integration_test_teardown.yml "${CMD_ARGS[@]}"
+
+}
+
+trap "cleanup" EXIT
+
+# Setup Environment
+ansible-playbook -c local aws_ssm_integration_test_setup.yml "$@"
+
+# Export the AWS Keys
+set +x
+. ./aws-env-vars.sh
+set -x
+
+cd ../connection
+
+# Execute Integration tests for Linux
+INVENTORY=../connection_aws_ssm/inventory-linux.aws_ssm ./test.sh \
+    -e target_hosts=aws_ssm \
+    -e local_tmp=/tmp/ansible-local \
+    -e remote_tmp=/tmp/ansible-remote \
+    -e action_prefix= \
+    "$@"
+
+# Execute Integration tests for Windows
+INVENTORY=../connection_aws_ssm/inventory-windows.aws_ssm ./test.sh \
+    -e target_hosts=aws_ssm \
+    -e local_tmp=/tmp/ansible-local \
+    -e remote_tmp=c:/windows/temp/ansible-remote \
+    -e action_prefix=win_ \
+    "$@"
diff --git a/test/units/plugins/connection/test_aws_ssm.py b/test/units/plugins/connection/test_aws_ssm.py
new file mode 100644
index 00000000000..bcea207e78d
--- /dev/null
+++ b/test/units/plugins/connection/test_aws_ssm.py
@@ -0,0 +1,194 @@
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+from io import StringIO
+import pytest
+import sys
+from ansible import constants as C
+from ansible.compat.selectors import SelectorKey, EVENT_READ
+from units.compat import unittest
+from units.compat.mock import patch, MagicMock, PropertyMock
+from ansible.errors import AnsibleError, AnsibleConnectionFailure, AnsibleFileNotFound
+from ansible.module_utils.six.moves import shlex_quote
+from ansible.module_utils._text import to_bytes
+from ansible.playbook.play_context import PlayContext
+from ansible.plugins.connection import aws_ssm
+from ansible.plugins.loader import connection_loader
+
+
+@pytest.mark.skipif(sys.version_info < (2, 7), reason="requires Python 2.7 or higher")
+class TestConnectionBaseClass(unittest.TestCase):
+
+    @patch('os.path.exists')
+    @patch('subprocess.Popen')
+    @patch('select.poll')
+    @patch('boto3.client')
+    def test_plugins_connection_aws_ssm_start_session(self, boto_client, s_poll, s_popen, mock_ospe):
+        pc = PlayContext()
+        new_stdin = StringIO()
+        conn = connection_loader.get('aws_ssm', pc, new_stdin)
+        conn.get_option = MagicMock()
+        conn.get_option.side_effect = ['i1234', 'executable', 'abcd', 'i1234']
+        conn.host = 'abc'
+        mock_ospe.return_value = True
+        boto3 = MagicMock()
+        boto3.client('ssm').return_value = MagicMock()
+        conn.start_session = MagicMock()
+        conn._session_id = MagicMock()
+        conn._session_id.return_value = 's1'
+        s_popen.return_value.stdin.write = MagicMock()
+        s_poll.return_value = MagicMock()
+        s_poll.return_value.register = MagicMock()
+        s_popen.return_value.poll = MagicMock()
+        s_popen.return_value.poll.return_value = None
+        conn._stdin_readline = MagicMock()
+        conn._stdin_readline.return_value = 'abc123'
+        conn.SESSION_START = 'abc'
+        conn.start_session()
+
+    @patch('random.choice')
+    def test_plugins_connection_aws_ssm_exec_command(self, r_choice):
+        pc = PlayContext()
+        new_stdin = StringIO()
+        conn = connection_loader.get('aws_ssm', pc, new_stdin)
+        r_choice.side_effect = ['a', 'a', 'a', 'a', 'a', 'b', 'b', 'b', 'b', 'b']
+        conn.MARK_LENGTH = 5
+        conn._session = MagicMock()
+        conn._session.stdin.write = MagicMock()
+        conn._wrap_command = MagicMock()
+        conn._wrap_command.return_value = 'cmd1'
+        conn._flush_stderr = MagicMock()
+        conn._windows = MagicMock()
+        conn._windows.return_value = True
+        sudoable = True
+        conn._session.poll = MagicMock()
+        conn._session.poll.return_value = None
+        remaining = 0
+        conn._timeout = MagicMock()
+        conn._poll_stdout = MagicMock()
+        conn._poll_stdout.poll = MagicMock()
+        conn._poll_stdout.poll.return_value = True
+        conn._session.stdout = MagicMock()
+        conn._session.stdout.readline = MagicMock()
+        begin = True
+        mark_end = 'a'
+        line = ['a', 'b']
+        conn._post_process = MagicMock()
+        conn._post_process.return_value = 'test'
+        conn._session.stdout.readline.side_effect = iter(['aaaaa\n', 'Hi\n', '0\n', 'bbbbb\n'])
+        conn.get_option = MagicMock()
+        conn.get_option.return_value = 1
+        cmd = MagicMock()
+        returncode = 'a'
+        stdout = 'b'
+        return (returncode, stdout, conn._flush_stderr)
+
+    def test_plugins_connection_aws_ssm_prepare_terminal(self):
+        pc = PlayContext()
+        new_stdin = StringIO()
+        conn = connection_loader.get('aws_ssm', pc, new_stdin)
+        conn.is_windows = MagicMock()
+        conn.is_windows.return_value = True
+
+    def test_plugins_connection_aws_ssm_wrap_command(self):
+        pc = PlayContext()
+        new_stdin = StringIO()
+        conn = connection_loader.get('aws_ssm', pc, new_stdin)
+        conn.is_windows = MagicMock()
+        conn.is_windows.return_value = True
+        return('windows1')
+
+    def test_plugins_connection_aws_ssm_post_process(self):
+        pc = PlayContext()
+        new_stdin = StringIO()
+        conn = connection_loader.get('aws_ssm', pc, new_stdin)
+        conn.is_windows = MagicMock()
+        conn.is_windows.return_value = True
+        success = 3
+        fail = 2
+        conn.stdout = MagicMock()
+        returncode = 0
+        return(returncode, conn.stdout)
+
+    @patch('subprocess.Popen')
+    def test_plugins_connection_aws_ssm_flush_stderr(self, s_popen):
+        pc = PlayContext()
+        new_stdin = StringIO()
+        conn = connection_loader.get('aws_ssm', pc, new_stdin)
+        conn.poll_stderr = MagicMock()
+        conn.poll_stderr.register = MagicMock()
+        conn.stderr = None
+        s_popen.poll().return_value = 123
+        return(conn.stderr)
+
+    @patch('boto3.client')
+    def test_plugins_connection_aws_ssm_get_url(self, boto):
+        pc = PlayContext()
+        new_stdin = StringIO()
+        conn = connection_loader.get('aws_ssm', pc, new_stdin)
+        boto3 = MagicMock()
+        boto3.client('s3').return_value = MagicMock()
+        boto3.generate_presigned_url.return_value = MagicMock()
+        return (boto3.generate_presigned_url.return_value)
+
+    @patch('os.path.exists')
+    def test_plugins_connection_aws_ssm_put_file(self, mock_ospe):
+        pc = PlayContext()
+        new_stdin = StringIO()
+        conn = connection_loader.get('aws_ssm', pc, new_stdin)
+        conn._connect = MagicMock()
+        conn._file_transport_command = MagicMock()
+        conn._file_transport_command.return_value = (0, 'stdout', 'stderr')
+        res, stdout, stderr = conn.put_file('/in/file', '/out/file')
+
+    def test_plugins_connection_aws_ssm_fetch_file(self):
+        pc = PlayContext()
+        new_stdin = StringIO()
+        conn = connection_loader.get('aws_ssm', pc, new_stdin)
+        conn._connect = MagicMock()
+        conn._file_transport_command = MagicMock()
+        conn._file_transport_command.return_value = (0, 'stdout', 'stderr')
+        res, stdout, stderr = conn.fetch_file('/in/file', '/out/file')
+
+    @patch('subprocess.check_output')
+    @patch('boto3.client')
+    def test_plugins_connection_file_transport_command(self, boto_client, s_check_output):
+        pc = PlayContext()
+        new_stdin = StringIO()
+        conn = connection_loader.get('aws_ssm', pc, new_stdin)
+        conn.get_option = MagicMock()
+        conn.get_option.side_effect = ['1', '2', '3', '4', '5']
+        conn._get_url = MagicMock()
+        conn._get_url.side_effect = ['url1', 'url2']
+        boto3 = MagicMock()
+        boto3.client('s3').return_value = MagicMock()
+        conn.get_option.return_value = 1
+        ssm_action = 'get'
+        get_command = MagicMock()
+        put_command = MagicMock()
+        conn.exec_command = MagicMock()
+        conn.exec_command.return_value = (put_command, None, False)
+        conn.download_fileobj = MagicMock()
+        (returncode, stdout, stderr) = conn.exec_command(put_command, in_data=None, sudoable=False)
+        returncode = 0
+        (returncode, stdout, stderr) = conn.exec_command(get_command, in_data=None, sudoable=False)
+
+    @patch('subprocess.check_output')
+    def test_plugins_connection_aws_ssm_close(self, s_check_output):
+        pc = PlayContext()
+        new_stdin = StringIO()
+        conn = connection_loader.get('aws_ssm', pc, new_stdin)
+        conn.instance_id = "i-12345"
+        conn._session_id = True
+        conn.get_option = MagicMock()
+        conn.get_option.side_effect = ["/abc", "pqr"]
+        conn._session = MagicMock()
+        conn._session.terminate = MagicMock()
+        conn._session.communicate = MagicMock()
+        conn._terminate_session = MagicMock()
+        conn._terminate_session.return_value = ''
+        conn._session_id = MagicMock()
+        conn._session_id.return_value = 'a'
+        conn._client = MagicMock()
+        conn.close()