Configure Sentry

Added sentry as a dependency.

dags/common/pull_ftp.py

@@ -135,6 +135,8 @@ def _differential_pull(
     sftp_files = s_ftp.list_files(excluded_directories=excluded_directories)
     s3_files = repo.get_all_raw_filenames()
     diff_files = list(filter(lambda x: x not in s3_files, sftp_files))
+    logger.msg("Differential")
+    logger.msg(diff_files)
     return migrate_files(diff_files, s_ftp, repo, logger)
 
 

requirements-airflow.txt

@@ -1,3 +1,3 @@
 -c https://raw.githubusercontent.com/apache/airflow/constraints-2.8.3/constraints-3.10.txt
-apache-airflow[celery, postgres, redis, cncf.kubernetes]==2.8.3
+apache-airflow[celery, postgres, redis, cncf.kubernetes, sentry]==2.8.3
 

tests/integration/iop/test_iop_dag_pull_sftp.py

@@ -22,7 +22,7 @@ def dag_was_paused(dag):
 
 
 @pytest.fixture
-def iop_empty_repo():
+def iop_gen_empty_repo():
     repo = IOPRepository()
     repo.delete_all()
     yield repo
@@ -33,8 +33,8 @@ def test_dag_loaded(dag):
     assert len(dag.tasks) == 2
 
 
-def test_dag_run(dag, dag_was_paused: bool, iop_empty_repo):
-    assert len(iop_empty_repo.find_all()) == 0
+def test_dag_run(dag, dag_was_paused: bool, iop_gen_empty_repo):
+    assert len(iop_gen_empty_repo.find_all()) == 0
     dag.clear()
     dag.test()
     expected_files = [
@@ -85,17 +85,17 @@ def test_dag_run(dag, dag_was_paused: bool, iop_empty_repo):
         {"xml": "extracted/aca95c/aca95c.xml"},
     ]
 
-    assert sorted(iop_empty_repo.find_all(), key=lambda x: x.get("xml", "")) == sorted(
+    assert sorted(iop_gen_empty_repo.find_all(), key=lambda x: x.get("xml", "")) == sorted(
         expected_files, key=lambda x: x.get("xml", "")
     )
 
 
-def test_dag_migrate_from_FTP(iop_empty_repo):
-    assert len(iop_empty_repo.find_all()) == 0
+def test_dag_migrate_from_FTP(iop_gen_empty_repo):
+    assert len(iop_gen_empty_repo.find_all()) == 0
     with IOPSFTPService() as sftp:
         migrate_from_ftp(
             sftp,
-            iop_empty_repo,
+            iop_gen_empty_repo,
             get_logger().bind(class_name="test_logge"),
             **{
                 "params": {
@@ -157,10 +157,16 @@ def test_dag_migrate_from_FTP(iop_empty_repo):
             },
             {"xml": "extracted/aca95c/aca95c.xml"},
         ]
-        for (file_from_repo, expected_file) in zip(
-            iop_empty_repo.find_all(), expected_files
-        ):
-            assert file_from_repo == expected_file
+
+        assert len(iop_gen_empty_repo.find_all()) == len(expected_files)
+
+        iop_pdf_files = sorted(item["pdf"] for item in iop_gen_empty_repo.find_all() if "pdf" in item)
+        expected_pdf_files = sorted(item["pdf"] for item in expected_files if "pdf" in item)
+        assert iop_pdf_files == expected_pdf_files
+
+        iop_xml_files = sorted(item["xml"] for item in iop_gen_empty_repo.find_all() if "xml" in item)
+        expected_xml_files = sorted(item["xml"] for item in expected_files if "xml" in item)
+        assert iop_xml_files == expected_xml_files
 
 
 def test_dag_trigger_file_processing():

tests/integration/iop/test_repo.py

@@ -3,6 +3,7 @@
 from iop.sftp_service import IOPSFTPService
 from pytest import fixture
 from structlog import get_logger
+import time
 
 
 @fixture
@@ -30,6 +31,9 @@ def test_pull_from_sftp(iop_empty_repo):
                 }
             }
         )
+
+        time.sleep(5)
+
         expected_files = [
             {
                 "pdf": "extracted/2022-07-30T03_02_01_content/1674-1137/1674-1137_46/1674-1137_46_8/1674-1137_46_8_085001/cpc_46_8_085001.pdf",
@@ -77,7 +81,17 @@ def test_pull_from_sftp(iop_empty_repo):
             {"xml": "extracted/aca95c/aca95c.xml"},
 
         ]
-        assert iop_empty_repo.find_all() == expected_files
+
+        assert len(iop_empty_repo.find_all()) == len(expected_files)
+
+        iop_pdf_files = sorted(item["pdf"] for item in iop_empty_repo.find_all() if "pdf" in item)
+        expected_pdf_files = sorted(item["pdf"] for item in expected_files if "pdf" in item)
+        assert iop_pdf_files == expected_pdf_files
+
+        iop_xml_files = sorted(item["xml"] for item in iop_empty_repo.find_all() if "xml" in item)
+        expected_xml_files = sorted(item["xml"] for item in expected_files if "xml" in item)
+        assert iop_xml_files == expected_xml_files
+
         assert sorted(iop_empty_repo.get_all_raw_filenames()) == sorted(
             [
                 "2022-07-30T03_02_01_content.zip",