add http tests for the study and its flow run use case (#2691)

SantanM · hengxian-jiang · Copilot · web-flow · commit d54a840e4539 · 2026-07-24T03:07:13.000Z
* Use tokenStudyCode instead of studyId in strategus flow * Update flows to use tokenStudyCode * Remove unintended changes * Adapt tokenStudyCode on functions side * More align in tokenStudyCode * Verify and fix * add strategus http tests, with hardcoded treatment patterns specification * Add strategus analysis integration test as a separate daily scheduled GHA workflow * Fix workflow branch trigger to use current branch for testing * Add Prefect diagnostics on strategus flow run failure in CI * Add Prefect diagnostics on strategus flow run failure in CI * fix env in gh action * fix env in gh action * add a patch for treatment patterns for this version * change databaseconnector remote ref * add flow-hades build; to revert * use cache schema name instead hardcoded schemaname * concat databasename and schemaname * add print logs, to revert * use cache_id in DBDao * add cache_id placeholder and in active places * fix trexdao * fix strategus DuckDB schema qualification * align strategus results cache routing * Fix Strategus execution schema qualification (#2832) Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> * WIP, todo: revert * use cdmsettings for schemaname * successful flow execution on local * revert building docker image * reorder steps in gh action * add a comment * fixes for results persistence * add missing env * add back the env * use _shared env * add default value to the flow * fix copilot comment * change sqlrender to alp-dqd branch * remove branch in GHA on event * use env Signed-off-by: Santan Maddi <55120902+SantanM@users.noreply.github.com> --------- Signed-off-by: Santan Maddi <55120902+SantanM@users.noreply.github.com> Co-authored-by: hengxian-jiang <hengxian.jiang@u.nus.edu> Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
diff --git a/.github/workflows/_test-http-duckdb.yml b/.github/workflows/_test-http-duckdb.yml
@@ -377,7 +377,7 @@ jobs:
           echo "isTestEnv: ${isTestEnv:-false}"
           echo "TESTSCHEMA: ${TESTSCHEMA}"
           echo "HTTPTEST_DB_DIALECT: ${HTTPTEST_DB_DIALECT}"
-          yarn test-specs --test_schema_name=$TESTSCHEMA
+          yarn test-specs --test_schema_name=$TESTSCHEMA '--file_pattern=^(?!strategus_analysis_test).*_test\.js$'
 
       - name: Mark HTTP duckdb attempt success
         if: success()
diff --git a/.github/workflows/strategus-analysis-integration-test.yml b/.github/workflows/strategus-analysis-integration-test.yml
diff --git a/README.md b/README.md
@@ -15,6 +15,7 @@
 [![Commuity](https://img.shields.io/badge/Community-Slack-d9dbec)](https://join.slack.com/t/data2evidence/shared_invite/zt-3vabnh2qr-vMev2VfLI2Sl1YA27gGVig)
 [![GitHub Activity](https://img.shields.io/github.qkg1.topmit-activity/m/ohdsi/d2e?label=Commit%20activty&logo=github&color=d9dbec)](https://github.com/ohdsi/d2e/graphs/contributors)
 [![D2E CLI Version](https://img.shields.io/npm/v/d2e?label=D2E%20installer&logo=npm&color=d9dbec)](https://www.npmjs.com/package/d2e)
+[![Strategus Analysis Tests](https://github.com/ohdsi/d2e/actions/workflows/strategus-analysis-integration-test.yml/badge.svg?branch=develop)](https://github.com/ohdsi/d2e/actions/workflows/strategus-analysis-integration-test.yml)
 
 ## Overview  
 
diff --git a/plugins/flows/_shared_flow_utils/dao/trexdao.py b/plugins/flows/_shared_flow_utils/dao/trexdao.py
@@ -484,7 +484,9 @@ def get_database_connector_connection_string(self) -> str:
         host = self.tenant_configs.host
         port = self.tenant_configs.port
 
-        return f"{DialectDrivers.jdbc.trex}://{host}:{port}/{self.database_code}?preferQueryMode=simple&autocommit=true"
+        # Match Python's _get_connection: connect on cache_id so unqualified queries resolve there.
+        jdbc_dbname = self.cache_id or self.database_code
+        return f"{DialectDrivers.jdbc.trex}://{host}:{port}/{jdbc_dbname}?preferQueryMode=simple&autocommit=true"
 
     def get_database_connector_dbms_val(self) -> str:
         return DialectDrivers.database_connector.trex
diff --git a/plugins/flows/hades/renv.lock b/plugins/flows/hades/renv.lock
@@ -934,7 +934,7 @@
       "RemoteHost": "api.github.com",
       "RemoteUsername": "data2evidence",
       "RemoteRepo": "d2e-DatabaseConnector",
-      "RemoteRef": "v6.3.2-0.0.1"
+      "RemoteRef": "alp-dqd"
     },
     "DeepPatientLevelPrediction": {
       "Package": "DeepPatientLevelPrediction",
@@ -2699,8 +2699,7 @@
       "RemoteHost": "api.github.com",
       "RemoteUsername": "data2evidence",
       "RemoteRepo": "d2e-SqlRender",
-      "RemoteRef": "6573c8044a9ddc27da5222a50dc4d9ecdbc59525",
-      "RemoteSha": "6573c8044a9ddc27da5222a50dc4d9ecdbc59525"
+      "RemoteRef": "alp-dqd"
     },
     "TTR": {
       "Package": "TTR",
diff --git a/plugins/flows/hades/strategus_plugin/dc-local.yml b/plugins/flows/hades/strategus_plugin/dc-local.yml
@@ -1,8 +1,9 @@
-# docker compose -f flows/hades/strategus_plugin/dc-local.yml up -d
+# docker compose -f ./plugins/flows/hades/strategus_plugin/dc-local.yml up -d
 services:
   app:
     platform: linux/amd64
     image: ghcr.io/ohdsi/d2e/flow-hades:develop
+    # entrypoint: tail -f /dev/null
     # with debugger listening
     # entrypoint: python -m debugpy --listen 0.0.0.0:9234 --wait-for-client -m flows.hades.strategus_plugin.flow
     entrypoint: /app/.venv/bin/python -m pytest /app/flows/hades/strategus_plugin/tests/
diff --git a/plugins/flows/hades/strategus_plugin/flow.py b/plugins/flows/hades/strategus_plugin/flow.py
@@ -6,6 +6,7 @@
 
 from prefect import flow, task
 from prefect.context import TaskRunContext, FlowRunContext, get_run_context
+from prefect.variables import Variable
 from prefect.artifacts import create_markdown_artifact
 
 from .hooks import generate_nodes_flow_hook, execute_nodes_flow_hook, node_task_execution_hook
@@ -53,6 +54,7 @@ def strategus_plugin(json_graph, options):
     update_results_schema = _options.get('updateResultsSchema', True)
     databaseCode = options.get('databaseCode', None)
     datasetId = options.get('datasetId', None)
+    cacheId = options.get('cacheId', None)
     studyName = options.get("studyName", "")
     tokenStudyCode = options.get('tokenStudyCode', None)
 
@@ -91,11 +93,18 @@ def strategus_plugin(json_graph, options):
         )
         # updateResultsSchema option will drop the existing schema before uploading new results
         if(update_results_schema):
-            drop_strategus_results(options)
+            results_db_settings = {
+                'databaseCode': Variable.get('trex_strategus_results_db_name', 'strategus_results'),
+                'cacheId': cacheId,
+                "datasetID": datasetId,
+                "tokenStudyCode": tokenStudyCode
+            }
+            drop_strategus_results(results_db_settings)
 
         if(upload_results):
             result_db_settings = {
-                'database_code': databaseCode,
+                'database_code': Variable.get('trex_strategus_results_db_name', 'strategus_results'),
+                'cache_id': cacheId,
                 "dataset_id": datasetId,
                 "token_study_code": tokenStudyCode
             }
@@ -215,6 +224,7 @@ def runStrategus(json_graph, options):
     token_study_code = options.get('tokenStudyCode', None)
     datasetId = options.get('datasetId', None)
     database_code = options.get('databaseCode', None)
+    cache_id = options.get('cacheId', None)
     schema_name = options.get('schemaName', None)
     upload_results = options.get('uploadResults', False)
     update_results_schema = options.get('updateResultsSchema', True)
@@ -228,7 +238,7 @@ def runStrategus(json_graph, options):
     if(not schema_name):
        raise Exception('Schema name is missing')
     
-    dbSettings = { "database_code": database_code, "schema_name": schema_name, "dataset_id": datasetId, "token_study_code": token_study_code }
+    dbSettings = { "database_code": database_code, "cache_id": cache_id, "schema_name": schema_name, "dataset_id": datasetId, "token_study_code": token_study_code }
     base_path = f'/tmp/{flow_run_id}'
     work_folder = f'{base_path}/work'
     path_to_results = f'{base_path}/results'
@@ -243,6 +253,8 @@ def runStrategus(json_graph, options):
 
     analysisSpec = json.dumps(analysisSpec)
     defaultExecutionSettings = getRCdmExecutionSettings({
+        "cacheId": cache_id,
+        "dbName": cache_id if cache_id else database_code,
         "schemaName": schema_name,
         "workFolder": work_folder,
         "resultsFolder": path_to_results
@@ -252,11 +264,16 @@ def runStrategus(json_graph, options):
     execute_r_strategus(analysisSpec, executionSettings, dbSettings)
     # updateResultsSchema option will drop the existing schema before uploading new results
     if(update_results_schema):
-        drop_strategus_results(options)
+        drop_strategus_results({
+            'databaseCode': Variable.get('trex_strategus_results_db_name', 'strategus_results'),
+            'cacheId': cache_id,
+            'tokenStudyCode': token_study_code
+        })
 
     if(upload_results):
         result_db_settings = {
-            'database_code': database_code,
+            'database_code': Variable.get('trex_strategus_results_db_name', 'strategus_results'),
+            'cache_id': cache_id,
             "dataset_id": datasetId,
             "token_study_code": token_study_code
         }
@@ -297,6 +314,7 @@ def drop_strategus_results(options):
 
     drop_strategus_results_schema(dbSettings={
         'database_code': database_code,
+        'cache_id': options.get('cacheId', None),
         'token_study_code': token_study_code
     })
 
diff --git a/plugins/flows/hades/strategus_plugin/nodes.py b/plugins/flows/hades/strategus_plugin/nodes.py
@@ -1186,10 +1186,16 @@ def task(self, nodes, results, task_run_context):
 
                 databaseConnectorJarFolder = '/app/inst/drivers'
                 os.environ['DATABASECONNECTOR_JAR_FOLDER'] = databaseConnectorJarFolder
-                dbSettings = { "database_code": self.flowOptions["databaseCode"], "schema_name": self.flowOptions["schemaName"], "dataset_id": self.flowOptions["datasetId"] }
+                dbSettings = {
+                    "database_code": self.flowOptions["databaseCode"],
+                    "cache_id": self.flowOptions.get("cacheId"),
+                    "schema_name": self.flowOptions["schemaName"],
+                    "dataset_id": self.flowOptions["datasetId"],
+                }
                 dbdao = DBDao(
                     dialect=SupportedDatabaseDialects.TREX if USE_TREX_CONNECTION else None,
-                    database_code=dbSettings['database_code']
+                    database_code=dbSettings['database_code'],
+                    cache_id=dbSettings.get('cache_id', None)
                 )
                 db_credentials = dbdao.tenant_configs
                 rDatabaseConnector = ro.packages.importr('DatabaseConnector')
@@ -1207,6 +1213,7 @@ def task(self, nodes, results, task_run_context):
                 path_to_results = f'{base_path}/results'
 
                 executionSettings = getRCdmExecutionSettings({
+                    "cacheId": dbSettings.get("cache_id"),
                     "schemaName": dbSettings['schema_name'],
                     "workFolder": work_folder,
                     "resultsFolder": path_to_results
@@ -1241,22 +1248,27 @@ def task(self, nodes, results, task_run_context):
 def get_strategus_node(options):
     return StrategusNode({"id": str(uuid.uuid4()), "type": "strategus_node", "flowOptions": options})
 
+
 @flow(name="execute-r-strategus",
       log_prints=True)
 def execute_r_strategus(analysisSpec: str, executionSettings, dbSettings):
     with ro.default_converter.context():
         try:
             ro.r(set_trex_env_var(USE_TREX_CONNECTION))
             database_code = dbSettings['database_code']
+            cache_id = dbSettings.get('cache_id', None)
             rParallelLogger = importr('ParallelLogger')
             rDatabaseConnector = importr('DatabaseConnector')
             databaseConnectorJarFolder = '/app/inst/drivers'
 
             dbdao = DBDao(
                 dialect=SupportedDatabaseDialects.TREX if USE_TREX_CONNECTION else None,
-                database_code=database_code
+                database_code=database_code,
+                cache_id=cache_id
             )
             db_credentials = dbdao.tenant_configs
+            print(f"Connecting to database with connection string: {dbdao.get_database_connector_connection_string()}")
+
             rConnectionDetails = rDatabaseConnector.createConnectionDetails(
                 dbms=dbdao.get_database_connector_dbms_val(), 
                 connectionString=dbdao.get_database_connector_connection_string(),
@@ -1296,6 +1308,9 @@ def execute_r_strategus(analysisSpec: str, executionSettings, dbSettings):
 def execute(rSpec, rExecutionSettings, rConnectionDetails):
     with ro.default_converter.context():
         ro.r(set_trex_env_var(USE_TREX_CONNECTION))
+        _patch = "/app/flows/hades/strategus_plugin/patch_treatment_patterns.R"
+        if os.path.exists(_patch):
+            ro.r(f'source("{_patch}")')
         ro.r('cat("Max Java Heap Size (GB): ", .jcall(.jnew("java/lang/Runtime"), "J", "maxMemory") / 1e9, "\\n")')
         rStrategus = importr('Strategus')
         rStrategus.execute(connectionDetails = rConnectionDetails, analysisSpecifications = rSpec, executionSettings = rExecutionSettings)
@@ -1307,7 +1322,7 @@ def upload_strategus_results(analysisSpec: str, path_to_results, dbSettings):
         try:
             ro.r(set_trex_env_var(USE_TREX_CONNECTION))
             database_code = dbSettings['database_code']
-            results_schema = f'results_{validate_token_study_code(dbSettings["token_study_code"])}'
+            results_schema = f'{database_code}.results_{validate_token_study_code(dbSettings["token_study_code"])}'
             rStrategus = importr('Strategus')
             rParallelLogger = importr('ParallelLogger')
             rDatabaseConnector = importr('DatabaseConnector')
@@ -1353,6 +1368,8 @@ def upload_strategus_results(analysisSpec: str, path_to_results, dbSettings):
                 );
                 """
                 dbdao.execute_sql(create_table_sql)
+            else:
+                print(f'Schema {results_schema} already exists, skipping creation of results datamodel')
 
             # uploadResults logs are not captured by default
             # so we override the consolewrite_print callback to capture the logs
@@ -1431,8 +1448,8 @@ def getRCdmExecutionSettings(settings) -> str:
             rCohortGenerator = importr('CohortGenerator')
 
             rExecutionSettings = rStrategus.createCdmExecutionSettings(
-                workDatabaseSchema = settings['schemaName'],
-                cdmDatabaseSchema = settings['schemaName'],
+                workDatabaseSchema = f"{settings['dbName']}.{settings['schemaName']}",
+                cdmDatabaseSchema = f"{settings['dbName']}.{settings['schemaName']}",
                 cohortTableNames = rCohortGenerator.getCohortTableNames(cohortTable = "cohort"),
                 workFolder = settings['workFolder'],
                 resultsFolder = settings['resultsFolder'],
diff --git a/plugins/flows/hades/strategus_plugin/patch_treatment_patterns.R b/plugins/flows/hades/strategus_plugin/patch_treatment_patterns.R
@@ -0,0 +1,73 @@
+local({
+  ns <- getNamespace("TreatmentPatterns")
+  cls <- get("CDMInterface", envir = ns)
+
+  patched <- function(cohorts, cohortTableName, andromeda, andromedaTableName, minEraDuration) {
+    targetCohortId <- getCohortIds(cohorts, "target")
+    n <- unlist(lapply(cohortTableName, function(cohortTable) {
+      as.numeric(unlist(DatabaseConnector::renderTranslateQuerySql(
+        connection = private$.connection,
+        sql = "\n        SELECT COUNT(*)\n        FROM @resultSchema.@cohortTable\n        WHERE cohort_definition_id IN (@cohortIds)\n        GROUP BY subject_id;",
+        resultSchema = private$.resultSchema, cohortTable = cohortTable, cohortIds = cohorts$cohortId
+      )))
+    }))
+    private$dbAppendAttrition(n, andromeda, sort(cohorts$cohortId))
+
+    sql <- lapply(cohortTableName, function(tableName) {
+      SqlRender::loadRenderTranslateSql(
+        sqlFilename = "selectData.sql", packageName = "TreatmentPatterns",
+        dbms = private$.connection@dbms, tempEmulationSchema = private$.tempEmulationSchema,
+        resultSchema = private$.resultSchema, cdmSchema = private$.cdmSchema,
+        cohortTable = tableName, cohortIds = cohorts$cohortId, minEraDuration = minEraDuration
+      )
+    })
+
+    # FIX: loadRenderTranslateSql appends a trailing semicolon. Embedding it directly
+    # into sprintf("FROM (\n  %s\n) a") places the semicolon inside the subquery, which
+    # causes splitSql to split mid-statement producing an incomplete CREATE TEMP TABLE.
+    sql <- lapply(sql, function(s) trimws(sub(";[[:space:]]*$", "", s)))
+
+    renderedSql <- paste(sql, collapse = "\nUNION ALL\n")
+    DatabaseConnector::renderTranslateExecuteSql(
+      connection = private$.connection, oracleTempSchema = private$.tempEmulationSchema,
+      sql = sprintf("DROP TABLE IF EXISTS #tp_dbc_cohort_table;\n\n          SELECT *\n          INTO #tp_dbc_cohort_table\n          FROM (\n            %s\n          ) a;", renderedSql),
+      tempEmulationSchema = private$.tempEmulationSchema
+    )
+
+    DatabaseConnector::renderTranslateQuerySqlToAndromeda(
+      connection = private$.connection, andromeda = andromeda,
+      andromedaTableName = andromedaTableName, tempEmulationSchema = private$.tempEmulationSchema,
+      sql = "\n        SELECT\n          #tp_dbc_cohort_table.cohort_definition_id AS cohort_definition_id,\n          #tp_dbc_cohort_table.subject_id AS subject_id,\n          #tp_dbc_cohort_table.cohort_start_date,\n          #tp_dbc_cohort_table.cohort_end_date,\n          #tp_dbc_cohort_table.age,\n          #tp_dbc_cohort_table.sex,\n          #tp_dbc_cohort_table.subject_id_origin\n        FROM #tp_dbc_cohort_table\n        INNER JOIN (\n          SELECT #tp_dbc_cohort_table.subject_id\n          FROM #tp_dbc_cohort_table\n          WHERE #tp_dbc_cohort_table.cohort_definition_id IN (@targetCohortId)\n        ) AS cross_sec\n        ON cross_sec.subject_id = #tp_dbc_cohort_table.subject_id",
+      targetCohortId = targetCohortId
+    )
+
+    names(andromeda[[andromedaTableName]]) <- tolower(names(andromeda[[andromedaTableName]]))
+    if (inherits(private$.connection, "DatabaseConnectorJdbcConnection") &&
+        private$.connection@dbms == "postgresql") {
+      andromeda[[andromedaTableName]] <- andromeda[[andromedaTableName]] %>%
+        dplyr::mutate(
+          cohort_start_date = dplyr::sql("datediff('day', DATE '1970-01-01', cohort_start_date)"),
+          cohort_end_date   = dplyr::sql("datediff('day', DATE '1970-01-01', cohort_end_date)")
+        )
+    } else {
+      andromeda[[andromedaTableName]] <- andromeda[[andromedaTableName]] %>%
+        dplyr::mutate(
+          cohort_start_date = as.integer(.data$cohort_start_date),
+          cohort_end_date   = as.integer(.data$cohort_end_date)
+        )
+    }
+
+    n <- andromeda[[andromedaTableName]] %>%
+      dplyr::group_by(.data$subject_id) %>% dplyr::summarise(n = dplyr::n()) %>% dplyr::pull()
+    appendAttrition(
+      toAdd = data.frame(number_records = sum(n), number_subjects = length(n), reason_id = 2,
+        reason = sprintf("Removing records < minEraDuration (%s)", minEraDuration),
+        time_stamp = as.numeric(Sys.time())),
+      andromeda = andromeda
+    )
+    return(andromeda)
+  }
+
+  cls$set("private", "dbconFetchCohortTable", patched, overwrite = TRUE)
+  message("Patched TreatmentPatterns::CDMInterface$dbconFetchCohortTable (trailing semicolon fix)")
+})
diff --git a/plugins/functions/alp-dataflow-gen-init/src/env.ts b/plugins/functions/alp-dataflow-gen-init/src/env.ts
@@ -42,6 +42,7 @@ export const env = {
     trex_sql_port: _env.TREX__SQL__PORT,
     trex_sql_dbname: _env.TREX__SQL__DBNAME,
     trex_sql_user: _env.TREX__SQL__USER,
+    trex_strategus_results_db_name: _env.TREX__STRATEGUS_RESULTS_DB_NAME || "strategus_results",
     is_dev_env: _env.PREFECT__LOCAL_DEBUG === "true" || false, // Default to false if not set
     logs_debug_enable: _env.PREFECT__LOGS_DEBUG_ENABLED === "true" || false, // Default to false if not set
     fhir_database_code: _env.FHIR_DATABASE_CODE,
diff --git a/plugins/functions/jobplugins/src/services/PrefectService.ts b/plugins/functions/jobplugins/src/services/PrefectService.ts
@@ -240,7 +240,7 @@ export class PrefectService {
           {},
           {
             mode: "upload-results-from-storage",
-            databaseCode,
+            databaseCode: env.TREX__STRATEGUS_RESULTS_DB_NAME,
             cacheId,
             tokenStudyCode,
             datasetId,
@@ -301,7 +301,7 @@ export class PrefectService {
           {},
           {
             mode: "drop-results",
-            databaseCode,
+            databaseCode: env.TREX__STRATEGUS_RESULTS_DB_NAME,
             cacheId,
             tokenStudyCode,
             datasetId,
diff --git a/plugins/functions/package.json b/plugins/functions/package.json
@@ -589,6 +589,7 @@
           "TREX__SQL__PORT": "${TREX__SQL__PORT:-5433}",
           "TREX__SQL__USER": "${TREX__SQL__USER:-postgres}",
           "TREX__SQL__DBNAME": "${TREX__SQL__DBNAME:-postgres}",
+          "TREX__STRATEGUS_RESULTS_DB_NAME": "${TREX__STRATEGUS_RESULTS_DB_NAME:-strategus_results}",
           "INSTALL_SQLALCHEMY": "${INSTALL_SQLALCHEMY:-}",
           "REFRESH_TOKEN__ENDPOINT": "${REFRESH_TOKEN__ENDPOINT}",
           "REFRESH_TOKEN__CLIENT_ID": "${REFRESH_TOKEN__CLIENT_ID}",
diff --git a/plugins/functions/strategus-analysis/src/analysis/services.ts b/plugins/functions/strategus-analysis/src/analysis/services.ts
diff --git a/tests/backend_integration_tests/pg/specs/configs/strategus_analysis_json/treatment_pathways.json b/tests/backend_integration_tests/pg/specs/configs/strategus_analysis_json/treatment_pathways.json
diff --git a/tests/backend_integration_tests/pg/specs/strategus_analysis_test.js b/tests/backend_integration_tests/pg/specs/strategus_analysis_test.js
