Source Facebook Marketing: support insights streams via async queries (…

…#1990)

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/e7778cfc-e97c-4458-9ecb-b4f2bba8946c.json

@@ -2,6 +2,6 @@
   "sourceDefinitionId": "e7778cfc-e97c-4458-9ecb-b4f2bba8946c",
   "name": "Facebook Marketing",
   "dockerRepository": "airbyte/source-facebook-marketing",
-  "dockerImageTag": "0.1.2",
+  "dockerImageTag": "0.1.3",
   "documentationUrl": "https://hub.docker.com/r/airbyte/source-facebook-marketing"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -76,7 +76,7 @@
 - sourceDefinitionId: e7778cfc-e97c-4458-9ecb-b4f2bba8946c
   name: Facebook Marketing
   dockerRepository: airbyte/source-facebook-marketing
-  dockerImageTag: 0.1.2
+  dockerImageTag: 0.1.3
   documentationUrl: https://hub.docker.com/r/airbyte/source-facebook-marketing
 - sourceDefinitionId: 57eb1576-8f52-463d-beb6-2e107cdf571d
   name: Hubspot

airbyte-integrations/bases/base-python/base_python/source.py

@@ -69,7 +69,7 @@ def check(self, logger: AirbyteLogger, config: json) -> AirbyteConnectionStatus:
         return AirbyteConnectionStatus(status=Status.SUCCEEDED)
 
     def read(
-        self, logger: AirbyteLogger, config: json, catalog: ConfiguredAirbyteCatalog, state: Dict[str, any]
+        self, logger: AirbyteLogger, config: json, catalog: ConfiguredAirbyteCatalog, state: Dict[str, any] = {}
     ) -> Generator[AirbyteMessage, None, None]:
         client = self._get_client(config)
 

airbyte-integrations/connectors/source-facebook-marketing/Dockerfile

@@ -12,5 +12,5 @@ COPY $CODE_PATH ./$CODE_PATH
 COPY setup.py ./
 RUN pip install .
 
-LABEL io.airbyte.version=0.1.2
+LABEL io.airbyte.version=0.1.3
 LABEL io.airbyte.name=airbyte/source-facebook-marketing

airbyte-integrations/connectors/source-facebook-marketing/build.gradle

@@ -1,13 +1,26 @@
 plugins {
     id 'airbyte-python'
     id 'airbyte-docker'
-    id 'airbyte-source-test'
+    id 'airbyte-standard-source-test-file'
 }
 
 airbytePython {
     moduleDirectory 'source_facebook_marketing'
 }
 
+airbyteStandardSourceTestFile {
+    specPath = "source_facebook_marketing/spec.json"
+    configPath = "secrets/config.json"
+    configuredCatalogPath = "sample_files/configured_catalog.json"
+}
+
+task("pythonIntegrationTests", type: PythonTask, dependsOn: installTestReqs) {
+    module = "pytest"
+    command = "-s integration_tests"
+}
+
+integrationTest.dependsOn("pythonIntegrationTests")
+
 dependencies {
     implementation files(project(':airbyte-integrations:bases:base-python').airbyteDocker.outputs)
 }

airbyte-integrations/connectors/source-facebook-marketing/integration_tests/integration_test.py

@@ -0,0 +1,58 @@
+"""
+MIT License
+
+Copyright (c) 2020 Airbyte
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+"""
+
+import json
+
+from airbyte_protocol import ConfiguredAirbyteCatalog, Type
+from base_python import AirbyteLogger
+from source_facebook_marketing.source import SourceFacebookMarketing
+
+config = json.loads(open("secrets/config.json", "r").read())
+
+
+class TestFacebookMarketingSource:
+    def test_ad_insights_streams_outputs_records(self):
+        catalog = self._read_catalog("sample_files/configured_catalog_adsinsights.json")
+        self._run_sync_test(config, catalog)
+
+    def test_ad_creatives_stream_outputs_records(self):
+        catalog = self._read_catalog("sample_files/configured_catalog_adcreatives.json")
+        self._run_sync_test(config, catalog)
+
+    @staticmethod
+    def _read_catalog(path):
+        return ConfiguredAirbyteCatalog.parse_raw(open(path, "r").read())
+
+    @staticmethod
+    def _run_sync_test(conf, catalog):
+        records = []
+        state = []
+        for message in SourceFacebookMarketing().read(AirbyteLogger(), conf, catalog):
+            if message.type == Type.RECORD:
+                records.append(message)
+            elif message.type == Type.STATE:
+                state.append(message)
+
+        assert len(records) > 0
+        assert len(state) > 0