pypilot/test_web_api.py at master · wbijen/pypilot · 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
import json
import unittest

from web import web as web_module


class FakePypilotClient:
    def __init__(self, values=None, connect=True):
        self.connection = False
        self.connect = connect
        self.values = values or {}
        self.sent = []
        self.requested = []
        self.disconnected = False

    def poll(self, timeout=0):
        if self.connect:
            self.connection = True

    def send(self, message):
        self.sent.append(message)

    def watch(self, name, value=True):
        self.requested.append(name)

    def receive(self):
        values = {}
        while self.requested:
            name = self.requested.pop(0)
            if name in self.values:
                values[name] = self.values[name]
        return values

    def disconnect(self):
        self.disconnected = True


class WebApiTests(unittest.TestCase):
    def setUp(self):
        self.created_clients = []
        web_module.app.config['TESTING'] = True
        web_module.app.config['PYPILOT_CLIENT_FACTORY'] = self.make_client
        self.app = web_module.app.test_client()

    def tearDown(self):
        web_module.app.config.pop('PYPILOT_CLIENT_FACTORY', None)

    def make_client(self):
        client = FakePypilotClient(values={
            'ap.enabled': True,
            'profile': {'name': 'cruise'},
        })
        self.created_clients.append(client)
        return client

    def test_api_can_set_and_get_values(self):
        response = self.app.post('/api/pypilot', json={
            'set': {
                'ap.enabled': True,
                'profile': {'name': 'cruise'},
            },
            'get': ['ap.enabled', 'profile'],
        })

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.get_json(), {
            'ok': True,
            'values': {
                'ap.enabled': True,
                'profile': {'name': 'cruise'},
            },
        })

        client = self.created_clients[0]
        sent = {}
        for message in client.sent:
            name, value = message.rstrip().split('=', 1)
            sent[name] = json.loads(value)
        self.assertEqual(sent, {
            'ap.enabled': True,
            'profile': {'name': 'cruise'},
        })
        self.assertTrue(client.disconnected)

    def test_api_rejects_invalid_payload(self):
        response = self.app.post('/api/pypilot', json={'set': 'ap.enabled'})

        self.assertEqual(response.status_code, 400)
        self.assertFalse(response.get_json()['ok'])

    def test_api_reports_connection_failure(self):
        def disconnected_client():
            client = FakePypilotClient(connect=False)
            self.created_clients.append(client)
            return client

        web_module.app.config['PYPILOT_CLIENT_FACTORY'] = disconnected_client

        response = self.app.post('/api/pypilot', json={'get': ['ap.enabled']})

        self.assertEqual(response.status_code, 503)
        self.assertEqual(response.get_json(), {
            'ok': False,
            'message': 'Unable to connect to pypilot.',
        })
        self.assertTrue(self.created_clients[-1].disconnected)


if __name__ == '__main__':
    unittest.main()