address comments from @sherifnada

airbyte-integrations/bases/standard-test/README.md

@@ -1,7 +1,46 @@
-# Running Standard Source Tests
+# Standard tests
+This package uses pytest to discover, configure and execute the tests.
+It implemented as a pytest plugin.
+
+It adds new configuration option `--standard_test_config` - path to configuration file (by default is current folder). 
+Configuration stored in YaML format and validated by pydantic.
+
+Example configuration can be found in `sample_files/` folder:
+```yaml
+connector_image: <your_image>
+tests:
+  spec:
+    - spec_path: "<connector_folder>/spec.json"
+  connection:
+    - config_path: "secrets/config.json"
+      status: "succeed"
+    - config_path: "sample_files/invalid_config.json"
+      status: "exception"
+  discovery:
+    - config_path: "secrets/config.json"
+  basic_read:
+    - config_path: "secrets/config.json"
+      configured_catalog_path: "sample_files/configured_catalog.json"
+      validate_output_from_all_streams: true
+  incremental:
+    - config_path: "secrets/config.json"
+      configured_catalog_path: "sample_files/configured_catalog.json"
+      state_path: "sample_files/abnormal_state.json"
+      cursor_paths:
+        subscription_changes: ["timestamp"]
+        email_events: ["timestamp"]
+  full_refresh:
+    - config_path: "secrets/config.json"
+      configured_catalog_path: "sample_files/configured_catalog.json"
+```
+# Running
 ```bash
-python -m pytest standard_test/tests --standard_test_config=../../connectors/source-hubspot/ -vvv
+python -m pytest standard_test/tests --standard_test_config=<path_to_your_connector> -vvv
 ```
+_Note: this will assume that docker image for connector is already built_
 
-## How to
-TODO
+Using Gradle
+```bash
+./gradlew :airbyte-integrations:connectors:source-<name>:standardTest
+```
+_Note: this will also build docker image for connector_

airbyte-integrations/bases/standard-test/sample_files/standard_test_config.yml

@@ -4,7 +4,9 @@ tests:
     - spec_path: "source_hubspot/spec.json"
   connection:
     - config_path: "secrets/config.json"
-      invalid_config_path: "sample_files/invalid_config.json"
+      status: "succeed"
+    - config_path: "sample_files/invalid_config.json"
+      status: "exception"
   discovery:
     - config_path: "secrets/config.json"
   basic_read:

airbyte-integrations/bases/standard-test/standard_test/config.py

@@ -24,6 +24,7 @@
 
 from typing import List, Mapping, Optional
 
+from enum import Enum
 from pydantic import BaseModel, Field
 
 config_path: str = Field(default="secrets/config.json", description="Path to a JSON object representing a valid connector configuration")
@@ -44,8 +45,13 @@ class SpecTestConfig(BaseConfig):
 
 
 class ConnectionTestConfig(BaseConfig):
+    class Status(Enum):
+        Succeed = 'succeed'
+        Failed = 'failed'
+        Exception = 'exception'
+
     config_path: str = config_path
-    invalid_config_path: str = invalid_config_path
+    status: Status = Field(Status.Succeed, description="Indicate if connection check should succeed with provided config")
 
 
 class DiscoveryTestConfig(BaseConfig):

airbyte-integrations/bases/standard-test/standard_test/conftest.py

@@ -25,10 +25,11 @@
 import copy
 import json
 from pathlib import Path
-from typing import Optional
+from typing import Optional, MutableMapping, Any
 
 import pytest
 from airbyte_protocol import AirbyteCatalog, ConfiguredAirbyteCatalog, ConnectorSpecification
+from standard_test.config import Config
 from standard_test.connector_runner import ConnectorRunner
 from standard_test.utils import load_config
 
@@ -41,32 +42,32 @@ def base_path_fixture(pytestconfig, standard_test_config):
     return Path(pytestconfig.getoption("--standard_test_config")).absolute()
 
 
-@pytest.fixture(name="standard_test_config")
-def standard_test_config_fixture(pytestconfig):
+@pytest.fixture(name="standard_test_config", scope="session")
+def standard_test_config_fixture(pytestconfig) -> Config:
     """Fixture with test's config"""
     return load_config(pytestconfig.getoption("--standard_test_config"))
 
 
 @pytest.fixture(name="connector_config_path")
-def connector_config_path_fixture(inputs, base_path):
+def connector_config_path_fixture(inputs, base_path) -> Path:
     """Fixture with connector's config path"""
     return Path(base_path) / getattr(inputs, "config_path")
 
 
 @pytest.fixture(name="invalid_connector_config_path")
-def invalid_connector_config_path_fixture(inputs, base_path):
+def invalid_connector_config_path_fixture(inputs, base_path) -> Path:
     """Fixture with connector's config path"""
     return Path(base_path) / getattr(inputs, "invalid_config_path")
 
 
 @pytest.fixture(name="connector_spec_path")
-def connector_spec_path_fixture(inputs, base_path):
+def connector_spec_path_fixture(inputs, base_path) -> Path:
     """Fixture with connector's specification path"""
     return Path(base_path) / getattr(inputs, "spec_path")
 
 
 @pytest.fixture(name="configured_catalog_path")
-def configured_catalog_path_fixture(inputs, base_path):
+def configured_catalog_path_fixture(inputs, base_path) -> Optional[str]:
     """Fixture with connector's configured_catalog path"""
     if getattr(inputs, "configured_catalog_path"):
         return Path(base_path) / getattr(inputs, "configured_catalog_path")
@@ -88,27 +89,27 @@ def catalog_fixture(configured_catalog: ConfiguredAirbyteCatalog) -> Optional[Ai
 
 
 @pytest.fixture(name="image_tag")
-def image_tag_fixture(standard_test_config):
+def image_tag_fixture(standard_test_config) -> str:
     return standard_test_config.connector_image
 
 
 @pytest.fixture(name="connector_config")
-def connector_config_fixture(base_path, connector_config_path):
+def connector_config_fixture(base_path, connector_config_path) -> MutableMapping[str, Any]:
     with open(str(connector_config_path), "r") as file:
         contents = file.read()
     return json.loads(contents)
 
 
 @pytest.fixture(name="invalid_connector_config")
-def invalid_connector_config_fixture(base_path, invalid_connector_config_path):
+def invalid_connector_config_fixture(base_path, invalid_connector_config_path) -> MutableMapping[str, Any]:
     """TODO: implement default value - generate from valid config"""
     with open(str(invalid_connector_config_path), "r") as file:
         contents = file.read()
     return json.loads(contents)
 
 
 @pytest.fixture(name="malformed_connector_config")
-def malformed_connector_config_fixture(connector_config):
+def malformed_connector_config_fixture(connector_config) -> MutableMapping[str, Any]:
     """TODO: drop required field, add extra"""
     malformed_config = copy.deepcopy(connector_config)
     return malformed_config
@@ -124,7 +125,7 @@ def docker_runner_fixture(image_tag, tmp_path) -> ConnectorRunner:
     return ConnectorRunner(image_tag, volume=tmp_path)
 
 
-@pytest.fixture(name="validate_output_from_all_streams")
-def validate_output_from_all_streams_fixture(inputs):
-    """Fixture to provide value of `validate output from all streams` flag"""
-    return getattr(inputs, "validate_output_from_all_streams")
+@pytest.fixture(scope="session", autouse=True)
+def pull_docker_image(standard_test_config) -> None:
+    """Startup fixture to pull docker image"""
+    ConnectorRunner(image_tag=standard_test_config.connector_image)

airbyte-integrations/bases/standard-test/standard_test/connector_runner.py

@@ -25,16 +25,16 @@
 import json
 import logging
 from pathlib import Path
-from typing import Iterable, Mapping, Optional
+from typing import Iterable, Mapping, Optional, List
 
 import docker
 from airbyte_protocol import AirbyteMessage, ConfiguredAirbyteCatalog
 
 
 class ConnectorRunner:
-    def __init__(self, name: str, volume: Path):
-        self._name = name
+    def __init__(self, image_name: str, volume: Path):
         self._client = docker.from_env()
+        self._image = self._client.images.pull(image_name)
         self._runs = 0
         self._volume_base = volume
 
@@ -68,27 +68,27 @@ def _prepare_volumes(self, config: Optional[Mapping], state: Optional[Mapping],
         }
         return volumes
 
-    def call_spec(self, **kwargs):
+    def call_spec(self, **kwargs) -> List[AirbyteMessage]:
         cmd = "spec"
         output = list(self.run(cmd=cmd, **kwargs))
         return output
 
-    def call_check(self, config, **kwargs):
+    def call_check(self, config, **kwargs) -> List[AirbyteMessage]:
         cmd = "check --config tap_config.json"
         output = list(self.run(cmd=cmd, config=config, **kwargs))
         return output
 
-    def call_discover(self, config, **kwargs):
+    def call_discover(self, config, **kwargs) -> List[AirbyteMessage]:
         cmd = "discover --config tap_config.json"
         output = list(self.run(cmd=cmd, config=config, **kwargs))
         return output
 
-    def call_read(self, config, catalog, **kwargs):
+    def call_read(self, config, catalog, **kwargs) -> List[AirbyteMessage]:
         cmd = "read --config tap_config.json --catalog catalog.json"
         output = list(self.run(cmd=cmd, config=config, catalog=catalog, **kwargs))
         return output
 
-    def call_read_with_state(self, config, catalog, state, **kwargs):
+    def call_read_with_state(self, config, catalog, state, **kwargs) -> List[AirbyteMessage]:
         cmd = "read --config tap_config.json --catalog catalog.json --state state.json"
         output = list(self.run(cmd=cmd, config=config, catalog=catalog, state=state, **kwargs))
         return output
@@ -97,8 +97,9 @@ def run(self, cmd, config=None, state=None, catalog=None, **kwargs) -> Iterable[
         self._runs += 1
         volumes = self._prepare_volumes(config, state, catalog)
         logs = self._client.containers.run(
-            image=self._name, command=cmd, working_dir="/data", volumes=volumes, network="host", stdout=True, stderr=True, **kwargs
+            image=self._image, command=cmd, working_dir="/data", volumes=volumes, network="host", stdout=True, stderr=True, **kwargs
         )
+        logging.info("Running docker, folders: %s", volumes)
         for line in logs.decode("utf-8").splitlines():
             logging.info(AirbyteMessage.parse_raw(line).type)
             yield AirbyteMessage.parse_raw(line)

airbyte-integrations/bases/standard-test/standard_test/tests/test_core.py

@@ -28,6 +28,7 @@
 from airbyte_protocol import ConnectorSpecification, Status, Type
 from docker.errors import ContainerError
 from standard_test.base import BaseTest
+from standard_test.config import BasicReadTestConfig, ConnectionTestConfig
 from standard_test.connector_runner import ConnectorRunner
 
 
@@ -44,26 +45,25 @@ def test_spec(self, connector_spec: ConnectorSpecification, docker_runner: Conne
 
 @pytest.mark.timeout(30)
 class TestConnection(BaseTest):
-    def test_check(self, connector_config, docker_runner: ConnectorRunner):
-        output = docker_runner.call_check(config=connector_config)
-        con_messages = [message for message in output if message.type == Type.CONNECTION_STATUS]
+    def test_check(self, connector_config, inputs: ConnectionTestConfig, docker_runner: ConnectorRunner):
+        if inputs.status == ConnectionTestConfig.Status.Succeed:
+            output = docker_runner.call_check(config=connector_config)
+            con_messages = [message for message in output if message.type == Type.CONNECTION_STATUS]
 
-        assert len(con_messages) == 1, "Connection status message should be emitted exactly once"
-        assert con_messages[0].connectionStatus.status == Status.SUCCEEDED
+            assert len(con_messages) == 1, "Connection status message should be emitted exactly once"
+            assert con_messages[0].connectionStatus.status == Status.SUCCEEDED
+        elif inputs.status == ConnectionTestConfig.Status.Failed:
+            output = docker_runner.call_check(config=connector_config)
+            con_messages = [message for message in output if message.type == Type.CONNECTION_STATUS]
 
-    # def test_check_without_network(self, connector_config, docker_runner: ConnectorRunner):
-    #     output = docker_runner.call_check(config=connector_config, network_disabled=True)
-    #     con_messages = [message for message in output if message.type == Type.CONNECTION_STATUS]
-    #
-    #     assert len(con_messages) == 1, "Connection status message should be emitted exactly once"
-    #     assert con_messages[0].connectionStatus.status == Status.FAILED
+            assert len(con_messages) == 1, "Connection status message should be emitted exactly once"
+            assert con_messages[0].connectionStatus.status == Status.FAILED
+        elif inputs.status == ConnectionTestConfig.Status.Exception:
+            with pytest.raises(ContainerError) as err:
+                docker_runner.call_check(config=connector_config)
 
-    def test_check_with_invalid_config(self, invalid_connector_config, docker_runner: ConnectorRunner):
-        with pytest.raises(ContainerError) as err:
-            docker_runner.call_check(config=invalid_connector_config)
-
-        assert err.value.exit_status != 0, "Connector should exit with error code"
-        assert "Traceback" in err.value.stderr.decode("utf-8"), "Connector should print exception"
+            assert err.value.exit_status != 0, "Connector should exit with error code"
+            assert "Traceback" in err.value.stderr.decode("utf-8"), "Connector should print exception"
 
 
 @pytest.mark.timeout(30)
@@ -83,7 +83,7 @@ def test_discover(self, connector_config, catalog, docker_runner: ConnectorRunne
 
 @pytest.mark.timeout(300)
 class TestBasicRead(BaseTest):
-    def test_read(self, connector_config, configured_catalog, validate_output_from_all_streams, docker_runner: ConnectorRunner):
+    def test_read(self, connector_config, configured_catalog, inputs: BasicReadTestConfig, docker_runner: ConnectorRunner):
         output = docker_runner.call_read(connector_config, configured_catalog)
         records = [message.record for message in output if message.type == Type.RECORD]
         counter = Counter(record.stream for record in records)
@@ -94,7 +94,7 @@ def test_read(self, connector_config, configured_catalog, validate_output_from_a
 
         assert records, "At least one record should be read using provided catalog"
 
-        if validate_output_from_all_streams:
+        if inputs.validate_output_from_all_streams:
             assert (
                 not streams_without_records
             ), f"All streams should return some records, streams without records: {streams_without_records}"

airbyte-integrations/bases/standard-test/standard_test/tests/test_full_refresh.py

@@ -31,7 +31,7 @@
 from standard_test.utils import full_refresh_only_catalog
 
 
-@pytest.mark.timeout(300)
+@pytest.mark.timeout(20 * 60)
 class TestFullRefresh(BaseTest):
     def test_sequential_reads(self, connector_config, configured_catalog, docker_runner: ConnectorRunner):
         configured_catalog = full_refresh_only_catalog(configured_catalog)