test(acceptance): define missing steps

Author: parfeon

pytest.ini

@@ -1,5 +1,7 @@
 import json
 from behave import then
+from soupsieve.util import lower
+
 from pubnub.exceptions import PubNubException
 
 
@@ -18,6 +20,11 @@ def step_impl(context, channel):
     context.token_resource = context.parsed_token["resources"]["channels"].get(channel.strip("'"))
     assert context.token_resource
 
+@then("token {data_type} permission {permission}")
+def step_impl(context, data_type, permission):
+    assert context.token_resource
+    assert context.token_resource[lower(permission)] == True
+
 
 @then("the token contains the authorized UUID {test_uuid}")
 def step_impl(context, test_uuid):
@@ -80,6 +87,75 @@ def step_impl(context):
     context.pam_call_error = json.loads(context.pam_call_result._errormsg)
 
 
+@then("the error status code is {error_code}")
+def step_impl(context, error_code):
+    assert context.pam_call_error['status'] == int(error_code)
+
+
+@then("the auth error message is '{error_message}'")
+@then("the error message is '{error_message}'")
+def step_impl(context, error_message):
+    if 'message' in context.pam_call_error:
+        assert context.pam_call_error['message'] == error_message
+    elif 'error' in context.pam_call_error and 'message' in context.pam_call_error['error']:
+        assert context.pam_call_error['error']['message'] == error_message
+    else:
+        raise AssertionError("Unexpected payload: {}".format(context.pam_call_error))
+
+
+@then("the error detail message is not empty")
+def step_impl(context, details_message):
+    if 'error' in context.pam_call_error and 'details' in context.pam_call_error['error']:
+        assert len(context.pam_call_error['error']['details']) > 0
+        assert 'message' in context.pam_call_error['error']['details'][0]
+        assert len(context.pam_call_error['error']['details'][0]['message']) > 0
+    else:
+        raise AssertionError("Unexpected payload: {}".format(context.pam_call_error))
+
+
+@then("the error detail message is '{details_message}'")
+def step_impl(context, details_message):
+    if 'error' in context.pam_call_error and 'details' in context.pam_call_error['error']:
+        assert len(context.pam_call_error['error']['details']) > 0
+        assert 'message' in context.pam_call_error['error']['details'][0]
+        assert context.pam_call_error['error']['details'][0]['message'] == details_message
+    else:
+        raise AssertionError("Unexpected payload: {}".format(context.pam_call_error))
+
+
+@then("the error detail location is '{details_location}'")
+def step_impl(context, details_location):
+    if 'error' in context.pam_call_error and 'details' in context.pam_call_error['error']:
+        assert len(context.pam_call_error['error']['details']) > 0
+        assert 'location' in context.pam_call_error['error']['details'][0]
+        assert context.pam_call_error['error']['details'][0]['location'] == details_location
+    else:
+        raise AssertionError("Unexpected payload: {}".format(context.pam_call_error))
+
+
+@then("the error detail location type is '{details_location_type}'")
+def step_impl(context, details_location_type):
+    if 'error' in context.pam_call_error and 'details' in context.pam_call_error['error']:
+        assert len(context.pam_call_error['error']['details']) > 0
+        assert 'locationType' in context.pam_call_error['error']['details'][0]
+        assert context.pam_call_error['error']['details'][0]['locationType'] == details_location_type
+    else:
+        raise AssertionError("Unexpected payload: {}".format(context.pam_call_error))
+
+
+@then("the error service is '{error_service}'")
+def step_impl(context, error_service):
+    assert context.pam_call_error['service'] == error_service
+
+
+@then("the error source is '{error_source}'")
+def step_impl(context, error_source):
+    if 'error' in context.pam_call_error and 'source' in context.pam_call_error['error']:
+        assert context.pam_call_error['error']['source'] == error_source
+    else:
+        raise AssertionError("Unexpected payload: {}".format(context.pam_call_error))
+
+
 @then("the result is successful")
 def step_impl(context):
     assert context.publish_result.result.timetoken

tests/acceptance/pam/steps/then_steps.py

@@ -1,3 +1,4 @@
+import asyncio
 import pytest
 
 from pubnub.models.consumer.signal import PNSignalResult
@@ -52,12 +53,21 @@ async def test_change_uuid_no_lock():
     assert isinstance(envelope.status, PNStatus)
 
 
-def test_uuid_validation_at_init():
+@pytest.fixture
+def event_loop():
+    loop = asyncio.new_event_loop()
+    try:
+        yield loop
+    finally:
+        loop.run_until_complete(asyncio.sleep(0))
+        loop.close()
+
+def test_uuid_validation_at_init(event_loop):
     with pytest.raises(AssertionError) as exception:
         pnconf = PNConfiguration()
         pnconf.publish_key = "demo"
         pnconf.subscribe_key = "demo"
-        PubNubAsyncio(pnconf)
+        PubNubAsyncio(pnconf, custom_event_loop=event_loop)
 
     assert str(exception.value) == 'UUID missing or invalid type'
 

tests/integrational/asyncio/test_change_uuid.py

@@ -1,3 +1,4 @@
+import asyncio
 import pytest
 
 from pubnub.pubnub_asyncio import PubNubAsyncio
@@ -7,13 +8,24 @@
 from tests.helper import pnconf_mc_copy
 from tests.integrational.vcr_helper import pn_vcr
 
+@pytest.fixture
+def event_loop():
+    loop = asyncio.new_event_loop()
+    try:
+        yield loop
+    finally:
+        loop.run_until_complete(asyncio.sleep(0))
+        loop.close()
 
 @pytest.fixture
-def pn():
+def pn(event_loop):
     config = pnconf_mc_copy()
     config.enable_subscribe = False
-    pn = PubNubAsyncio(config)
-    yield pn
+    pn = PubNubAsyncio(config, custom_event_loop=event_loop)
+    try:
+        yield pn
+    finally:
+        event_loop.run_until_complete(pn.stop())
 
 
 @pn_vcr.use_cassette(