pyrox.dev
1from typing import Any, Dict
2import json
3
4start_all()
5machine.wait_for_unit("netbox.target")
6machine.wait_until_succeeds("journalctl --since -1m --unit netbox --grep Listening")
7
8test_objects = {
9 "sites": {
10 "test-site": {
11 "name": "Test site",
12 "slug": "test-site"
13 },
14 "test-site-two": {
15 "name": "Test site 2",
16 "slug": "test-site-second-edition"
17 }
18 },
19 "prefixes": {
20 "v4-with-updated-desc": {
21 "prefix": "192.0.2.0/24",
22 "class_type": "Prefix",
23 "family": { "label": "IPv4" },
24 "scope": {
25 "__typename": "SiteType",
26 "id": "1",
27 "description": "Test site description"
28 }
29 },
30 "v6-cidr-32": {
31 "prefix": "2001:db8::/32",
32 "class_type": "Prefix",
33 "family": { "label": "IPv6" },
34 "scope": {
35 "__typename": "SiteType",
36 "id": "1",
37 "description": "Test site description"
38 }
39 },
40 "v6-cidr-48": {
41 "prefix": "2001:db8:c0fe::/48",
42 "class_type": "Prefix",
43 "family": { "label": "IPv6" },
44 "scope": {
45 "__typename": "SiteType",
46 "id": "1",
47 "description": "Test site description"
48 }
49 }
50 }
51}
52
53def compare(a: str, b: str):
54 differences = [(x - y) for (x,y) in list(zip(
55 list(map(int, a.split('.'))),
56 list(map(int, b.split('.')))
57 ))]
58 for d in differences:
59 if d != 0:
60 return d
61 return 0
62
63with subtest("Home screen loads"):
64 machine.succeed(
65 "curl -sSfL http://[::1]:8001 | grep '<title>Home | NetBox</title>'"
66 )
67
68with subtest("Staticfiles are generated"):
69 machine.succeed("test -e /var/lib/netbox/static/netbox.js")
70
71with subtest("Superuser can be created"):
72 machine.succeed(
73 "netbox-manage createsuperuser --noinput --username netbox --email netbox@example.com"
74 )
75 # Django doesn't have a "clean" way of inputting the password from the command line
76 machine.succeed("cat '${changePassword}' | netbox-manage shell")
77
78machine.wait_for_unit("network.target")
79
80with subtest("Home screen loads from nginx"):
81 machine.succeed(
82 "curl -sSfL http://localhost | grep '<title>Home | NetBox</title>'"
83 )
84
85with subtest("Staticfiles can be fetched"):
86 machine.succeed("curl -sSfL http://localhost/static/netbox.js")
87 machine.succeed("curl -sSfL http://localhost/static/docs/")
88
89def login(username: str, password: str):
90 encoded_data = json.dumps({"username": username, "password": password})
91 uri = "/users/tokens/provision/"
92 result = json.loads(
93 machine.succeed(
94 "curl -sSfL "
95 "-X POST "
96 "-H 'Accept: application/json' "
97 "-H 'Content-Type: application/json' "
98 f"'http://localhost/api{uri}' "
99 f"--data '{encoded_data}'"
100 )
101 )
102 return result["key"]
103
104with subtest("Can login"):
105 auth_token = login("netbox", "netbox")
106
107def get(uri: str):
108 return json.loads(
109 machine.succeed(
110 "curl -sSfL "
111 "-H 'Accept: application/json' "
112 f"-H 'Authorization: Token {auth_token}' "
113 f"'http://localhost/api{uri}'"
114 )
115 )
116
117def delete(uri: str):
118 return machine.succeed(
119 "curl -sSfL "
120 f"-X DELETE "
121 "-H 'Accept: application/json' "
122 f"-H 'Authorization: Token {auth_token}' "
123 f"'http://localhost/api{uri}'"
124 )
125
126
127def data_request(uri: str, method: str, data: Dict[str, Any]):
128 encoded_data = json.dumps(data)
129 return json.loads(
130 machine.succeed(
131 "curl -sSfL "
132 f"-X {method} "
133 "-H 'Accept: application/json' "
134 "-H 'Content-Type: application/json' "
135 f"-H 'Authorization: Token {auth_token}' "
136 f"'http://localhost/api{uri}' "
137 f"--data '{encoded_data}'"
138 )
139 )
140
141def post(uri: str, data: Dict[str, Any]):
142 return data_request(uri, "POST", data)
143
144def patch(uri: str, data: Dict[str, Any]):
145 return data_request(uri, "PATCH", data)
146
147# Retrieve netbox version
148netbox_version = get("/status/")["netbox-version"]
149
150with subtest("Can create objects"):
151 result = post("/dcim/sites/", {"name": "Test site", "slug": "test-site"})
152 site_id = result["id"]
153
154
155 for prefix in test_objects["prefixes"].values():
156 if compare(netbox_version, '4.2.0') >= 0:
157 post("/ipam/prefixes/", {
158 "prefix": prefix["prefix"],
159 "scope_id": site_id,
160 "scope_type": "dcim." + prefix["scope"]["__typename"].replace("Type", "").lower()
161 })
162 prefix["scope"]["id"] = str(site_id)
163 else:
164 post("/ipam/prefixes/", {
165 "prefix": prefix["prefix"],
166 "site": str(site_id),
167 })
168
169 result = post(
170 "/dcim/manufacturers/",
171 {"name": "Test manufacturer", "slug": "test-manufacturer"}
172 )
173 manufacturer_id = result["id"]
174
175 # Had an issue with device-types before NetBox 3.4.0
176 result = post(
177 "/dcim/device-types/",
178 {
179 "model": "Test device type",
180 "manufacturer": manufacturer_id,
181 "slug": "test-device-type",
182 },
183 )
184 device_type_id = result["id"]
185
186with subtest("Can list objects"):
187 result = get("/dcim/sites/")
188
189 assert result["count"] == 1
190 assert result["results"][0]["id"] == site_id
191 assert result["results"][0]["name"] == "Test site"
192 assert result["results"][0]["description"] == ""
193
194 result = get("/dcim/device-types/")
195 assert result["count"] == 1
196 assert result["results"][0]["id"] == device_type_id
197 assert result["results"][0]["model"] == "Test device type"
198
199with subtest("Can update objects"):
200 new_description = "Test site description"
201 patch(f"/dcim/sites/{site_id}/", {"description": new_description})
202 result = get(f"/dcim/sites/{site_id}/")
203 assert result["description"] == new_description
204
205with subtest("Can delete objects"):
206 # Delete a device-type since no object depends on it
207 delete(f"/dcim/device-types/{device_type_id}/")
208
209 result = get("/dcim/device-types/")
210 assert result["count"] == 0
211
212def request_graphql(query: str):
213 return machine.succeed(
214 "curl -sSfL "
215 "-H 'Accept: application/json' "
216 "-H 'Content-Type: application/json' "
217 f"-H 'Authorization: Token {auth_token}' "
218 "'http://localhost/graphql/' "
219 f"--data '{json.dumps({"query": query})}'"
220 )
221
222
223if compare(netbox_version, '4.2.0') >= 0:
224 with subtest("Can use the GraphQL API (NetBox 4.2.0+)"):
225 graphql_query = '''query {
226 prefix_list {
227 prefix
228 class_type
229 family {
230 label
231 }
232 scope {
233 __typename
234 ... on SiteType {
235 id
236 description
237 }
238 }
239 }
240 }
241 '''
242
243 answer = request_graphql(graphql_query)
244 result = json.loads(answer)
245 assert len(result["data"]["prefix_list"]) == 3
246 assert test_objects["prefixes"]["v4-with-updated-desc"] in result["data"]["prefix_list"]
247 assert test_objects["prefixes"]["v6-cidr-32"] in result["data"]["prefix_list"]
248 assert test_objects["prefixes"]["v6-cidr-48"] in result["data"]["prefix_list"]
249
250if compare(netbox_version, '4.2.0') < 0:
251 with subtest("Can use the GraphQL API (Netbox <= 4.2.0)"):
252 answer = request_graphql('''query {
253 prefix_list {
254 prefix
255 site {
256 id
257 }
258 }
259 }
260 ''')
261 result = json.loads(answer)
262 print(result["data"]["prefix_list"][0])
263 assert result["data"]["prefix_list"][0]["prefix"] == test_objects["prefixes"]["v4-with-updated-desc"]["prefix"]
264 assert int(result["data"]["prefix_list"][0]["site"]["id"]) == int(test_objects["prefixes"]["v4-with-updated-desc"]["scope"]["id"])
265
266with subtest("Can login with LDAP"):
267 machine.wait_for_unit("openldap.service")
268 login("alice", "${testPassword}")
269
270with subtest("Can associate LDAP groups"):
271 result = get("/users/users/?username=${testUser}")
272
273 assert result["count"] == 1
274 assert any(group["name"] == "${testGroup}" for group in result["results"][0]["groups"])