[Issue 3266] mod factories opp attach staging

api/tests/src/data_migration/transformation/subtask/test_transform_opportunity_attachment.py

@@ -0,0 +1,53 @@
+import os
+
+from src.db.models.foreign.opportunity import Topportunity as FTopportunity
+from src.db.models.staging.opportunity import Topportunity as STopportunity
+from tests.src.db.models.factories import ForeignTopportunityFactory, StagingTopportunityFactory
+
+
+def test_uploading_attachment_staging(db_session, enable_factory_create):
+    opp = StagingTopportunityFactory.create(my_attachment=b"Testing saving to db")
+    db_session.commit()
+    db_session.expire_all()
+
+    db_opp = (
+        db_session.query(STopportunity)
+        .filter(STopportunity.opportunity_id == opp.opportunity_id)
+        .one_or_none()
+    )
+
+    try:
+        with open("out_file.txt", "wb") as outfile:
+            outfile.write(db_opp.my_attachment)
+
+        with open("out_file.txt", "rb") as infile:
+            file_content = infile.read()
+            assert file_content == db_opp.my_attachment
+    finally:
+        # Cleanup: delete the file after verifying
+        if os.path.exists("out_file.txt"):
+            os.remove("out_file.txt")
+
+
+def test_uploading_attachment_foreign(db_session, enable_factory_create):
+    opp = ForeignTopportunityFactory.create(my_attachment=b"Testing saving to db")
+    db_session.commit()
+    db_session.expire_all()
+
+    db_opp = (
+        db_session.query(FTopportunity)
+        .filter(FTopportunity.opportunity_id == opp.opportunity_id)
+        .one_or_none()
+    )
+
+    try:
+        with open("out_file.txt", "wb") as outfile:
+            outfile.write(db_opp.my_attachment)
+
+        with open("out_file.txt", "rb") as infile:
+            file_content = infile.read()
+            assert file_content == db_opp.my_attachment
+    finally:
+        # Cleanup: delete the file after verifying
+        if os.path.exists("out_file.txt"):
+            os.remove("out_file.txt")

api/tests/src/db/models/factories.py

@@ -1155,6 +1155,14 @@ class Params:
             category_explanation=None,
         )
 
+    # Set non-model fields after creation
+    @factory.post_generation
+    def my_attachment(self, create, extracted):
+        if extracted:
+            self.my_attachment = extracted
+        else:
+            self.my_attachment = b"Test attachment"
+
 
 class StagingTopportunityCfdaFactory(TopportunityCfdaFactory, AbstractStagingFactory):
     class Meta:
@@ -1363,6 +1371,40 @@ class Meta:
     creator_id = factory.Faker("first_name")
 
 
+class StagingTopportunityAttachmentFactory(TopportunityFactory, AbstractStagingFactory):
+    class Meta:
+        model = staging.attachment.TsynopsisAttachment
+
+    class Params:
+        # Trait to set all nullable fields to None
+        all_fields_null = factory.Trait(
+            att_revision_number=None,
+            att_type=None,
+            file_lob=None,
+            creator_id=None,
+            last_upd_id=None,
+            syn_att_folder_id=None,
+        )
+
+    syn_att_id: factory.Sequence(lambda n: n)
+    opportunity_id: int = factory.LazyAttribute(lambda s: s.synopsis.opportunity_id)
+    mime_type = factory.Faker("mime_type")
+    link_url = factory.Faker("link_url")
+    file_name = factory.Faker("file_name")
+    file_desc = factory.Faker("sentence")
+    file_lob_size = factory.Faker("random_int", min=1000, max=10000000)
+
+    create_date = factory.Faker("date_time_between", start_date="-1y", end_date="now")
+    created_date = factory.LazyAttribute(
+        lambda o: fake.date_time_between(start_date=o.created_at, end_date="now")
+    )
+    last_upd_date = factory.LazyAttribute(
+        lambda o: fake.date_time_between(start_date=o.created_at, end_date="now")
+    )
+    creator_id = factory.Faker("creator_id")
+    last_upd_id = factory.Faker("last_upd_id")
+
+
 ####################################
 # Transfer Table Factories
 ####################################
@@ -1427,6 +1469,12 @@ def _setup_next_sequence(cls):
         size=lambda: random.randint(1, 3),
     )
 
+    # Set non-model fields after creation
+    @factory.post_generation
+    def my_attachment(self, create, extracted):
+        if extracted:
+            self.my_attachment = extracted
+
 
 class ForeignTopportunityCfdaFactory(TopportunityCfdaFactory):
     class Meta:
@@ -1593,6 +1641,40 @@ class Meta:
     revision_number = factory.LazyAttribute(lambda s: s.synopsis.revision_number)
 
 
+class ForeignTopportunityAttachmentFactory(TopportunityFactory, AbstractStagingFactory):
+    class Meta:
+        model = foreign.attachment.tsynopsisattachment
+
+    class Params:
+        # Trait to set all nullable fields to None
+        all_fields_null = factory.Trait(
+            att_revision_number=None,
+            att_type=None,
+            file_lob=None,
+            creator_id=None,
+            last_upd_id=None,
+            syn_att_folder_id=None,
+        )
+
+    syn_att_id: factory.Sequence(lambda n: n)
+    opportunity_id: int = factory.LazyAttribute(lambda s: s.synopsis.opportunity_id)
+    mime_type = factory.Faker("mime_type")
+    link_url = factory.Faker("link_url")
+    file_name = factory.Faker("file_name")
+    file_desc = factory.Faker("sentence")
+    file_lob_size = factory.Faker("random_int", min=1000, max=10000000)
+
+    create_date = factory.Faker("date_time_between", start_date="-1y", end_date="now")
+    created_date = factory.LazyAttribute(
+        lambda o: fake.date_time_between(start_date=o.created_at, end_date="now")
+    )
+    last_upd_date = factory.LazyAttribute(
+        lambda o: fake.date_time_between(start_date=o.created_at, end_date="now")
+    )
+    creator_id = factory.Faker("creator_id")
+    last_upd_id = factory.Faker("last_upd_id")
+
+
 ##
 # Pseudo-factories
 ##