Extend api with account and user management

2025-08-02 02:33:44 +00:00 · 2020-01-27 12:38:38 +00:00
parent 1cd423041c
commit 4d391ccb34
10 changed files with 1730 additions and 8 deletions
--- a/tests/integration/api/management/test_admin_user.py
+++ b/tests/integration/api/management/test_admin_user.py
@@ -0,0 +1,367 @@
+
+import json
+from tests.fixtures import (    # noqa: F401
+    client, initial_data, basic_auth_admin_headers,
+    test_admin_user, test_user, account_data, user1_data,
+)
+from . import IntegrationApiManagement
+
+
+class TestIntegrationApiManagementAdminUser(IntegrationApiManagement):
+
+    def test_accounts_empty_get(
+            self, client, initial_data,                         # noqa: F811
+            basic_auth_admin_headers):                          # noqa: F811
+        res = client.get("/api/v1/pdnsadmin/accounts",
+                         headers=basic_auth_admin_headers)
+        data = res.get_json(force=True)
+        assert res.status_code == 200
+        assert data == []
+
+    def test_users_empty_get(
+            self, client, initial_data,                         # noqa: F811
+            test_admin_user, test_user,                         # noqa: F811
+            basic_auth_admin_headers):                          # noqa: F811
+        res = client.get("/api/v1/pdnsadmin/users",
+                         headers=basic_auth_admin_headers)
+        data = res.get_json(force=True)
+        assert res.status_code == 200
+        # Initally contains 2 records
+        assert len(data) == 2
+        for user in data:
+            assert user["username"] in (test_admin_user, test_user)
+
+    def test_accounts(
+            self, client, initial_data,                         # noqa: F811
+            account_data,                                       # noqa: F811
+            basic_auth_admin_headers):                          # noqa: F811
+        account_name = account_data["name"]
+        self.client = client
+        self.basic_auth_admin_headers = basic_auth_admin_headers
+
+        # Create account
+        res = client.post(
+            "/api/v1/pdnsadmin/accounts",
+            headers=basic_auth_admin_headers,
+            data=json.dumps(account_data),
+            content_type="application/json",
+        )
+        data = res.get_json(force=True)
+        assert res.status_code == 201
+
+        # Check account
+        data = self.check_account(account_data)
+        account_id = data["id"]
+
+        updated = account_data.copy()
+        # Update and check values
+        for upd_key in ["description", "contact", "mail"]:
+            upd_value = "upd-{}".format(account_data[upd_key])
+
+            # Update
+            data = {"name": account_name, upd_key: upd_value}
+            res = client.put(
+                "/api/v1/pdnsadmin/accounts/{}".format(account_id),
+                data=json.dumps(data),
+                headers=basic_auth_admin_headers,
+                content_type="application/json",
+            )
+            assert res.status_code == 204
+            updated[upd_key] = upd_value
+
+            # Check
+            data = self.check_account(updated)
+
+        # Update to defaults
+        res = client.put(
+            "/api/v1/pdnsadmin/accounts/{}".format(account_id),
+            data=json.dumps(account_data),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 204
+
+        # Check account
+        res = client.get(
+            "/api/v1/pdnsadmin/accounts/{}".format(account_name),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        data = res.get_json(force=True)
+        assert res.status_code == 200
+        assert len(data) == 1
+        data = data[0]
+        account_id = data["id"]
+        for key, value in account_data.items():
+            assert data[key] == value
+
+        # Cleanup (delete account)
+        res = client.delete(
+            "/api/v1/pdnsadmin/accounts/{}".format(account_id),
+            data=json.dumps(account_data),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 204
+
+        # Get non-existing account (should fail)
+        data = self.get_account(account_name, status_code=404)
+
+        # Update non-existing account (should fail)
+        res = client.put(
+            "/api/v1/pdnsadmin/accounts/{}".format(account_id),
+            data=json.dumps(account_data),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 404
+
+        # Delete non-existing account (should fail)
+        res = client.delete(
+            "/api/v1/pdnsadmin/accounts/{}".format(account_id),
+            data=json.dumps(account_data),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 404
+
+    def test_users(
+            self, client, initial_data,                         # noqa: F811
+            user1_data,                                         # noqa: F811
+            basic_auth_admin_headers):                          # noqa: F811
+        user1name = user1_data["username"]
+        self.client = client
+        self.basic_auth_admin_headers = basic_auth_admin_headers
+
+        # Create user (user1)
+        res = client.post(
+            "/api/v1/pdnsadmin/users",
+            headers=basic_auth_admin_headers,
+            data=json.dumps(user1_data),
+            content_type="application/json",
+        )
+        data = res.get_json(force=True)
+        assert res.status_code == 201
+        assert len(data) == 1
+
+        # Check user
+        user1 = self.check_user(user1_data, data[0])
+        user1_id = user1["id"]
+
+        updated = user1_data.copy()
+        # Update and check values
+        for upd_key in ["firstname", "lastname", "email"]:
+            upd_value = "upd-{}".format(user1_data[upd_key])
+
+            # Update
+            data = {"username": user1name, upd_key: upd_value}
+            res = client.put(
+                "/api/v1/pdnsadmin/users/{}".format(user1_id),
+                data=json.dumps(data),
+                headers=basic_auth_admin_headers,
+                content_type="application/json",
+            )
+            assert res.status_code == 204
+            updated[upd_key] = upd_value
+
+            # Check
+            data = self.check_user(updated)
+
+        # Update to defaults
+        res = client.put(
+            "/api/v1/pdnsadmin/users/{}".format(user1_id),
+            data=json.dumps(user1_data),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 204
+
+        # Check user
+        self.check_user(user1_data)
+
+        # Cleanup (delete user)
+        res = client.delete(
+            "/api/v1/pdnsadmin/users/{}".format(user1_id),
+            data=json.dumps(user1_data),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 204
+
+        # Get non-existing user (should fail)
+        data = self.get_user(user1name, status_code=404)
+
+        # Update non-existing user (should fail)
+        res = client.put(
+            "/api/v1/pdnsadmin/users/{}".format(user1_id),
+            data=json.dumps(user1_data),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 404
+
+        # Delete non-existing user (should fail)
+        res = client.delete(
+            "/api/v1/pdnsadmin/users/{}".format(user1_id),
+            data=json.dumps(user1_data),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 404
+
+    def test_account_users(
+            self, client, initial_data,                         # noqa: F811
+            test_user, account_data, user1_data,                # noqa: F811
+            basic_auth_admin_headers):                          # noqa: F811
+        self.client = client
+        self.basic_auth_admin_headers = basic_auth_admin_headers
+        test_user_id = self.get_user(test_user)["id"]
+
+        # Create account
+        res = client.post(
+            "/api/v1/pdnsadmin/accounts",
+            headers=basic_auth_admin_headers,
+            data=json.dumps(account_data),
+            content_type="application/json",
+        )
+        data = res.get_json(force=True)
+        assert res.status_code == 201
+
+        # Check account
+        data = self.check_account(account_data)
+        account_id = data["id"]
+
+        # Create user1
+        res = client.post(
+            "/api/v1/pdnsadmin/users",
+            headers=basic_auth_admin_headers,
+            data=json.dumps(user1_data),
+            content_type="application/json",
+        )
+        data = res.get_json(force=True)
+        assert res.status_code == 201
+        assert len(data) == 1
+
+        # Check user
+        user1 = self.check_user(user1_data, data[0])
+        user1_id = user1["id"]
+
+        # Assert test account has no users
+        res = client.get(
+            "/api/v1/pdnsadmin/accounts/users/{}".format(account_id),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        data = res.get_json(force=True)
+        assert res.status_code == 200
+        assert data == []
+
+        # Assert unlinking an unlinked account fails
+        res = client.delete(
+            "/api/v1/pdnsadmin/accounts/users/{}/{}".format(
+                account_id, user1_id),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 404
+
+        # Link user to account
+        res = client.put(
+            "/api/v1/pdnsadmin/accounts/users/{}/{}".format(
+                account_id, user1_id),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 204
+
+        # Check user is linked to account
+        res = client.get(
+            "/api/v1/pdnsadmin/accounts/users/{}".format(account_id),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        data = res.get_json(force=True)
+        assert res.status_code == 200
+        assert len(data) == 1
+        self.check_user(user1_data, data[0])
+
+        # Unlink user from account
+        res = client.delete(
+            "/api/v1/pdnsadmin/accounts/users/{}/{}".format(
+                account_id, user1_id),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 204
+
+        # Check user is unlinked from account
+        res = client.get(
+            "/api/v1/pdnsadmin/accounts/users/{}".format(account_id),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        data = res.get_json(force=True)
+        assert res.status_code == 200
+        assert data == []
+
+        # Unlink unlinked user from account (should fail)
+        res = client.delete(
+            "/api/v1/pdnsadmin/accounts/users/{}/{}".format(
+                account_id, user1_id),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 404
+
+        # Cleanup (delete user)
+        res = client.delete(
+            "/api/v1/pdnsadmin/users/{}".format(user1_id),
+            data=json.dumps(user1_data),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 204
+
+        # Link non-existing user to account (should fail)
+        res = client.put(
+            "/api/v1/pdnsadmin/accounts/users/{}/{}".format(
+                account_id, user1_id),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 404
+
+        # Unlink non-exiting user from account (should fail)
+        res = client.delete(
+            "/api/v1/pdnsadmin/accounts/users/{}/{}".format(
+                account_id, user1_id),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 404
+
+        # Cleanup (delete account)
+        res = client.delete(
+            "/api/v1/pdnsadmin/accounts/{}".format(account_id),
+            data=json.dumps(account_data),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 204
+
+        # List users in non-existing account (should fail)
+        res = client.get(
+            "/api/v1/pdnsadmin/accounts/users/{}".format(account_id),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 404
+
+        # Link existing user to non-existing account (should fail)
+        res = client.put(
+            "/api/v1/pdnsadmin/accounts/users/{}/{}".format(
+                account_id, test_user_id),
+            headers=basic_auth_admin_headers,
+            content_type="application/json",
+        )
+        assert res.status_code == 404