Create base Join tests

Author: gwaramadze

tests/test_quixstreams/test_dataframe/test_joins/test_base.py

@@ -0,0 +1,39 @@
+import pytest
+
+from quixstreams.dataframe.joins.base import Join
+from quixstreams.models.topics.exceptions import TopicPartitionsMismatch
+
+
+def test_how_invalid_value():
+    with pytest.raises(ValueError, match='Invalid "how" value'):
+        Join(how="invalid", on_merge="raise", grace_ms=1)
+
+
+def test_on_merge_invalid_value():
+    with pytest.raises(ValueError, match='Invalid "on_merge"'):
+        Join(how="inner", on_merge="invalid", grace_ms=1)
+
+
+def test_mismatching_partitions_fails(topic_manager_topic_factory, create_sdf):
+    left_topic = topic_manager_topic_factory()
+    right_topic = topic_manager_topic_factory(partitions=2)
+    left_sdf, right_sdf = create_sdf(left_topic), create_sdf(right_topic)
+    join = Join(how="inner", on_merge="raise", grace_ms=1)
+
+    with pytest.raises(TopicPartitionsMismatch):
+        join.join(left_sdf, right_sdf)
+
+
+def test_self_join_not_supported(topic_manager_topic_factory, create_sdf):
+    match = "Joining dataframes originating from the same topic is not yet supported."
+    join = Join(how="inner", on_merge="raise", grace_ms=1)
+
+    # The very same sdf object
+    sdf = create_sdf(topic_manager_topic_factory())
+    with pytest.raises(ValueError, match=match):
+        join.join(sdf, sdf)
+
+    # Same topic, different branch
+    sdf2 = sdf.apply(lambda _: _)
+    with pytest.raises(ValueError, match=match):
+        join.join(sdf, sdf2)

tests/test_quixstreams/test_dataframe/test_joins/test_join_asof.py

@@ -2,7 +2,6 @@
 
 import pytest
 
-from quixstreams.models.topics.exceptions import TopicPartitionsMismatch
 from quixstreams.state.exceptions import StoreAlreadyRegisteredError
 
 
@@ -71,25 +70,6 @@ def test_how(
         )
         assert joined_value == expected
 
-    def test_how_invalid_value(self, topic_manager_topic_factory, create_sdf):
-        left_topic = topic_manager_topic_factory()
-        right_topic = topic_manager_topic_factory()
-        left_sdf, right_sdf = create_sdf(left_topic), create_sdf(right_topic)
-
-        match = 'Invalid "how" value'
-        with pytest.raises(ValueError, match=match):
-            left_sdf.join_asof(right_sdf, how="invalid")
-
-    def test_mismatching_partitions_fails(
-        self, topic_manager_topic_factory, create_sdf
-    ):
-        left_topic = topic_manager_topic_factory()
-        right_topic = topic_manager_topic_factory(partitions=2)
-        left_sdf, right_sdf = create_sdf(left_topic), create_sdf(right_topic)
-
-        with pytest.raises(TopicPartitionsMismatch):
-            left_sdf.join_asof(right_sdf)
-
     @pytest.mark.parametrize(
         "on_merge, right, left, expected",
         [
@@ -165,15 +145,6 @@ def test_on_merge(
             )
             assert joined_value == [(expected, b"key", 2, None)]
 
-    def test_on_merge_invalid_value(self, topic_manager_topic_factory, create_sdf):
-        left_topic = topic_manager_topic_factory()
-        right_topic = topic_manager_topic_factory()
-        left_sdf, right_sdf = create_sdf(left_topic), create_sdf(right_topic)
-
-        match = 'Invalid "on_merge"'
-        with pytest.raises(ValueError, match=match):
-            left_sdf.join_asof(right_sdf, on_merge="invalid")
-
     def test_on_merge_callback(
         self, topic_manager_topic_factory, create_sdf, assign_partition, publish
     ):
@@ -223,22 +194,6 @@ def test_grace_ms(
         assert publish_left(timestamp=4) == []
         assert publish_left(timestamp=5) == [({"left": 4, "right": 2}, b"key", 5, None)]
 
-    def test_self_join_not_supported(self, topic_manager_topic_factory, create_sdf):
-        topic = topic_manager_topic_factory()
-        match = (
-            "Joining dataframes originating from the same topic is not yet supported."
-        )
-
-        # The very same sdf object
-        sdf = create_sdf(topic)
-        with pytest.raises(ValueError, match=match):
-            sdf.join_asof(sdf)
-
-        # Same topic, different branch
-        sdf2 = sdf.apply(lambda v: v)
-        with pytest.raises(ValueError, match=match):
-            sdf.join_asof(sdf2)
-
     def test_join_same_topic_multiple_times_fails(
         self, topic_manager_topic_factory, create_sdf
     ):

tests/test_quixstreams/test_dataframe/test_joins/test_join_interval.py

@@ -5,7 +5,6 @@
 import pytest
 
 from quixstreams.dataframe.joins.base import OnOverlap
-from quixstreams.models.topics.exceptions import TopicPartitionsMismatch
 
 
 @dataclass
@@ -274,22 +273,3 @@ def test_backward_ms_greater_than_grace_ms(
             match="The backward_ms must not be greater than the grace_ms to avoid losing data.",
         ):
             left_sdf.join_interval(right_sdf, grace_ms=1, backward_ms=2)
-
-    def test_how_invalid_value(self, topic_manager_topic_factory, create_sdf):
-        left_topic = topic_manager_topic_factory()
-        right_topic = topic_manager_topic_factory()
-        left_sdf, right_sdf = create_sdf(left_topic), create_sdf(right_topic)
-
-        match = 'Invalid "how" value'
-        with pytest.raises(ValueError, match=match):
-            left_sdf.join_interval(right_sdf, how="invalid")
-
-    def test_mismatching_partitions_fails(
-        self, topic_manager_topic_factory, create_sdf
-    ):
-        left_topic = topic_manager_topic_factory()
-        right_topic = topic_manager_topic_factory(partitions=2)
-        left_sdf, right_sdf = create_sdf(left_topic), create_sdf(right_topic)
-
-        with pytest.raises(TopicPartitionsMismatch):
-            left_sdf.join_interval(right_sdf)