python/tests/integrational/asyncio/test_state.py at 856af80b8db0079df7d75658642be6ad7c0f3822 · pubnub/python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
import asyncio
import pytest
import logging
import pubnub as pn

from pubnub.models.consumer.presence import PNSetStateResult, PNGetStateResult
from pubnub.pubnub_asyncio import PubNubAsyncio
from tests.helper import pnconf, pnconf_copy, pnconf_sub_copy, pnconf_pam_env_copy
from tests.integrational.vcr_asyncio_sleeper import get_sleeper, VCR599Listener
from tests.integrational.vcr_helper import pn_vcr


pn.set_stream_logger('pubnub', logging.DEBUG)


@pn_vcr.use_cassette(
    'tests/integrational/fixtures/asyncio/state/single_channel.yaml',
    filter_query_parameters=['uuid', 'pnsdk'],
    match_on=['method', 'host', 'path', 'state_object_in_query'])
@pytest.mark.asyncio
async def test_single_channelx(event_loop):
    pubnub = PubNubAsyncio(pnconf_copy(), custom_event_loop=event_loop)
    ch = 'test-state-asyncio-ch'
    pubnub.config.uuid = 'test-state-asyncio-uuid'
    state = {"name": "Alex", "count": 5}

    env = await pubnub.set_state() \
        .channels(ch) \
        .state(state) \
        .future()

    assert env.result.state['name'] == "Alex"
    assert env.result.state['count'] == 5

    env = await pubnub.get_state() \
        .channels(ch) \
        .future()

    assert env.result.channels[ch]['name'] == "Alex"
    assert env.result.channels[ch]['count'] == 5

    await pubnub.stop()


@get_sleeper('tests/integrational/fixtures/asyncio/state/single_channel_with_subscription.yaml')
@pn_vcr.use_cassette(
    'tests/integrational/fixtures/asyncio/state/single_channel_with_subscription.yaml',
    filter_query_parameters=['uuid', 'pnsdk'],
    match_on=['method', 'host', 'path', 'state_object_in_query'])
@pytest.mark.asyncio
async def test_single_channel_with_subscription(event_loop, sleeper=asyncio.sleep):
    pnconf = pnconf_sub_copy()
    pnconf.set_presence_timeout(12)
    pubnub = PubNubAsyncio(pnconf, custom_event_loop=event_loop)
    ch = 'test-state-asyncio-ch'
    pubnub.config.uuid = 'test-state-asyncio-uuid'
    state = {"name": "Alex", "count": 5}

    callback = VCR599Listener(1)
    pubnub.add_listener(callback)
    pubnub.subscribe().channels(ch).execute()

    await callback.wait_for_connect()
    await sleeper(20)

    env = await pubnub.set_state() \
        .channels(ch) \
        .state(state) \
        .future()

    assert env.result.state['name'] == "Alex"
    assert env.result.state['count'] == 5

    env = await pubnub.get_state() \
        .channels(ch) \
        .future()

    assert env.result.channels[ch]['name'] == "Alex"
    assert env.result.channels[ch]['count'] == 5

    pubnub.unsubscribe().channels(ch).execute()
    await callback.wait_for_disconnect()

    await pubnub.stop()


@pn_vcr.use_cassette(
    'tests/integrational/fixtures/asyncio/state/multiple_channel.yaml',
    filter_query_parameters=['uuid', 'pnsdk'],
    match_on=['method', 'host', 'path', 'state_object_in_query'])
@pytest.mark.asyncio
async def test_multiple_channels(event_loop):
    pubnub = PubNubAsyncio(pnconf, custom_event_loop=event_loop)
    ch1 = 'test-state-asyncio-ch1'
    ch2 = 'test-state-asyncio-ch2'
    pubnub.config.uuid = 'test-state-asyncio-uuid'
    state = {"name": "Alex", "count": 5}

    env = await pubnub.set_state() \
        .channels([ch1, ch2]) \
        .state(state) \
        .future()

    assert env.result.state['name'] == "Alex"
    assert env.result.state['count'] == 5

    env = await pubnub.get_state() \
        .channels([ch1, ch2]) \
        .future()

    assert env.result.channels[ch1]['name'] == "Alex"
    assert env.result.channels[ch2]['name'] == "Alex"
    assert env.result.channels[ch1]['count'] == 5
    assert env.result.channels[ch2]['count'] == 5

    await pubnub.stop()


@pytest.mark.asyncio
async def test_state_super_admin_call(event_loop):
    pnconf = pnconf_pam_env_copy()
    pubnub = PubNubAsyncio(pnconf, custom_event_loop=event_loop)
    ch1 = 'test-state-asyncio-ch1'
    ch2 = 'test-state-asyncio-ch2'
    pubnub.config.uuid = 'test-state-asyncio-uuid-|.*$'
    state = {"name": "Alex", "count": 5}

    env = await pubnub.set_state() \
        .channels([ch1, ch2]) \
        .state(state) \
        .future()
    assert isinstance(env.result, PNSetStateResult)

    env = await pubnub.get_state() \
        .channels([ch1, ch2]) \
        .future()
    assert isinstance(env.result, PNGetStateResult)

    await pubnub.stop()