mirror of
https://github.com/CHN-beta/nixpkgs.git
synced 2026-01-12 02:40:31 +08:00
nixos/tests/netbox: Adjust tests for 4.2.0+, drop test for 3.6
This changes the previously used `site_id` in the tests' REST-API usage to `scope_type`+`scope_id`. Netbox 3.6 is deprecated and no longer available, the test for it has been removed.
This commit is contained in:
transcaffeine
@@ -782,10 +782,9 @@ in {
|
||||
networking.scripted = handleTest ./networking/networkd-and-scripted.nix { networkd = false; };
|
||||
networking.networkd = handleTest ./networking/networkd-and-scripted.nix { networkd = true; };
|
||||
networking.networkmanager = handleTest ./networking/networkmanager.nix {};
|
||||
netbox_3_6 = handleTest ./web-apps/netbox.nix { netbox = pkgs.netbox_3_6; };
|
||||
netbox_3_7 = handleTest ./web-apps/netbox.nix { netbox = pkgs.netbox_3_7; };
|
||||
netbox_4_1 = handleTest ./web-apps/netbox.nix { netbox = pkgs.netbox_4_1; };
|
||||
netbox_4_2 = handleTest ./web-apps/netbox.nix { netbox = pkgs.netbox_4_2; };
|
||||
netbox_3_7 = handleTest ./web-apps/netbox/default.nix { netbox = pkgs.netbox_3_7; };
|
||||
netbox_4_1 = handleTest ./web-apps/netbox/default.nix { netbox = pkgs.netbox_4_1; };
|
||||
netbox_4_2 = handleTest ./web-apps/netbox/default.nix { netbox = pkgs.netbox_4_2; };
|
||||
netbox-upgrade = handleTest ./web-apps/netbox-upgrade.nix {};
|
||||
# TODO: put in networking.nix after the test becomes more complete
|
||||
networkingProxy = handleTest ./networking-proxy.nix {};
|
||||
|
||||
@@ -1,331 +0,0 @@
|
||||
let
|
||||
ldapDomain = "example.org";
|
||||
ldapSuffix = "dc=example,dc=org";
|
||||
|
||||
ldapRootUser = "admin";
|
||||
ldapRootPassword = "foobar";
|
||||
|
||||
testUser = "alice";
|
||||
testPassword = "verySecure";
|
||||
testGroup = "netbox-users";
|
||||
in
|
||||
import ../make-test-python.nix (
|
||||
{
|
||||
lib,
|
||||
pkgs,
|
||||
netbox,
|
||||
...
|
||||
}:
|
||||
{
|
||||
name = "netbox";
|
||||
|
||||
meta = with lib.maintainers; {
|
||||
maintainers = [
|
||||
minijackson
|
||||
];
|
||||
};
|
||||
|
||||
nodes.machine =
|
||||
{ config, ... }:
|
||||
{
|
||||
virtualisation.memorySize = 2048;
|
||||
services.netbox = {
|
||||
enable = true;
|
||||
package = netbox;
|
||||
secretKeyFile = pkgs.writeText "secret" ''
|
||||
abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789
|
||||
'';
|
||||
|
||||
enableLdap = true;
|
||||
ldapConfigPath = pkgs.writeText "ldap_config.py" ''
|
||||
import ldap
|
||||
from django_auth_ldap.config import LDAPSearch, PosixGroupType
|
||||
|
||||
AUTH_LDAP_SERVER_URI = "ldap://localhost/"
|
||||
|
||||
AUTH_LDAP_USER_SEARCH = LDAPSearch(
|
||||
"ou=accounts,ou=posix,${ldapSuffix}",
|
||||
ldap.SCOPE_SUBTREE,
|
||||
"(uid=%(user)s)",
|
||||
)
|
||||
|
||||
AUTH_LDAP_GROUP_SEARCH = LDAPSearch(
|
||||
"ou=groups,ou=posix,${ldapSuffix}",
|
||||
ldap.SCOPE_SUBTREE,
|
||||
"(objectClass=posixGroup)",
|
||||
)
|
||||
AUTH_LDAP_GROUP_TYPE = PosixGroupType()
|
||||
|
||||
# Mirror LDAP group assignments.
|
||||
AUTH_LDAP_MIRROR_GROUPS = True
|
||||
|
||||
# For more granular permissions, we can map LDAP groups to Django groups.
|
||||
AUTH_LDAP_FIND_GROUP_PERMS = True
|
||||
'';
|
||||
};
|
||||
|
||||
services.nginx = {
|
||||
enable = true;
|
||||
|
||||
recommendedProxySettings = true;
|
||||
|
||||
virtualHosts.netbox = {
|
||||
default = true;
|
||||
locations."/".proxyPass = "http://localhost:${toString config.services.netbox.port}";
|
||||
locations."/static/".alias = "/var/lib/netbox/static/";
|
||||
};
|
||||
};
|
||||
|
||||
# Adapted from the sssd-ldap NixOS test
|
||||
services.openldap = {
|
||||
enable = true;
|
||||
settings = {
|
||||
children = {
|
||||
"cn=schema".includes = [
|
||||
"${pkgs.openldap}/etc/schema/core.ldif"
|
||||
"${pkgs.openldap}/etc/schema/cosine.ldif"
|
||||
"${pkgs.openldap}/etc/schema/inetorgperson.ldif"
|
||||
"${pkgs.openldap}/etc/schema/nis.ldif"
|
||||
];
|
||||
"olcDatabase={1}mdb" = {
|
||||
attrs = {
|
||||
objectClass = [
|
||||
"olcDatabaseConfig"
|
||||
"olcMdbConfig"
|
||||
];
|
||||
olcDatabase = "{1}mdb";
|
||||
olcDbDirectory = "/var/lib/openldap/db";
|
||||
olcSuffix = ldapSuffix;
|
||||
olcRootDN = "cn=${ldapRootUser},${ldapSuffix}";
|
||||
olcRootPW = ldapRootPassword;
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
declarativeContents = {
|
||||
${ldapSuffix} = ''
|
||||
dn: ${ldapSuffix}
|
||||
objectClass: top
|
||||
objectClass: dcObject
|
||||
objectClass: organization
|
||||
o: ${ldapDomain}
|
||||
|
||||
dn: ou=posix,${ldapSuffix}
|
||||
objectClass: top
|
||||
objectClass: organizationalUnit
|
||||
|
||||
dn: ou=accounts,ou=posix,${ldapSuffix}
|
||||
objectClass: top
|
||||
objectClass: organizationalUnit
|
||||
|
||||
dn: uid=${testUser},ou=accounts,ou=posix,${ldapSuffix}
|
||||
objectClass: person
|
||||
objectClass: posixAccount
|
||||
userPassword: ${testPassword}
|
||||
homeDirectory: /home/${testUser}
|
||||
uidNumber: 1234
|
||||
gidNumber: 1234
|
||||
cn: ""
|
||||
sn: ""
|
||||
|
||||
dn: ou=groups,ou=posix,${ldapSuffix}
|
||||
objectClass: top
|
||||
objectClass: organizationalUnit
|
||||
|
||||
dn: cn=${testGroup},ou=groups,ou=posix,${ldapSuffix}
|
||||
objectClass: posixGroup
|
||||
gidNumber: 2345
|
||||
memberUid: ${testUser}
|
||||
'';
|
||||
};
|
||||
};
|
||||
|
||||
users.users.nginx.extraGroups = [ "netbox" ];
|
||||
|
||||
networking.firewall.allowedTCPPorts = [ 80 ];
|
||||
};
|
||||
|
||||
testScript =
|
||||
let
|
||||
changePassword = pkgs.writeText "change-password.py" ''
|
||||
from users.models import User
|
||||
u = User.objects.get(username='netbox')
|
||||
u.set_password('netbox')
|
||||
u.save()
|
||||
'';
|
||||
in
|
||||
''
|
||||
from typing import Any, Dict
|
||||
import json
|
||||
|
||||
start_all()
|
||||
machine.wait_for_unit("netbox.target")
|
||||
machine.wait_until_succeeds("journalctl --since -1m --unit netbox --grep Listening")
|
||||
|
||||
with subtest("Home screen loads"):
|
||||
machine.succeed(
|
||||
"curl -sSfL http://[::1]:8001 | grep '<title>Home | NetBox</title>'"
|
||||
)
|
||||
|
||||
with subtest("Staticfiles are generated"):
|
||||
machine.succeed("test -e /var/lib/netbox/static/netbox.js")
|
||||
|
||||
with subtest("Superuser can be created"):
|
||||
machine.succeed(
|
||||
"netbox-manage createsuperuser --noinput --username netbox --email netbox@example.com"
|
||||
)
|
||||
# Django doesn't have a "clean" way of inputting the password from the command line
|
||||
machine.succeed("cat '${changePassword}' | netbox-manage shell")
|
||||
|
||||
machine.wait_for_unit("network.target")
|
||||
|
||||
with subtest("Home screen loads from nginx"):
|
||||
machine.succeed(
|
||||
"curl -sSfL http://localhost | grep '<title>Home | NetBox</title>'"
|
||||
)
|
||||
|
||||
with subtest("Staticfiles can be fetched"):
|
||||
machine.succeed("curl -sSfL http://localhost/static/netbox.js")
|
||||
machine.succeed("curl -sSfL http://localhost/static/docs/")
|
||||
|
||||
def login(username: str, password: str):
|
||||
encoded_data = json.dumps({"username": username, "password": password})
|
||||
uri = "/users/tokens/provision/"
|
||||
result = json.loads(
|
||||
machine.succeed(
|
||||
"curl -sSfL "
|
||||
"-X POST "
|
||||
"-H 'Accept: application/json' "
|
||||
"-H 'Content-Type: application/json' "
|
||||
f"'http://localhost/api{uri}' "
|
||||
f"--data '{encoded_data}'"
|
||||
)
|
||||
)
|
||||
return result["key"]
|
||||
|
||||
with subtest("Can login"):
|
||||
auth_token = login("netbox", "netbox")
|
||||
|
||||
def get(uri: str):
|
||||
return json.loads(
|
||||
machine.succeed(
|
||||
"curl -sSfL "
|
||||
"-H 'Accept: application/json' "
|
||||
f"-H 'Authorization: Token {auth_token}' "
|
||||
f"'http://localhost/api{uri}'"
|
||||
)
|
||||
)
|
||||
|
||||
def delete(uri: str):
|
||||
return machine.succeed(
|
||||
"curl -sSfL "
|
||||
f"-X DELETE "
|
||||
"-H 'Accept: application/json' "
|
||||
f"-H 'Authorization: Token {auth_token}' "
|
||||
f"'http://localhost/api{uri}'"
|
||||
)
|
||||
|
||||
|
||||
def data_request(uri: str, method: str, data: Dict[str, Any]):
|
||||
encoded_data = json.dumps(data)
|
||||
return json.loads(
|
||||
machine.succeed(
|
||||
"curl -sSfL "
|
||||
f"-X {method} "
|
||||
"-H 'Accept: application/json' "
|
||||
"-H 'Content-Type: application/json' "
|
||||
f"-H 'Authorization: Token {auth_token}' "
|
||||
f"'http://localhost/api{uri}' "
|
||||
f"--data '{encoded_data}'"
|
||||
)
|
||||
)
|
||||
|
||||
def post(uri: str, data: Dict[str, Any]):
|
||||
return data_request(uri, "POST", data)
|
||||
|
||||
def patch(uri: str, data: Dict[str, Any]):
|
||||
return data_request(uri, "PATCH", data)
|
||||
|
||||
with subtest("Can create objects"):
|
||||
result = post("/dcim/sites/", {"name": "Test site", "slug": "test-site"})
|
||||
site_id = result["id"]
|
||||
|
||||
# Example from:
|
||||
# http://netbox.extra.cea.fr/static/docs/integrations/rest-api/#creating-a-new-object
|
||||
post("/ipam/prefixes/", {"prefix": "192.0.2.0/24", "site": site_id})
|
||||
|
||||
result = post(
|
||||
"/dcim/manufacturers/",
|
||||
{"name": "Test manufacturer", "slug": "test-manufacturer"}
|
||||
)
|
||||
manufacturer_id = result["id"]
|
||||
|
||||
# Had an issue with device-types before NetBox 3.4.0
|
||||
result = post(
|
||||
"/dcim/device-types/",
|
||||
{
|
||||
"model": "Test device type",
|
||||
"manufacturer": manufacturer_id,
|
||||
"slug": "test-device-type",
|
||||
},
|
||||
)
|
||||
device_type_id = result["id"]
|
||||
|
||||
with subtest("Can list objects"):
|
||||
result = get("/dcim/sites/")
|
||||
|
||||
assert result["count"] == 1
|
||||
assert result["results"][0]["id"] == site_id
|
||||
assert result["results"][0]["name"] == "Test site"
|
||||
assert result["results"][0]["description"] == ""
|
||||
|
||||
result = get("/dcim/device-types/")
|
||||
assert result["count"] == 1
|
||||
assert result["results"][0]["id"] == device_type_id
|
||||
assert result["results"][0]["model"] == "Test device type"
|
||||
|
||||
with subtest("Can update objects"):
|
||||
new_description = "Test site description"
|
||||
patch(f"/dcim/sites/{site_id}/", {"description": new_description})
|
||||
result = get(f"/dcim/sites/{site_id}/")
|
||||
assert result["description"] == new_description
|
||||
|
||||
with subtest("Can delete objects"):
|
||||
# Delete a device-type since no object depends on it
|
||||
delete(f"/dcim/device-types/{device_type_id}/")
|
||||
|
||||
result = get("/dcim/device-types/")
|
||||
assert result["count"] == 0
|
||||
|
||||
with subtest("Can use the GraphQL API"):
|
||||
encoded_data = json.dumps({
|
||||
"query": "query { prefix_list { prefix, site { id, description } } }",
|
||||
})
|
||||
result = json.loads(
|
||||
machine.succeed(
|
||||
"curl -sSfL "
|
||||
"-H 'Accept: application/json' "
|
||||
"-H 'Content-Type: application/json' "
|
||||
f"-H 'Authorization: Token {auth_token}' "
|
||||
"'http://localhost/graphql/' "
|
||||
f"--data '{encoded_data}'"
|
||||
)
|
||||
)
|
||||
|
||||
assert len(result["data"]["prefix_list"]) == 1
|
||||
assert result["data"]["prefix_list"][0]["prefix"] == "192.0.2.0/24"
|
||||
assert result["data"]["prefix_list"][0]["site"]["id"] == str(site_id)
|
||||
assert result["data"]["prefix_list"][0]["site"]["description"] == new_description
|
||||
|
||||
with subtest("Can login with LDAP"):
|
||||
machine.wait_for_unit("openldap.service")
|
||||
login("alice", "${testPassword}")
|
||||
|
||||
with subtest("Can associate LDAP groups"):
|
||||
result = get("/users/users/?username=${testUser}")
|
||||
|
||||
assert result["count"] == 1
|
||||
assert any(group["name"] == "${testGroup}" for group in result["results"][0]["groups"])
|
||||
'';
|
||||
}
|
||||
)
|
||||
164
nixos/tests/web-apps/netbox/default.nix
Normal file
164
nixos/tests/web-apps/netbox/default.nix
Normal file
@@ -0,0 +1,164 @@
|
||||
let
|
||||
ldapDomain = "example.org";
|
||||
ldapSuffix = "dc=example,dc=org";
|
||||
|
||||
ldapRootUser = "admin";
|
||||
ldapRootPassword = "foobar";
|
||||
|
||||
testUser = "alice";
|
||||
testPassword = "verySecure";
|
||||
testGroup = "netbox-users";
|
||||
in
|
||||
import ../../make-test-python.nix (
|
||||
{
|
||||
lib,
|
||||
pkgs,
|
||||
netbox,
|
||||
...
|
||||
}:
|
||||
{
|
||||
name = "netbox";
|
||||
|
||||
meta = with lib.maintainers; {
|
||||
maintainers = [
|
||||
minijackson
|
||||
];
|
||||
};
|
||||
|
||||
skipTypeCheck = true;
|
||||
|
||||
nodes.machine =
|
||||
{ config, ... }:
|
||||
{
|
||||
virtualisation.memorySize = 2048;
|
||||
services.netbox = {
|
||||
enable = true;
|
||||
package = netbox;
|
||||
secretKeyFile = pkgs.writeText "secret" ''
|
||||
abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789
|
||||
'';
|
||||
|
||||
enableLdap = true;
|
||||
ldapConfigPath = pkgs.writeText "ldap_config.py" ''
|
||||
import ldap
|
||||
from django_auth_ldap.config import LDAPSearch, PosixGroupType
|
||||
|
||||
AUTH_LDAP_SERVER_URI = "ldap://localhost/"
|
||||
|
||||
AUTH_LDAP_USER_SEARCH = LDAPSearch(
|
||||
"ou=accounts,ou=posix,${ldapSuffix}",
|
||||
ldap.SCOPE_SUBTREE,
|
||||
"(uid=%(user)s)",
|
||||
)
|
||||
|
||||
AUTH_LDAP_GROUP_SEARCH = LDAPSearch(
|
||||
"ou=groups,ou=posix,${ldapSuffix}",
|
||||
ldap.SCOPE_SUBTREE,
|
||||
"(objectClass=posixGroup)",
|
||||
)
|
||||
AUTH_LDAP_GROUP_TYPE = PosixGroupType()
|
||||
|
||||
# Mirror LDAP group assignments.
|
||||
AUTH_LDAP_MIRROR_GROUPS = True
|
||||
|
||||
# For more granular permissions, we can map LDAP groups to Django groups.
|
||||
AUTH_LDAP_FIND_GROUP_PERMS = True
|
||||
'';
|
||||
};
|
||||
|
||||
services.nginx = {
|
||||
enable = true;
|
||||
|
||||
recommendedProxySettings = true;
|
||||
|
||||
virtualHosts.netbox = {
|
||||
default = true;
|
||||
locations."/".proxyPass = "http://localhost:${toString config.services.netbox.port}";
|
||||
locations."/static/".alias = "/var/lib/netbox/static/";
|
||||
};
|
||||
};
|
||||
|
||||
# Adapted from the sssd-ldap NixOS test
|
||||
services.openldap = {
|
||||
enable = true;
|
||||
settings = {
|
||||
children = {
|
||||
"cn=schema".includes = [
|
||||
"${pkgs.openldap}/etc/schema/core.ldif"
|
||||
"${pkgs.openldap}/etc/schema/cosine.ldif"
|
||||
"${pkgs.openldap}/etc/schema/inetorgperson.ldif"
|
||||
"${pkgs.openldap}/etc/schema/nis.ldif"
|
||||
];
|
||||
"olcDatabase={1}mdb" = {
|
||||
attrs = {
|
||||
objectClass = [
|
||||
"olcDatabaseConfig"
|
||||
"olcMdbConfig"
|
||||
];
|
||||
olcDatabase = "{1}mdb";
|
||||
olcDbDirectory = "/var/lib/openldap/db";
|
||||
olcSuffix = ldapSuffix;
|
||||
olcRootDN = "cn=${ldapRootUser},${ldapSuffix}";
|
||||
olcRootPW = ldapRootPassword;
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
declarativeContents = {
|
||||
${ldapSuffix} = ''
|
||||
dn: ${ldapSuffix}
|
||||
objectClass: top
|
||||
objectClass: dcObject
|
||||
objectClass: organization
|
||||
o: ${ldapDomain}
|
||||
|
||||
dn: ou=posix,${ldapSuffix}
|
||||
objectClass: top
|
||||
objectClass: organizationalUnit
|
||||
|
||||
dn: ou=accounts,ou=posix,${ldapSuffix}
|
||||
objectClass: top
|
||||
objectClass: organizationalUnit
|
||||
|
||||
dn: uid=${testUser},ou=accounts,ou=posix,${ldapSuffix}
|
||||
objectClass: person
|
||||
objectClass: posixAccount
|
||||
userPassword: ${testPassword}
|
||||
homeDirectory: /home/${testUser}
|
||||
uidNumber: 1234
|
||||
gidNumber: 1234
|
||||
cn: ""
|
||||
sn: ""
|
||||
|
||||
dn: ou=groups,ou=posix,${ldapSuffix}
|
||||
objectClass: top
|
||||
objectClass: organizationalUnit
|
||||
|
||||
dn: cn=${testGroup},ou=groups,ou=posix,${ldapSuffix}
|
||||
objectClass: posixGroup
|
||||
gidNumber: 2345
|
||||
memberUid: ${testUser}
|
||||
'';
|
||||
};
|
||||
};
|
||||
|
||||
users.users.nginx.extraGroups = [ "netbox" ];
|
||||
|
||||
networking.firewall.allowedTCPPorts = [ 80 ];
|
||||
};
|
||||
|
||||
testScript =
|
||||
let
|
||||
changePassword = pkgs.writeText "change-password.py" ''
|
||||
from users.models import User
|
||||
u = User.objects.get(username='netbox')
|
||||
u.set_password('netbox')
|
||||
u.save()
|
||||
'';
|
||||
in
|
||||
builtins.replaceStrings
|
||||
[ "$\{changePassword}" "$\{testUser}" "$\{testPassword}" "$\{testGroup}" ]
|
||||
[ "${changePassword}" "${testUser}" "${testPassword}" "${testGroup}" ]
|
||||
(lib.readFile "${./testScript.py}");
|
||||
}
|
||||
)
|
||||
274
nixos/tests/web-apps/netbox/testScript.py
Normal file
274
nixos/tests/web-apps/netbox/testScript.py
Normal file
@@ -0,0 +1,274 @@
|
||||
from typing import Any, Dict
|
||||
import json
|
||||
|
||||
start_all()
|
||||
machine.wait_for_unit("netbox.target")
|
||||
machine.wait_until_succeeds("journalctl --since -1m --unit netbox --grep Listening")
|
||||
|
||||
test_objects = {
|
||||
"sites": {
|
||||
"test-site": {
|
||||
"name": "Test site",
|
||||
"slug": "test-site"
|
||||
},
|
||||
"test-site-two": {
|
||||
"name": "Test site 2",
|
||||
"slug": "test-site-second-edition"
|
||||
}
|
||||
},
|
||||
"prefixes": {
|
||||
"v4-with-updated-desc": {
|
||||
"prefix": "192.0.2.0/24",
|
||||
"class_type": "Prefix",
|
||||
"family": { "label": "IPv4" },
|
||||
"scope": {
|
||||
"__typename": "SiteType",
|
||||
"id": "1",
|
||||
"description": "Test site description"
|
||||
}
|
||||
},
|
||||
"v6-cidr-32": {
|
||||
"prefix": "2001:db8::/32",
|
||||
"class_type": "Prefix",
|
||||
"family": { "label": "IPv6" },
|
||||
"scope": {
|
||||
"__typename": "SiteType",
|
||||
"id": "1",
|
||||
"description": "Test site description"
|
||||
}
|
||||
},
|
||||
"v6-cidr-48": {
|
||||
"prefix": "2001:db8:c0fe::/48",
|
||||
"class_type": "Prefix",
|
||||
"family": { "label": "IPv6" },
|
||||
"scope": {
|
||||
"__typename": "SiteType",
|
||||
"id": "1",
|
||||
"description": "Test site description"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
def compare(a: str, b: str):
|
||||
differences = [(x - y) for (x,y) in list(zip(
|
||||
list(map(int, a.split('.'))),
|
||||
list(map(int, b.split('.')))
|
||||
))]
|
||||
for d in differences:
|
||||
if d != 0:
|
||||
return d
|
||||
return 0
|
||||
|
||||
with subtest("Home screen loads"):
|
||||
machine.succeed(
|
||||
"curl -sSfL http://[::1]:8001 | grep '<title>Home | NetBox</title>'"
|
||||
)
|
||||
|
||||
with subtest("Staticfiles are generated"):
|
||||
machine.succeed("test -e /var/lib/netbox/static/netbox.js")
|
||||
|
||||
with subtest("Superuser can be created"):
|
||||
machine.succeed(
|
||||
"netbox-manage createsuperuser --noinput --username netbox --email netbox@example.com"
|
||||
)
|
||||
# Django doesn't have a "clean" way of inputting the password from the command line
|
||||
machine.succeed("cat '${changePassword}' | netbox-manage shell")
|
||||
|
||||
machine.wait_for_unit("network.target")
|
||||
|
||||
with subtest("Home screen loads from nginx"):
|
||||
machine.succeed(
|
||||
"curl -sSfL http://localhost | grep '<title>Home | NetBox</title>'"
|
||||
)
|
||||
|
||||
with subtest("Staticfiles can be fetched"):
|
||||
machine.succeed("curl -sSfL http://localhost/static/netbox.js")
|
||||
machine.succeed("curl -sSfL http://localhost/static/docs/")
|
||||
|
||||
def login(username: str, password: str):
|
||||
encoded_data = json.dumps({"username": username, "password": password})
|
||||
uri = "/users/tokens/provision/"
|
||||
result = json.loads(
|
||||
machine.succeed(
|
||||
"curl -sSfL "
|
||||
"-X POST "
|
||||
"-H 'Accept: application/json' "
|
||||
"-H 'Content-Type: application/json' "
|
||||
f"'http://localhost/api{uri}' "
|
||||
f"--data '{encoded_data}'"
|
||||
)
|
||||
)
|
||||
return result["key"]
|
||||
|
||||
with subtest("Can login"):
|
||||
auth_token = login("netbox", "netbox")
|
||||
|
||||
def get(uri: str):
|
||||
return json.loads(
|
||||
machine.succeed(
|
||||
"curl -sSfL "
|
||||
"-H 'Accept: application/json' "
|
||||
f"-H 'Authorization: Token {auth_token}' "
|
||||
f"'http://localhost/api{uri}'"
|
||||
)
|
||||
)
|
||||
|
||||
def delete(uri: str):
|
||||
return machine.succeed(
|
||||
"curl -sSfL "
|
||||
f"-X DELETE "
|
||||
"-H 'Accept: application/json' "
|
||||
f"-H 'Authorization: Token {auth_token}' "
|
||||
f"'http://localhost/api{uri}'"
|
||||
)
|
||||
|
||||
|
||||
def data_request(uri: str, method: str, data: Dict[str, Any]):
|
||||
encoded_data = json.dumps(data)
|
||||
return json.loads(
|
||||
machine.succeed(
|
||||
"curl -sSfL "
|
||||
f"-X {method} "
|
||||
"-H 'Accept: application/json' "
|
||||
"-H 'Content-Type: application/json' "
|
||||
f"-H 'Authorization: Token {auth_token}' "
|
||||
f"'http://localhost/api{uri}' "
|
||||
f"--data '{encoded_data}'"
|
||||
)
|
||||
)
|
||||
|
||||
def post(uri: str, data: Dict[str, Any]):
|
||||
return data_request(uri, "POST", data)
|
||||
|
||||
def patch(uri: str, data: Dict[str, Any]):
|
||||
return data_request(uri, "PATCH", data)
|
||||
|
||||
# Retrieve netbox version
|
||||
netbox_version = get("/status/")["netbox-version"]
|
||||
|
||||
with subtest("Can create objects"):
|
||||
result = post("/dcim/sites/", {"name": "Test site", "slug": "test-site"})
|
||||
site_id = result["id"]
|
||||
|
||||
|
||||
for prefix in test_objects["prefixes"].values():
|
||||
if compare(netbox_version, '4.2.0') >= 0:
|
||||
post("/ipam/prefixes/", {
|
||||
"prefix": prefix["prefix"],
|
||||
"scope_id": site_id,
|
||||
"scope_type": "dcim." + prefix["scope"]["__typename"].replace("Type", "").lower()
|
||||
})
|
||||
prefix["scope"]["id"] = str(site_id)
|
||||
else:
|
||||
post("/ipam/prefixes/", {
|
||||
"prefix": prefix["prefix"],
|
||||
"site": str(site_id),
|
||||
})
|
||||
|
||||
result = post(
|
||||
"/dcim/manufacturers/",
|
||||
{"name": "Test manufacturer", "slug": "test-manufacturer"}
|
||||
)
|
||||
manufacturer_id = result["id"]
|
||||
|
||||
# Had an issue with device-types before NetBox 3.4.0
|
||||
result = post(
|
||||
"/dcim/device-types/",
|
||||
{
|
||||
"model": "Test device type",
|
||||
"manufacturer": manufacturer_id,
|
||||
"slug": "test-device-type",
|
||||
},
|
||||
)
|
||||
device_type_id = result["id"]
|
||||
|
||||
with subtest("Can list objects"):
|
||||
result = get("/dcim/sites/")
|
||||
|
||||
assert result["count"] == 1
|
||||
assert result["results"][0]["id"] == site_id
|
||||
assert result["results"][0]["name"] == "Test site"
|
||||
assert result["results"][0]["description"] == ""
|
||||
|
||||
result = get("/dcim/device-types/")
|
||||
assert result["count"] == 1
|
||||
assert result["results"][0]["id"] == device_type_id
|
||||
assert result["results"][0]["model"] == "Test device type"
|
||||
|
||||
with subtest("Can update objects"):
|
||||
new_description = "Test site description"
|
||||
patch(f"/dcim/sites/{site_id}/", {"description": new_description})
|
||||
result = get(f"/dcim/sites/{site_id}/")
|
||||
assert result["description"] == new_description
|
||||
|
||||
with subtest("Can delete objects"):
|
||||
# Delete a device-type since no object depends on it
|
||||
delete(f"/dcim/device-types/{device_type_id}/")
|
||||
|
||||
result = get("/dcim/device-types/")
|
||||
assert result["count"] == 0
|
||||
|
||||
def request_graphql(query: str):
|
||||
return machine.succeed(
|
||||
"curl -sSfL "
|
||||
"-H 'Accept: application/json' "
|
||||
"-H 'Content-Type: application/json' "
|
||||
f"-H 'Authorization: Token {auth_token}' "
|
||||
"'http://localhost/graphql/' "
|
||||
f"--data '{json.dumps({"query": query})}'"
|
||||
)
|
||||
|
||||
|
||||
if compare(netbox_version, '4.2.0') >= 0:
|
||||
with subtest("Can use the GraphQL API (NetBox 4.2.0+)"):
|
||||
graphql_query = '''query {
|
||||
prefix_list {
|
||||
prefix
|
||||
class_type
|
||||
family {
|
||||
label
|
||||
}
|
||||
scope {
|
||||
__typename
|
||||
... on SiteType {
|
||||
id
|
||||
description
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
'''
|
||||
|
||||
answer = request_graphql(graphql_query)
|
||||
result = json.loads(answer)
|
||||
assert len(result["data"]["prefix_list"]) == 3
|
||||
assert test_objects["prefixes"]["v4-with-updated-desc"] in result["data"]["prefix_list"]
|
||||
assert test_objects["prefixes"]["v6-cidr-32"] in result["data"]["prefix_list"]
|
||||
assert test_objects["prefixes"]["v6-cidr-48"] in result["data"]["prefix_list"]
|
||||
|
||||
if compare(netbox_version, '4.2.0') < 0:
|
||||
with subtest("Can use the GraphQL API (Netbox <= 4.2.0)"):
|
||||
answer = request_graphql('''query {
|
||||
prefix_list {
|
||||
prefix
|
||||
site {
|
||||
id
|
||||
}
|
||||
}
|
||||
}
|
||||
''')
|
||||
result = json.loads(answer)
|
||||
print(result["data"]["prefix_list"][0])
|
||||
assert result["data"]["prefix_list"][0]["prefix"] == test_objects["prefixes"]["v4-with-updated-desc"]["prefix"]
|
||||
assert int(result["data"]["prefix_list"][0]["site"]["id"]) == int(test_objects["prefixes"]["v4-with-updated-desc"]["scope"]["id"])
|
||||
|
||||
with subtest("Can login with LDAP"):
|
||||
machine.wait_for_unit("openldap.service")
|
||||
login("alice", "${testPassword}")
|
||||
|
||||
with subtest("Can associate LDAP groups"):
|
||||
result = get("/users/users/?username=${testUser}")
|
||||
|
||||
assert result["count"] == 1
|
||||
assert any(group["name"] == "${testGroup}" for group in result["results"][0]["groups"])
|
||||
Reference in New Issue
Block a user