Added char limit to name, address, and parent name fields.

db_revisions/versions/6dd77f09fae6_add_char_limit_to_financial_institution_.py

@@ -0,0 +1,43 @@
+"""add char limit to financial institution table fields
+
+Revision ID: 6dd77f09fae6
+Revises: 2ad66aaf4583
+Create Date: 2024-11-05 09:28:35.233356
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "6dd77f09fae6"
+down_revision: Union[str, None] = "2ad66aaf4583"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("financial_institutions", schema=None) as batch_op:
+        batch_op.alter_column("name", type_=sa.String(255), nullable=False)
+        batch_op.alter_column("hq_address_street_1", type_=sa.String(255), nullable=False)
+        batch_op.alter_column("hq_address_street_2", type_=sa.String(255), nullable=True)
+        batch_op.alter_column("hq_address_street_3", type_=sa.String(255), nullable=True)
+        batch_op.alter_column("hq_address_street_4", type_=sa.String(255), nullable=True)
+        batch_op.alter_column("hq_address_city", type_=sa.String(255), nullable=False)
+        batch_op.alter_column("parent_legal_name", type_=sa.String(255), nullable=True)
+        batch_op.alter_column("top_holder_legal_name", type_=sa.String(255), nullable=True)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("financial_institutions", schema=None) as batch_op:
+        batch_op.alter_column("name", type_=sa.String, nullable=False)
+        batch_op.alter_column("hq_address_street_1", type_=sa.String, nullable=False)
+        batch_op.alter_column("hq_address_street_2", type_=sa.String, nullable=True)
+        batch_op.alter_column("hq_address_street_3", type_=sa.String, nullable=True)
+        batch_op.alter_column("hq_address_street_4", type_=sa.String, nullable=True)
+        batch_op.alter_column("hq_address_city", type_=sa.String, nullable=False)
+        batch_op.alter_column("parent_legal_name", type_=sa.String, nullable=True)
+        batch_op.alter_column("top_holder_legal_name", type_=sa.String, nullable=True)

src/regtech_user_fi_management/entities/models/dao.py

@@ -41,7 +41,7 @@ class FinancialInstitutionDao(AuditMixin, Base):
     version: Mapped[int] = mapped_column(nullable=False, default=0)
     __mapper_args__ = {"version_id_col": version, "version_id_generator": False}
     lei: Mapped[str] = mapped_column(String(20), unique=True, index=True, primary_key=True)
-    name: Mapped[str] = mapped_column(index=True)
+    name: Mapped[str] = mapped_column(String(255), index=True)
     is_active: Mapped[bool] = mapped_column(index=True)
     domains: Mapped[List["FinancialInstitutionDomainDao"]] = relationship(
         "FinancialInstitutionDomainDao", back_populates="fi", lazy="selectin"
@@ -53,19 +53,19 @@ class FinancialInstitutionDao(AuditMixin, Base):
     hmda_institution_type_id: Mapped[str] = mapped_column(ForeignKey("hmda_institution_type.id"), nullable=True)
     hmda_institution_type: Mapped["HMDAInstitutionTypeDao"] = relationship(lazy="selectin")
     sbl_institution_types: Mapped[List[SblTypeMappingDao]] = relationship(lazy="selectin", cascade="all, delete-orphan")
-    hq_address_street_1: Mapped[str]
-    hq_address_street_2: Mapped[str] = mapped_column(nullable=True)
-    hq_address_street_3: Mapped[str] = mapped_column(nullable=True)
-    hq_address_street_4: Mapped[str] = mapped_column(nullable=True)
-    hq_address_city: Mapped[str]
+    hq_address_street_1: Mapped[str] = mapped_column(String(255))
+    hq_address_street_2: Mapped[str] = mapped_column(String(255), nullable=True)
+    hq_address_street_3: Mapped[str] = mapped_column(String(255), nullable=True)
+    hq_address_street_4: Mapped[str] = mapped_column(String(255), nullable=True)
+    hq_address_city: Mapped[str] = mapped_column(String(255))
     hq_address_state_code: Mapped[str] = mapped_column(ForeignKey("address_state.code"), nullable=True)
     hq_address_state: Mapped["AddressStateDao"] = relationship(lazy="selectin")
     hq_address_zip: Mapped[str] = mapped_column(String(5))
     parent_lei: Mapped[str] = mapped_column(String(20), nullable=True)
-    parent_legal_name: Mapped[str] = mapped_column(nullable=True)
+    parent_legal_name: Mapped[str] = mapped_column(String(255), nullable=True)
     parent_rssd_id: Mapped[int] = mapped_column(nullable=True)
     top_holder_lei: Mapped[str] = mapped_column(String(20), nullable=True)
-    top_holder_legal_name: Mapped[str] = mapped_column(nullable=True)
+    top_holder_legal_name: Mapped[str] = mapped_column(String(255), nullable=True)
     top_holder_rssd_id: Mapped[int] = mapped_column(nullable=True)
     modified_by: Mapped[str] = mapped_column()
 

src/regtech_user_fi_management/entities/models/dto.py

@@ -1,7 +1,7 @@
 from regtech_user_fi_management.config import regex_configs
 
 from typing import Generic, List, Set, Sequence
-from pydantic import BaseModel, model_validator
+from pydantic import BaseModel, model_validator, Field
 from typing import TypeVar
 
 T = TypeVar("T")
@@ -65,18 +65,18 @@ class FinancialInstitutionDto(FinancialInstitutionBase):
     primary_federal_regulator_id: str | None = None
     hmda_institution_type_id: str | None = None
     sbl_institution_types: List[SblTypeAssociationDto | str] = []
-    hq_address_street_1: str
-    hq_address_street_2: str | None = None
-    hq_address_street_3: str | None = None
-    hq_address_street_4: str | None = None
-    hq_address_city: str
+    hq_address_street_1: str = Field(max_length=255)
+    hq_address_street_2: str | None = Field(None, max_length=255)
+    hq_address_street_3: str | None = Field(None, max_length=255)
+    hq_address_street_4: str | None = Field(None, max_length=255)
+    hq_address_city: str = Field(None, max_length=255)
     hq_address_state_code: str | None = None
     hq_address_zip: str
     parent_lei: str | None = None
-    parent_legal_name: str | None = None
+    parent_legal_name: str | None = Field(None, max_length=255)
     parent_rssd_id: int | None = None
     top_holder_lei: str | None = None
-    top_holder_legal_name: str | None = None
+    top_holder_legal_name: str | None = Field(None, max_length=255)
     top_holder_rssd_id: int | None = None
     version: int | None = None
 

tests/entities/repos/test_institutions_repo.py

@@ -1,5 +1,6 @@
 import pytest
 from pytest_mock import MockerFixture
+from pydantic import ValidationError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from regtech_user_fi_management.entities.models.dto import (
@@ -410,3 +411,79 @@ async def test_update_sbl_institution_types_inst_non_exist(
         res = await repo.update_sbl_types(transaction_session, self.auth_user, test_lei, sbl_types)
         commit_spy.assert_not_called()
         assert res is None
+
+    async def test_add_institution_invalid_field_length(self, transaction_session: AsyncSession):
+
+        out_of_range_text = (
+            "Lorem ipsum dolor sit amet, consectetuer adipiscing elit. Aenean commodo ligula eget "
+            "dolor. Aenean massa. Cum sociis natoque penatibus et magnis dis parturient montes, "
+            "nascetur ridiculus mus. Donec quam felis, ultricies nec, pellentesque eu, pretium quis..."
+        )
+        with pytest.raises(Exception) as e:
+            await repo.upsert_institution(
+                transaction_session,
+                FinancialInstitutionDto(
+                    name="New Bank 123",
+                    lei="NEWBANK1230000000000",
+                    is_active=True,
+                    tax_id="65-4321987",
+                    rssd_id=6543,
+                    primary_federal_regulator_id="FRI3",
+                    hmda_institution_type_id="HIT3",
+                    sbl_institution_types=[SblTypeAssociationDto(id="1")],
+                    hq_address_street_1=out_of_range_text,
+                    hq_address_street_2="",
+                    hq_address_street_3="",
+                    hq_address_street_4="",
+                    hq_address_city="Test City 3",
+                    hq_address_state_code="FL",
+                    hq_address_zip="22222",
+                    parent_lei="0123PARENTNEWBANK123",
+                    parent_legal_name="PARENT NEW BANK 123",
+                    parent_rssd_id=76543,
+                    top_holder_lei="TOPHOLDNEWBANKLEI123",
+                    top_holder_legal_name="TOP HOLDER NEW BANK LEI 123",
+                    top_holder_rssd_id=876543,
+                    modified_by="test_user_id",
+                ),
+                self.auth_user,
+            )
+        assert isinstance(e.value, ValidationError)
+
+    async def test_update_institution_invalid_field_length(self, transaction_session: AsyncSession):
+        out_of_range_text = (
+            "Lorem ipsum dolor sit amet, consectetuer adipiscing elit. Aenean commodo ligula eget "
+            "dolor. Aenean massa. Cum sociis natoque penatibus et magnis dis parturient montes, "
+            "nascetur ridiculus mus. Donec quam felis, ultricies nec, pellentesque eu, pretium quis..."
+        )
+
+        with pytest.raises(Exception) as e:
+            await repo.upsert_institution(
+                transaction_session,
+                FinancialInstitutionDto(
+                    name="New Bank 123",
+                    lei="NEWBANK1230000000000",
+                    is_active=True,
+                    tax_id="65-4321987",
+                    rssd_id=6543,
+                    primary_federal_regulator_id="FRI3",
+                    hmda_institution_type_id="HIT3",
+                    sbl_institution_types=[SblTypeAssociationDto(id="1")],
+                    hq_address_street_1=out_of_range_text,
+                    hq_address_street_2="",
+                    hq_address_street_3="",
+                    hq_address_street_4="",
+                    hq_address_city="Test City 3",
+                    hq_address_state_code="FL",
+                    hq_address_zip="22222",
+                    parent_lei="0123PARENTNEWBANK123",
+                    parent_legal_name="PARENT NEW BANK 123",
+                    parent_rssd_id=76543,
+                    top_holder_lei="TOPHOLDNEWBANKLEI123",
+                    top_holder_legal_name=out_of_range_text,
+                    top_holder_rssd_id=876543,
+                    modified_by="test_user_id",
+                ),
+                self.auth_user,
+            )
+        assert isinstance(e.value, ValidationError)

tests/migrations/test_migrations.py

@@ -67,3 +67,21 @@ def test_migration_to_2ad66aaf4583(alembic_runner: MigrationContext, alembic_eng
 
     assert "hq_address_street_3" in [c["name"] for c in inspector.get_columns("financial_institutions")]
     assert "hq_address_street_4" in [c["name"] for c in inspector.get_columns("financial_institutions")]
+
+
+def test_migration_to_6dd77f09fae6(alembic_runner: MigrationContext, alembic_engine: Engine):
+
+    alembic_runner.migrate_up_to("6dd77f09fae6")
+
+    inspector = sqlalchemy.inspect(alembic_engine)
+
+    columns = inspector.get_columns("financial_institutions")
+
+    assert [c for c in columns if c["name"] == "name"][0]["type"].length == 255
+    assert [c for c in columns if c["name"] == "hq_address_street_1"][0]["type"].length == 255
+    assert [c for c in columns if c["name"] == "hq_address_street_2"][0]["type"].length == 255
+    assert [c for c in columns if c["name"] == "hq_address_street_3"][0]["type"].length == 255
+    assert [c for c in columns if c["name"] == "hq_address_street_4"][0]["type"].length == 255
+    assert [c for c in columns if c["name"] == "hq_address_city"][0]["type"].length == 255
+    assert [c for c in columns if c["name"] == "parent_legal_name"][0]["type"].length == 255
+    assert [c for c in columns if c["name"] == "top_holder_legal_name"][0]["type"].length == 255