This repository has been archived by the owner on Feb 6, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
test: added tests and examples
- Loading branch information
Showing
5 changed files
with
165 additions
and
78 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,49 @@ | ||
""" | ||
user_management.py | ||
""" | ||
from ipfabric import IPFClient | ||
from ipfabric.settings import UserMgmt | ||
|
||
|
||
if __name__ == '__main__': | ||
ipf = IPFClient('https://demo3.ipfabric.io/') # Token must have User Management Permissions | ||
usermgmt = UserMgmt(client=ipf) | ||
|
||
print(usermgmt.users[0]) | ||
""" | ||
username='justin' scope=['read', 'write', 'settings'] email='justin.jeffrey@ipfabric.io' user_id='1108612054' | ||
local=True sso_provider=None domains='' custom_scope=True ldap_id=None | ||
""" | ||
print() | ||
|
||
user = usermgmt.add_user(username='Test', email='test@ipfabric.io', password='8characters', | ||
scope=['read', 'write', 'settings', 'team']) | ||
print(user) | ||
""" | ||
username='Test' scope=['read', 'write', 'settings', 'team'] email='test@ipfabric.io' user_id='1168572704' | ||
local=None sso_provider=None domains=None custom_scope=True ldap_id=None | ||
""" | ||
print() | ||
|
||
print(usermgmt.get_user_by_id(user_id=user.user_id)) | ||
""" | ||
username='Test' scope=['read', 'write', 'settings', 'team'] email='test@ipfabric.io' user_id='1168572704' | ||
local=None sso_provider=None domains=None custom_scope=True ldap_id=None | ||
""" | ||
print() | ||
|
||
print(usermgmt.get_users(username=user.username)) | ||
""" | ||
[User(username='Test', scope=['read', 'write', 'settings', 'team'], email='test@ipfabric.io', | ||
user_id='1168572704', local=True, sso_provider=None, domains='', custom_scope=True, ldap_id=None)] | ||
""" | ||
print() | ||
|
||
print(usermgmt.delete_user(user_id=user.user_id)) | ||
""" | ||
[User(username='justin', scope=['read', 'write', 'settings'], email='justin.jeffrey@ipfabric.io', | ||
user_id='1108612054', local=True, sso_provider=None, domains='', custom_scope=True, ldap_id=None), | ||
User(username='vector', scope=['read', 'write', 'settings'], email='vector@vector.pl', user_id='1083776225', | ||
local=True, sso_provider=None, domains='', custom_scope=True, ldap_id=None), ] | ||
""" | ||
print() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,6 +1,6 @@ | ||
from .api_tokens import APIToken | ||
from .authentication import Authentication | ||
from .seeds import Seeds | ||
from .users import Users | ||
from .user_mgmt import UserMgmt | ||
|
||
__all__ = [Authentication, Seeds, APIToken, Users] | ||
__all__ = [Authentication, Seeds, APIToken, UserMgmt] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,80 @@ | ||
import logging | ||
from typing import Any, Optional | ||
|
||
from pydantic import Field, BaseModel | ||
|
||
from ipfabric.tools.helpers import create_regex | ||
|
||
logger = logging.getLogger() | ||
|
||
|
||
class User(BaseModel): | ||
username: str | ||
scope: list | ||
email: str | ||
user_id: str = Field(alias='id') | ||
local: Optional[bool] = Field(alias='isLocal') | ||
sso_provider: Optional[Any] = Field(alias='ssoProvider') | ||
domains: Optional[Any] = Field(alias='domainSuffixes') | ||
custom_scope: bool = Field(alias='customScope') | ||
ldap_id: Any = Field(alias='ldapId') | ||
|
||
|
||
class UserMgmt: | ||
def __init__(self, client): | ||
self.client: Any = client | ||
self.users = self.get_users() | ||
|
||
def get_users(self, username: str = None): | ||
""" | ||
Gets all users or filters on one of the options. | ||
:param username: str: Username to filter | ||
:return: List of users | ||
""" | ||
payload = { | ||
"columns": ["id", "isLocal", "username", "ssoProvider", "ldapId", | ||
"domainSuffixes", "email", "customScope", "scope"] | ||
} | ||
if username: | ||
payload['filters'] = {"username": ["reg", create_regex(username)]} | ||
users = self.client._ipf_pager('tables/users', payload) | ||
return [User(**user) for user in users] | ||
|
||
def get_user_by_id(self, user_id: str): | ||
""" | ||
Gets a user by ID | ||
:param user_id: Union[str, int]: User ID to filter | ||
:return: User | ||
""" | ||
resp = self.client.get('users/' + str(user_id)) | ||
resp.raise_for_status() | ||
return User(**resp.json()) | ||
|
||
def add_user(self, username: str, email: str, password: str, scope: list): | ||
""" | ||
Adds a user | ||
:param username: str: Username | ||
:param email: str: Email | ||
:param password: str: Must be 8 characters | ||
:param scope: list: Accepted values: ['read', 'write', 'settings', 'team'] | ||
:return: User | ||
""" | ||
if len(password) < 8: | ||
raise SyntaxError("Password must be 8 characters.") | ||
if not all(x in ['read', 'write', 'settings', 'team'] for x in scope): | ||
raise SyntaxError("Only accepted scopes are ['read', 'write', 'settings', 'team']") | ||
resp = self.client.post('users', | ||
json={"username": username, "email": email, "password": password, "scope": scope}) | ||
resp.raise_for_status() | ||
user_id = resp.json()['id'] | ||
return self.get_user_by_id(user_id) | ||
|
||
def delete_user(self, user_id: str): | ||
""" | ||
Deletes a user and returns list of remaining users | ||
:param user_id: | ||
:return: | ||
""" | ||
resp = self.client.delete('users/' + str(user_id)) | ||
resp.raise_for_status() | ||
return self.get_users() |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,33 +1,45 @@ | ||
import unittest | ||
from unittest.mock import MagicMock | ||
|
||
from ipfabric.settings import users | ||
from ipfabric.settings import user_mgmt | ||
|
||
|
||
class TestUsers(unittest.TestCase): | ||
def setUp(self) -> None: | ||
mock = MagicMock() | ||
mock._ipf_pager.return_value = [{'id': '1108612054', 'isLocal': True, 'username': 'justin', | ||
'ssoProvider': None, 'domainSuffixes': '', | ||
'email': 'justin.jeffrey@ipfabric.io', 'customScope': True, | ||
'scope': ['read', 'write', 'settings']}] | ||
mock.get().json.return_value = {'id': '1108612054', 'isLocal': True, 'username': 'justin', | ||
'ssoProvider': None, 'domainSuffixes': '', | ||
'email': 'justin.jeffrey@ipfabric.io', 'customScope': True, | ||
'scope': ['read', 'write', 'settings']} | ||
self.usermgmt = user_mgmt.UserMgmt(mock) | ||
|
||
def test_users(self): | ||
user = users.Users(MagicMock()) | ||
user.client._ipf_pager.return_value = [{'id': '1108612054', 'isLocal': True, 'username': 'justin', | ||
'ssoProvider': None, 'domainSuffixes': '', | ||
'email': 'justin.jeffrey@ipfabric.io', 'customScope': True, | ||
'scope': ['read', 'write', 'settings']}] | ||
u = user.users[0] | ||
self.assertIsInstance(u, users.User) | ||
u = self.usermgmt.users[0] | ||
self.assertIsInstance(u, user_mgmt.User) | ||
|
||
def test_get_users(self): | ||
user = users.Users(MagicMock()) | ||
user.client._ipf_pager.return_value = [{'id': '1108612054', 'isLocal': True, 'username': 'justin', | ||
'ssoProvider': None, 'domainSuffixes': '', | ||
'email': 'justin.jeffrey@ipfabric.io', 'customScope': True, | ||
'scope': ['read', 'write', 'settings']}] | ||
u = user.get_users('test')[0] | ||
self.assertIsInstance(u, users.User) | ||
u = self.usermgmt.get_users('test')[0] | ||
self.assertIsInstance(u, user_mgmt.User) | ||
|
||
def test_get_user_by_id(self): | ||
user = users.Users(MagicMock()) | ||
user.client._ipf_pager.return_value = [{'id': '1108612054', 'isLocal': True, 'username': 'justin', | ||
'ssoProvider': None, 'domainSuffixes': '', | ||
'email': 'justin.jeffrey@ipfabric.io', 'customScope': True, | ||
'scope': ['read', 'write', 'settings']}] | ||
u = user.get_user_by_id('1108612054') | ||
self.assertIsInstance(u, users.User) | ||
u = self.usermgmt.get_user_by_id('1108612054') | ||
self.assertIsInstance(u, user_mgmt.User) | ||
|
||
def test_add_user(self): | ||
self.usermgmt.client.post().json.return_value = {'id': 1} | ||
u = self.usermgmt.add_user('test', 'test', 'test1234', ['read']) | ||
self.assertIsInstance(u, user_mgmt.User) | ||
|
||
def test_add_user_fail(self): | ||
with self.assertRaises(SyntaxError) as err: | ||
self.usermgmt.add_user('test', 'test', 'test', ['read']) | ||
with self.assertRaises(SyntaxError) as err: | ||
self.usermgmt.add_user('test', 'test', 'test1234', ['hello']) | ||
|
||
def test_delete_user(self): | ||
u = self.usermgmt.delete_user('test1234') | ||
self.assertIsInstance(u[0], user_mgmt.User) |