controller/rootfs/api/tests/test_services.py at a75b226f5082218158d5098db230c7054d08b4cb · drycc/controller · 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
140
from django.contrib.auth.models import User
from django.core.cache import cache
from rest_framework.authtoken.models import Token

from api.models import App

from api.tests import DryccTransactionTestCase


class TestServices(DryccTransactionTestCase):

    """Tests push notification from build system"""

    fixtures = ['tests.json']

    def setUp(self):
        self.user = User.objects.get(username='autotest')
        self.token = Token.objects.get(user=self.user).key
        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token)

    def tearDown(self):
        # make sure every test has a clean slate for k8s mocking
        cache.clear()

    def test_settings_maintenance(self):
        """
        Test that maintenance can be applied
        """
        app_id = self.create_app()
        app = App.objects.get(id=app_id)

        # create service
        response = self.client.post(
            '/v2/apps/{}/services'.format(app_id),
            {'procfile_type': 'test', 'path_pattern': '/testep/notify'}
        )
        self.assertEqual(response.status_code, 201, response.data)

        # maintenance mode  & routable sequence
        settings = {'maintenance': True}
        response = self.client.post(
            '/v2/apps/{app_id}/settings'.format(**locals()),
            settings)
        self.assertEqual(response.status_code, 201, response.data)
        self.assertTrue(response.data['maintenance'])
        self.assertTrue(app.appsettings_set.latest().maintenance)

        settings = {'routable': False}
        response = self.client.post(
            '/v2/apps/{app_id}/settings'.format(**locals()),
            settings)
        self.assertEqual(response.status_code, 201, response.data)
        self.assertTrue(app.appsettings_set.latest().maintenance)

        settings['maintenance'] = False
        response = self.client.post(
            '/v2/apps/{app_id}/settings'.format(**locals()),
            settings)
        self.assertEqual(response.status_code, 201, response.data)
        self.assertFalse(response.data['maintenance'])
        self.assertFalse(app.appsettings_set.latest().maintenance)

        response = self.client.post(
            '/v2/apps/{app_id}/settings'.format(**locals()),
            settings)
        self.assertEqual(response.status_code, 409, response.data)
        self.assertFalse(app.appsettings_set.latest().maintenance)

    def test_service_basic_ops(self):
        """Test basic service operations."""
        app_id = self.create_app()

        # list non-existing services
        response = self.client.get('/v2/apps/{}/services'.format(app_id))
        self.assertEqual(response.status_code, 200, response.data)
        self.assertEqual(len(response.data['services']), 0)
        # create 1st service
        response = self.client.post(
            '/v2/apps/{}/services'.format(app_id),
            {'procfile_type': 'test', 'path_pattern': '/testep/notify'}
        )
        self.assertEqual(response.status_code, 201, response.data)
        # list 1st service
        response = self.client.get('/v2/apps/{}/services'.format(app_id))
        self.assertEqual(response.status_code, 200, response.data)
        self.assertEqual(len(response.data['services']), 1)
        expected1 = {
            'procfile_type': 'test',
            'path_pattern': '/testep/notify'
        }
        self.assertDictContainsSubset(expected1, response.data['services'][0])
        # update 1st service
        response = self.client.post(
            '/v2/apps/{}/services'.format(app_id),
            {'procfile_type': 'test', 'path_pattern': '/testep/notify_new'}
        )
        self.assertEqual(response.status_code, 201, response.data)
        # list 1st service and get new value
        response = self.client.get('/v2/apps/{}/services'.format(app_id))
        self.assertEqual(response.status_code, 200, response.data)
        self.assertEqual(len(response.data['services']), 1)
        expected1 = {
            'procfile_type': 'test',
            'path_pattern': '/testep/notify_new'
        }
        self.assertDictContainsSubset(expected1, response.data['services'][0])
        # create 2nd service
        response = self.client.post(
            '/v2/apps/{}/services'.format(app_id),
            {'procfile_type': 'test2', 'path_pattern': '/testep2/notify'}
        )
        self.assertEqual(response.status_code, 201, response.data)
        # list two services
        response = self.client.get('/v2/apps/{}/services'.format(app_id))
        self.assertEqual(response.status_code, 200, response.data)
        self.assertEqual(len(response.data['services']), 2)
        expected2 = {
            'procfile_type': 'test2',
            'path_pattern': '/testep2/notify'
        }
        self.assertDictContainsSubset(expected2, response.data['services'][0])
        self.assertDictContainsSubset(expected1, response.data['services'][1])
        # delete 1st
        response = self.client.delete(
            '/v2/apps/{}/services'.format(app_id),
            {'procfile_type': 'test'}
        )
        self.assertEqual(response.status_code, 204, response.data)
        # delete 2nd
        response = self.client.delete(
            '/v2/apps/{}/services'.format(app_id),
            {'procfile_type': 'test2'}
        )
        self.assertEqual(response.status_code, 204, response.data)
        # delete non-existing (1st again)
        response = self.client.delete(
            '/v2/apps/{}/services'.format(app_id),
            {'procfile_type': 'test'}
        )
        self.assertEqual(response.status_code, 404, response.data)