Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 0 additions & 20 deletions app/commands/user.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,6 @@
import app.services.output_manager.help_page as user_help
import app.services.output_manager.message_handler as mhandler
from app.services.user_authentication.decorator import require_login_session

# from app.services.user_authentication.user_login_logout import user_login
from app.services.user_authentication.user_login_logout import user_device_id_login
from app.services.user_authentication.user_login_logout import user_logout
from app.services.user_authentication.user_login_logout import validate_user_device_login
Expand All @@ -21,24 +19,6 @@ def cli():
pass


# COMMENT user password login temporally.
# @click.command()
# @click.option(
# '-U', '--username', prompt='Username', help=(user_help.user_help_page(user_help.UserHELP.USER_LOGIN_USERNAME))
# )
# @click.option(
# '-P',
# '--password',
# prompt='Password',
# help=(user_help.user_help_page(user_help.UserHELP.USER_LOGIN_PASSWORD)),
# hide_input=True,
# )
# @doc(user_help.user_help_page(user_help.UserHELP.USER_LOGIN))
# def login(username, password):
# user_login(username, password)
# mhandler.SrvOutPutHandler.login_success()


@click.command()
@doc(user_help.user_help_page(user_help.UserHELP.USER_LOGIN))
def login():
Expand Down
3 changes: 3 additions & 0 deletions app/configs/user_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,9 @@ def clear(self):
}
self.save()

def is_logged_in(self) -> bool:
return bool(self.access_token and self.refresh_token)

@property
def username(self):
return decryption(self.config['USER']['username'], self.secret)
Expand Down
7 changes: 2 additions & 5 deletions app/services/user_authentication/token_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,13 +16,10 @@
class SrvTokenManager(metaclass=MetaService):
def __init__(self):
user_config = UserConfig()
has_user = user_config.config.has_section('USER')
has_access_token = user_config.config.has_option('USER', 'access_token')
has_refresh_token = user_config.config.has_option('USER', 'refresh_token')
if has_user and has_access_token and has_refresh_token:
if user_config.is_logged_in():
self.config = user_config
else:
raise (Exception('Login session not found, please login first.'))
raise Exception('Login session not found, please login first.')

def update_token(self, access_token, refresh_token):
self.config.access_token = access_token
Expand Down
32 changes: 1 addition & 31 deletions app/services/user_authentication/user_login_logout.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,33 +17,6 @@
from app.services.output_manager.message_handler import SrvOutPutHandler


def user_login(username, password):
url = AppConfig.Connections.url_authn
user_config = UserConfig()
request_body = {'username': username, 'password': password}
headers = {'Content-Type': 'application/json'}
response = requests.post(url, json=request_body, headers=headers)
if response.status_code == 200:
res_to_dict = response.json()
user_config.username = username
user_config.password = password
user_config.access_token = res_to_dict['result']['access_token']
user_config.refresh_token = res_to_dict['result']['refresh_token']
user_config.last_active = str(int(time.time()))
user_config.hpc_token = ''
user_config.session_id = 'cli-' + str(uuid4())
user_config.save()
elif response.status_code == 401:
res_to_dict = []
SrvErrorHandler.customized_handle(ECustomizedError.INVALID_CREDENTIALS, True)
else:
if response.text:
SrvErrorHandler.default_handle(response.text, True)
res_to_dict = response.json()
SrvErrorHandler.default_handle(response.content, True)
return res_to_dict


def user_device_id_login() -> Dict[str, Any]:
"""Get device code URL for user login."""

Expand Down Expand Up @@ -104,10 +77,7 @@ def validate_user_device_login(device_code: str, expires: int, interval: int) ->

def check_is_login(if_print: bool = True) -> bool:
user_config = UserConfig()
has_username = user_config.config.has_option('USER', 'username')
has_access_token = user_config.config.has_option('USER', 'access_token')
has_refresh_token = user_config.config.has_option('USER', 'refresh_token')
if has_username and has_access_token and has_refresh_token and user_config.username != '':
if user_config.is_logged_in():
return True
else:
SrvErrorHandler.customized_handle(ECustomizedError.LOGIN_SESSION_INVALID, if_print) if if_print else None
Expand Down
57 changes: 27 additions & 30 deletions poetry.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ click = "7.1.2"
PyJWT = "1.7.1"
questionary = "^1.10.0"
tqdm = "4.56.0"
cryptography = "3.1.1"
cryptography = "3.4.8"
urllib3 = "^1.26.14"
requests = "^2.28.2"
pydantic = "^1.9.2"
Expand Down
53 changes: 1 addition & 52 deletions tests/app/services/user_authentication/test_user_login_logout.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,64 +6,13 @@

from app.configs.app_config import AppConfig
from app.configs.user_config import UserConfig
from app.resources.custom_error import Error
from app.services.output_manager.error_handler import ECustomizedError
from app.services.user_authentication.user_login_logout import check_is_login
from app.services.user_authentication.user_login_logout import user_device_id_login
from app.services.user_authentication.user_login_logout import user_login
from app.services.user_authentication.user_login_logout import validate_user_device_login


def test_user_login_success(requests_mock):
requests_mock.post(
'http://service_auth',
json={
'code': 200,
'error_msg': '',
'page': 0,
'total': 1,
'num_of_pages': 1,
'result': {
'access_token': 'fake-token',
'expires_in': 300,
'refresh_expires_in': 360,
'refresh_token': 'refresh-token',
'token_type': 'Bearer',
'not-before-policy': 0,
'session_state': 'session-state',
'scope': 'roles groups profile email',
},
},
)
res = user_login('username', 'password')
assert res.get('code') == 200
assert res['result'].get('access_token') == 'fake-token'
assert res['result'].get('refresh_token') == 'refresh-token'
assert res.get('error_msg') == ''


def test_user_login_wrong_password(requests_mock, capsys):
requests_mock.post(
'http://service_auth',
json={
'code': 401,
'error_msg': "401: b'{\"error\":\"invalid_grant\",\"error_description\":\"Invalid user credentials\"}'",
'page': 0,
'total': 1,
'num_of_pages': 1,
'result': [],
},
status_code=401,
)
with pytest.raises(SystemExit):
user_login('username', 'password')
out, err = capsys.readouterr()
assert out == Error.error_msg.get(ECustomizedError.INVALID_CREDENTIALS.name, 'Unknown error.') + '\n'
assert err == ''


def test_check_is_not_login(mocker):
mocker.patch('configparser.ConfigParser.has_option', return_value=False)
mocker.patch('app.configs.user_config.UserConfig.is_logged_in', return_value=False)
expected_result = False
with pytest.raises(SystemExit):
actual = check_is_login()
Expand Down