swindon 0.7.8

An HTTP edge (frontend) server with smart websockets support 
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264

import aiohttp
import asyncio
import json
import pytest
import re

from aiohttp import ClientSession
from async_timeout import timeout
from unittest import mock


def assert_auth(req):
    assert req.path == '/swindon/authorize_connection'
    assert req.headers["Host"] == "swindon.internal"
    assert req.headers['Content-Type'] == 'application/json'
    assert re.match('^swindon/(\d+\.){2}\d+$', req.headers['User-Agent'])
    assert 'Authorization' not in req.headers


async def auth(handler, auth_data):
    req = await handler.request()
    assert req.path == '/swindon/authorize_connection'
    meta, *tail = await req.json()
    assert 'connection_id' in meta
    cid = meta['connection_id']
    ws = await handler.json_response(auth_data)
    hello = await ws.receive_json()
    assert hello == ['hello', {}, auth_data]
    return cid, ws


async def put(url, loop):
    async with ClientSession(loop=loop) as s:
        async with s.put(url,
            headers={"Content-Type": "application/json"}) as resp:
            assert resp.status == 204


async def delete(url, loop):
    async with ClientSession(loop=loop) as s:
        async with s.delete(url) as resp:
            assert resp.status == 204


async def post(url, data, loop):
    async with ClientSession(loop=loop) as s:
        async with s.post(url,
            headers={"Content-Type": "application/json"},
            data=data) as resp:
            assert resp.status == 204


async def test_simple_replication(swindon_two, proxy_server, loop):
    peerA, peerB = swindon_two
    # await asyncio.sleep(1.5, loop=loop)  # wait reconnect

    urlA = peerA.url / 'swindon-lattice'
    urlB = peerB.url / 'swindon-lattice'
    async with proxy_server(port=peerA.proxy.port) as proxy:
        handlerA = proxy.swindon_lattice(urlA, timeout=1)
        cid1, ws1 = await auth(
            handlerA, {'user_id': 'replicated-user:1', 'username': 'John'})

        url = peerA.api3 / 'v1/connection' / cid1 / 'subscriptions' / 'general'
        await put(url, loop)

        handlerB = proxy.swindon_lattice(urlB, timeout=1)
        cid2, ws2 = await auth(
            handlerB, {'user_id': 'replicated-user:1', 'username': 'John'})

        url = peerB.api3 / 'v1/connection' / cid2 / 'subscriptions' / 'general'
        await put(url, loop)

        # subscribe both to some topic and publish into one peer.

        data = b'{"test": "message"}'
        await post(peerB.api3 / 'v1/publish/general', data, loop)

        msg1 = await ws1.receive_json()
        msg2 = await ws2.receive_json()
        assert msg1 == [
            'message', {'topic': 'general'}, {'test': 'message'}]
        assert msg2 == [
            'message', {'topic': 'general'}, {'test': 'message'}]


@pytest.mark.parametrize("through", ["peerA", "peerB"])
async def test_non_local_connections(swindon_two, proxy_server, loop, through):
    peerA, peerB = swindon_two
    if through == 'peerA':
        subscribe_peer = peerA
    else:
        subscribe_peer = peerB

    # await asyncio.sleep(1.5, loop=loop)

    # The use case:
    # * client A connect to peer A; receive peer-local cid(0);
    # * client B connect to peer B; receive peer-lcoal cid(0);
    # * backend sends request to peer A to subscribe client A to some topic
    # * action replicated to peer B;
    # Expected result:
    # client B not subscribed to that topic
    urlA = peerA.url / 'swindon-lattice'
    urlB = peerB.url / 'swindon-lattice'
    async with proxy_server(port=peerA.proxy.port) as proxy:
        handlerA = proxy.swindon_lattice(urlA, timeout=1)
        cid1, ws1 = await auth(
            handlerA, {'user_id': 'replicated-user:1', 'username': 'John'})

        handlerB = proxy.swindon_lattice(urlB, timeout=1)
        cid2, ws2 = await auth(
            handlerB, {'user_id': 'replicated-user:2', 'username': 'John'})

        # Subscribe only first user;
        url = subscribe_peer.api3 / 'v1/connection'
        url = url / cid1 / 'subscriptions/general'
        await put(url, loop)
        await asyncio.sleep(0.1)  # wait for replication
        # publish some data
        data = b'{"hello": "world"}'
        await post(peerA.api3 / 'v1/publish/general', data, loop)

        with timeout(1):
            msg1 = await ws1.receive_json()
            assert msg1 == [
                "message", {"topic": "general"}, {"hello": "world"}]

        with pytest.raises(asyncio.TimeoutError):
            with timeout(1, loop=loop):
                assert await ws2.receive_json() is None


@pytest.mark.parametrize("through", ["peerA", "peerB"])
async def test_topic_unsubscribe(swindon_two, proxy_server, loop,
                                 user_id, through):
    peerA, peerB = swindon_two
    if through == 'peerA':
        control = peerA
    else:
        control = peerB
    urlA = peerA.url / 'swindon-lattice'
    # urlB = peerB.url / 'swindon-lattice'
    async with proxy_server(port=peerA.proxy.port) as proxy:
        handlerA = proxy.swindon_lattice(urlA, timeout=1)
        cid, ws = await auth(handlerA, {"user_id": user_id})

        topic_url = control.api3 / 'v1/connection'
        topic_url = topic_url / cid / 'subscriptions/xxxx'
        await put(topic_url, loop)

        # publish some data
        data = b'["hello", "from", "peerA"]'
        await post(peerA.api3 / 'v1/publish/xxxx', data, loop)
        msg = await ws.receive_json()
        assert msg == [
            "message", {"topic": "xxxx"}, ["hello", "from", "peerA"]]

        data = b'["hello", "from", "peerB"]'
        await post(peerB.api3 / 'v1/publish/xxxx', data, loop)
        msg = await ws.receive_json()
        assert msg == [
            "message", {"topic": "xxxx"}, ["hello", "from", "peerB"]]

        topic_url = control.api3 / 'v1/connection'
        topic_url = topic_url / cid / 'subscriptions/xxxx'
        await delete(topic_url, loop)
        # XXX: publish can be received earlier than unsubscribe "replicated"
        await asyncio.sleep(.05, loop=loop)

        # publish some data
        data = b'["hello", "world"]'
        await post(peerA.api3 / 'v1/publish/xxxx', data, loop)
        await post(peerB.api3 / 'v1/publish/xxxx', data, loop)
        with pytest.raises(asyncio.TimeoutError):
            with timeout(1, loop=loop):
                assert await ws.receive_json() is None


@pytest.mark.parametrize('by_user_id', [True, False])
async def test_swindon_user(proxy_server, swindon_two, loop,
        user_id, user_id2, by_user_id):

    peerA, peerB = swindon_two

    urlA = peerA.url / 'swindon-lattice'
    urlB = peerB.url / 'swindon-lattice'
    async with proxy_server(port=peerA.proxy.port) as proxy:
        handlerA = proxy.swindon_lattice(urlA, timeout=1)
        req = await handlerA.request()
        assert_auth(req)
        meta, args, kwargs = await req.json()
        cid1 = meta['connection_id']

        async with aiohttp.ClientSession(loop=loop) as s:
            subscr_url = peerA.api3 / 'v1/connection' / cid1 / 'users'
            data = json.dumps([user_id]).encode('utf-8')
            async with s.put(subscr_url,
                    headers={'Content-Type': 'application/json'},
                    data=data) as resp:
                assert resp.status == 204

        ws1 = await handlerA.json_response({
            "user_id": user_id, "username": "Jack"})
        hello = await ws1.receive_json()
        assert hello == [
            'hello', {}, {'user_id': user_id, 'username': 'Jack'}]
        msg = await ws1.receive_json()
        assert msg == ['lattice',
            {'namespace': 'swindon.user'},
            {user_id: {'status_register': [mock.ANY, 'active']}}]

        # Log in second user but subscribe him later

        handlerB = proxy.swindon_lattice(urlB, timeout=1)
        req = await handlerB.request()
        assert_auth(req)
        meta, args, kwargs = await req.json()
        cid2 = meta['connection_id']

        ws2 = await handlerB.json_response({
            "user_id": user_id2, "username": "John"})
        hello = await ws2.receive_json()
        assert hello == [
            'hello', {}, {'user_id': user_id2, 'username': 'John'}]

        # Now subscribe both users to each other
        async with aiohttp.ClientSession(loop=loop) as s:
            data = json.dumps([user_id, user_id2]).encode('utf-8')

            subscr_url = peerB.api3 / 'v1/connection' / cid2 / 'users'
            async with s.put(subscr_url,
                    headers={'Content-Type': 'application/json'},
                    data=data) as resp:
                assert resp.status == 204

            if by_user_id:
                subscr_url = peerB.api3 / 'v1/user' / user_id / 'users'
                async with s.put(subscr_url,
                    headers={'Content-Type': 'application/json'},
                    data=data) as resp:
                    assert resp.status == 204
            else:
                subscr_url = peerB.api3 / 'v1/connection' / cid1 / 'users'
                async with s.put(subscr_url,
                        headers={'Content-Type': 'application/json'},
                        data=data) as resp:
                    assert resp.status == 204

        with timeout(1):
            msg = await ws1.receive_json()
        # TODO(tailhook) this user is fully sent on resubscription
        #                we can optimize it to send only new users
        assert msg == ['lattice',
            {'namespace': 'swindon.user'},
            {user_id: {'status_register': [mock.ANY, 'active']},
             user_id2: {'status_register': [mock.ANY, 'active']}}]

        with timeout(1):
            msg = await ws2.receive_json()
        assert msg == ['lattice',
            {'namespace': 'swindon.user'},
            {user_id: {'status_register': [mock.ANY, 'active']},
             user_id2: {'status_register': [mock.ANY, 'active']}}]