Skip to content

Commit 446eae3

Browse files
alafanecherealafanechere
authored
airbyte-ci: Introduce --only-step option for connector tests (#34276)
1 parent e0adbe8 commit 446eae3

File tree

5 files changed

+150
-6
lines changed

5 files changed

+150
-6
lines changed

‎airbyte-ci/connectors/pipelines/README.md

+6-1
Original file line numberDiff line numberDiff line change
@@ -223,6 +223,9 @@ Test certified connectors:
223223
Test connectors changed on the current branch:
224224
`airbyte-ci connectors --modified test`
225225

226+
Run acceptance test only on the modified connectors, just run its full refresh tests:
227+
`airbyte-ci connectors --modified test --only-step="acceptance" --acceptance.-k=test_full_refresh`
228+
226229
#### What it runs
227230

228231
```mermaid
@@ -261,11 +264,12 @@ flowchart TD
261264
| Option | Multiple | Default value | Description |
262265
| ------------------------------------------------------- | -------- | ------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
263266
| `--skip-step/-x` | True | | Skip steps by id e.g. `-x unit -x acceptance` |
267+
| `--only-step/-k` | True | | Only run specific steps by id e.g. `-k unit -k acceptance` |
264268
| `--fail-fast` | False | False | Abort after any tests fail, rather than continuing to run additional tests. Use this setting to confirm a known bug is fixed (or not), or when you only require a pass/fail result. |
265269
| `--code-tests-only` | True | False | Skip any tests not directly related to code updates. For instance, metadata checks, version bump checks, changelog verification, etc. Use this setting to help focus on code quality during development. |
266270
| `--concurrent-cat` | False | False | Make CAT tests run concurrently using pytest-xdist. Be careful about source or destination API rate limits. |
267271
| `--<step-id>.<extra-parameter>=<extra-parameter-value>` | True | | You can pass extra parameters for specific test steps. More details in the extra parameters section below |
268-
| `--ci-requirements` | False | | | Output the CI requirements as a JSON payload. It is used to determine the CI runner to use.
272+
| `--ci-requirements` | False | | | Output the CI requirements as a JSON payload. It is used to determine the CI runner to use.
269273

270274
Note:
271275

@@ -539,6 +543,7 @@ E.G.: running `pytest` on a specific test folder:
539543

540544
| Version | PR | Description |
541545
| ------- | ---------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------- |
546+
| 3.4.0 | [#34276](https://github.com/airbytehq/airbyte/pull/34276) | Introduce `--only-step` option for connector tests. |
542547
| 3.3.0 | [#34218](https://github.com/airbytehq/airbyte/pull/34218) | Introduce `--ci-requirements` option for client defined CI runners. |
543548
| 3.2.0 | [#34050](https://github.com/airbytehq/airbyte/pull/34050) | Connector test steps can take extra parameters |
544549
| 3.1.3 | [#34136](https://github.com/airbytehq/airbyte/pull/34136) | Fix issue where dagger excludes were not being properly applied |

‎airbyte-ci/connectors/pipelines/pipelines/airbyte_ci/connectors/test/commands.py

+15-2
Original file line numberDiff line numberDiff line change
@@ -53,10 +53,19 @@
5353
@click.option(
5454
"--skip-step",
5555
"-x",
56+
"skip_steps",
5657
multiple=True,
5758
type=click.Choice([step_id.value for step_id in CONNECTOR_TEST_STEP_ID]),
5859
help="Skip a step by name. Can be used multiple times to skip multiple steps.",
5960
)
61+
@click.option(
62+
"--only-step",
63+
"-k",
64+
"only_steps",
65+
multiple=True,
66+
type=click.Choice([step_id.value for step_id in CONNECTOR_TEST_STEP_ID]),
67+
help="Only run specific step by name. Can be used multiple times to keep multiple steps.",
68+
)
6069
@click.argument(
6170
"extra_params", nargs=-1, type=click.UNPROCESSED, callback=argument_parsing.build_extra_params_mapping(CONNECTOR_TEST_STEP_ID)
6271
)
@@ -66,14 +75,17 @@ async def test(
6675
code_tests_only: bool,
6776
fail_fast: bool,
6877
concurrent_cat: bool,
69-
skip_step: List[str],
78+
skip_steps: List[str],
79+
only_steps: List[str],
7080
extra_params: Dict[CONNECTOR_TEST_STEP_ID, STEP_PARAMS],
7181
) -> bool:
7282
"""Runs a test pipeline for the selected connectors.
7383
7484
Args:
7585
ctx (click.Context): The click context.
7686
"""
87+
if only_steps and skip_steps:
88+
raise click.UsageError("Cannot use both --only-step and --skip-step at the same time.")
7789
if ctx.obj["is_ci"]:
7890
fail_if_missing_docker_hub_creds(ctx)
7991
if ctx.obj["is_ci"] and ctx.obj["pull_request"] and ctx.obj["pull_request"].draft:
@@ -89,7 +101,8 @@ async def test(
89101

90102
run_step_options = RunStepOptions(
91103
fail_fast=fail_fast,
92-
skip_steps=[CONNECTOR_TEST_STEP_ID(step_id) for step_id in skip_step],
104+
skip_steps=[CONNECTOR_TEST_STEP_ID(step_id) for step_id in skip_steps],
105+
keep_steps=[CONNECTOR_TEST_STEP_ID(step_id) for step_id in only_steps],
93106
step_params=extra_params,
94107
)
95108
connectors_tests_contexts = [

‎airbyte-ci/connectors/pipelines/pipelines/helpers/execution/run_steps.py

+65-2
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88

99
import inspect
1010
from dataclasses import dataclass, field
11-
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Tuple, Union
11+
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Optional, Set, Tuple, Union
1212

1313
import anyio
1414
import asyncer
@@ -27,16 +27,78 @@ class InvalidStepConfiguration(Exception):
2727
pass
2828

2929

30+
def _get_dependency_graph(steps: STEP_TREE) -> Dict[str, List[str]]:
31+
"""
32+
Get the dependency graph of a step tree.
33+
"""
34+
dependency_graph: Dict[str, List[str]] = {}
35+
for step in steps:
36+
if isinstance(step, StepToRun):
37+
dependency_graph[step.id] = step.depends_on
38+
elif isinstance(step, list):
39+
nested_dependency_graph = _get_dependency_graph(list(step))
40+
dependency_graph = {**dependency_graph, **nested_dependency_graph}
41+
else:
42+
raise Exception(f"Unexpected step type: {type(step)}")
43+
44+
return dependency_graph
45+
46+
47+
def _get_transitive_dependencies_for_step_id(
48+
dependency_graph: Dict[str, List[str]], step_id: str, visited: Optional[Set[str]] = None
49+
) -> List[str]:
50+
"""Get the transitive dependencies for a step id.
51+
52+
Args:
53+
dependency_graph (Dict[str, str]): The dependency graph to use.
54+
step_id (str): The step id to get the transitive dependencies for.
55+
visited (Optional[Set[str]], optional): The set of visited step ids. Defaults to None.
56+
57+
Returns:
58+
List[str]: List of transitive dependencies as step ids.
59+
"""
60+
if visited is None:
61+
visited = set()
62+
63+
if step_id not in visited:
64+
visited.add(step_id)
65+
66+
dependencies: List[str] = dependency_graph.get(step_id, [])
67+
for dependency in dependencies:
68+
dependencies.extend(_get_transitive_dependencies_for_step_id(dependency_graph, dependency, visited))
69+
70+
return dependencies
71+
else:
72+
return []
73+
74+
3075
@dataclass
3176
class RunStepOptions:
3277
"""Options for the run_step function."""
3378

3479
fail_fast: bool = True
3580
skip_steps: List[str] = field(default_factory=list)
81+
keep_steps: List[str] = field(default_factory=list)
3682
log_step_tree: bool = True
3783
concurrency: int = 10
3884
step_params: Dict[CONNECTOR_TEST_STEP_ID, STEP_PARAMS] = field(default_factory=dict)
3985

86+
def __post_init__(self) -> None:
87+
if self.skip_steps and self.keep_steps:
88+
raise ValueError("Cannot use both skip_steps and keep_steps at the same time")
89+
90+
def get_step_ids_to_skip(self, runnables: STEP_TREE) -> List[str]:
91+
if self.skip_steps:
92+
return self.skip_steps
93+
if self.keep_steps:
94+
step_ids_to_keep = set(self.keep_steps)
95+
dependency_graph = _get_dependency_graph(runnables)
96+
all_step_ids = set(dependency_graph.keys())
97+
for step_id in self.keep_steps:
98+
step_ids_to_keep.update(_get_transitive_dependencies_for_step_id(dependency_graph, step_id))
99+
return list(all_step_ids - step_ids_to_keep)
100+
return []
101+
40102

41103
@dataclass(frozen=True)
42104
class StepToRun:
@@ -217,6 +279,7 @@ async def run_steps(
217279
if not runnables:
218280
return results
219281

282+
step_ids_to_skip = options.get_step_ids_to_skip(runnables)
220283
# Log the step tree
221284
if options.log_step_tree:
222285
main_logger.info(f"STEP TREE: {runnables}")
@@ -232,7 +295,7 @@ async def run_steps(
232295
steps_to_evaluate, remaining_steps = _get_next_step_group(runnables)
233296

234297
# Remove any skipped steps
235-
steps_to_run, results = _filter_skipped_steps(steps_to_evaluate, options.skip_steps, results)
298+
steps_to_run, results = _filter_skipped_steps(steps_to_evaluate, step_ids_to_skip, results)
236299

237300
# Run all steps in list concurrently
238301
semaphore = anyio.Semaphore(options.concurrency)

‎airbyte-ci/connectors/pipelines/pyproject.toml

+1-1
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
44

55
[tool.poetry]
66
name = "pipelines"
7-
version = "3.3.0"
7+
version = "3.4.0"
88
description = "Packaged maintained by the connector operations team to perform CI for connectors' pipelines"
99
authors = ["Airbyte <contact@airbyte.io>"]
1010

‎airbyte-ci/connectors/pipelines/tests/test_helpers/test_execution/test_run_steps.py

+63
Original file line numberDiff line numberDiff line change
@@ -359,3 +359,66 @@ async def test_run_steps_with_params():
359359
TestStep.accept_extra_params = True
360360
await run_steps(steps, options=options)
361361
assert steps[0].step.params_as_cli_options == ["--param1=value1"]
362+
363+
364+
class TestRunStepOptions:
365+
def test_init(self):
366+
options = RunStepOptions()
367+
assert options.fail_fast is True
368+
assert options.concurrency == 10
369+
assert options.skip_steps == []
370+
assert options.step_params == {}
371+
372+
options = RunStepOptions(fail_fast=False, concurrency=1, skip_steps=["step1"], step_params={"step1": {"--param1": ["value1"]}})
373+
assert options.fail_fast is False
374+
assert options.concurrency == 1
375+
assert options.skip_steps == ["step1"]
376+
assert options.step_params == {"step1": {"--param1": ["value1"]}}
377+
378+
with pytest.raises(ValueError):
379+
RunStepOptions(skip_steps=["step1"], keep_steps=["step2"])
380+
381+
@pytest.mark.parametrize(
382+
"step_tree, options, expected_skipped_ids",
383+
[
384+
(
385+
[
386+
[StepToRun(id="step1", step=TestStep(test_context)), StepToRun(id="step2", step=TestStep(test_context))],
387+
StepToRun(id="step3", step=TestStep(test_context)),
388+
StepToRun(id="step4", step=TestStep(test_context), depends_on=["step3", "step1"]),
389+
StepToRun(id="step5", step=TestStep(test_context)),
390+
],
391+
RunStepOptions(keep_steps=["step4"]),
392+
{"step2", "step5"},
393+
),
394+
(
395+
[
396+
[StepToRun(id="step1", step=TestStep(test_context)), StepToRun(id="step2", step=TestStep(test_context))],
397+
StepToRun(id="step3", step=TestStep(test_context)),
398+
[
399+
StepToRun(id="step4", step=TestStep(test_context), depends_on=["step1"]),
400+
StepToRun(id="step6", step=TestStep(test_context), depends_on=["step4", "step5"]),
401+
],
402+
StepToRun(id="step5", step=TestStep(test_context), depends_on=["step3"]),
403+
],
404+
RunStepOptions(keep_steps=["step6"]),
405+
{"step2"},
406+
),
407+
(
408+
[
409+
[StepToRun(id="step1", step=TestStep(test_context)), StepToRun(id="step2", step=TestStep(test_context))],
410+
StepToRun(id="step3", step=TestStep(test_context)),
411+
[
412+
StepToRun(id="step4", step=TestStep(test_context), depends_on=["step1"]),
413+
StepToRun(id="step6", step=TestStep(test_context), depends_on=["step4", "step5"]),
414+
],
415+
StepToRun(id="step5", step=TestStep(test_context), depends_on=["step3"]),
416+
],
417+
RunStepOptions(skip_steps=["step1"]),
418+
{"step1"},
419+
),
420+
],
421+
)
422+
def test_get_step_ids_to_skip(self, step_tree, options, expected_skipped_ids):
423+
skipped_ids = options.get_step_ids_to_skip(step_tree)
424+
assert set(skipped_ids) == expected_skipped_ids

0 commit comments

Comments
 (0)