{{ connector_name }} test report
-
@@ -159,19 +167,23 @@
{% endif %}
- {{ artifact.name }} + {% endfor %} +
+ {% if step_result_to_artifact_links[step_result.step.title] %}
+
diff --git a/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/upgrade_cdk/pipeline.py b/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/upgrade_cdk/pipeline.py
index 6824f9d25550..8af4baab6a75 100644
--- a/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/upgrade_cdk/pipeline.py
+++ b/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/upgrade_cdk/pipeline.py
@@ -59,9 +59,7 @@ async def _run(self) -> StepResult:
status=StepStatus.FAILURE,
stdout="Could not export diff to local git repo.",
)
- return StepResult(
- step=self, status=StepStatus.SUCCESS, stdout=f"Updated CDK version to {self.new_version}", output_artifact=diff
- )
+ return StepResult(step=self, status=StepStatus.SUCCESS, stdout=f"Updated CDK version to {self.new_version}", output=diff)
except ValueError as e:
return StepResult(
step=self,
diff --git a/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/format/format_command.py b/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/format/format_command.py
index aeb17b08e504..5787c2c67132 100644
--- a/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/format/format_command.py
+++ b/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/format/format_command.py
@@ -124,7 +124,7 @@ async def invoke(self, ctx: click.Context, click_pipeline_context: ClickPipeline
container = self.get_format_container_fn(dagger_client, dir_to_format)
command_result = await self.get_format_command_result(dagger_client, container, dir_to_format)
- if (formatted_code_dir := command_result.output_artifact) and self.export_formatted_code:
+ if (formatted_code_dir := command_result.output) and self.export_formatted_code:
await formatted_code_dir.export(self.LOCAL_REPO_PATH)
if self._enable_logging:
@@ -203,9 +203,7 @@ async def get_format_command_result(
if await dir_with_modified_files.entries():
modified_files = await list_files_in_directory(dagger_client, dir_with_modified_files)
self.logger.debug(f"Modified files: {modified_files}")
- return CommandResult(
- command=self, status=StepStatus.FAILURE, stdout=stdout, stderr=stderr, output_artifact=dir_with_modified_files
- )
+ return CommandResult(command=self, status=StepStatus.FAILURE, stdout=stdout, stderr=stderr, output=dir_with_modified_files)
return CommandResult(command=self, status=StepStatus.SUCCESS, stdout=stdout, stderr=stderr)
except dagger.ExecError as e:
return CommandResult(command=self, status=StepStatus.FAILURE, stderr=e.stderr, stdout=e.stdout, exc_info=e)
diff --git a/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/steps/gradle.py b/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/steps/gradle.py
index 0fd139e20c33..456e6c3f1aa3 100644
--- a/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/steps/gradle.py
+++ b/airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/steps/gradle.py
@@ -1,19 +1,18 @@
#
# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
#
-import html
-import re
from abc import ABC
-from typing import Any, ClassVar, List, Optional, Tuple
+from datetime import datetime
+from typing import Any, ClassVar, List, Optional, Tuple, cast
import pipelines.dagger.actions.system.docker
-import xmltodict
-from dagger import CacheSharingMode, CacheVolume, Container, QueryError
+from dagger import CacheSharingMode, CacheVolume, Container, ExecError
from pipelines.airbyte_ci.connectors.context import ConnectorContext
from pipelines.consts import AMAZONCORRETTO_IMAGE
from pipelines.dagger.actions import secrets
from pipelines.hacks import never_fail_exec
-from pipelines.helpers.utils import sh_dash_c
+from pipelines.helpers.utils import dagger_directory_as_zip_file, sh_dash_c
+from pipelines.models.artifacts import Artifact
from pipelines.models.steps import Step, StepResult
@@ -36,7 +35,7 @@ class GradleTask(Step, ABC):
gradle_task_name: ClassVar[str]
bind_to_docker_host: ClassVar[bool] = False
mount_connector_secrets: ClassVar[bool] = False
- with_test_report: ClassVar[bool] = False
+ with_test_artifacts: ClassVar[bool] = False
accept_extra_params = True
@property
@@ -193,9 +192,18 @@ async def _run(self, *args: Any, **kwargs: Any) -> StepResult:
connector_gradle_task = f":airbyte-integrations:connectors:{self.context.connector.technical_name}:{self.gradle_task_name}"
gradle_command = self._get_gradle_command(connector_gradle_task, task_options=self.params_as_cli_options)
gradle_container = gradle_container.with_(never_fail_exec([gradle_command]))
- return await self.get_step_result(gradle_container)
- async def get_step_result(self, container: Container) -> StepResult:
+ # Collect the test artifacts, if applicable.
+ artifacts = []
+ if self.with_test_artifacts:
+ if test_logs := await self._collect_test_logs(gradle_container):
+ artifacts.append(test_logs)
+ if test_results := await self._collect_test_results(gradle_container):
+ artifacts.append(test_results)
+
+ return await self.get_step_result(gradle_container, artifacts)
+
+ async def get_step_result(self, container: Container, outputs: List[Artifact]) -> StepResult:
step_result = await super().get_step_result(container)
# Decorate with test report, if applicable.
return StepResult(
@@ -203,76 +211,75 @@ async def get_step_result(self, container: Container) -> StepResult:
status=step_result.status,
stdout=step_result.stdout,
stderr=step_result.stderr,
- report=await self._collect_test_report(container),
- output_artifact=step_result.output_artifact,
+ output=step_result.output,
+ artifacts=outputs,
)
- async def _collect_test_report(self, gradle_container: Container) -> Optional[str]:
- if not self.with_test_report:
+ async def _collect_test_logs(self, gradle_container: Container) -> Optional[Artifact]:
+ """
+ Exports the java docs from the container into the host filesystem.
+ The docs in the container are expected to be in build/test-logs, and will end up test-artifact directory by default
+ One can change the destination directory by setting the outputs
+ """
+ test_logs_dir_name_in_container = "test-logs"
+ test_logs_dir_name_in_zip = f"test-logs-{datetime.fromtimestamp(cast(float, self.context.pipeline_start_timestamp)).isoformat()}-{self.context.git_branch}-{self.gradle_task_name}".replace(
+ "/", "_"
+ )
+ if (
+ test_logs_dir_name_in_container
+ not in await gradle_container.directory(f"{self.context.connector.code_directory}/build").entries()
+ ):
+ self.context.logger.warn(f"No {test_logs_dir_name_in_container} found directory in the build folder")
return None
+ try:
+ zip_file = await (
+ dagger_directory_as_zip_file(
+ self.dagger_client,
+ await gradle_container.directory(f"{self.context.connector.code_directory}/build/{test_logs_dir_name_in_container}"),
+ test_logs_dir_name_in_zip,
+ )
+ )
+ return Artifact(
+ name=f"{test_logs_dir_name_in_zip}.zip",
+ content=zip_file,
+ content_type="application/zip",
+ to_upload=True,
+ )
+ except ExecError as e:
+ self.context.logger.error(str(e))
+ return None
- junit_xml_path = f"{self.context.connector.code_directory}/build/test-results/{self.gradle_task_name}"
- testsuites = []
+ async def _collect_test_results(self, gradle_container: Container) -> Optional[Artifact]:
+ """
+ Exports the junit test reports from the container into the host filesystem.
+ The docs in the container are expected to be in build/test-results, and will end up test-artifact directory by default
+ Only the XML files generated by junit are downloaded into the host filesystem
+ One can change the destination directory by setting the outputs
+ """
+ test_results_dir_name_in_container = "test-results"
+ test_results_dir_name_in_zip = f"test-results-{datetime.fromtimestamp(cast(float, self.context.pipeline_start_timestamp)).isoformat()}-{self.context.git_branch}-{self.gradle_task_name}".replace(
+ "/", "_"
+ )
+ if (
+ test_results_dir_name_in_container
+ not in await gradle_container.directory(f"{self.context.connector.code_directory}/build").entries()
+ ):
+ self.context.logger.warn(f"No {test_results_dir_name_in_container} found directory in the build folder")
+ return None
try:
- junit_xml_dir = await gradle_container.directory(junit_xml_path)
- for file_name in await junit_xml_dir.entries():
- if file_name.endswith(".xml"):
- junit_xml = await junit_xml_dir.file(file_name).contents()
- # This will be embedded in the HTML report in a Artifacts
+-
+ {% for artifact in step_result_to_artifact_links[step_result.step.title] %}
+
{% if step_result.stdout %}
- Standard output:
+ Standard output():
{{ step_result.stdout }}
{% endif %}
{% if step_result.stderr %}
- Standard error:
+ Standard error():
{{ step_result.stderr }}
{% endif %}
- {% if step_result.report %}
- Report:
- {{ step_result.report }}
- {% endif %}
block.
- # The java logging backend will have already taken care of masking any secrets.
- # Nothing to do in that regard.
- try:
- if testsuite := xmltodict.parse(junit_xml):
- testsuites.append(testsuite)
- except Exception as e:
- self.context.logger.error(str(e))
- self.context.logger.warn(f"Failed to parse junit xml file {file_name}.")
- except QueryError as e:
+ zip_file = await (
+ dagger_directory_as_zip_file(
+ self.dagger_client,
+ await gradle_container.directory(f"{self.context.connector.code_directory}/build/{test_results_dir_name_in_container}"),
+ test_results_dir_name_in_zip,
+ )
+ )
+ return Artifact(
+ name=f"{test_results_dir_name_in_zip}.zip",
+ content=zip_file,
+ content_type="application/zip",
+ to_upload=True,
+ )
+ except ExecError as e:
self.context.logger.error(str(e))
- self.context.logger.warn(f"Failed to retrieve junit test results from {junit_xml_path} gradle container.")
return None
- return render_junit_xml(testsuites)
-
-
-MAYBE_STARTS_WITH_XML_TAG = re.compile("^ *<")
-ESCAPED_ANSI_COLOR_PATTERN = re.compile(r"\?\[0?m|\?\[[34][0-9]m")
-
-
-def render_junit_xml(testsuites: List[Any]) -> str:
- """Renders the JUnit XML report as something readable in the HTML test report."""
- # Transform the dict contents.
- indent = " "
- for testsuite in testsuites:
- testsuite = testsuite.get("testsuite")
- massage_system_out_and_err(testsuite, indent, 4)
- if testcases := testsuite.get("testcase"):
- if not isinstance(testcases, list):
- testcases = [testcases]
- for testcase in testcases:
- massage_system_out_and_err(testcase, indent, 5)
- # Transform back to XML string.
- # Try to respect the JUnit XML test result schema.
- root = {"testsuites": {"testsuite": testsuites}}
- xml = xmltodict.unparse(root, pretty=True, short_empty_elements=True, indent=indent)
- # Escape < and > and so forth to make them render properly, but not in the log messages.
- # These lines will already have been escaped by xmltodict.unparse.
- lines = xml.splitlines()
- for idx, line in enumerate(lines):
- if MAYBE_STARTS_WITH_XML_TAG.match(line):
- lines[idx] = html.escape(line)
- return "\n".join(lines)
-
-
-def massage_system_out_and_err(d: dict, indent: str, indent_levels: int) -> None:
- """Makes the system-out and system-err text prettier."""
- if d:
- for key in ["system-out", "system-err"]:
- if s := d.get(key):
- lines = s.splitlines()
- s = ""
- for line in lines:
- stripped = line.strip()
- if stripped:
- s += "\n" + indent * indent_levels + ESCAPED_ANSI_COLOR_PATTERN.sub("", line.strip())
- s = s + "\n" + indent * (indent_levels - 1) if s else None
- d[key] = s
diff --git a/airbyte-ci/connectors/pipelines/pipelines/helpers/utils.py b/airbyte-ci/connectors/pipelines/pipelines/helpers/utils.py
index e275b42610bf..cd8d31dd4d40 100644
--- a/airbyte-ci/connectors/pipelines/pipelines/helpers/utils.py
+++ b/airbyte-ci/connectors/pipelines/pipelines/helpers/utils.py
@@ -19,7 +19,7 @@
import anyio
import asyncclick as click
import asyncer
-from dagger import Client, Config, Container, ExecError, File, ImageLayerCompression, Platform, Secret
+from dagger import Client, Config, Container, Directory, ExecError, File, ImageLayerCompression, Platform, Secret
from more_itertools import chunked
if TYPE_CHECKING:
@@ -353,3 +353,24 @@ def java_log_scrub_pattern(secrets_to_mask: List[str]) -> str:
":": ":",
},
)
+
+
+def dagger_directory_as_zip_file(dagger_client: Client, directory: Directory, directory_name: str) -> File:
+ """Compress a directory and return a File object representing the zip file.
+
+ Args:
+ dagger_client (Client): The dagger client.
+ directory (Path): The directory to compress.
+ directory_name (str): The name of the directory.
+
+ Returns:
+ File: The File object representing the zip file.
+ """
+ return (
+ dagger_client.container()
+ .from_("alpine:3.19.1")
+ .with_exec(sh_dash_c(["apk update", "apk add zip"]))
+ .with_mounted_directory(f"/{directory_name}", directory)
+ .with_exec(["zip", "-r", "/zipped.zip", f"/{directory_name}"])
+ .file("/zipped.zip")
+ )
diff --git a/airbyte-ci/connectors/pipelines/pipelines/models/artifacts.py b/airbyte-ci/connectors/pipelines/pipelines/models/artifacts.py
new file mode 100644
index 000000000000..f1deafd445e7
--- /dev/null
+++ b/airbyte-ci/connectors/pipelines/pipelines/models/artifacts.py
@@ -0,0 +1,47 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+import dagger
+from pipelines.consts import GCS_PUBLIC_DOMAIN
+from pipelines.dagger.actions import remote_storage
+
+
+@dataclass(kw_only=True)
+class Artifact:
+ """A dataclass to represent an artifact produced by a pipeline execution."""
+
+ name: str
+ content_type: str
+ content: dagger.File
+ to_upload: bool = True
+ local_path: Optional[Path] = None
+ gcs_url: Optional[str] = None
+
+ async def save_to_local_path(self, path: Path) -> Path:
+ exported = await self.content.export(str(path))
+ if exported:
+ self.local_path = path
+ return path
+ else:
+ raise Exception(f"Failed to save artifact {self.name} to local path {path}")
+
+ async def upload_to_gcs(self, dagger_client: dagger.Client, bucket: str, key: str, gcs_credentials: dagger.Secret) -> str:
+ gcs_cp_flags = [f'--content-disposition=filename="{self.name}"']
+ if self.content_type is not None:
+ gcs_cp_flags = gcs_cp_flags + [f"--content-type={self.content_type}"]
+
+ report_upload_exit_code, _, _ = await remote_storage.upload_to_gcs(
+ dagger_client=dagger_client,
+ file_to_upload=self.content,
+ key=key,
+ bucket=bucket,
+ gcs_credentials=gcs_credentials,
+ flags=gcs_cp_flags,
+ )
+ if report_upload_exit_code != 0:
+ raise Exception(f"Failed to upload artifact {self.name} to GCS. Exit code: {report_upload_exit_code}.")
+ self.gcs_url = f"{GCS_PUBLIC_DOMAIN}/{bucket}/{key}"
+ return f"{GCS_PUBLIC_DOMAIN}/{bucket}/{key}"
diff --git a/airbyte-ci/connectors/pipelines/pipelines/models/reports.py b/airbyte-ci/connectors/pipelines/pipelines/models/reports.py
index 206e1c44e9e1..4cf5c33f8055 100644
--- a/airbyte-ci/connectors/pipelines/pipelines/models/reports.py
+++ b/airbyte-ci/connectors/pipelines/pipelines/models/reports.py
@@ -7,17 +7,17 @@
from __future__ import annotations
import json
+import time
import typing
from dataclasses import dataclass, field
from datetime import datetime, timedelta
+from pathlib import Path
from typing import List
-import anyio
-from anyio import Path
from connector_ops.utils import console # type: ignore
-from pipelines.consts import GCS_PUBLIC_DOMAIN, LOCAL_REPORTS_PATH_ROOT
-from pipelines.dagger.actions import remote_storage
-from pipelines.helpers.utils import format_duration
+from pipelines.consts import LOCAL_REPORTS_PATH_ROOT
+from pipelines.helpers.utils import format_duration, slugify
+from pipelines.models.artifacts import Artifact
from pipelines.models.steps import StepResult, StepStatus
from rich.console import Group
from rich.panel import Panel
@@ -44,6 +44,10 @@ class Report:
def report_output_prefix(self) -> str:
return self.pipeline_context.report_output_prefix
+ @property
+ def report_dir_path(self) -> Path:
+ return Path(f"{LOCAL_REPORTS_PATH_ROOT}/{self.report_output_prefix}")
+
@property
def json_report_file_name(self) -> str:
return self.filename + ".json"
@@ -84,42 +88,52 @@ def lead_duration(self) -> timedelta:
def remote_storage_enabled(self) -> bool:
return self.pipeline_context.is_ci
- async def save_local(self, filename: str, content: str) -> Path:
- """Save the report files locally."""
- local_path = anyio.Path(f"{LOCAL_REPORTS_PATH_ROOT}/{self.report_output_prefix}/{filename}")
- await local_path.parents[0].mkdir(parents=True, exist_ok=True)
- await local_path.write_text(content)
- return local_path
-
- async def save_remote(self, local_path: Path, remote_key: str, content_type: str) -> int:
- assert self.pipeline_context.ci_report_bucket is not None, "The ci_report_bucket must be set to save reports."
-
- gcs_cp_flags = None if content_type is None else [f"--content-type={content_type}"]
- local_file = self.pipeline_context.dagger_client.host().directory(".", include=[str(local_path)]).file(str(local_path))
- report_upload_exit_code, _, _ = await remote_storage.upload_to_gcs(
- dagger_client=self.pipeline_context.dagger_client,
- file_to_upload=local_file,
- key=remote_key,
- bucket=self.pipeline_context.ci_report_bucket,
- gcs_credentials=self.pipeline_context.ci_gcs_credentials_secret,
- flags=gcs_cp_flags,
- )
- gcs_uri = "gs://" + self.pipeline_context.ci_report_bucket + "/" + remote_key
- public_url = f"{GCS_PUBLIC_DOMAIN}/{self.pipeline_context.ci_report_bucket}/{remote_key}"
- if report_upload_exit_code != 0:
- self.pipeline_context.logger.error(f"Uploading {local_path} to {gcs_uri} failed.")
- else:
- self.pipeline_context.logger.info(f"Uploading {local_path} to {gcs_uri} succeeded. Public URL: {public_url}")
- return report_upload_exit_code
-
async def save(self) -> None:
- """Save the report files."""
-
- local_json_path = await self.save_local(self.json_report_file_name, self.to_json())
- absolute_path = await local_json_path.absolute()
+ self.report_dir_path.mkdir(parents=True, exist_ok=True)
+ await self.save_json_report()
+ await self.save_step_result_artifacts()
+
+ async def save_json_report(self) -> None:
+ """Save the report as JSON, upload it to GCS if the pipeline is running in CI"""
+
+ json_report_path = self.report_dir_path / self.json_report_file_name
+ report_dir = self.pipeline_context.dagger_client.host().directory(str(self.report_dir_path))
+ local_json_report_file = report_dir.with_new_file(self.json_report_file_name, self.to_json()).file(self.json_report_file_name)
+ json_report_artifact = Artifact(name="JSON Report", content_type="application/json", content=local_json_report_file)
+ await json_report_artifact.save_to_local_path(json_report_path)
+ absolute_path = json_report_path.absolute()
self.pipeline_context.logger.info(f"Report saved locally at {absolute_path}")
- if self.remote_storage_enabled:
- await self.save_remote(local_json_path, self.json_report_remote_storage_key, "application/json")
+ if self.remote_storage_enabled and self.pipeline_context.ci_report_bucket and self.pipeline_context.ci_gcs_credentials_secret:
+ gcs_url = await json_report_artifact.upload_to_gcs(
+ dagger_client=self.pipeline_context.dagger_client,
+ bucket=self.pipeline_context.ci_report_bucket,
+ key=self.json_report_remote_storage_key,
+ gcs_credentials=self.pipeline_context.ci_gcs_credentials_secret,
+ )
+ self.pipeline_context.logger.info(f"JSON Report uploaded to {gcs_url}")
+
+ async def save_step_result_artifacts(self) -> None:
+ local_artifacts_dir = self.report_dir_path / "artifacts"
+ local_artifacts_dir.mkdir(parents=True, exist_ok=True)
+ # TODO: concurrent save and upload
+ for step_result in self.steps_results:
+ for artifact in step_result.artifacts:
+ step_artifacts_dir = local_artifacts_dir / slugify(step_result.step.title)
+ step_artifacts_dir.mkdir(parents=True, exist_ok=True)
+ await artifact.save_to_local_path(step_artifacts_dir / artifact.name)
+ if (
+ self.remote_storage_enabled
+ and self.pipeline_context.ci_report_bucket
+ and self.pipeline_context.ci_gcs_credentials_secret
+ ):
+ upload_time = int(time.time())
+ gcs_url = await artifact.upload_to_gcs(
+ dagger_client=self.pipeline_context.dagger_client,
+ bucket=self.pipeline_context.ci_report_bucket,
+ key=f"{self.report_output_prefix}/artifacts/{slugify(step_result.step.title)}/{upload_time}_{artifact.name}",
+ gcs_credentials=self.pipeline_context.ci_gcs_credentials_secret,
+ )
+ self.pipeline_context.logger.info(f"Artifact {artifact.name} for {step_result.step.title} uploaded to {gcs_url}")
def to_json(self) -> str:
"""Create a JSON representation of the report.
diff --git a/airbyte-ci/connectors/pipelines/pipelines/models/steps.py b/airbyte-ci/connectors/pipelines/pipelines/models/steps.py
index edc97079a32f..86be5550f713 100644
--- a/airbyte-ci/connectors/pipelines/pipelines/models/steps.py
+++ b/airbyte-ci/connectors/pipelines/pipelines/models/steps.py
@@ -19,6 +19,7 @@
from pipelines import main_logger
from pipelines.helpers import sentry_utils
from pipelines.helpers.utils import format_duration, get_exec_result
+from pipelines.models.artifacts import Artifact
if TYPE_CHECKING:
from typing import Any, ClassVar, Optional, Union
@@ -73,7 +74,8 @@ class Result:
stdout: Optional[str] = None
report: Optional[str] = None
exc_info: Optional[Exception] = None
- output_artifact: Any = None
+ output: Any = None
+ artifacts: List[Artifact] = field(default_factory=list)
@property
def success(self) -> bool:
@@ -387,7 +389,7 @@ def get_step_status_from_exit_code(
else:
return StepStatus.FAILURE
- async def get_step_result(self, container: Container) -> StepResult:
+ async def get_step_result(self, container: Container, *args: Any, **kwargs: Any) -> StepResult:
"""Concurrent retrieval of exit code, stdout and stdout of a container.
Create a StepResult object from these objects.
@@ -404,7 +406,7 @@ async def get_step_result(self, container: Container) -> StepResult:
status=self.get_step_status_from_exit_code(exit_code),
stderr=stderr,
stdout=stdout,
- output_artifact=container,
+ output=container,
)
def _get_timed_out_step_result(self) -> StepResult:
diff --git a/airbyte-ci/connectors/pipelines/pyproject.toml b/airbyte-ci/connectors/pipelines/pyproject.toml
index d6cca9affdc0..cc7abba8285d 100644
--- a/airbyte-ci/connectors/pipelines/pyproject.toml
+++ b/airbyte-ci/connectors/pipelines/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
[tool.poetry]
name = "pipelines"
-version = "4.3.2"
+version = "4.4.0"
description = "Packaged maintained by the connector operations team to perform CI for connectors' pipelines"
authors = ["Airbyte "]
diff --git a/airbyte-ci/connectors/pipelines/tests/test_bases.py b/airbyte-ci/connectors/pipelines/tests/test_bases.py
index a109e1e33a0b..15808f2c88de 100644
--- a/airbyte-ci/connectors/pipelines/tests/test_bases.py
+++ b/airbyte-ci/connectors/pipelines/tests/test_bases.py
@@ -39,7 +39,7 @@ async def test_run_with_timeout(self, test_context):
assert step_result.status == timed_out_step_result.status
assert step_result.stdout == timed_out_step_result.stdout
assert step_result.stderr == timed_out_step_result.stderr
- assert step_result.output_artifact == timed_out_step_result.output_artifact
+ assert step_result.output == timed_out_step_result.output
assert step.retry_count == step.max_retries + 1
@pytest.mark.parametrize(
diff --git a/airbyte-ci/connectors/pipelines/tests/test_build_image/test_python_connectors.py b/airbyte-ci/connectors/pipelines/tests/test_build_image/test_python_connectors.py
index bb8ac23a10ea..084911d82cf1 100644
--- a/airbyte-ci/connectors/pipelines/tests/test_build_image/test_python_connectors.py
+++ b/airbyte-ci/connectors/pipelines/tests/test_build_image/test_python_connectors.py
@@ -117,14 +117,14 @@ async def test__run_using_base_image_with_mocks(self, mocker, test_context_with_
container_built_from_base.with_exec.assert_called_with(["spec"])
assert step_result.status is StepStatus.SUCCESS
for platform in all_platforms:
- assert step_result.output_artifact[platform] == container_built_from_base
+ assert step_result.output[platform] == container_built_from_base
@pytest.mark.slow
async def test_building_from_base_image_for_real(self, test_context_with_real_connector_using_base_image, current_platform):
step = python_connectors.BuildConnectorImages(test_context_with_real_connector_using_base_image)
step_result = await step._run()
step_result.status is StepStatus.SUCCESS
- built_container = step_result.output_artifact[current_platform]
+ built_container = step_result.output[current_platform]
assert await built_container.env_variable("AIRBYTE_ENTRYPOINT") == " ".join(
build_customization.get_entrypoint(step.context.connector)
)
@@ -146,7 +146,7 @@ async def test_building_from_base_image_with_customization_for_real(
step = python_connectors.BuildConnectorImages(test_context_with_real_connector_using_base_image_with_build_customization)
step_result = await step._run()
step_result.status is StepStatus.SUCCESS
- built_container = step_result.output_artifact[current_platform]
+ built_container = step_result.output[current_platform]
assert await built_container.env_variable("MY_PRE_BUILD_ENV_VAR") == "my_pre_build_env_var_value"
assert await built_container.env_variable("MY_POST_BUILD_ENV_VAR") == "my_post_build_env_var_value"
@@ -161,7 +161,7 @@ async def test__run_using_base_dockerfile_with_mocks(self, mocker, test_context_
container_built_from_dockerfile.with_exec.assert_called_with(["spec"])
assert step_result.status is StepStatus.SUCCESS
for platform in all_platforms:
- assert step_result.output_artifact[platform] == container_built_from_dockerfile
+ assert step_result.output[platform] == container_built_from_dockerfile
async def test_building_from_dockerfile_for_real(self, test_context_with_real_connector_without_base_image):
step = python_connectors.BuildConnectorImages(test_context_with_real_connector_without_base_image)
diff --git a/airbyte-ci/connectors/pipelines/tests/test_helpers/test_execution/test_run_steps.py b/airbyte-ci/connectors/pipelines/tests/test_helpers/test_execution/test_run_steps.py
index 37bfa991eee0..2d0193676b1b 100644
--- a/airbyte-ci/connectors/pipelines/tests/test_helpers/test_execution/test_run_steps.py
+++ b/airbyte-ci/connectors/pipelines/tests/test_helpers/test_execution/test_run_steps.py
@@ -299,7 +299,7 @@ async def test_run_steps_passes_results():
StepToRun(
id=CONNECTOR_TEST_STEP_ID.ACCEPTANCE,
step=AcceptanceTests(context, True),
- args=lambda results: {"connector_under_test_container": results[CONNECTOR_TEST_STEP_ID.BUILD].output_artifact[LOCAL_BUILD_PLATFORM]},
+ args=lambda results: {"connector_under_test_container": results[CONNECTOR_TEST_STEP_ID.BUILD].output[LOCAL_BUILD_PLATFORM]},
depends_on=[CONNECTOR_TEST_STEP_ID.BUILD],
),
@@ -309,23 +309,23 @@ class Simple(Step):
title = "Test Step"
async def _run(self, arg1, arg2) -> StepResult:
- output_artifact = f"{arg1}:{arg2}"
- return StepResult(step=self, status=StepStatus.SUCCESS, output_artifact=output_artifact)
+ output = f"{arg1}:{arg2}"
+ return StepResult(step=self, status=StepStatus.SUCCESS, output=output)
async def async_args(results):
- return {"arg1": results["step2"].output_artifact, "arg2": "4"}
+ return {"arg1": results["step2"].output, "arg2": "4"}
steps = [
[StepToRun(id="step1", step=Simple(test_context), args={"arg1": "1", "arg2": "2"})],
- [StepToRun(id="step2", step=Simple(test_context), args=lambda results: {"arg1": results["step1"].output_artifact, "arg2": "3"})],
+ [StepToRun(id="step2", step=Simple(test_context), args=lambda results: {"arg1": results["step1"].output, "arg2": "3"})],
[StepToRun(id="step3", step=Simple(test_context), args=async_args)],
]
results = await run_steps(steps)
- assert results["step1"].output_artifact == "1:2"
- assert results["step2"].output_artifact == "1:2:3"
- assert results["step3"].output_artifact == "1:2:3:4"
+ assert results["step1"].output == "1:2"
+ assert results["step2"].output == "1:2:3"
+ assert results["step3"].output == "1:2:3:4"
@pytest.mark.anyio
diff --git a/airbyte-ci/connectors/pipelines/tests/test_publish.py b/airbyte-ci/connectors/pipelines/tests/test_publish.py
index 471c89ae6ece..f5cb73d0ca9e 100644
--- a/airbyte-ci/connectors/pipelines/tests/test_publish.py
+++ b/airbyte-ci/connectors/pipelines/tests/test_publish.py
@@ -278,12 +278,12 @@ async def test_run_connector_publish_pipeline_when_image_does_not_exist(
name="check_connector_image_does_not_exist_result", status=StepStatus.SUCCESS
)
- # have output_artifact.values return []
+ # have output.values return []
built_connector_platform = mocker.Mock()
built_connector_platform.values.return_value = ["linux/amd64"]
publish_pipeline.steps.run_connector_build.return_value = mocker.Mock(
- name="build_connector_for_publish_result", status=build_step_status, output_artifact=built_connector_platform
+ name="build_connector_for_publish_result", status=build_step_status, output=built_connector_platform
)
publish_pipeline.PushConnectorImageToRegistry.return_value.run.return_value = mocker.Mock(
diff --git a/airbyte-ci/connectors/pipelines/tests/test_tests/test_python_connectors.py b/airbyte-ci/connectors/pipelines/tests/test_tests/test_python_connectors.py
index f6b718be42f3..f53b43ebe57a 100644
--- a/airbyte-ci/connectors/pipelines/tests/test_tests/test_python_connectors.py
+++ b/airbyte-ci/connectors/pipelines/tests/test_tests/test_python_connectors.py
@@ -48,7 +48,7 @@ def context_for_certified_connector_with_setup(self, mocker, certified_connector
@pytest.fixture
async def certified_container_with_setup(self, context_for_certified_connector_with_setup, current_platform):
result = await BuildConnectorImages(context_for_certified_connector_with_setup).run()
- return result.output_artifact[current_platform]
+ return result.output[current_platform]
@pytest.fixture
def context_for_connector_with_poetry(self, mocker, connector_with_poetry, dagger_client, current_platform):
@@ -69,7 +69,7 @@ def context_for_connector_with_poetry(self, mocker, connector_with_poetry, dagge
@pytest.fixture
async def container_with_poetry(self, context_for_connector_with_poetry, current_platform):
result = await BuildConnectorImages(context_for_connector_with_poetry).run()
- return result.output_artifact[current_platform]
+ return result.output[current_platform]
async def test__run_for_setup_py(self, context_for_certified_connector_with_setup, certified_container_with_setup):
# Assume that the tests directory is available
@@ -80,7 +80,7 @@ async def test__run_for_setup_py(self, context_for_certified_connector_with_setu
"Total coverage:" in result.stdout
), "The pytest-cov package should be installed in the test environment and test coverage report should be displayed."
assert "Required test coverage of" in result.stdout, "A test coverage threshold should be defined for certified connectors."
- pip_freeze_output = await result.output_artifact.with_exec(["pip", "freeze"], skip_entrypoint=True).stdout()
+ pip_freeze_output = await result.output.with_exec(["pip", "freeze"], skip_entrypoint=True).stdout()
assert (
context_for_certified_connector_with_setup.connector.technical_name in pip_freeze_output
), "The connector should be installed in the test environment."
@@ -93,7 +93,7 @@ async def test__run_for_poetry(self, context_for_connector_with_poetry, containe
assert isinstance(result, StepResult)
# We only check for the presence of "test session starts" because we have no guarantee that the tests will pass
assert "test session starts" in result.stdout or "test session starts" in result.stderr, "The pytest tests should have started."
- pip_freeze_output = await result.output_artifact.with_exec(["poetry", "run", "pip", "freeze"], skip_entrypoint=True).stdout()
+ pip_freeze_output = await result.output.with_exec(["poetry", "run", "pip", "freeze"], skip_entrypoint=True).stdout()
assert (
context_for_connector_with_poetry.connector.technical_name in pip_freeze_output