different fixtures to use for regression tests on past versions

Author: emasab

tests/integration/conftest.py

@@ -26,12 +26,14 @@
 work_dir = os.path.dirname(os.path.realpath(__file__))
 
 
-def create_trivup_cluster():
-    return TrivupFixture({'with_sr': True,
-                          'debug': True,
-                          'cp_version': 'latest',
-                          'broker_conf': ['transaction.state.log.replication.factor=1',
-                                          'transaction.state.log.min.isr=1']})
+def create_trivup_cluster(conf={}):
+    trivup_fixture_conf = {'with_sr': True,
+                           'debug': True,
+                           'cp_version': 'latest',
+                           'broker_conf': ['transaction.state.log.replication.factor=1',
+                                           'transaction.state.log.min.isr=1']}
+    trivup_fixture_conf.update(conf)
+    return TrivupFixture(trivup_fixture_conf)
 
 
 def create_byo_cluster(conf):
@@ -41,8 +43,11 @@ def create_byo_cluster(conf):
     return ByoFixture(conf)
 
 
-@pytest.fixture(scope="package")
-def kafka_cluster():
+def kafka_cluster_fixture(
+    brokers_env="BROKERS",
+    sr_url_env="SR_URL",
+    trivup_cluster_conf={}
+):
     """
     If BROKERS environment variable is set to a CSV list of bootstrap servers
     an existing cluster is used.
@@ -52,22 +57,28 @@ def kafka_cluster():
     If BROKERS is not set a TrivUp cluster is created and used.
     """
 
-    bootstraps = os.environ.get("BROKERS", "")
+    bootstraps = os.environ.get(brokers_env, "")
     if bootstraps != "":
         conf = {"bootstrap.servers": bootstraps}
-        sr_url = os.environ.get("SR_URL", "")
+        sr_url = os.environ.get(sr_url_env, "")
         if sr_url != "":
             conf["schema.registry.url"] = sr_url
         print("Using ByoFixture with config from env variables: ", conf)
         cluster = create_byo_cluster(conf)
     else:
-        cluster = create_trivup_cluster()
+        cluster = create_trivup_cluster(trivup_cluster_conf)
     try:
         yield cluster
     finally:
         cluster.stop()
 
 
+@pytest.fixture(scope="package")
+def kafka_cluster():
+    for fixture in kafka_cluster_fixture():
+        yield fixture
+
+
 @pytest.fixture()
 def load_file():
     def get_handle(name):

tests/integration/schema_registry/test_api_client.py

@@ -21,6 +21,22 @@
 
 from confluent_kafka.schema_registry import Schema
 from confluent_kafka.schema_registry.error import SchemaRegistryError
+from tests.integration.conftest import kafka_cluster_fixture
+
+
+@pytest.fixture(scope="module")
+def kafka_cluster_cp_7_0_1():
+    """
+    Returns a Trivup cluster with CP version 7.0.1.
+    SR version 7.0.1 is the last returning 500 instead of 422
+    for the invalid schema passed to test_api_get_register_schema_invalid
+    """
+    for fixture in kafka_cluster_fixture(
+        brokers_env="BROKERS_7_0_1",
+        sr_url_env="SR_URL_7_0_1",
+        trivup_cluster_conf={'cp_version': '7.0.1'}
+    ):
+        yield fixture
 
 
 def _subject_name(prefix):
@@ -157,15 +173,28 @@ def test_api_get_registration_subject_not_found(kafka_cluster, load_file):
     assert e.value.error_code == 40401
 
 
-def test_api_get_register_schema_invalid(kafka_cluster, load_file):
+@pytest.mark.parametrize("kafka_cluster_name, http_status_code, error_code", [
+    ["kafka_cluster_cp_7_0_1", 500, 500],
+    ["kafka_cluster", 422, 42201],
+])
+def test_api_get_register_schema_invalid(
+        kafka_cluster_name,
+        http_status_code,
+        error_code,
+        load_file,
+        request):
     """
     Attempts to obtain registration information with an invalid schema
+    with different CP versions.
 
     Args:
-        kafka_cluster (KafkaClusterFixture): Kafka Cluster fixture
+        kafka_cluster_name (str): name of the Kafka Cluster fixture to use
+        http_status_code (int): HTTP status return code expected in this version
+        error_code (int): error code expected in this version
         load_file (callable(str)): Schema fixture constructor
-
+        request (FixtureRequest): PyTest object giving access to the test context
     """
+    kafka_cluster = request.getfixturevalue(kafka_cluster_name)
     sr = kafka_cluster.schema_registry()
     subject = _subject_name("registration_invalid_schema")
     schema = Schema(load_file('basic_schema.avsc'), schema_type='AVRO')
@@ -177,8 +206,8 @@ def test_api_get_register_schema_invalid(kafka_cluster, load_file):
     with pytest.raises(SchemaRegistryError, match="Invalid schema") as e:
         sr.lookup_schema(subject, schema2)
 
-    assert e.value.http_status_code == 422
-    assert e.value.error_code == 42201
+    assert e.value.http_status_code == http_status_code
+    assert e.value.error_code == error_code
 
 
 def test_api_get_subjects(kafka_cluster, load_file):