test(smoke-test): updates to smoke-tests

docker/profiles/docker-compose.frontend.yml

@@ -10,6 +10,7 @@ x-datahub-frontend-service: &datahub-frontend-service
     - ${DATAHUB_LOCAL_FRONTEND_ENV:-empty2.env}
   environment: &datahub-frontend-service-env
     KAFKA_BOOTSTRAP_SERVER: broker:29092
+    DATAHUB_GMS_HOST: ${DATAHUB_GMS_HOST:-datahub-gms}
   volumes:
     - ${HOME}/.datahub/plugins:/etc/datahub/plugins
 

docker/profiles/docker-compose.gms.yml

@@ -40,6 +40,7 @@ x-kafka-env: &kafka-env
   # KAFKA_SCHEMAREGISTRY_URL=http://schema-registry:8081
   SCHEMA_REGISTRY_TYPE: INTERNAL
   KAFKA_SCHEMAREGISTRY_URL: http://datahub-gms:8080/schema-registry/api/
+  SPRING_KAFKA_CONSUMER_AUTO_OFFSET_RESET: ${SPRING_KAFKA_CONSUMER_AUTO_OFFSET_RESET:-earliest}
 
 x-datahub-quickstart-telemetry-env: &datahub-quickstart-telemetry-env
   DATAHUB_SERVER_TYPE: ${DATAHUB_SERVER_TYPE:-quickstart}

docker/profiles/docker-compose.prerequisites.yml

@@ -234,7 +234,7 @@ services:
     env_file: kafka-broker/env/docker.env
     environment:
       KAFKA_NODE_ID: 1
-      KAFKA_ADVERTISED_LISTENERS: BROKER://broker:29092,EXTERNAL://broker:9092
+      KAFKA_ADVERTISED_LISTENERS: BROKER://broker:29092,EXTERNAL://localhost:9092
       KAFKA_LISTENERS: BROKER://broker:29092,EXTERNAL://broker:9092,CONTROLLER://broker:39092
       KAFKA_INTER_BROKER_LISTENER_NAME: BROKER
       KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER

smoke-test/build.gradle

@@ -44,12 +44,19 @@ task yarnInstall(type: YarnTask) {
   environment = ['NODE_OPTIONS': '--openssl-legacy-provider']
   args = ['install', '--cwd', "${project.rootDir}/smoke-test/tests/cypress"]
 }
+
 task cypressLint(type: YarnTask, dependsOn: yarnInstall) {
   environment = ['NODE_OPTIONS': '--openssl-legacy-provider']
   // TODO: Run a full lint instead of just format.
   args = ['--cwd', "${project.rootDir}/smoke-test/tests/cypress", 'run', 'format']
 }
 
+task cypressLintFix(type: YarnTask, dependsOn: yarnInstall) {
+  environment = ['NODE_OPTIONS': '--openssl-legacy-provider']
+  // TODO: Run a full lint instead of just format.
+  args = ['--cwd', "${project.rootDir}/smoke-test/tests/cypress", 'run', 'format', '--write']
+}
+
 task installDev(type: Exec) {
     inputs.file file('pyproject.toml')
     inputs.file file('requirements.txt')
@@ -86,10 +93,7 @@ task pythonLintFix(type: Exec, dependsOn: installDev) {
  */
 task noCypressSuite0(type: Exec, dependsOn: [installDev, ':metadata-ingestion:installDev']) {
     environment 'RUN_QUICKSTART', 'false'
-    environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
-    environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
     environment 'TEST_STRATEGY', 'no_cypress_suite0'
-    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',
@@ -99,10 +103,7 @@ task noCypressSuite0(type: Exec, dependsOn: [installDev, ':metadata-ingestion:in
 
 task noCypressSuite1(type: Exec, dependsOn: [installDev, ':metadata-ingestion:installDev']) {
     environment 'RUN_QUICKSTART', 'false'
-    environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
-    environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
     environment 'TEST_STRATEGY', 'no_cypress_suite1'
-    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',
@@ -112,10 +113,7 @@ task noCypressSuite1(type: Exec, dependsOn: [installDev, ':metadata-ingestion:in
 
 task cypressSuite1(type: Exec, dependsOn: [installDev, ':metadata-ingestion:installDev']) {
     environment 'RUN_QUICKSTART', 'false'
-    environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
-    environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
     environment 'TEST_STRATEGY', 'cypress_suite1'
-    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',
@@ -125,10 +123,7 @@ task cypressSuite1(type: Exec, dependsOn: [installDev, ':metadata-ingestion:inst
 
 task cypressRest(type: Exec, dependsOn: [installDev, ':metadata-ingestion:installDev']) {
     environment 'RUN_QUICKSTART', 'false'
-    environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
-    environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
     environment 'TEST_STRATEGY', 'cypress_rest'
-    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',
@@ -141,9 +136,6 @@ task cypressRest(type: Exec, dependsOn: [installDev, ':metadata-ingestion:instal
  */
 task cypressDev(type: Exec, dependsOn: [installDev, ':metadata-ingestion:installDev']) {
     environment 'RUN_QUICKSTART', 'false'
-    environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
-    environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
-    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',
@@ -156,13 +148,18 @@ task cypressDev(type: Exec, dependsOn: [installDev, ':metadata-ingestion:install
  */
 task cypressData(type: Exec, dependsOn: [installDev, ':metadata-ingestion:installDev']) {
     environment 'RUN_QUICKSTART', 'false'
-    environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
-    environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
     environment 'RUN_UI', 'false'
-    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',
             "source ${venv_name}/bin/activate && set -x && " +
                     "./cypress-dev.sh"
-}
+}
+
+task lint {
+    dependsOn pythonLint, cypressLint
+}
+
+task lintFix {
+    dependsOn pythonLintFix
+}

smoke-test/cypress-dev.sh

@@ -10,9 +10,9 @@ fi
 
 source venv/bin/activate
 
-export KAFKA_BROKER_CONTAINER="datahub-kafka-broker-1"
-export KAFKA_BOOTSTRAP_SERVER="broker:9092"
-export ELASTIC_ID_HASH_ALGO="MD5"
+# set environment variables for the test
+source ./set-test-env-vars.sh
+
 python -c 'from tests.cypress.integration_test import ingest_data; ingest_data()'
 
 cd tests/cypress

smoke-test/run-quickstart.sh

@@ -10,17 +10,19 @@ source venv/bin/activate
 mkdir -p ~/.datahub/plugins/frontend/auth/
 echo "test_user:test_pass" >> ~/.datahub/plugins/frontend/auth/user.props
 
+echo "DATAHUB_VERSION = $DATAHUB_VERSION"
 DATAHUB_SEARCH_IMAGE="${DATAHUB_SEARCH_IMAGE:=opensearchproject/opensearch}"
 DATAHUB_SEARCH_TAG="${DATAHUB_SEARCH_TAG:=2.9.0}"
 XPACK_SECURITY_ENABLED="${XPACK_SECURITY_ENABLED:=plugins.security.disabled=true}"
 ELASTICSEARCH_USE_SSL="${ELASTICSEARCH_USE_SSL:=false}"
 USE_AWS_ELASTICSEARCH="${USE_AWS_ELASTICSEARCH:=true}"
 ELASTIC_ID_HASH_ALGO="${ELASTIC_ID_HASH_ALGO:=MD5}"
 
-echo "DATAHUB_VERSION = $DATAHUB_VERSION"
+
 DATAHUB_TELEMETRY_ENABLED=false  \
 DOCKER_COMPOSE_BASE="file://$( dirname "$DIR" )" \
 DATAHUB_SEARCH_IMAGE="$DATAHUB_SEARCH_IMAGE" DATAHUB_SEARCH_TAG="$DATAHUB_SEARCH_TAG" \
 XPACK_SECURITY_ENABLED="$XPACK_SECURITY_ENABLED" ELASTICSEARCH_USE_SSL="$ELASTICSEARCH_USE_SSL" \
 USE_AWS_ELASTICSEARCH="$USE_AWS_ELASTICSEARCH" \
-datahub docker quickstart --version ${DATAHUB_VERSION} --standalone_consumers --dump-logs-on-failure --kafka-setup
+DATAHUB_VERSION=${DATAHUB_VERSION}  \
+docker compose --project-directory ../docker/profiles	--profile quickstart-consumers up -d --quiet-pull --wait --wait-timeout 900

smoke-test/set-cypress-creds.sh

@@ -2,4 +2,4 @@
 
 export CYPRESS_ADMIN_USERNAME=${ADMIN_USERNAME:-datahub}
 export CYPRESS_ADMIN_PASSWORD=${ADMIN_PASSWORD:-datahub}
-export CYPRESS_ADMIN_DISPLAYNAME=${ADMIN_DISPLAYNAME:-DataHub}
+export CYPRESS_ADMIN_DISPLAYNAME=${ADMIN_DISPLAYNAME:-DataHub}

smoke-test/set-test-env-vars.sh

@@ -0,0 +1,3 @@
+export DATAHUB_KAFKA_SCHEMA_REGISTRY_URL=http://localhost:8080/schema-registry/api
+export DATAHUB_GMS_URL=http://localhost:8080
+export ELASTIC_ID_HASH_ALGO="MD5"

smoke-test/smoke.sh

@@ -16,16 +16,23 @@ cd "$DIR"
 
 if [ "${RUN_QUICKSTART:-true}" == "true" ]; then
     source ./run-quickstart.sh
+else
+  mkdir -p ~/.datahub/plugins/frontend/auth/
+  echo "test_user:test_pass" >> ~/.datahub/plugins/frontend/auth/user.props
+  echo "datahub:datahub" > ~/.datahub/plugins/frontend/auth/user.props
+
+  python3 -m venv venv
+  source venv/bin/activate
+  python -m pip install --upgrade pip uv>=0.1.10 wheel setuptools
+  uv pip install -r requirements.txt
 fi
 
-source venv/bin/activate
-
 (cd ..; ./gradlew :smoke-test:yarnInstall)
 
 source ./set-cypress-creds.sh
 
-export DATAHUB_GMS_URL=http://localhost:8080
-export ELASTIC_ID_HASH_ALGO="MD5"
+# set environment variables for the test
+source ./set-test-env-vars.sh
 
 # no_cypress_suite0, no_cypress_suite1, cypress_suite1, cypress_rest
 if [[ -z "${TEST_STRATEGY}" ]]; then

smoke-test/test_e2e.py

@@ -21,6 +21,7 @@
     get_frontend_session,
     get_admin_credentials,
     get_root_urn,
+    wait_for_writes_to_sync,
 )
 
 bootstrap_sample_data = "../metadata-ingestion/examples/mce_files/bootstrap_mce.json"
@@ -150,11 +151,13 @@ def _ensure_group_not_present(urn: str, frontend_session) -> Any:
 def test_ingestion_via_rest(wait_for_healthchecks):
     ingest_file_via_rest(bootstrap_sample_data)
     _ensure_user_present(urn=get_root_urn())
+    wait_for_writes_to_sync()
 
 
 @pytest.mark.dependency(depends=["test_healthchecks"])
 def test_ingestion_usage_via_rest(wait_for_healthchecks):
     ingest_file_via_rest(usage_sample_data)
+    wait_for_writes_to_sync()
 
 
 @pytest.mark.dependency(depends=["test_healthchecks"])
@@ -185,6 +188,7 @@ def test_ingestion_via_kafka(wait_for_healthchecks):
     # Since Kafka emission is asynchronous, we must wait a little bit so that
     # the changes are actually processed.
     time.sleep(kafka_post_ingestion_wait_sec)
+    wait_for_writes_to_sync()
 
 
 @pytest.mark.dependency(
@@ -196,6 +200,7 @@ def test_ingestion_via_kafka(wait_for_healthchecks):
 )
 def test_run_ingestion(wait_for_healthchecks):
     # Dummy test so that future ones can just depend on this one.
+    wait_for_writes_to_sync()
     pass
 
 
@@ -1384,7 +1389,9 @@ def test_native_user_endpoints(frontend_session):
     unauthenticated_get_invite_token_response = unauthenticated_session.post(
         f"{get_frontend_url()}/api/v2/graphql", json=get_invite_token_json
     )
-    assert unauthenticated_get_invite_token_response.status_code == HTTPStatus.UNAUTHORIZED
+    assert (
+        unauthenticated_get_invite_token_response.status_code == HTTPStatus.UNAUTHORIZED
+    )
 
     unauthenticated_create_reset_token_json = {
         "query": """mutation createNativeUserResetToken($input: CreateNativeUserResetTokenInput!) {\n
@@ -1399,7 +1406,10 @@ def test_native_user_endpoints(frontend_session):
         f"{get_frontend_url()}/api/v2/graphql",
         json=unauthenticated_create_reset_token_json,
     )
-    assert unauthenticated_create_reset_token_response.status_code == HTTPStatus.UNAUTHORIZED
+    assert (
+        unauthenticated_create_reset_token_response.status_code
+        == HTTPStatus.UNAUTHORIZED
+    )
 
     # cleanup steps
     json = {

smoke-test/tests/consistency_utils.py

@@ -8,14 +8,31 @@
 ELASTICSEARCH_REFRESH_INTERVAL_SECONDS: int = int(
     os.getenv("ELASTICSEARCH_REFRESH_INTERVAL_SECONDS", 5)
 )
-KAFKA_BROKER_CONTAINER: str = str(
-    os.getenv("KAFKA_BROKER_CONTAINER", "datahub-broker-1")
-)
 KAFKA_BOOTSTRAP_SERVER: str = str(os.getenv("KAFKA_BOOTSTRAP_SERVER", "broker:29092"))
 
 logger = logging.getLogger(__name__)
 
 
+def infer_kafka_broker_container() -> str:
+    cmd = "docker ps --format '{{.Names}}' | grep broker"
+    completed_process = subprocess.run(
+        cmd,
+        capture_output=True,
+        shell=True,
+        text=True,
+    )
+    result = str(completed_process.stdout)
+    lines = result.splitlines()
+    if len(lines) == 0:
+        raise ValueError("No Kafka broker containers found")
+    return lines[0]
+
+
+KAFKA_BROKER_CONTAINER: str = str(
+    os.getenv("KAFKA_BROKER_CONTAINER", infer_kafka_broker_container())
+)
+
+
 def wait_for_writes_to_sync(max_timeout_in_sec: int = 120) -> None:
     if USE_STATIC_SLEEP:
         time.sleep(ELASTICSEARCH_REFRESH_INTERVAL_SECONDS)
@@ -44,7 +61,9 @@ def wait_for_writes_to_sync(max_timeout_in_sec: int = 120) -> None:
             if maximum_lag == 0:
                 lag_zero = True
         except ValueError:
-            logger.warning(f"Error reading kafka lag using command: {cmd}")
+            logger.warning(
+                f"Error reading kafka lag using command: {cmd}", exc_info=True
+            )
 
     if not lag_zero:
         logger.warning(