Unverified Commit 963f97c0 authored by misuzu's avatar misuzu Committed by GitHub
Browse files

netbox_4_2: 4.2.3 -> 4.2.6, nixos/tests/netbox: Adjust tests for Netbox 4.2.5+ (#394329)

parents 9b339964 42c621ba
Loading
Loading
Loading
Loading
+3 −4
Original line number Diff line number Diff line
@@ -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 {};
+164 −0
Original line number Diff line number Diff line
@@ -9,7 +9,7 @@ let
  testPassword = "verySecure";
  testGroup = "netbox-users";
in
import ../make-test-python.nix (
import ../../make-test-python.nix (
  {
    lib,
    pkgs,
@@ -25,6 +25,8 @@ import ../make-test-python.nix (
      ];
    };

    skipTypeCheck = true;

    nodes.machine =
      { config, ... }:
      {
@@ -154,178 +156,9 @@ import ../make-test-python.nix (
          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"])
      '';
      builtins.replaceStrings
        [ "$\{changePassword}" "$\{testUser}" "$\{testPassword}" "$\{testGroup}" ]
        [ "${changePassword}" "${testUser}" "${testPassword}" "${testGroup}" ]
        (lib.readFile "${./testScript.py}");
  }
)
+274 −0
Original line number Diff line number Diff line
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"])
+2 −2
Original line number Diff line number Diff line
@@ -14,7 +14,7 @@ let
in
py.pkgs.buildPythonApplication rec {
  pname = "netbox";
  version = "4.2.3";
  version = "4.2.6";

  format = "other";

@@ -22,7 +22,7 @@ py.pkgs.buildPythonApplication rec {
    owner = "netbox-community";
    repo = "netbox";
    tag = "v${version}";
    hash = "sha256-vdH/R88Vtu+xRLjETK0h+E4WoYRoseP0r+wROi8nMcM=";
    hash = "sha256-SOGVMaqAYc+DeyeF5ZQ4TQr9RIhWH23Lwth3h0Y3Dtg=";
  };

  patches = [