自動テスト実装 Ｖメーカー卸組織展開表 ※PK項目がないテーブルはINSERTを期待値とする

ecs/jskult-batch-daily/tests/batch/vjsk/vjsk_load/test_vjsk_load.py

@@ -1,3 +1,4 @@
+import time
 from os import path
 
 import pytest
@@ -597,7 +598,12 @@ class TestImportFileToDb:
         # srcテーブル結果が期待値通りかを突合
         assert_table_results(result_src, assert_list, ignore_columns)
 
-        # assertion2 (update 2row +insert 2row)
+        # assertion2 (update 0row +insert 4row) ※PK項目がないテーブルなのですべてinsertになる
+
+        assetion1_done_dt = self.db.execute_select("select SYSDATE()")[0]["SYSDATE()"]
+
+        # assertion2でinsertされたレコードをdwh_upd_dtで判断するため、assertion1からの実行間隔を明確に空けるためにスリープを挟む
+        time.sleep(3)
 
         # 処理実行
         target_dict = {
@@ -616,14 +622,14 @@ class TestImportFileToDb:
         assert_table_results(result_org, assert_list, ignore_columns)
         # srcテーブル結果(orgテーブル結果のPK値で一致するもの)を取得
         result_src = self.db.execute_select(
-            f"select * from {table_name_src} s inner join {table_name_org} o on (s.v_cd_1 = o.v_cd_1 and s.v_cd_2 = o.v_cd_2)")
+            f"select * from {table_name_src} where dwh_upd_dt > :dt_value", {"dt_value": assetion1_done_dt})
         # 突合から除外する項目
         ignore_columns = ['dwh_upd_dt']
         # srcテーブル結果が期待値通りかを突合
         assert_table_results(result_src, assert_list, ignore_columns)
 
-        # srcテーブル結果のレコード件数 (insert 4row + update 2row + insert 2row = 6row)
+        # srcテーブル結果のレコード件数 (insert 4row + update 0row + insert 4row = 8row) ※PK項目がないテーブルなのですべてinsertになる
         result_src_count = self.db.execute_select(f"select count(*) from {table_name_src} ")
-        assert result_src_count[0]['count(*)'] == 6
+        assert result_src_count[0]['count(*)'] == 8
 
         # teardown