test

integration-tests/test_redis_to_relational.py

@@ -23,12 +23,17 @@
 ])
 def test_redis_to_relational_db(db_type: str, schema_name: Optional[str]):
     """Reads data from a Redis stream and writes it to a relational database."""
+    job_name = f"tests.redis_to_{db_type}"
+
     try:
         redis_container = redis_utils.get_redis_oss_container(redis_utils.REDIS_PORT)
         redis_container.start()
 
         redis_utils.add_to_emp_stream(redis_utils.get_redis_client("localhost", redis_utils.REDIS_PORT))
 
+        with pytest.raises(ValueError):
+            run_job(job_name)
+
         if db_type == "db2":
             db_container = db_utils.get_db2_container("hr", "my_user", "my_pass")
         elif db_type == "mysql":
@@ -52,7 +57,7 @@ def test_redis_to_relational_db(db_type: str, schema_name: Optional[str]):
         db_utils.insert_to_emp_table(engine, schema_name)
         db_utils.insert_to_address_table(engine, schema_name)
 
-        run_job(f"tests.redis_to_{db_type}")
+        run_job(job_name)
         check_results(engine, schema_name)
     finally:
         with suppress(Exception):