Merge pull request #38 from livestreamx/update-dependencies

update deps + some fixes?

makefile

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

overhave/db/base.py

@@ -94,7 +94,7 @@ def _get_query_cls(
         except AttributeError:
             pass
 
-    return so.Query(mapper, session)  # type: ignore[arg-type]
+    return so.Query(mapper, session)
 
 
 Session = so.sessionmaker(query_cls=_get_query_cls)

overhave/storage/emulation_storage.py

@@ -1,7 +1,7 @@
 import abc
 import logging
 import socket
-from typing import Any, List, cast
+from typing import List, cast
 
 import orjson
 import sqlalchemy as sa
@@ -56,7 +56,7 @@ def has_running_emulation_with_user(self, test_user_id: int) -> bool:
 class EmulationStorage(IEmulationStorage):
     """Class for emulation runs storage."""
 
-    def __init__(self, settings: OverhaveEmulationSettings, redis: "Redis[Any]"):
+    def __init__(self, settings: OverhaveEmulationSettings, redis: Redis):
         self._redis = redis
         self._settings = settings
         self._redis.set(self._settings.redis_ports_key, orjson.dumps([]))

overhave/transport/redis/deps.py

@@ -1,5 +1,6 @@
 import logging
 from functools import cache
+from typing import cast
 
 from redis import Redis
 from redis.sentinel import Sentinel
@@ -9,22 +10,25 @@
 logger = logging.getLogger(__name__)
 
 
-def make_sentinel_master(settings: OverhaveRedisSentinelSettings) -> Redis:  # type: ignore
+def make_sentinel_master(settings: OverhaveRedisSentinelSettings) -> Redis:
     logger.info("Connecting to redis through sentinel %s", settings.urls)
     url_tuples = [(url.host, url.port) for url in settings.urls if url.host is not None and url.port is not None]
     sentinel = Sentinel(url_tuples, socket_timeout=settings.socket_timeout.total_seconds(), retry_on_timeout=True)
-    return sentinel.master_for(settings.master_set, password=settings.password, db=settings.db)
+    return cast(Redis, sentinel.master_for(settings.master_set, password=settings.password, db=settings.db))
 
 
-def make_regular_redis(redis_settings: OverhaveRedisSettings) -> Redis:  # type: ignore
-    return Redis.from_url(
-        redis_settings.url.human_repr(),
-        db=redis_settings.db,
-        socket_timeout=redis_settings.socket_timeout.total_seconds(),
+def make_regular_redis(redis_settings: OverhaveRedisSettings) -> Redis:
+    return cast(
+        Redis,
+        Redis.from_url(
+            redis_settings.url.human_repr(),
+            db=redis_settings.db,
+            socket_timeout=redis_settings.socket_timeout.total_seconds(),
+        ),
     )
 
 
-def make_redis(redis_settings: BaseRedisSettings) -> Redis:  # type: ignore  # noqa: E501
+def make_redis(redis_settings: BaseRedisSettings) -> Redis:
     if isinstance(redis_settings, OverhaveRedisSentinelSettings):
         return make_sentinel_master(redis_settings)
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "overhave"
-version = "5.0.2"
+version = "5.1.0"
 description = "Overhave - web-framework for BDD"
 readme = "README.rst"
 authors = [
@@ -37,7 +37,7 @@ WTForms = ">=2.2"
 python-ldap = "^3.2"
 ldap3 = "^2.6"
 wsgi_intercept = "^1.8"
-redis = "^4.2.0"
+redis = "^5.0.0"
 httptools = "^0.5.0"
 python-dateutil = "^2.8.1"
 alembic = "^1.4.3"
@@ -57,9 +57,9 @@ sqlalchemy-utils = "^0.41.1"
 sqlalchemy-utc = "^0.14.0"
 prometheus-client = ">=0.16.0"
 pydantic-settings = "^2.0.1"
-typer = "^0.9.0"
+typer = "^0.7.0"
 tenacity = "^8.2.3"
-pytz = "^2023.3.post1"
+pytz = "^2020.1"
 fastapi = ">=0.99.0"
 sqlalchemy = "^2.0.17"
 flask-admin = "^1.6.1"

tests/integration/admin/conftest.py

@@ -1,5 +1,6 @@
 import os
 import tempfile
+from copy import deepcopy
 from pathlib import Path
 from typing import Callable
 from uuid import uuid1
@@ -92,7 +93,11 @@ def test_report_with_index(test_report_without_index: Path, faker: Faker) -> Pat
 def test_app(
     clean_proxy_manager: Callable[[], IProxyManager], patched_app_admin_factory: IAdminFactory
 ) -> OverhaveAdminApp:
-    return overhave_app(factory=patched_app_admin_factory)
+    app = overhave_app(factory=patched_app_admin_factory)
+    ctx = app.app_context()
+    ctx.push()
+    app.config["SERVER_NAME"] = "localhost"
+    return app
 
 
 @pytest.fixture()
@@ -109,7 +114,11 @@ def test_client(test_app: OverhaveAdminApp) -> FlaskClient:
 
 @pytest.fixture()
 def test_authorized_user(test_client: FlaskClient, service_system_user: SystemUserModel) -> SystemUserModel:
+    validate = deepcopy(LoginForm.validate_on_submit)
+    get_user = deepcopy(LoginForm.get_user)
     LoginForm.validate_on_submit = lambda self: True
     LoginForm.get_user = lambda self: AdminPanelUser(user_data=service_system_user)
     test_client.post("/login")
-    return service_system_user
+    yield service_system_user
+    LoginForm.validate_on_submit = validate
+    LoginForm.get_user = get_user

tests/integration/admin/views/test_login_view.py

@@ -4,6 +4,7 @@
 from flask.testing import FlaskClient
 
 from overhave import OverhaveAdminApp
+from tests.db_utils import create_test_session
 
 
 @pytest.fixture(scope="module")
@@ -23,7 +24,9 @@ def test_show_flash_with_chat_for_unregistered_user(
         mock_support_chat_url: None,
     ) -> None:
         test_app.config["WTF_CSRF_ENABLED"] = False
-        response = test_client.post("/login", data={"username": "kek", "password": "12345"}, follow_redirects=True)
+        with create_test_session():
+            response = test_client.post("/login", data={"username": "kek", "password": "12345"}, follow_redirects=True)
+
         assert (
             "Username 'kek' is not registered! Please contact the <a href='https://localhost'>support channel</a>!"
             in response.data.decode("utf-8")
@@ -35,7 +38,8 @@ def test_show_flash_without_chat_for_unregistered_user(
         test_client: FlaskClient,
     ) -> None:
         test_app.config["WTF_CSRF_ENABLED"] = False
-        response = test_client.post("/login", data={"username": "kek", "password": "12345"}, follow_redirects=True)
+        with create_test_session():
+            response = test_client.post("/login", data={"username": "kek", "password": "12345"}, follow_redirects=True)
 
         assert "Username 'kek' is not registered!" in response.data.decode(
             "utf-8"