DM-46397: Extend Qserv integration test to use HTTP-based push mode catalog ingest

.github/workflows/ci.yml

@@ -324,7 +324,7 @@ jobs:
 
       - name: Run integration tests of the HTTP frontend
         run: |
-          ./admin/local/cli/qserv --log-level DEBUG itest-http \
+          ./admin/local/cli/qserv --log-level DEBUG itest-http --reload --load-http \
           --qserv-image ${{ needs.image-names.outputs.qserv-image }} \
           --mariadb-image ${{ needs.image-names.outputs.mariadb-image }}
 

admin/tools/docker/base/Dockerfile

@@ -171,7 +171,8 @@ RUN pip3 install \
     sqlalchemy \
     sqlalchemy-stubs \
     types-PyYAML \
-    types-requests
+    types-requests \
+    requests_toolbelt
 
 RUN useradd --create-home --uid 1000 --shell /bin/bash qserv
 
@@ -282,4 +283,5 @@ RUN pip3 install \
     "mysql-connector-python==8.0.31" \
     pyyaml \
     requests \
+    requests_toolbelt \
     sqlalchemy

src/admin/python/lsst/qserv/admin/cli/_integration_test.py

@@ -114,6 +114,7 @@ def run_integration_tests(
     unload: bool,
     load: Optional[bool],
     reload: bool,
+    load_http: bool,
     cases: List[str],
     run_tests: bool,
     tests_yaml: str,
@@ -133,6 +134,8 @@ def run_integration_tests(
         database matches the qserv database.)
     reload : `bool`
         Remove test databases and re-add them.
+    load_http : `bool`
+        The protocol to use for loading the data.
     cases : `list` [`str`]
         Run (and load/reload data if those flags are set) these test cases only.
     run_tests : `bool`
@@ -197,6 +200,7 @@ def run_integration_tests(
             admin_auth_key=tests_data["repl-admin-auth-key"],
             cases=cases,
             load=load,
+            load_http=load_http,
         )
 
     if run_tests:
@@ -225,6 +229,7 @@ def run_integration_tests_http(
     unload: bool,
     load: Optional[bool],
     reload: bool,
+    load_http: bool,
     cases: List[str],
     run_tests: bool,
     tests_yaml: str,
@@ -244,6 +249,8 @@ def run_integration_tests_http(
         database matches the qserv database.)
     reload : `bool`
         Remove test databases and re-add them.
+    load_http : `bool`
+        The protocol to use for loading the data.
     cases : `list` [`str`]
         Run (and load/reload data if those flags are set) these test cases only.
     run_tests : `bool`
@@ -308,6 +315,7 @@ def run_integration_tests_http(
             admin_auth_key=tests_data["repl-admin-auth-key"],
             cases=cases,
             load=load,
+            load_http=load_http,
         )
 
     if run_tests:

src/admin/python/lsst/qserv/admin/cli/entrypoint.py

@@ -51,6 +51,7 @@
     option_options_file,
     option_mysql_monitor_password,
     option_reload,
+    option_load_http,
     option_repl_auth_key,
     option_repl_admin_auth_key,
     option_repl_connection,
@@ -326,12 +327,13 @@ def entrypoint(log_level: str) -> None:
 @entrypoint.command()
 @option_repl_auth_key()
 @click.argument("repl_ctrl_uri")
-def load_simple(repl_ctrl_uri: str, repl_auth_key: str) -> None:
+@option_load_http()
+def load_simple(repl_ctrl_uri: str, repl_auth_key: str, load_http: bool) -> None:
     """Load a small test dataset into qserv.
 
     REPL_CTRL_URI is the uri to the replication controller.
     """
-    script.load_simple(repl_ctrl_uri, auth_key=repl_auth_key)
+    script.load_simple(repl_ctrl_uri, auth_key=repl_auth_key, load_http=load_http)
 
 
 @entrypoint.command()
@@ -342,6 +344,7 @@ def load_simple(repl_ctrl_uri: str, repl_auth_key: str) -> None:
 @option_unload()
 @option_load()
 @option_reload()
+@option_load_http()
 @option_case()
 @option_run_tests()
 @option_tests_yaml()
@@ -351,6 +354,7 @@ def integration_test(
     unload: bool,
     load: Optional[bool],
     reload: bool,
+    load_http: bool,
     cases: List[str],
     run_tests: bool,
     tests_yaml: str,
@@ -366,6 +370,7 @@ def integration_test(
         unload=unload,
         load=load,
         reload=reload,
+        load_http=load_http,
         cases=cases,
         run_tests=run_tests,
         tests_yaml=tests_yaml,
@@ -383,6 +388,7 @@ def integration_test(
 @option_unload()
 @option_load()
 @option_reload()
+@option_load_http()
 @option_case()
 @option_run_tests()
 @option_tests_yaml()
@@ -392,6 +398,7 @@ def integration_test_http(
     unload: bool,
     load: Optional[bool],
     reload: bool,
+    load_http: bool,
     cases: List[str],
     run_tests: bool,
     tests_yaml: str,
@@ -407,6 +414,7 @@ def integration_test_http(
         unload=unload,
         load=load,
         reload=reload,
+        load_http=load_http,
         cases=cases,
         run_tests=run_tests,
         tests_yaml=tests_yaml,

src/admin/python/lsst/qserv/admin/cli/options.py

@@ -266,6 +266,12 @@ def __call__(self, f: Callable) -> Callable:
     is_flag=True,
 )
 
+option_load_http = partial(
+    click.option,
+    "--load-http",
+    help="HTTP-based table loading protocol. Used with --load and --reload",
+    is_flag=True,
+)
 
 option_run_tests = partial(
     click.option,

src/admin/python/lsst/qserv/admin/cli/script.py

@@ -962,7 +962,7 @@ def delete_database(
     repl.delete_database(database, admin)
 
 
-def load_simple(repl_ctrl_uri: str, auth_key: str) -> None:
+def load_simple(repl_ctrl_uri: str, auth_key: str, load_http: bool) -> None:
     """Load a simple predefined database into qserv.
 
     The database is called "test101" and have a table called Object with one row.
@@ -973,6 +973,8 @@ def load_simple(repl_ctrl_uri: str, auth_key: str) -> None:
         The uri to the replication controller service.
     auth_key : `str`
         The authorizaiton key for the replication-ingest system.
+    load_http : `bool`
+        If true, the database will be loaded using the http interface.
     """
     repl = ReplicationInterface(repl_ctrl_uri, auth_key)
 
@@ -1025,10 +1027,15 @@ def load_simple(repl_ctrl_uri: str, auth_key: str) -> None:
     chunk_location = repl.ingest_chunk_config(transaction_id, "0")
     repl.ingest_data_file(
         transaction_id,
+        "0",
+        False,
         chunk_location.host,
         chunk_location.port,
+        chunk_location.http_host,
+        chunk_location.http_port,
         data_file=data_file,
         table=table,
+        load_http=load_http,
     )
     repl.commit_transaction(transaction_id)
     repl.publish_database(database)
@@ -1039,6 +1046,7 @@ def integration_test(
     unload: bool,
     load: Optional[bool],
     reload: bool,
+    load_http: bool,
     cases: List[str],
     run_tests: bool,
     tests_yaml: str,
@@ -1051,6 +1059,7 @@ def integration_test(
         unload=unload,
         load=load,
         reload=reload,
+        load_http=load_http,
         cases=cases,
         run_tests=run_tests,
         tests_yaml=tests_yaml,
@@ -1064,6 +1073,7 @@ def integration_test_http(
     unload: bool,
     load: Optional[bool],
     reload: bool,
+    load_http: bool,
     cases: List[str],
     run_tests: bool,
     tests_yaml: str,
@@ -1076,6 +1086,7 @@ def integration_test_http(
         unload=unload,
         load=load,
         reload=reload,
+        load_http=load_http,
         cases=cases,
         run_tests=run_tests,
         tests_yaml=tests_yaml,

src/admin/python/lsst/qserv/admin/itest_load.py

@@ -338,6 +338,7 @@ def _load_database(
     repl_ctrl_uri: str,
     auth_key: str,
     admin_auth_key: str,
+    load_http: bool,
 ) -> None:
     """Load a database.
 
@@ -355,6 +356,8 @@ def _load_database(
         The authorizaiton key for the replication-ingest system.
     admin_auth_key : `str`
         The admin authorizaiton key for the replication-ingest system.
+    load_http : `bool`, optional
+        The protocol to use for loading the data.
     """
     _log.info(f"Loading database %s for test %s auth_key %s admin_auth_key %s", load_db.name, load_db.id, auth_key, admin_auth_key)
     repl = ReplicationInterface(repl_ctrl_uri, auth_key, admin_auth_key)
@@ -404,12 +407,14 @@ def do_ingest_table_config() -> None:
                         table=table,
                         chunks_folder=staging_dir,
                         chunk_info_file=os.path.join(staging_dir, chunk_info_file),
+                        load_http=load_http,
                     )
                 else:
                     repl.ingest_table_data(
                         transaction_id=transaction_id,
                         table=table,
                         data_file=data_file,
+                        load_http=load_http,
                     )
                 repl.commit_transaction(transaction_id)
 
@@ -528,6 +533,7 @@ def load(
     test_cases_data: List[Dict[Any, Any]],
     ref_db_admin: str,
     load: Optional[bool],
+    load_http: bool,
     cases: Optional[List[str]],
     auth_key: str,
     admin_auth_key: str,
@@ -550,6 +556,8 @@ def load(
         unload == True then will not load the database, otherwise if `None` will
         load the database if it is not yet loaded into qserv (assumes the ref
         database matches the qserv database.)
+    load_http : `bool`
+        The protocol to use for loading the data.
     cases : `list` [`str`], optional
         Restrict loading to these test cases if provided.
     auth_key : `str`
@@ -570,7 +578,7 @@ def load(
     for case_data in cases_data:
         load_db = LoadDb(case_data)
         if load == True or (load is None and load_db.name not in qserv_dbs):
-            _load_database(load_db, ref_db_uri, ref_db_admin, repl_ctrl_uri, auth_key, admin_auth_key)
+            _load_database(load_db, ref_db_uri, ref_db_admin, repl_ctrl_uri, auth_key, admin_auth_key, load_http=load_http)
 
 
 def remove(

src/admin/python/lsst/qserv/admin/qservCli/launch.py

@@ -930,6 +930,7 @@ def integration_test(
     unload: bool,
     load: Optional[bool],
     reload: bool,
+    load_http: bool,
     cases: List[str],
     run_tests: bool,
     tests_yaml: str,
@@ -970,6 +971,8 @@ def integration_test(
         will load test databases that are not loaded yet.
     reload : bool
         Remove and reload test data. Same as passing `unload=True` and `load=True`.
+    load_http : bool
+        Table loading protocol. If True, use the HTTP protocol to load tables.
     cases : List[str]
         Run this/these test cases only. If list is empty list will run all the cases.
     run_tests : bool
@@ -1019,6 +1022,7 @@ def integration_test(
     for opt, var in (
         ("--unload", unload),
         ("--reload", reload),
+        ("--load-http", load_http),
     ):
         if var:
             args.append(opt)
@@ -1060,6 +1064,7 @@ def integration_test_http(
     unload: bool,
     load: Optional[bool],
     reload: bool,
+    load_http: bool,
     cases: List[str],
     run_tests: bool,
     tests_yaml: str,
@@ -1091,6 +1096,8 @@ def integration_test_http(
         If True do not run the command; print what would have been run.
     project : `str`
         The name used for qserv instance customizations.
+    load_http : bool
+        Table loading protocol. If True, use the HTTP protocol to load tables.
     cases : List[str]
         Run this/these test cases only. If list is empty list will run all the cases.
     run_tests : bool
@@ -1140,6 +1147,7 @@ def integration_test_http(
     for opt, var in (
         ("--unload", unload),
         ("--reload", reload),
+        ("--load-http", load_http),
     ):
         if var:
             args.append(opt)
@@ -1182,6 +1190,7 @@ def itest(
     unload: bool,
     load: Optional[bool],
     reload: bool,
+    load_http: bool,
     cases: List[str],
     run_tests: bool,
     tests_yaml: str,
@@ -1226,6 +1235,7 @@ def itest(
             unload,
             load,
             reload,
+            load_http,
             cases,
             run_tests,
             tests_yaml,
@@ -1251,6 +1261,7 @@ def itest_http(
     unload: bool,
     load: Optional[bool],
     reload: bool,
+    load_http: bool,
     cases: List[str],
     run_tests: bool,
     tests_yaml: str,
@@ -1284,6 +1295,7 @@ def itest_http(
             unload,
             load,
             reload,
+            load_http,
             cases,
             run_tests,
             tests_yaml,