Merge pull request #7 from hexfrost/beta-0.1.6

Beta 0.1.6

Author: kaziamov

pyproject.toml

@@ -22,7 +22,7 @@ dependencies = [
     "sqlalchemy-utils>=0.41.2",
 ]
 name = "hexfrost-toolbox"
-version = "0.1.6"
+version = "0.1.6b"
 description = "Open source library with useful utils for fast development"
 readme = "README.md"
 

tests/fixtures/database.py

@@ -3,28 +3,22 @@
 from toolbox.sqlalchemy.connection import DatabaseConnectionSettings
 
 
-@pytest.fixture(autouse=True)
+@pytest.fixture(autouse=True, scope="session")
 def db_settings():
-    from pydantic import BaseModel
-    data = dict(
+    data = DatabaseConnectionSettings(
         POSTGRES_USER="postgres",
         POSTGRES_PASSWORD = "postgres",
         POSTGRES_HOST = "0.0.0.0",
         POSTGRES_PORT = "5432",
         POSTGRES_DB = "postgres"
     )
-    class TestSettings(BaseModel, DatabaseConnectionSettings):
-        pass
-
-    return TestSettings(**data)
+    return data
 
 
-@pytest.fixture(autouse=True)
+@pytest.fixture(autouse=True, scope="session")
 async def temp_db(db_settings):
     from toolbox.testing import temporary_database
-    from toolbox.sqlalchemy.models import BaseModel
-    async with temporary_database(settings=db_settings, base_model=BaseModel):
+    from toolbox.sqlalchemy.models import BaseDatabaseModel
+    async with temporary_database(settings=db_settings, base_model=BaseDatabaseModel):
         yield
         pass
-
-

tests/sqlalchemy/test_repositories.py

@@ -0,0 +1,98 @@
+import pytest
+from sqlalchemy import Column, Integer, String, orm, LargeBinary
+
+from toolbox.schemes import SensitiveDataScheme, SensitiveFieldData
+from toolbox.sqlalchemy.connection import DatabaseConnectionManager, DatabaseConnectionSettings
+from toolbox.sqlalchemy.repositories import AbstractDatabaseCrudManager
+
+
+class BaseDatabaseModel(orm.DeclarativeBase):
+    pass
+
+
+class TestSQLAlchemyModel(BaseDatabaseModel):
+    __tablename__ = "test_items"
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String, index=True)
+    sensitive_field = Column(LargeBinary)
+
+
+class TestPydanticModel(SensitiveDataScheme):
+    _sensitive_attributes = ["sensitive_field"]
+
+    name: str
+    sensitive_field: SensitiveFieldData
+
+
+class TestItemRepository(AbstractDatabaseCrudManager):
+    _alchemy_model = TestSQLAlchemyModel
+    _pydantic_model = TestPydanticModel
+
+
+@pytest.fixture(autouse=True, scope="session")
+def db_settings():
+    data = DatabaseConnectionSettings(
+        POSTGRES_USER="postgres",
+        POSTGRES_PASSWORD="postgres",
+        POSTGRES_HOST="0.0.0.0",
+        POSTGRES_PORT="5432",
+        POSTGRES_DB="postgres"
+    )
+    return data
+
+
+@pytest.fixture(autouse=True, scope="session")
+async def temp_db(db_settings):
+    from toolbox.testing import temporary_database
+    from toolbox.sqlalchemy.models import BaseDatabaseModel
+    async with temporary_database(settings=db_settings, base_model=BaseDatabaseModel) as db_connection:
+        yield db_connection
+        pass
+
+
+@pytest.fixture(autouse=True, scope="session")
+def database_connection(db_settings):
+    dc = DatabaseConnectionManager(settings=db_settings)
+    return dc
+
+class MockCipherManager:
+    def encrypt(self, value):
+        return f"{value}_secret".encode()
+
+    def decrypt(self, value):
+        return value.decode()[:-7]
+
+
+async def test_add_one_sensitive_field(temp_db, database_connection):
+    new_obj = TestPydanticModel(
+        name="test_item",
+        sensitive_field="test_sensitive_field"
+    )
+
+    TestPydanticModel.set_cipher_suite_manager(MockCipherManager())
+    async with database_connection.get_db_session() as conn:
+        from sqlalchemy.schema import CreateTable
+        try:
+            await conn.execute(CreateTable(TestSQLAlchemyModel.__table__))
+        except:
+            pass
+
+        all_ = await TestItemRepository.get_all(conn)
+        assert len(all_) == 0
+
+        await TestItemRepository.add_one(conn, new_obj)
+
+        import asyncpg
+        another_conn = await asyncpg.connect(dsn=database_connection._get_settings().get_dsn())
+        raw_results = await another_conn.fetch(f"SELECT * FROM {TestSQLAlchemyModel.__table__}")
+        await another_conn.close()
+
+        assert len(raw_results) == 1
+        result = dict(zip(raw_results[0].keys(), raw_results[0].values()))
+
+        assert result["sensitive_field"] == "test_sensitive_field_secret".encode()
+
+        all_ = await TestItemRepository.get_all(conn)
+        assert len(all_) == 1
+        obj = all_[0]
+        assert obj.sensitive_field == "test_sensitive_field"

tests/test_alchemy.py

@@ -1,30 +1,31 @@
-import pytest
 from unittest.mock import MagicMock, AsyncMock, patch
-from sqlalchemy import select, String
+
+import pytest
 from sqlalchemy.orm import mapped_column, Mapped
 from sqlalchemy.sql.selectable import Select
 
+from sqlalchemy import select, String
+from toolbox.schemes import SensitiveDataScheme
+from toolbox.sqlalchemy.models import BaseDatabaseModel
 from toolbox.sqlalchemy.repositories import AbstractDatabaseCrudManager
-from toolbox.sqlalchemy.models import BaseModel
-from toolbox.schemes import BaseScheme, SensitiveDataScheme
 
 
-class TestPydanticModel(BaseScheme):
+class TestPydanticModel(SensitiveDataScheme):
     name: str
     value: str
 
     def encrypt_fields(self):
         return self
 
 
-class TestSQLAlchemyModel(BaseModel):
+class TestSQLAlchemyDatabaseModel(BaseDatabaseModel):
     __tablename__ = "test_table"
     name: Mapped[str] = mapped_column(String, primary_key=True)
     value: Mapped[str] = mapped_column(String)
 
 
 class TestCrudManager(AbstractDatabaseCrudManager):
-    _alchemy_model = TestSQLAlchemyModel
+    _alchemy_model = TestSQLAlchemyDatabaseModel
     _pydantic_model = TestPydanticModel
 
 
@@ -57,7 +58,7 @@ async def test_add_one(mock_db_manager, mock_session, test_data):
 
     session.add.assert_called_once()
     added_model = session.add.call_args[0][0]
-    assert isinstance(added_model, TestSQLAlchemyModel)
+    assert isinstance(added_model, TestSQLAlchemyDatabaseModel)
     assert added_model.name == test_data.name
     assert added_model.value == test_data.value
 
@@ -71,8 +72,8 @@ async def test_get_all(mock_db_manager, mock_session):
     session_maker, session = mock_session
 
     test_models = [
-        TestSQLAlchemyModel(name="test1", value="value1"),
-        TestSQLAlchemyModel(name="test2", value="value2")
+        TestSQLAlchemyDatabaseModel(name="test1", value="value1"),
+        TestSQLAlchemyDatabaseModel(name="test2", value="value2")
     ]
 
     db_operation_result_mock = MagicMock()
@@ -87,8 +88,7 @@ async def test_get_all(mock_db_manager, mock_session):
     assert isinstance(query, Select)
 
     assert len(result) == 2
-    assert all(isinstance(item, TestSQLAlchemyModel) for item in result)
-    assert result == test_models
+    assert all(isinstance(item, TestPydanticModel) for item in result)
 
 
 @pytest.mark.asyncio

tests/test_database.py

@@ -1,9 +1,8 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
-from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
 
+from sqlalchemy import select
 from tests.fixtures.database import db_settings, temp_db
 from tests.fixtures.db_connect import database_connector