tests: enable Airflow 3 integration tests for OPA (#642)

* wip: working tests (except auth: opa, oidc)

* wip: changing python test files

* single health.py for all airflow versions

* single metrics.py for all airflow versions

* update tests with new commons scripts

* tests: use "airflow-latest" instead of "airflow"

* cleanup and code comments

* restore deleted env

* use correct webserver service

* restore operator to release list

* test: fix oidc

* wip: get logging tests to work post-merge

* make env-vars version-specific

* fixed resolution of webserver url for execution api

* test: update opa test

* relaxed default resources

* test: opa almost working

* update test defs for oidc/opa

* changelog

* successful run of opa tests with airflow 3

* split opa tests for 2 and 3 in separate files

* Update tests/templates/kuttl/logging/51-assert.yaml.j2

Co-authored-by: Malte Sander <[email protected]>

* code review changes

* code review changes

* move containerdebug cmd to a function

* replaced random key with hard-coded one with comment

* cleanup and better comment

* for 3.x: only scheduler updates FAB permissions and restrict workers

* update Airlow 3 test dimension

---------

Co-authored-by: Andrew Kenworthy <[email protected]>
Co-authored-by: Andrew Kenworthy <[email protected]>
Co-authored-by: Malte Sander <[email protected]>

Author: 3 people

tests/templates/kuttl/opa/30-install-airflow.yaml.j2

@@ -24,11 +24,11 @@ metadata:
   name: airflow
 spec:
   image:
-{% if test_scenario['values']['airflow-non-experimental'].find(",") > 0 %}
-    custom: "{{ test_scenario['values']['airflow-non-experimental'].split(',')[1] }}"
-    productVersion: "{{ test_scenario['values']['airflow-non-experimental'].split(',')[0] }}"
+{% if test_scenario['values']['airflow'].find(",") > 0 %}
+    custom: "{{ test_scenario['values']['airflow'].split(',')[1] }}"
+    productVersion: "{{ test_scenario['values']['airflow'].split(',')[0] }}"
 {% else %}
-    productVersion: "{{ test_scenario['values']['airflow-non-experimental'] }}"
+    productVersion: "{{ test_scenario['values']['airflow'] }}"
 {% endif %}
     pullPolicy: IfNotPresent
   clusterConfig:
@@ -53,6 +53,7 @@ spec:
     configOverrides:
       webserver_config.py:
         WTF_CSRF_ENABLED: "False" # Allow "POST /login/" without CSRF token
+        AUTH_OPA_CACHE_MAXSIZE_DEFAULT: "0" # disable decision caching for easy debugging
     roleGroups:
       default:
         replicas: 1

tests/templates/kuttl/opa/31-opa-rules.yaml

@@ -12,11 +12,16 @@ data:
     default is_authorized_configuration := false
     default is_authorized_connection := false
     default is_authorized_dag := false
+    # This is no longer present in Airflow 3
     default is_authorized_dataset := false
     default is_authorized_pool := false
     default is_authorized_variable := false
     default is_authorized_view := false
     default is_authorized_custom_view := false
+    # These are new in Airflow 3
+    default is_authorized_backfill := false
+    default is_authorized_asset := false
+    default is_authorized_asset_alias := false
 
     # Allow the user "airflow" to create test users
     # POST /auth/fab/v1/users
@@ -26,6 +31,42 @@ data:
 
         input.user.name == "airflow"
     }
+    is_authorized_configuration if {
+        input.user.name == "airflow"
+    }
+    is_authorized_configuration if {
+        input.user.name == "airflow"
+    }
+    is_authorized_connection if {
+        input.user.name == "airflow"
+    }
+    is_authorized_dag if {
+        input.user.name == "airflow"
+    }
+    is_authorized_dataset if {
+        input.user.name == "airflow"
+    }
+    is_authorized_pool if {
+        input.user.name == "airflow"
+    }
+    is_authorized_variable if {
+        input.user.name == "airflow"
+    }
+    is_authorized_view if {
+        input.user.name == "airflow"
+    }
+    is_authorized_custom_view if {
+        input.user.name == "airflow"
+    }
+    is_authorized_backfill if {
+        input.user.name == "airflow"
+    }
+    is_authorized_asset if {
+        input.user.name == "airflow"
+    }
+    is_authorized_asset_alias if {
+        input.user.name == "airflow"
+    }
 
     # GET /api/v1/config
     is_authorized_configuration if {
@@ -72,7 +113,9 @@ data:
     is_authorized_dag if {
         input.method == "GET"
         input.access_entity == "RUN"
-        input.details.id == null
+        # Airflow 2 sets this to null
+        # Ignore for now so this rule can be used with Airflow 2 and 3
+        # input.details.id == "~"
 
         input.user.name == "jane.doe"
     }
@@ -148,3 +191,24 @@ data:
 
         input.user.name == "jane.doe"
     }
+
+    # GET /api/v2/backfills
+    is_authorized_backfill if {
+        input.method == "GET"
+
+        input.user.name == "jane.doe"
+    }
+
+    # GET /api/v2/assets
+    is_authorized_asset if {
+        input.method == "GET"
+
+        input.user.name == "jane.doe"
+    }
+
+    # GET /api/v2/assets/aliases
+    is_authorized_asset_alias if {
+        input.method == "GET"
+
+        input.user.name == "jane.doe"
+    }

tests/templates/kuttl/opa/32-create-users.yaml

@@ -0,0 +1,23 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+metadata:
+  name: create-users
+timeout: 300
+commands:
+  - script: |
+      kubectl exec -n $NAMESPACE airflow-webserver-default-0 -- airflow users create \
+      --username "jane.doe" \
+      --firstname "Jane" \
+      --lastname "Doe" \
+      --email "[email protected]" \
+      --password "T8mn72D9" \
+      --role "User"
+
+      kubectl exec -n $NAMESPACE airflow-webserver-default-0 -- airflow users create \
+        --username "richard.roe" \
+        --firstname "Richard" \
+        --lastname "Roe" \
+        --email "[email protected]" \
+        --password "NvfpU518" \
+        --role "User"

tests/templates/kuttl/opa/41-check-authorization.yaml renamed to tests/templates/kuttl/opa/41-check-authorization.yaml.j2

@@ -6,5 +6,9 @@ metadata:
 commands:
   - script: >
       kubectl cp
-      41_check-authorization.py
+{% if test_scenario['values']['airflow'].startswith("2") %}
+      41_check-authorization_2.py
+{% else %}
+      41_check-authorization_3.py
+{% endif %}
       $NAMESPACE/test-runner-0:/stackable/check-authorization.py

tests/templates/kuttl/opa/41_check-authorization.py renamed to tests/templates/kuttl/opa/41_check-authorization_2.py

@@ -29,14 +29,6 @@
 url = "http://airflow-webserver-default:8080"
 
 
-def create_user(user):
-    requests.post(
-        f"{url}/auth/fab/v1/users",
-        auth=("airflow", "airflow"),
-        json=user,
-    )
-
-
 def check_api_authorization_for_user(
     user, expected_status_code, method, endpoint, data=None, api="api/v1"
 ):
@@ -152,10 +144,6 @@ def test_is_authorized_custom_view():
     )
 
 
-# Create test users
-create_user(user_jane_doe)
-create_user(user_richard_roe)
-
 test_is_authorized_configuration()
 test_is_authorized_connection()
 test_is_authorized_dag()

tests/templates/kuttl/opa/41_check-authorization_3.py

@@ -0,0 +1,188 @@
+import logging
+import sys
+
+import requests
+
+logging.basicConfig(
+    level="DEBUG", format="%(asctime)s %(levelname)s: %(message)s", stream=sys.stdout
+)
+
+log = logging.getLogger(__name__)
+
+# user to headers mapping
+headers: dict[str, dict[str, str]] = {}
+
+# Jane Doe has access to specific resources.
+user_jane_doe = {
+    "first_name": "Jane",
+    "last_name": "Doe",
+    "username": "jane.doe",
+    "email": "[email protected]",
+    "roles": [{"name": "User"}],
+    "password": "T8mn72D9",
+}
+# Richard Roe has no access.
+user_richard_roe = {
+    "first_name": "Richard",
+    "last_name": "Roe",
+    "username": "richard.roe",
+    "email": "[email protected]",
+    "roles": [{"name": "User"}],
+    "password": "NvfpU518",
+}
+
+url = "http://airflow-webserver-default:8080"
+api = "api/v2"
+url_login = f"{url}/auth/login"
+
+
+def obtain_access_token(user: dict[str, str]) -> str:
+    token_url = f"{url}/auth/token"
+
+    data = {"username": user["username"], "password": user["password"]}
+
+    headers = {"Content-Type": "application/json"}
+
+    response = requests.post(token_url, headers=headers, json=data)
+
+    if response.status_code == 200 or response.status_code == 201:
+        token_data = response.json()
+        access_token = token_data["access_token"]
+        log.info(f"Got access token: {access_token}")
+        return access_token
+    else:
+        log.error(
+            f"Failed to obtain access token: {response.status_code} - {response.text}"
+        )
+        sys.exit(1)
+
+
+def assert_status_code(msg, left, right):
+    if left != right:
+        raise AssertionError(f"{msg}\n\tleft: {left}\n\tright: {right}")
+
+
+def check_api_authorization_for_user(
+    user, expected_status_code, method, endpoint, data=None
+):
+    api_url = f"{url}/{api}"
+
+    response = requests.request(
+        method, f"{api_url}/{endpoint}", headers=headers[user["email"]], json=data
+    )
+
+    assert_status_code(
+        f"Unexpected status code for {user["email"]=}",
+        response.status_code,
+        expected_status_code,
+    )
+
+
+def check_api_authorization(method, endpoint, expected_status_code=200, data=None):
+    check_api_authorization_for_user(
+        user_jane_doe, expected_status_code, method=method, endpoint=endpoint, data=data
+    )
+    check_api_authorization_for_user(
+        user_richard_roe, 403, method=method, endpoint=endpoint, data=data
+    )
+
+
+def check_website_authorization_for_user(user, expected_status_code):
+    username = user["username"]
+    password = user["password"]
+    with requests.Session() as session:
+        login_response = session.post(
+            url_login,
+            data=f"username={username}&password={password}",
+            allow_redirects=True,
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+        )
+        assert login_response.ok, f"Login for {username} failed"
+        home_response = session.get(f"{url}/home", allow_redirects=True)
+        assert_status_code(
+            f"GET /home for user [{username}] failed",
+            home_response.status_code,
+            expected_status_code,
+        )
+
+
+def test_is_authorized_configuration():
+    # section == null
+    check_api_authorization("GET", "config")
+    # section != null
+    check_api_authorization("GET", "config/section/core/option/dags_folder")
+
+
+def test_is_authorized_connection():
+    # conn_id == null
+    check_api_authorization("GET", "connections")
+
+
+def test_is_authorized_dag():
+    # access_entity == null and id == null
+    # There is no API endpoint to test this case.
+
+    # access_entity == null and id != null
+    check_api_authorization("GET", "dags/example_trigger_target_dag")
+
+    # access_entity != null and id == null
+    # Check "GET /dags/~/dagRuns" because access to "GET /dags" is always allowed
+    check_api_authorization("GET", "dags/~/dagRuns")
+
+    # access_entity != null and id != null
+    check_api_authorization("GET", "dags/example_trigger_target_dag/dagRuns")
+
+
+def test_is_authorized_dataset():
+    # uri == null
+    check_api_authorization("GET", "datasets")
+    # uri != null
+    check_api_authorization("GET", "datasets/s3%3A%2F%2Fdag1%2Foutput_1.txt")
+
+
+def test_is_authorized_pool():
+    # name == null
+    check_api_authorization("GET", "pools")
+    # name != null
+    check_api_authorization("GET", "pools/default_pool")
+
+
+def test_is_authorized_variable():
+    # key != null
+    check_api_authorization(
+        "POST", "variables", 201, data={"key": "myVar", "value": "1"}
+    )
+    # key == null
+    check_api_authorization("GET", "variables/myVar")
+
+
+def test_is_authorized_asset():
+    # name == null
+    check_api_authorization("GET", "assets")
+    # name != null
+    check_api_authorization("GET", "assets/3")  ## 'test-asset' has id 3
+
+
+def test_is_authorized_view():
+    check_website_authorization_for_user(user_jane_doe, 200)
+    check_website_authorization_for_user(user_richard_roe, 200)
+
+
+access_token_jane_doe = obtain_access_token(user_jane_doe)
+headers[user_jane_doe["email"]] = {
+    "Authorization": f"Bearer {access_token_jane_doe}",
+    "Content-Type": "application/json",
+}
+access_token_richard_roe = obtain_access_token(user_richard_roe)
+headers[user_richard_roe["email"]] = {
+    "Authorization": f"Bearer {access_token_richard_roe}",
+    "Content-Type": "application/json",
+}
+
+test_is_authorized_configuration()
+test_is_authorized_connection()
+test_is_authorized_dag()
+test_is_authorized_pool()
+test_is_authorized_variable()
+test_is_authorized_view()
+test_is_authorized_asset()

tests/test-definition.yaml

@@ -18,11 +18,6 @@ dimensions:
       - 3.0.1
       # To use a custom image, add a comma and the full name after the product version
       # - 2.9.3,oci.stackable.tech/sandbox/airflow:2.9.3-stackable0.0.0-dev
-  - name: airflow-non-experimental
-    values:
-      - 2.10.5
-      # To use a custom image, add a comma and the full name after the product version
-      # - 2.9.3,oci.stackable.tech/sandbox/airflow:2.9.3-stackable0.0.0-dev
   - name: opa-latest
     values:
       - 1.4.2
@@ -66,7 +61,7 @@ tests:
       - openshift
   - name: opa
     dimensions:
-      - airflow-non-experimental
+      - airflow
       - opa-latest
       - openshift
   - name: resources