Merge pull request #255 from Tinkoff/systemuserstorage-upd

Updated SystemUserStorage

demo/demo.py

@@ -71,12 +71,13 @@ def _ensure_demo_app_has_features(settings_generator: OverhaveDemoSettingsGenera
     synchronizer = _create_synchronizer()
     with db.create_session() as session:
         create_db_features = session.query(db.Feature).first() is None
-    if not overhave_synchronizer_factory().system_user_storage.get_user_by_credits(
-        login=settings_generator.default_feature_user
-    ):
-        overhave_synchronizer_factory().system_user_storage.create_user(
-            login=settings_generator.default_feature_user, password=SecretStr(settings_generator.default_feature_user)
-        )
+        user_storage = overhave_synchronizer_factory().system_user_storage
+        if not user_storage.get_user_by_credits(session=session, login=settings_generator.default_feature_user):
+            user_storage.create_user(
+                session=session,
+                login=settings_generator.default_feature_user,
+                password=SecretStr(settings_generator.default_feature_user),
+            )
     try:
         synchronizer.synchronize(create_db_features=create_db_features)
     except BaseScenarioParserError:

makefile

@@ -1,7 +1,7 @@
 CODE = overhave
 VENV ?= .venv
 WORK_DIR ?= .
-MIN_COVERAGE ?= 88.0
+MIN_COVERAGE ?= 88.1
 PACKAGE_BUILD_DIR ?= dist
 PYTHON_VERSION ?= 3.11
 

overhave/admin/flask/login_manager.py

@@ -1,8 +1,8 @@
 import logging
+from dataclasses import dataclass
 
 from flask import redirect
 from flask_login import LoginManager
-from pydantic import BaseModel
 from werkzeug import Response
 
 from overhave import db
@@ -15,7 +15,8 @@ class UnauthorizedUserError(Exception):
     """Raises when user is not authorized and trying to get user fields such as role or login."""
 
 
-class AdminPanelUser(BaseModel):
+@dataclass(frozen=True)
+class AdminPanelUser:
     """SystemUserModel wrapper for flask_login."""
 
     user_data: SystemUserModel | None
@@ -32,21 +33,22 @@ def is_active(self) -> bool:
     def is_anonymous(self) -> bool:
         return self.user_data is None
 
+    @property
+    def _authorized_user(self) -> SystemUserModel:
+        if self.user_data is None:
+            raise UnauthorizedUserError("User is not authorized!")
+        return self.user_data
+
     def get_id(self) -> int:
-        return self._get_authorized_user_or_raise().id
+        return self._authorized_user.id
 
     @property
     def login(self) -> str:
-        return self._get_authorized_user_or_raise().login
+        return self._authorized_user.login
 
     @property
     def role(self) -> db.Role:
-        return self._get_authorized_user_or_raise().role
-
-    def _get_authorized_user_or_raise(self) -> SystemUserModel:
-        if self.user_data is None:
-            raise UnauthorizedUserError("User is not authorized!")
-        return self.user_data
+        return self._authorized_user.role
 
     def __unicode__(self) -> str:
         if self.user_data is not None:
@@ -67,7 +69,7 @@ def __init__(self, system_user_storage: ISystemUserStorage, login_view: str) ->
 
     def _get_user(self, user_id: int) -> AdminPanelUser:
         logger.info("Get user by id=%s...", user_id)
-        return AdminPanelUser(user_data=self._system_user_storage.get_user(user_id=user_id))
+        return AdminPanelUser(user_data=self._system_user_storage.get_user_model(user_id=user_id))
 
     def _unathorized_response(self) -> Response:
         return redirect(f"/{self.login_view}")

overhave/api/auth/regular.py

@@ -4,6 +4,7 @@
 from fastapi import security as fastapi_security
 from jose import JWTError
 
+from overhave import db
 from overhave.api.auth.models import AUTH_HEADERS
 from overhave.api.auth.token import get_token_data
 from overhave.api.deps import get_api_auth_settings, get_system_user_storage
@@ -29,7 +30,8 @@ def get_authorized_user(
             raise creds_exception
     except JWTError:
         raise creds_exception
-    user = storage.get_user_by_credits(login=token_data.username)
-    if user is None:
-        raise creds_exception
-    return user
+    with db.create_session() as session:
+        user = storage.get_user_by_credits(session=session, login=token_data.username)
+        if user is None:
+            raise creds_exception
+        return SystemUserModel.from_orm(user)

overhave/api/views/auth_views.py

@@ -4,6 +4,7 @@
 from fastapi import security as fastapi_security
 from pydantic.types import SecretStr
 
+from overhave import db
 from overhave.api.auth import AUTH_HEADERS, create_access_token
 from overhave.api.deps import get_api_auth_settings, get_system_user_storage
 from overhave.api.settings import OverhaveApiAuthSettings
@@ -16,13 +17,15 @@ def login_for_access_token(
     auth_settings: OverhaveApiAuthSettings = fastapi.Depends(get_api_auth_settings),
     storage: ISystemUserStorage = fastapi.Depends(get_system_user_storage),
 ) -> AuthToken:
-    user = storage.get_user_by_credits(login=form_data.username, password=SecretStr(form_data.password))
-    if not user:
-        raise fastapi.HTTPException(
-            status_code=HTTPStatus.UNAUTHORIZED,
-            detail="Incorrect username or password",
-            headers=AUTH_HEADERS.dict(),
-        )
+    with db.create_session() as session:
+        if not storage.get_user_by_credits(
+            session=session, login=form_data.username, password=SecretStr(form_data.password)
+        ):
+            raise fastapi.HTTPException(
+                status_code=HTTPStatus.UNAUTHORIZED,
+                detail="Incorrect username or password",
+                headers=AUTH_HEADERS.dict(),
+            )
     expires_at = get_current_time() + auth_settings.access_token_expire_timedelta
     access_token = create_access_token(auth_settings=auth_settings, username=form_data.username, expires_at=expires_at)
     return AuthToken(access_token=access_token, expires_at=expires_at)

overhave/entities/auth_managers/default.py

@@ -2,6 +2,7 @@
 
 from pydantic import SecretStr
 
+from overhave import db
 from overhave.entities.auth_managers.secret_mixin import AdminSecretMixin
 from overhave.storage import SystemUserModel
 
@@ -15,4 +16,8 @@ class DefaultAdminAuthorizationManager(AdminSecretMixin):
     """
 
     def authorize_user(self, username: str, password: SecretStr) -> SystemUserModel | None:
-        return self._system_user_storage.get_user_by_credits(login=username, password=password)
+        with db.create_session() as session:
+            user = self._system_user_storage.get_user_by_credits(session=session, login=username, password=password)
+            if user is None:
+                return None
+            return SystemUserModel.from_orm(user)

overhave/entities/auth_managers/ldap/manager.py

@@ -1,9 +1,10 @@
 import logging
 from pprint import pformat
 
+import sqlalchemy.orm as so
 from pydantic import SecretStr
 
-from overhave.db import Role
+from overhave import db
 from overhave.entities.auth_managers.base import BaseAdminAuthorizationManager
 from overhave.entities.auth_managers.ldap.settings import OverhaveLdapManagerSettings
 from overhave.storage import ISystemUserGroupStorage, ISystemUserStorage, SystemUserModel
@@ -36,11 +37,16 @@ def __init__(
         self._system_user_group_storage = system_user_group_storage
         self._ldap_authenticator = ldap_authenticator
 
-    def _reassign_role_if_neccessary(self, user: SystemUserModel, user_groups: list[str]) -> None:
-        if self._settings.ldap_admin_group not in user_groups:
-            return
-        user.role = Role.admin
-        self._system_user_storage.update_user_role(user_model=user)
+    @staticmethod
+    def _user_role_by_admin_group_entry(user_has_admin_group: bool) -> db.Role:
+        if user_has_admin_group:
+            return db.Role.admin
+        return db.Role.user
+
+    def _can_user_be_created(self, session: so.Session, user_has_admin_group: bool, user_groups: list[str]) -> bool:
+        return user_has_admin_group or self._system_user_group_storage.has_any_group(
+            session=session, user_groups=user_groups
+        )
 
     def authorize_user(self, username: str, password: SecretStr) -> SystemUserModel | None:
         logger.debug("Try to authorize user '%s'...", username)
@@ -49,14 +55,24 @@ def authorize_user(self, username: str, password: SecretStr) -> SystemUserModel
             logger.info("LDAP user '%s' does not exist!", username)
             return None
         logger.debug("LDAP user groups: \n %s", pformat(user_groups))
-        user = self._system_user_storage.get_user_by_credits(login=username)
-        if user is not None:
-            self._reassign_role_if_neccessary(user=user, user_groups=user_groups)
-            return user
-        logger.debug("Have not found user with username '%s'!", username)
-        if self._system_user_group_storage.has_any_group(user_groups) or self._settings.ldap_admin_group in user_groups:
-            user = self._system_user_storage.create_user(login=username)
-            self._reassign_role_if_neccessary(user=user, user_groups=user_groups)
-            return user
-        logger.debug("Received user groups (%s) are not supplied with database groups!", user_groups)
+
+        with db.create_session() as session:
+            user_has_admin_group = self._settings.ldap_admin_group in user_groups
+            intended_user_role = self._user_role_by_admin_group_entry(user_has_admin_group)
+
+            user = self._system_user_storage.get_user_by_credits(session=session, login=username)
+            if user is not None:
+                if user.role is db.Role.user and user.role is not intended_user_role:
+                    user.role = intended_user_role
+                    session.flush()
+                return SystemUserModel.from_orm(user)
+
+            logger.debug("Have not found user with username '%s'!", username)
+            if self._can_user_be_created(
+                session=session, user_has_admin_group=user_has_admin_group, user_groups=user_groups
+            ):
+                user = self._system_user_storage.create_user(session=session, login=username, role=intended_user_role)
+                return SystemUserModel.from_orm(user)
+
+        logger.debug("Received user groups (%s) are not supplied with exist groups!", user_groups)
         return None

overhave/entities/auth_managers/secret_mixin.py

@@ -30,10 +30,12 @@ def _get_secret() -> SecretStr:
         return SecretStr(uuid4().hex)
 
     def _make_secret(self) -> None:
-        user = self._system_user_storage.get_user_by_credits(login=_ADMIN_USERNAME)
-        if user is not None:
-            logger.info("Admin user already exists")
-            return
         secret = self._get_secret()
+        with db.create_session() as session:
+            if self._system_user_storage.get_user_by_credits(session=session, login=_ADMIN_USERNAME):
+                logger.info("Admin user already exists")
+                return
+            self._system_user_storage.create_user(
+                session=session, login=_ADMIN_USERNAME, password=secret, role=db.Role.admin
+            )
         logger.info("Generated admin secret: %s", secret.get_secret_value())
-        self._system_user_storage.create_user(login=_ADMIN_USERNAME, password=secret, role=db.Role.admin)

overhave/entities/auth_managers/simple.py

@@ -2,6 +2,7 @@
 
 from pydantic import SecretStr
 
+from overhave import db
 from overhave.entities.auth_managers.secret_mixin import AdminSecretMixin
 from overhave.storage import SystemUserModel
 
@@ -24,11 +25,12 @@ class SimpleAdminAuthorizationManager(AdminSecretMixin):
     """
 
     def authorize_user(self, username: str, password: SecretStr) -> SystemUserModel | None:
-        user = self._system_user_storage.get_user_by_credits(login=username)
-        if user is None:
-            user = self._system_user_storage.create_user(login=username, password=password)
-        if user.password is None:
-            raise NullablePasswordError(f"User with id={user.id} has not got password!")
-        if user.password.get_secret_value() == password.get_secret_value():
-            return user
+        with db.create_session() as session:
+            user = self._system_user_storage.get_user_by_credits(session=session, login=username)
+            if user is None:
+                user = self._system_user_storage.create_user(session=session, login=username, password=password)
+            if user.password is None:
+                raise NullablePasswordError(f"User with id={user.id} has not got password!")
+            if user.password == password.get_secret_value():
+                return SystemUserModel.from_orm(user)
         return None

overhave/metrics/client/container.py

@@ -36,6 +36,8 @@ def consume_redis_task(self, task_type: str) -> None:
 class TestRunOverhaveMetricContainer(BaseOverhaveMetricContainer):
     """Overhave prometheus metric container for test runs."""
 
+    __test__ = False
+
     def __init__(self, registry: CollectorRegistry):
         super().__init__(registry=registry)
         self._init_test_run_metrics()

overhave/scenario/compiler/compiler.py

@@ -1,10 +1,12 @@
+from typing import Sequence
+
 import allure
 from pytest_bdd import types as default_types
 
 from overhave.entities import OverhaveLanguageSettings
 from overhave.scenario.compiler.settings import OverhaveScenarioCompilerSettings
 from overhave.scenario.prefix_mixin import PrefixMixin
-from overhave.storage import TestExecutorContext
+from overhave.storage import TagModel, TestExecutorContext
 
 
 def generate_task_info(tasks: list[str], header: str | None) -> str:
@@ -13,10 +15,8 @@ def generate_task_info(tasks: list[str], header: str | None) -> str:
     return ""
 
 
-def generate_tags_list(context: TestExecutorContext) -> list[str] | None:
-    if feature_tags := [i.value for i in context.feature.feature_tags]:
-        return feature_tags
-    return None
+def _join_all_tags(lines: Sequence[str]) -> str:
+    return " ".join(lines).replace("  ", " ")
 
 
 class ScenarioCompilerError(Exception):
@@ -45,13 +45,9 @@ def _get_feature_type_tag(self, scenario_text: str, tag: str) -> str:
             return ""
         return f"{self._compilation_settings.tag_prefix}{tag}"
 
-    def _get_additional_tags(self, scenario_text: str, tags: list[str] | None) -> str:
-        if f"{self._compilation_settings.tag_prefix}{tags}" in scenario_text:
-            return ""
-        if tags is not None:
-            tags_with_prefix = (f"{self._compilation_settings.tag_prefix}{tag}" for tag in tags)
-            return f"{' '.join(tags_with_prefix)}"
-        return ""
+    def _get_additional_tags(self, scenario_text: str, tags: list[TagModel]) -> str:
+        tags_with_prefix = (f"{self._compilation_settings.tag_prefix}{tag.value}" for tag in tags)
+        return f"{' '.join(tag for tag in tags_with_prefix if tag not in scenario_text)}"
 
     def _get_severity_tag(self, severity: allure.severity_level) -> str:
         return f"{self._compilation_settings.severity_prefix}{severity.value}"
@@ -91,11 +87,16 @@ def _compile_header(self, context: TestExecutorContext) -> str:
         blocks_delimiter = f" {self._compilation_settings.blocks_delimiter} "
         if context.test_run.start is None:
             raise RuntimeError
+        joined_tags = _join_all_tags(
+            (
+                self._get_feature_type_tag(scenario_text=text, tag=context.feature.feature_type.name),
+                self._get_additional_tags(scenario_text=text, tags=context.feature.feature_tags),
+                self._get_severity_tag(severity=context.feature.severity),
+            )
+        )
         return "\n".join(
             (
-                f"{self._get_feature_type_tag(scenario_text=text, tag=context.feature.feature_type.name)} "
-                f"{self._get_additional_tags(scenario_text=text, tags=generate_tags_list(context))} "
-                f"{self._get_severity_tag(severity=context.feature.severity)}",
+                joined_tags,
                 f"{self._as_prefix(feature_prefix)} {context.feature.name}",
                 f"{self._compilation_settings.id_prefix} {context.feature.id}",
                 (

overhave/storage/system_user_group_storage.py

@@ -1,5 +1,7 @@
 import abc
 
+import sqlalchemy.orm as so
+
 from overhave import db
 
 
@@ -8,15 +10,13 @@ class ISystemUserGroupStorage(abc.ABC):
 
     @staticmethod
     @abc.abstractmethod
-    def has_any_group(user_groups: list[str]) -> bool:
+    def has_any_group(session: so.Session, user_groups: list[str]) -> bool:
         pass
 
 
 class SystemUserGroupStorage(ISystemUserGroupStorage):
     """Class for system user storage."""
 
     @staticmethod
-    def has_any_group(user_groups: list[str]) -> bool:
-        with db.create_session() as session:
-            group = session.query(db.GroupRole).filter(db.GroupRole.group.in_(user_groups)).first()
-            return group is not None
+    def has_any_group(session: so.Session, user_groups: list[str]) -> bool:
+        return session.query(db.GroupRole).filter(db.GroupRole.group.in_(user_groups)).first() is not None