3scale-api-python-crd/tests/integration/test_integration_services.py at ca475e1a77110db15076b40dd3cb1ac32e280d32 · mkudlej/3scale-api-python-crd · 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
141
142
143
144
145
146
147
148
149
150
151
152
from tests.integration import asserts
from threescale_api.resources import Proxy, Service
from .asserts import assert_resource, assert_resource_params


def test_3scale_url_is_set(api, url, token):
    assert url is not None
    assert token is not None
    assert api.url is not None


# tests important for CRD - CRU + list


def test_services_list(api, service):
    services = api.services.list()
    assert len(services) >= 1


def test_service_can_be_created(api, service_params, service):
    assert_resource(service)
    assert_resource_params(service, service_params)


def test_service_can_be_read(api, service_params, service):
    read = api.services.read(service.entity_id)
    asserts.assert_resource(read)
    asserts.assert_resource_params(read, service_params)


def test_service_can_be_read_by_name(api, service_params, service):
    account_name = service["system_name"]
    read = api.services[account_name]
    asserts.assert_resource(read)
    asserts.assert_resource_params(read, service_params)


# there is different syntax for setting up backend_version
def test_service_can_be_updated(api, service):
    des_changed = (service["description"] or "") + "_changed"
    service["description"] = des_changed
    service.update()
    assert service["description"] == des_changed
    updated = service.read()
    assert updated["description"] == des_changed
    assert service["description"] == des_changed


# end of tests important for CRD - CRU + list


def test_proxy_list(api, service: Service, proxy: Proxy):
    assert not isinstance(proxy, list)


# proxy object is doesn't contain error codes until they are configured.
# there is no default value in CRD
def test_service_get_proxy(api, service: Service, proxy: Proxy):
    assert "error_auth_failed" not in proxy.entity
    assert "error_auth_missing" not in proxy.entity
    assert "error_headers_auth_failed" not in proxy.entity
    assert "error_headers_auth_missing" not in proxy.entity
    assert "error_headers_limits_exceeded" not in proxy.entity
    assert "error_headers_no_match" not in proxy.entity
    assert "error_limits_exceeded" not in proxy.entity
    assert "error_no_match" not in proxy.entity
    assert "error_status_auth_failed" not in proxy.entity
    assert "error_status_auth_missing" not in proxy.entity
    assert "error_status_limits_exceeded" not in proxy.entity
    assert "error_status_no_match" not in proxy.entity


def test_service_set_proxy(api, service: Service, proxy: Proxy):
    updated = proxy.update(
        params=dict(error_status_no_match=403, error_status_auth_missing=403)
    )
    assert updated["error_status_auth_missing"] == 403
    assert updated["error_status_no_match"] == 403


def test_service_proxy_promote(service, proxy, backend_usage):
    service.proxy.list().deploy()
    service.proxy.list().promote()
    res = service.proxy.list().configs.latest(env="production")
    assert res is not None
    assert res["environment"] == "production"
    assert res["content"] is not None


def test_service_proxy_deploy(service, proxy, backend_usage):
    proxy.update(params=dict(error_status_no_match=405, error_status_auth_missing=405))
    proxy.deploy()
    res = proxy.configs.list(env="staging")
    proxy_config = res.entity["proxy_configs"][-1]["proxy_config"]
    assert proxy_config is not None
    assert proxy_config["environment"] == "sandbox"
    assert proxy_config["content"] is not None
    assert proxy_config["version"] > 1


def test_service_list_configs(service, proxy, backend_usage):
    proxy.update(params=dict(error_status_no_match=406, error_status_auth_missing=406))
    proxy.deploy()
    res = proxy.configs.list(env="staging")
    assert res
    item = res[0]
    assert item


def test_service_proxy_configs_version(service, proxy, backend_usage):
    config = service.proxy.list().configs.version(version=1)
    assert config
    assert config["environment"] == "sandbox"
    assert config["version"] == 1
    assert config["content"]


def test_service_proxy_configs_latest(service, proxy, backend_usage):
    config = service.proxy.list().configs.latest()
    assert config
    assert config["environment"] == "sandbox"
    assert config["version"]
    assert config["content"]


def test_service_proxy_configs_list_length(service, proxy, backend_usage, api_backend):
    configs = service.proxy.list().configs.list(env="sandbox")
    length = len(configs)
    proxy.update(params=dict(error_status_auth_failed=417, api_backend=api_backend))
    configs = service.proxy.list().configs.list(env="sandbox")
    assert len(configs) == length + 1


# there is no default mapping rule in service created from CRD
def test_service_mapping_rules(service):
    map_rules = service.mapping_rules.list()
    assert len(map_rules) == 0


def test_service_backend_usages_backend(backend_usage, backend):
    assert backend_usage.backend.entity_id == backend.entity_id


def test_service_active_docs(service, active_doc):
    assert all(
        [acs["service_id"] == service["id"] for acs in service.active_docs.list()]
    )


def test_service_annotation(service, api_origin):
    service_origin = api_origin.services.fetch(service.entity_id)
    assert service_origin['annotations']['managed_by'] == 'operator'