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
|
import json
from mock_pods import patch_k8s
from utilities import get_json
def test_get_pods(context, monkeypatch):
patch_k8s(monkeypatch)
import moon_orchestrator.server
server = moon_orchestrator.server.create_server()
_client = server.app.test_client()
req = _client.get("/pods")
assert req.status_code == 200
assert req.data
data = get_json(req.data)
assert isinstance(data, dict)
assert "pods" in data
def test_get_pods_failure(context, monkeypatch):
patch_k8s(monkeypatch)
import moon_orchestrator.server
server = moon_orchestrator.server.create_server()
_client = server.app.test_client()
req = _client.get("/pods/invalid")
assert req.status_code == 200
assert req.data
data = get_json(req.data)
assert isinstance(data, dict)
assert not data["pods"]
def test_add_pods_with_pipeline(context, monkeypatch):
patch_k8s(monkeypatch)
import moon_orchestrator.server
server = moon_orchestrator.server.create_server()
_client = server.app.test_client()
data = {
"keystone_project_id": context.get('project_id'),
"pdp_id": context.get('pdp_id'),
"security_pipeline": context.get('security_pipeline'),
}
req = _client.post("/pods", data=json.dumps(data),
headers={'Content-Type': 'application/json'})
assert req.status_code == 200
assert req.data
data = get_json(req.data)
assert isinstance(data, dict)
assert "pods" in data
assert data["pods"]
def test_add_pods_without_pipeline_with_bad_slave_name(context, monkeypatch):
patch_k8s(monkeypatch)
import moon_orchestrator.server
server = moon_orchestrator.server.create_server()
_client = server.app.test_client()
data = {
"slave_name": "test",
}
req = _client.post("/pods", data=json.dumps(data),
headers={'Content-Type': 'application/json'})
assert req.status_code == 400
assert req.data
data = get_json(req.data)
assert isinstance(data, dict)
assert 'The slave is unknown.' in data['message']
def test_add_pods_without_pipeline_with_good_slave_name(context, monkeypatch):
patch_k8s(monkeypatch)
import moon_orchestrator.server
server = moon_orchestrator.server.create_server()
_client = server.app.test_client()
data = {
"slave_name": "active_context",
}
req = _client.post("/pods", data=json.dumps(data),
headers={'Content-Type': 'application/json'})
assert req.status_code == 200
assert req.data
data = get_json(req.data)
assert isinstance(data, dict)
assert "pods" in data
assert data["pods"]
def test_add_pods_without_pipeline_without_slave_name(context, monkeypatch):
patch_k8s(monkeypatch)
import moon_orchestrator.server
server = moon_orchestrator.server.create_server()
_client = server.app.test_client()
data = {
}
req = _client.post("/pods", data=json.dumps(data),
headers={'Content-Type': 'application/json'})
assert req.status_code == 400
assert req.data
data = get_json(req.data)
assert isinstance(data, dict)
assert 'The slave is unknown.' in data['message']
def test_add_pods_with_no_data(context, monkeypatch):
patch_k8s(monkeypatch)
import moon_orchestrator.server
server = moon_orchestrator.server.create_server()
_client = server.app.test_client()
req = _client.post("/pods", data=json.dumps({}),
headers={'Content-Type': 'application/json'})
assert req.status_code == 400
assert req.data
data = get_json(req.data)
assert 'The slave is unknown.' in data['message']
def test_add_pods_with_no_policies_no_models(context, monkeypatch, no_requests):
patch_k8s(monkeypatch)
import moon_orchestrator.server
server = moon_orchestrator.server.create_server()
_client = server.app.test_client()
no_requests.get("http://manager:8082/policies",
json={'policies': {}})
no_requests.get("http://manager:8082/models",
json={'models': {}})
data = {
"keystone_project_id": context.get('project_id'),
"pdp_id": context.get('pdp_id'),
"security_pipeline": context.get('security_pipeline'),
}
req = _client.post("/pods", data=json.dumps(data),
headers={'Content-Type': 'application/json'})
assert req.status_code == 200
def test_delete_pods(context, monkeypatch):
# TODO
pass
|
