Issue 1:
Apache Airflow Provider(s)

amazon

Versions of Apache Airflow Providers

6.2.0

Apache Airflow version

2.5.0

Operating System

CentOS 7

Deployment

Other Docker-based deployment

Deployment details

No response

What happened

Upon upgrading to airflow 2.5.0, EMR Add Steps Operator now:

Succeeds at adding a step to our emr job.

tries to wait for the job to complete

times out, retries, and duplicates the steps on our emr cluster.

times out, and fails.

The cause seems to be an error in the operator:

https://github.com/apache/airflow/pull/27286/files#diff-94612f7f88a5f95ca9e3f23051f39f1e0b6a4cadac1f533615db5480b379ed34R115

Instead of passing wait_for_completion=True to the emr hook method add_job_flow_steps, it should probably pass self.wait_for_completion. Also, currently self.wait_for_completion seem to not be used anywhere in the project.

What you think should happen instead

The task should succeed once the EMR steps are added and move on to the next task in the dag.

How to reproduce

No response

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Issue 2:
Hello,

Overview

Currently, Helm Chart for Airflow has two types of tests: (Learn the best practices of Helm Chart testing)

Template testing (unit testing): these tests render the templates against various input values. These types of tests let us verify that the template rendered the expected resources in the manner you intended. These tests are fast to execute and can catch syntactic errors of your template

Integration testing: these tests take the rendered templates and deploy them to a real Kubernetes cluster. We can then verify the deployed infrastructure works as intended by hitting the endpoints or querying Kubernetes for the resources. These tests closely resemble an actual deployment and give us a close approximation of how it might behave when ready to push the chart to production.

In each case, we use a different framework:

For template testing, we use Helm-unittest

For integration testing, we use bash scripts and Pytest + kind cluster

Start reading here

Today I would like to talk about template testing. In my opinion, using helm-unittest is not a very good solution.

This framework is not very popular, so any contributor who wants to change must read the documentation first to add new tests. (People are lazy by nature, so they don't add these tests).

Another problem is the form - the YAML file. The YAML file is not very flexible and makes it very difficult to write more unusual tests. It would be more flexible to express the tests in a more normal programming language.

During the discussion with @dimberman, I prepared a small POC that shows how we can test the Helm Chart with Pytest + Python.

Here is link: #11533 (comment)

import subprocess
import unittest
from tempfile import NamedTemporaryFile

import yaml
from typing import Dict, Optional

OBJECT_COUNT_IN_BASIC_DEPLOYMENT = 22

class TestBaseChartTest(unittest.TestCase):
def render_chart(self, name, values: Optional[Dict] = None):
values = values or {}
with NamedTemporaryFile() as tmp_file:
content = yaml.dump(values)
tmp_file.write(content.encode())
tmp_file.flush()
templates = subprocess.check_output(["helm", "template", name, "..", '--values', tmp_file.name])
k8s_objects = yaml.load_all(templates)
k8s_objects = [k8s_object for k8s_object in k8s_objects if k8s_object]
return k8s_objects

def test_basic_deployments(self):
k8s_objects = self.render_chart("TEST-BASIC", {"chart": {'metadata': 'AA'}})
list_of_kind_names_tuples = [
(k8s_object['kind'], k8s_object['metadata']['name'])
for k8s_object in k8s_objects
]
self.assertEqual(
list_of_kind_names_tuples,
[
('ServiceAccount', 'TEST-BASIC-scheduler'),
('ServiceAccount', 'TEST-BASIC-webserver'),
('ServiceAccount', 'TEST-BASIC-worker'),
('Secret', 'TEST-BASIC-postgresql'),
('Secret', 'TEST-BASIC-airflow-metadata'),
('Secret', 'TEST-BASIC-airflow-result-backend'),
('ConfigMap', 'TEST-BASIC-airflow-config'),
('ClusterRole', 'TEST-BASIC-pod-launcher-role'),
('ClusterRoleBinding', 'TEST-BASIC-pod-launcher-rolebinding'),
('Service', 'TEST-BASIC-postgresql-headless'),
('Service', 'TEST-BASIC-postgresql'),
('Service', 'TEST-BASIC-statsd'),
('Service', 'TEST-BASIC-webserver'),
('Deployment', 'TEST-BASIC-scheduler'),
('Deployment', 'TEST-BASIC-statsd'),
('Deployment', 'TEST-BASIC-webserver'),
('StatefulSet', 'TEST-BASIC-postgresql'),
('Secret', 'TEST-BASIC-fernet-key'),
('Secret', 'TEST-BASIC-redis-password'),
('Secret', 'TEST-BASIC-broker-url'),
('Job', 'TEST-BASIC-create-user'),
('Job', 'TEST-BASIC-run-airflow-migrations')
]
)
self.assertEqual(OBJECT_COUNT_IN_BASIC_DEPLOYMENT, len(k8s_objects))

def test_basic_deployment_without_default_users(self):
k8s_objects = self.render_chart("TEST-BASIC", {"webserver": {'defaultUser': {'enabled': False}}})
list_of_kind_names_tuples = [
(k8s_object['kind'], k8s_object['metadata']['name'])
for k8s_object in k8s_objects
]
self.assertNotIn(('Job', 'TEST-BASIC-create-user'), list_of_kind_names_tuples)
self.assertEqual(OBJECT_COUNT_IN_BASIC_DEPLOYMENT - 1, len(k8s_objects))

Alternatively, we can also migrate to terratest, which has native Helm Chart integration but uses Go-lang.

Now there is a question for the community. What do you think about it? Should we migrate all tests to Pytestt? Is this the only reason contributors don't add unit tests? What else can we do to encourage contributors to write tests?

Best regards,

Kamil Breguła