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
|
#!/usr/bin/env python
# Ubuntu Continuous Integration Engine
# Copyright 2014 Canonical Ltd.
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License version 3, as
# published by the Free Software Foundation.
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import httplib2
import json
import unittest
import deployers
from ci_utils.ticket_states import TicketWorkflowStep
class TicketSystemTest(deployers.DeployerTest):
"""Integration tests for ticket-system service run on a juju deployment"""
def setUp(self):
super(TicketSystemTest, self).setUp()
# Adding get_ip_and_port('ts-django') in order to allow the deployment
# further time to settle before running the tests.
# http://pad.lv/1319077
self.get_ip_and_port('ci-airline-ts-django')
def get_server_status_and_content(self, url):
"""Issue an WebUI GET and return response and page content."""
final_url = url.format(
self.get_ip_and_port('ci-airline-webui-apache')[0])
client = httplib2.Http()
return client.request(final_url, 'GET')
def get_server_status_and_json_content(self, url):
"""Issue an WebUI GET and return response and JSON content."""
resp, content = self.get_server_status_and_content(url)
jcontent = json.loads(content)
return resp, jcontent
def create_ticket(self, ticket_content):
"""Create a new ticket via the API. """
url = 'http://{}/api/v1/ticket/'
final_url = url.format(
self.get_ip_and_port('ci-airline-webui-apache')[0])
client = httplib2.Http()
resp, content = client.request(
uri=final_url,
method='POST',
headers={'Content-Type': 'application/json'},
body=json.dumps(ticket_content),
)
return resp, content
def test_ticket_system_api(self):
# `Ticket` API is exposed via the WebUI.
url = 'http://{}/api/v1/ticket'
resp, content = self.get_server_status_and_json_content(url)
self.assertEqual('200', resp['status'], content)
def test_ticket_system_openid(self):
# Ticket System OpenID pages are exposed via the WebUI.
url = 'http://{}/openid/login/'
resp, content = self.get_server_status_and_content(url)
self.assertEqual('200', resp['status'], content)
def test_ticket_system_logout(self):
# Ticket System 'logout' is exposed via the WebUI as a redirect
# to <TS>/logout/, which redirects back to the WebUI root by
# default.
url = 'http://{}/logout/'
resp, content = self.get_server_status_and_content(url)
redirect = resp.previous
self.assertIsNotNone(redirect, '/logout/ is not redirecting.')
self.assertEqual('302', redirect['status'])
next_url = 'http://{}/'.format(
self.get_ip_and_port('ci-airline-webui-apache')[0])
self.assertEqual(next_url, redirect['location'])
def test_ticket_system_oauth(self):
# Ticket System OAuth pages are exposed via the WebUI.
url = 'http://{}/oauth2/authorize/'
resp, content = self.get_server_status_and_content(url)
self.assertEqual('200', resp['status'], content)
def test_ticket_system_admin(self):
# Ticket System Admin pages are exposed via the WebUI.
url = 'http://{}/admin/'
resp, content = self.get_server_status_and_content(url)
self.assertEqual('200', resp['status'], content)
def test_create_ticket(self):
ticket_content = {
"owner": "create@example.com",
"title": "Created ticket",
"description": "This ticket is for creation test",
"bug_id": "111111",
}
resp, content = self.create_ticket(ticket_content)
self.assertEqual('201', resp['status'], content)
def test_prohibit_delete(self):
ticket_content = {
'owner': "dont_delete@example.com",
"title": "Delete test ticket",
"description": "Attempt to delete ticket details",
"bug_id": "22222",
}
resp, content = self.create_ticket(ticket_content)
self.assertEqual('201', resp['status'], content)
ticket_loc = resp['location']
client = httplib2.Http()
resp, content = client.request(
uri=ticket_loc,
method='DELETE',
)
self.assertEqual('405', resp['status'], content)
resp, content = self. get_server_status_and_json_content(ticket_loc)
self.assertEqual('200', resp['status'], content)
def test_update_ticket_details(self):
# A `Ticket` can be updated by users.
ticket_content = {
'owner': "before_change@example.com",
"title": "Change test ticket",
"description": "Attempt to change ticket details",
"bug_id": "33333",
}
resp, content = self.create_ticket(ticket_content)
self.assertEqual('201', resp['status'], content)
ticket_loc = resp['location']
# Update the just-created ticket using a PATCH request (and
# respecting status constraints).
client = httplib2.Http()
resp, content = client.request(
uri=ticket_loc,
method='PATCH',
headers={'Content-Type': 'application/json'},
body=json.dumps({
'owner': 'after_change@example.com',
'current_workflow_step': TicketWorkflowStep.FAILED.value,
}),
)
self.assertEqual('202', resp['status'], content)
# Confirm the ticket was updated.
resp, content = self. get_server_status_and_json_content(ticket_loc)
self.assertEqual('200', resp['status'], content)
self.assertEqual('after_change@example.com', content['owner'])
self.assertEqual('Failed', content['current_workflow_step'])
if __name__ == "__main__":
unittest.main()
|