diff --git a/lib/ansible/plugins/test/core.py b/lib/ansible/plugins/test/core.py
index 4e5d15fb7f7..46357369060 100644
--- a/lib/ansible/plugins/test/core.py
+++ b/lib/ansible/plugins/test/core.py
@@ -26,11 +26,17 @@ from distutils.version import LooseVersion, StrictVersion
 
 from ansible import errors
 
+try:
+    from __main__ import display
+except ImportError:
+    from ansible.utils.display import Display
+    display = Display()
+
 
 def failed(result):
     ''' Test if task result yields failed '''
     if not isinstance(result, MutableMapping):
-        raise errors.AnsibleFilterError("The failed test expects a dictionary")
+        raise errors.AnsibleFilterError("The 'failed' test expects a dictionary")
     return result.get('failed', False)
 
 
@@ -42,7 +48,7 @@ def success(result):
 def changed(result):
     ''' Test if task result yields changed '''
     if not isinstance(result, MutableMapping):
-        raise errors.AnsibleFilterError("The changed test expects a dictionary")
+        raise errors.AnsibleFilterError("The 'changed' test expects a dictionary")
     if 'changed' not in result:
         changed = False
         if (
@@ -62,10 +68,24 @@ def changed(result):
 def skipped(result):
     ''' Test if task result yields skipped '''
     if not isinstance(result, MutableMapping):
-        raise errors.AnsibleFilterError("The skipped test expects a dictionary")
+        raise errors.AnsibleFilterError("The 'skipped' test expects a dictionary")
     return result.get('skipped', False)
 
 
+def finished(result):
+    ''' Test if async task has finished '''
+    if not isinstance(result, MutableMapping):
+        raise errors.AnsibleFilterError("The 'finished' test expects a dictionary")
+    if 'finished' in result:
+        # For async tasks return status
+        # NOTE: The value of finished it 0 or 1, not False or True :-/
+        return result.get('finished', 0) == 1
+    else:
+        # For non-async tasks warn user, but return as finished
+        display.warning("The 'finished' test expects an async task, but a non-async task was tested")
+        return True
+
+
 def regex(value='', pattern='', ignorecase=False, multiline=False, match_type='search'):
     ''' Expose `re` as a boolean filter using the `search` method by default.
         This is likely only useful for `search` and `match` which already
@@ -139,6 +159,9 @@ class TestModule(object):
             'skipped': skipped,
             'skip': skipped,
 
+            # async testing
+            'finished': finished,
+
             # regex
             'match': match,
             'search': search,
diff --git a/test/integration/targets/async/tasks/main.yml b/test/integration/targets/async/tasks/main.yml
index 16b9ea0b179..8f76d692778 100644
--- a/test/integration/targets/async/tasks/main.yml
+++ b/test/integration/targets/async/tasks/main.yml
@@ -33,13 +33,14 @@
         - "'cmd' in async_result"
         - "'delta' in async_result"
         - "'end' in async_result"
-        - "'finished' in async_result or async_result.finished == 1"
         - "'rc' in async_result"
         - "'start' in async_result"
         - "'stderr' in async_result"
         - "'stdout' in async_result"
         - "'stdout_lines' in async_result"
-        - "async_result.rc == 0"
+        - async_result.rc == 0
+        - async_result.finished == 1
+        - async_result is finished
 
 - name: test async without polling
   command: sleep 5
@@ -54,7 +55,8 @@
     that:
         - "'ansible_job_id' in async_result"
         - "'started' in async_result"
-        - "'finished' not in async_result or async_result.finished == 0"
+        - async_result.finished == 0
+        - async_result is not finished
 
 - name: test skipped task handling
   command: /bin/true
@@ -79,7 +81,7 @@
 - name: 'check on task started as a "fire-and-forget"'
   async_status: jid={{ fnf_task.ansible_job_id }}
   register: fnf_result
-  until: fnf_result.finished
+  until: fnf_result is finished
   retries: 10
   delay: 1
 
@@ -87,6 +89,7 @@
   assert:
     that:
     - fnf_result.finished
+    - fnf_result is finished
 
 - name: test graceful module failure
   async_test:
@@ -101,7 +104,8 @@
     that:
     - async_result.ansible_job_id is match('\d+\.\d+')
     - async_result.finished == 1
-    - async_result is changed == false
+    - async_result is finished
+    - async_result is not changed
     - async_result is failed
     - async_result.msg == 'failed gracefully'
 
@@ -118,8 +122,11 @@
     that:
     - async_result.ansible_job_id is match('\d+\.\d+')
     - async_result.finished == 1
+    - async_result is finished
     - async_result.changed == false
-    - async_result is failed == true
+    - async_result is not changed
+    - async_result.failed == true
+    - async_result is failed
     - async_result.stderr is search('failing via exception', multiline=True)
 
 - name: test leading junk before JSON
@@ -134,7 +141,9 @@
     that:
     - async_result.ansible_job_id is match('\d+\.\d+')
     - async_result.finished == 1
+    - async_result is finished
     - async_result.changed == true
+    - async_result is changed
     - async_result is successful
 
 - name: test trailing junk after JSON
@@ -149,6 +158,21 @@
     that:
     - async_result.ansible_job_id is match('\d+\.\d+')
     - async_result.finished == 1
+    - async_result is finished
     - async_result.changed == true
+    - async_result is changed
     - async_result is successful
     - async_result.warnings[0] is search('trailing junk after module output')
+
+# NOTE: This should report a warning that cannot be tested
+- name: test async properties on non-async task
+  command: sleep 1
+  register: non_async_result
+
+- name: validate response
+  assert:
+    that:
+    - non_async_result is successful
+    - non_async_result is changed
+    - non_async_result is finished
+    - "'ansible_job_id' not in non_async_result"