This repository has been archived by the owner on Oct 23, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 6
/
Copy pathtest_acl_sync.py
133 lines (105 loc) · 5.84 KB
/
test_acl_sync.py
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
import json
import mock
import pytest
import responses
from objectrocket.bases import BaseInstance
@pytest.yield_fixture(autouse=True)
def ensure_production_url(mongodb_sharded_instance, instance_acl_url):
"""Fixture that ensures that the proper production URLs are used in tests,
instead of the potentially overridden ones from environment variables.
See objectrocket.constants.OR_DEFAULT_API_URL
"""
with mock.patch.object(BaseInstance, '_url', new_callable=mock.PropertyMock) as mock_url:
type(mock_url).return_value = instance_acl_url.replace('acl_sync', '')
yield
@pytest.fixture
def instance_acl_url(mongodb_sharded_instance):
return "http://localhost:5050/v2/instances/{}/acl_sync".format(
mongodb_sharded_instance.name
)
class TestSetGetAclSync:
@responses.activate
def test_acl_sync_disables(self, client, mongodb_sharded_instance, instance_acl_url):
inst = mongodb_sharded_instance
responses.add(responses.PUT, instance_acl_url,
status=200,
body=json.dumps({'data': {'aws_acl_sync_enabled': False,
'rackspace_acl_sync_enabled': False}}),
content_type="application/json")
responses.add(responses.GET, instance_acl_url, status=200,
body=json.dumps({'data': {'aws_acl_sync_enabled': True,
'rackspace_acl_sync_enabled': True}}),
content_type="application/json")
response = inst.acl_sync(aws_sync=False, rackspace_sync=False)
assert isinstance(response, dict) is True
assert response.get('data', {}).get('aws_acl_sync_enabled', True) is False
assert response.get('data', {}).get('rackspace_acl_sync_enabled', True) is False
@responses.activate
def test_acl_sync_enables(self, client, mongodb_sharded_instance, instance_acl_url):
inst = mongodb_sharded_instance
responses.add(responses.PUT, instance_acl_url,
status=200,
body=json.dumps({'data': {'aws_acl_sync_enabled': True,
'rackspace_acl_sync_enabled': True}}),
content_type="application/json")
responses.add(responses.GET, instance_acl_url, status=200,
body=json.dumps({'data': {'aws_acl_sync_enabled': False,
'rackspace_acl_sync_enabled': False}}),
content_type="application/json")
response = inst.acl_sync(aws_sync=False, rackspace_sync=False)
assert isinstance(response, dict) is True
assert response.get('data', {}).get('aws_acl_sync_enabled', False) is True
assert response.get('data', {}).get('rackspace_acl_sync_enabled', False) is True
@responses.activate
def test_acl_sync_just_returns_status(self, client, mongodb_sharded_instance,
instance_acl_url):
inst = mongodb_sharded_instance
responses.add(responses.GET, instance_acl_url, status=200,
body=json.dumps({'data': {'aws_acl_sync_enabled': True,
'rackspace_acl_sync_enabled': True}}),
content_type="application/json")
responses.add(responses.PUT, instance_acl_url,
status=200,
body=json.dumps({'data': {'aws_acl_sync_enabled': False,
'rackspace_acl_sync_enabled': False}}),
content_type="application/json")
response = inst.acl_sync()
assert isinstance(response, dict) is True
assert response.get('data', {}).get('aws_acl_sync_enabled', True) is False
assert response.get('data', {}).get('rackspace_acl_sync_enabled', True) is False
@responses.activate
def test_acl_sync_raises(self, client, mongodb_sharded_instance, instance_acl_url):
inst = mongodb_sharded_instance
exception = Exception('Test Exception')
responses.add(responses.GET, instance_acl_url, status=500,
body=exception)
with pytest.raises(Exception) as exinfo:
inst.acl_sync()
assert exinfo is not None
assert exinfo.value.args[0] == 'Test Exception'
class TestRunAclSync:
@responses.activate
def test_run_sync_without_args_is_unchanged(self, client, mongodb_sharded_instance,
instance_acl_url):
inst = mongodb_sharded_instance
responses.add(responses.POST, instance_acl_url,
status=200,
body=json.dumps({'data': {'aws_acl_sync_state': 'unchanged',
'rackspace_acl_sync_state': 'unchanged'}}),
content_type="application/json")
response = inst.run_acl_sync()
assert isinstance(response, dict) is True
assert response.get('data', {}).get('aws_acl_sync_state', None) == 'unchanged'
assert response.get('data', {}).get('rackspace_acl_sync_state', None) == 'unchanged'
@responses.activate
def test_run_sync_starts(self, client, mongodb_sharded_instance, instance_acl_url):
inst = mongodb_sharded_instance
responses.add(responses.POST, instance_acl_url,
status=200,
body=json.dumps({'data': {'aws_acl_sync_state': 'started',
'rackspace_acl_sync_state': 'started'}}),
content_type="application/json")
response = inst.run_acl_sync(aws_sync=True, rackspace_sync=True)
assert isinstance(response, dict) is True
assert response.get('data', {}).get('aws_acl_sync_state', None) == 'started'
assert response.get('data', {}).get('rackspace_acl_sync_state', None) == 'started'