# -*- coding: utf8 -*- """Utilities to manage login workflows and states.""" from __future__ import annotations __all__ = ['LoginManager', 'LOGIN_MANAGER', 'StateManager', 'STATE_MANAGER'] import time import typing from qtpy import QtCore from anaconda_navigator import config from anaconda_navigator.config import preferences from anaconda_navigator.config import structures from anaconda_navigator.utils.logs import logger from anaconda_navigator.utils import singletons from anaconda_navigator.utils import telemetry from anaconda_navigator.utils import workers from . import api class LoginManager(QtCore.QObject): """Manage .cloud login flow.""" sig_login_started = QtCore.Signal() sig_login_succeeded = QtCore.Signal(workers.TaskResult) sig_login_canceled = QtCore.Signal(workers.TaskResult) sig_login_failed = QtCore.Signal(workers.TaskResult) sig_login_done = QtCore.Signal(workers.TaskResult) sig_logout = QtCore.Signal() sig_state_changed = QtCore.Signal(bool) sig_throttle_started = QtCore.Signal() sig_throttle_finished = QtCore.Signal() def __init__(self, parent: QtCore.QObject | None = None) -> None: """Initialize new instance of a :class:`~CloudLoginManager`.""" super().__init__(parent=parent) self.__pending: bool = False self.__worker: workers.TaskWorker | None = None self.__throttle: bool = False self.__throttle_timer: typing.Final[QtCore.QTimer] = QtCore.QTimer(self) self.__throttle_timer.setInterval(5_000) # 5 seconds self.__throttle_timer.setSingleShot(True) self.__throttle_timer.timeout.connect(self.__stop_throttle) # login def login(self, origin: str = 'unknown') -> None: """Start a login process.""" if self.__throttle: logger.debug('cloud login throttled') return logger.debug('cloud login initiated') telemetry.ANALYTICS.instance.event('cloud-login-requested', {'origin': origin}) if self.__worker is not None: self.__pending = True self.__worker.cancel() else: self.__login() def __login(self) -> None: """Initialize a worker to process login workflow.""" self.__start_throttle() self.__worker = typing.cast(workers.TaskWorker, api.CloudAPI().login.worker()) self.__worker.signals.sig_start.connect(self.sig_login_started) self.__worker.signals.sig_succeeded.connect(self.sig_login_succeeded) self.__worker.signals.sig_canceled.connect(self.sig_login_canceled) self.__worker.signals.sig_failed.connect(self.sig_login_failed) self.__worker.signals.sig_done.connect(self.__login_done) self.__worker.start() def __login_done(self, result: workers.TaskResult) -> None: """Process result of a login processing.""" self.__stop_throttle() self.__worker = None self.sig_login_done.emit(result) if result.status == workers.TaskStatus.SUCCEEDED: logger.debug('cloud login succeeded') telemetry.ANALYTICS.instance.event('cloud-login') self.sig_state_changed.emit(True) elif result.status == workers.TaskStatus.CANCELED: if self.__pending: self.__pending = False self.__login() logger.debug('cloud login canceled') telemetry.ANALYTICS.instance.event('cancel-cloud-login') elif result.status == workers.TaskStatus.FAILED: logger.debug('cloud login failed') telemetry.ANALYTICS.instance.event('cloud-login-failed') else: # just in case we add new status and forget to add it here logger.debug('cloud login done') # logout def logout(self) -> None: """Log out from .cloud account.""" if not api.CloudAPI().token: logger.debug('cloud logout ignored') return logger.debug('cloud logout initiated') if self.__worker is not None: self.__worker.cancel() api.CloudAPI().logout() telemetry.ANALYTICS.instance.event('cloud-logout') self.sig_logout.emit() self.sig_state_changed.emit(False) logger.debug('cloud logout succeeded') # throttle def __start_throttle(self) -> None: """ Start login process throttling. It doesn't allow to spam logins by giving each login attempt 5 seconds to start. """ if self.__throttle: return logger.debug('cloud login throttling started') self.__throttle = True self.__throttle_timer.start() self.sig_throttle_started.emit() def __stop_throttle(self) -> None: """Finish login throttling.""" if not self.__throttle: return logger.debug('cloud login throttling finished') self.__throttle_timer.stop() self.__throttle = False self.sig_throttle_finished.emit() LOGIN_MANAGER: typing.Final[singletons.Singleton[LoginManager]] = singletons.SingleInstanceOf(LoginManager) class StateManager: """Manager for the cloud login popup state.""" __slots__ = ('__delays', '__state', '__timestamp') CONF_GROUP: typing.Final[str] = 'internal' STATE_FIELD: typing.Final[str] = 'cloud_login_popup_state' TIMESTAMP_FIELD: typing.Final[str] = 'cloud_login_popup_ts' def __init__(self, delays: structures.Intervals[int] | None = None) -> None: """Initialize new instance of a :class:`~StateManager`.""" if delays is None: delays = preferences.CLOUD_LOGIN_POPUP_DELAYS self.__delays: typing.Final[structures.Intervals[int]] = delays self.__state: int = config.CONF.get(self.CONF_GROUP, self.STATE_FIELD, delays.first) self.__timestamp: int = config.CONF.get(self.CONF_GROUP, self.TIMESTAMP_FIELD, 0) @property def active(self) -> bool: # noqa: D401 """Current state allows login popup to be shown (user did not select "do not show again" previously).""" return self.state >= self.delays.first @property def delays(self) -> structures.Intervals[int]: # noqa: D401 """Configuration for delay periods between each cloud login popup.""" return self.__delays @property def pending(self) -> bool: # noqa: D401 """Cloud login popup should be shown now.""" return self.active and (time.time() >= self.timestamp) @property def state(self) -> int: # noqa: D401 """Index of the current state.""" return self.__state @state.setter def state(self, value: int) -> None: """Update :attr:`~StateManager.state` value.""" self.__state = value config.CONF.set(self.CONF_GROUP, self.STATE_FIELD, self.__state) @property def timestamp(self) -> int: # noqa: D401 """When next login popup should be shown.""" return self.__timestamp @timestamp.setter def timestamp(self, value: int) -> None: """update :attr:`~StateManager.timestamp` value.""" self.__timestamp = value config.CONF.set(self.CONF_GROUP, self.TIMESTAMP_FIELD, self.__timestamp) def after_login(self) -> None: """Update state after user successfully logged in.""" self.__update(self.delays.last, False) def after_logout(self) -> None: """Update state after user logged out.""" self.__update(self.delays.last, True) def before_popup(self) -> None: """Update state right before showing cloud login popup.""" self.__update(self.state + 1, True) def check(self, login: bool) -> None: """Check if current state properly reflects current cloud :code:`login` status.""" if self.active: if login: if (self.timestamp != 0) or (self.state < self.delays.last): self.after_login() elif self.timestamp == 0: self.after_logout() def do_not_show_again(self) -> None: """Do not show cloud login popup again.""" self.state = self.delays.first - 1 self.timestamp = 0 def __update(self, state: int, pending: bool) -> None: """ Update current state. Common interface for all other updates. """ if self.active: self.state = max(self.state, state) self.timestamp = int(self.active and pending) and int(time.time()) + self.delays.get(self.state) STATE_MANAGER: typing.Final[singletons.Singleton[StateManager]] = singletons.SingleInstanceOf(StateManager)