Data warehouse and Quickbase connection tests

Why these changes are being introduced:

This application requires connectivity to the Data
Warehouse and Quickbase.  A dedicated CLI command for testing
those connections is helpful for debugging environments in Dev,
Stage and Prod.

How this addresses that need:
* Adds test_connection() method on DWClient and QBClient classes
* Adds CLI command test-connections that runs this method and logs
output

Side effects of this change:
* None

Relevant ticket(s):
* https://mitlibraries.atlassian.net/browse/HRQB-18

Makefile

@@ -89,13 +89,3 @@ publish-stage: ## Only use in an emergency
 	docker login -u AWS -p $$(aws ecr get-login-password --region us-east-1) $(ECR_URL_STAGE)
 	docker push $(ECR_URL_STAGE):latest
 	docker push $(ECR_URL_STAGE):`git describe --always`
-
-
-## ---- Temporary Development Commands ---- ##
-
-docker-build: # build Docker container
-	docker build --platform linux/amd64 \
-	    -t hrqb-client:latest .
-
-docker-bash: # bash shell to docker container
-	docker run --entrypoint /bin/bash -it hrqb-client

hrqb/cli.py

@@ -8,6 +8,8 @@
 from hrqb.config import Config, configure_logger, configure_sentry
 from hrqb.tasks.pipelines import run_pipeline
 from hrqb.utils import click_argument_to_dict
+from hrqb.utils.data_warehouse import DWClient
+from hrqb.utils.quickbase import QBClient
 
 logger = logging.getLogger(__name__)
 
@@ -44,6 +46,41 @@ def ping(ctx: click.Context) -> None:
     )
 
 
+@main.command()
+@click.pass_context
+def test_connections(ctx: click.Context) -> None:
+    """Test connectivity with Data Warehouse and Quickbase."""
+    dw_success = False
+    qb_success = False
+
+    try:
+        if DWClient().test_connection():
+            logger.debug("Data Warehouse connection successful")
+            dw_success = True
+    except Exception as exc:  # noqa: BLE001
+        message = f"Data Warehouse connection failed: {exc}"
+        logger.error(message)  # noqa: TRY400
+    try:
+        if QBClient().test_connection():
+            logger.debug("Quickbase connection successful")
+            qb_success = True
+    except Exception as exc:  # noqa: BLE001
+        message = f"Quickbase connection failed: {exc}"
+        logger.error(message)  # noqa: TRY400
+
+    if dw_success and qb_success:
+        logger.info("All connections OK")
+    else:
+        logger.info("One or more connections failed")
+
+    logger.info(
+        "Total elapsed: %s",
+        str(
+            timedelta(seconds=perf_counter() - ctx.obj["START_TIME"]),
+        ),
+    )
+
+
 @click.group()
 @click.option(
     "-p",

hrqb/utils/data_warehouse.py

@@ -1,5 +1,7 @@
 """hrqb.utils.data_warehouse"""
 
+# ruff: noqa: PLR2004
+
 import logging
 
 import pandas as pd
@@ -45,6 +47,19 @@ def verify_connection_string_set(self) -> None:
             )
             raise AttributeError(message)
 
+    def test_connection(self) -> bool:
+        """Test connection to Data Warehouse.
+
+        Executes a simple SQL statement that should return a simple dataframe if a valid
+        connection is established.
+        """
+        if self.connection_string.startswith("sqlite"):
+            query = """select 1 as x, 2 as y"""
+        else:
+            query = """select 1 as x, 2 as y from dual"""
+        self.execute_query(query)
+        return True
+
     def init_engine(self) -> None:
         """Instantiate a SQLAlchemy engine if not already configured and set."""
         self.verify_connection_string_set()

hrqb/utils/quickbase.py

@@ -23,6 +23,18 @@ class QBClient:
     cache_results: bool = field(default=True)
     _cache: dict = field(factory=dict, repr=False)
 
+    def test_connection(self) -> bool:
+        """Test connection to Quickbase API.
+
+        Retrieves Quickbase app information and checks for known key to establish
+        successful authentication and app permissions.
+        """
+        results = self.get_app_info()
+        if results.get("id") != Config().QUICKBASE_APP_ID:
+            message = f"API returned unexpected response: {results}"
+            raise ValueError(message)
+        return True
+
     @property
     def request_headers(self) -> dict:
         return {

tests/conftest.py

@@ -27,7 +27,9 @@
 
 
 @pytest.fixture(autouse=True)
-def _test_env(monkeypatch, targets_directory, data_warehouse_connection_string):
+def _test_env(request, monkeypatch, targets_directory, data_warehouse_connection_string):
+    if request.node.get_closest_marker("integration"):
+        return
     monkeypatch.setenv("SENTRY_DSN", "None")
     monkeypatch.setenv("WORKSPACE", "test")
     monkeypatch.setenv("LUIGI_CONFIG_PATH", "hrqb/luigi.cfg")
@@ -205,9 +207,12 @@ def qbclient():
 
 
 @pytest.fixture(scope="session", autouse=True)
-def global_requests_mock():
-    with requests_mock.Mocker() as m:
-        yield m
+def global_requests_mock(request):
+    if any(item.get_closest_marker("integration") for item in request.node.items):
+        yield
+    else:
+        with requests_mock.Mocker() as m:
+            yield m
 
 
 @pytest.fixture
@@ -372,3 +377,31 @@ def data_warehouse_connection_string():
 @pytest.fixture
 def sqlite_dwclient():
     return DWClient(connection_string="sqlite:///:memory:", engine_parameters={})
+
+
+@pytest.fixture
+def _qbclient_connection_test_success():
+    with mock.patch.object(QBClient, "test_connection") as mocked_connection_test:
+        mocked_connection_test.return_value = True
+        yield
+
+
+@pytest.fixture
+def _dwclient_connection_test_success():
+    with mock.patch.object(DWClient, "test_connection") as mocked_connection_test:
+        mocked_connection_test.return_value = True
+        yield
+
+
+@pytest.fixture
+def _dwclient_connection_test_raise_exception():
+    with mock.patch.object(DWClient, "test_connection") as mocked_connection_test:
+        mocked_connection_test.side_effect = Exception("Intentional Error Here")
+        yield
+
+
+@pytest.fixture
+def _qbclient_connection_test_raise_exception():
+    with mock.patch.object(QBClient, "test_connection") as mocked_connection_test:
+        mocked_connection_test.side_effect = Exception("Intentional Error Here")
+        yield

tests/fixtures/qb_api_responses/getApp.json

@@ -3,7 +3,7 @@
   "dateFormat": "MM-DD-YYYY",
   "description": "My testing app",
   "hasEveryoneOnTheInternet": true,
-  "id": "bpqe82s1",
+  "id": "qb-app-def456",
   "name": "Testing App",
   "securityProperties": {
     "allowClone": false,

tests/test_cli.py

@@ -1,3 +1,4 @@
+import pytest
 from _pytest.logging import LogCaptureFixture
 from click.testing import Result
 
@@ -134,3 +135,54 @@ def test_cli_pipeline_run_and_remove_data_success(caplog, runner):
     ]
     for line in lines:
         assert text_in_logs_or_stdout(line, caplog, result)
+
+
+@pytest.mark.usefixtures(
+    "_dwclient_connection_test_success", "_qbclient_connection_test_success"
+)
+def test_cli_test_connections_success(caplog, runner):
+    caplog.set_level("DEBUG")
+    args = ["--verbose", "test-connections"]
+    result = runner.invoke(cli.main, args)
+    assert result.exit_code == OKAY_RESULT_CODE
+    lines = [
+        "Data Warehouse connection successful",
+        "Quickbase connection successful",
+        "All connections OK",
+    ]
+    for line in lines:
+        assert text_in_logs_or_stdout(line, caplog, result)
+
+
+@pytest.mark.usefixtures(
+    "_dwclient_connection_test_raise_exception", "_qbclient_connection_test_success"
+)
+def test_cli_test_connections_data_warehouse_connection_error(caplog, runner):
+    caplog.set_level("DEBUG")
+    args = ["--verbose", "test-connections"]
+    result = runner.invoke(cli.main, args)
+    assert result.exit_code == OKAY_RESULT_CODE
+    lines = [
+        "Data Warehouse connection failed: Intentional Error Here",
+        "Quickbase connection successful",
+        " One or more connections failed",
+    ]
+    for line in lines:
+        assert text_in_logs_or_stdout(line, caplog, result)
+
+
+@pytest.mark.usefixtures(
+    "_dwclient_connection_test_success", "_qbclient_connection_test_raise_exception"
+)
+def test_cli_test_connections_quickbase_connection_error(caplog, runner):
+    caplog.set_level("DEBUG")
+    args = ["--verbose", "test-connections"]
+    result = runner.invoke(cli.main, args)
+    assert result.exit_code == OKAY_RESULT_CODE
+    lines = [
+        "Data Warehouse connection successful",
+        "Quickbase connection failed: Intentional Error Here",
+        "One or more connections failed",
+    ]
+    for line in lines:
+        assert text_in_logs_or_stdout(line, caplog, result)

tests/test_connections.py

@@ -0,0 +1,32 @@
+"""tests.oracle.test_connections
+
+These integration tests will bypass monkey-patched env vars and require real values
+for the following env vars:
+    - QUICKBASE_API_URL
+    - QUICKBASE_API_TOKEN
+    - QUICKBASE_APP_ID
+    - DATA_WAREHOUSE_CONNECTION_STRING
+"""
+
+import oracledb
+import pytest
+
+from hrqb.utils.data_warehouse import DWClient
+from hrqb.utils.quickbase import QBClient
+
+
+@pytest.mark.integration()
+def test_integration_oracle_client_installed_success():
+    oracledb.init_oracle_client()
+
+
+@pytest.mark.integration()
+def test_integration_data_warehouse_connection_success():
+    dwclient = DWClient()
+    assert dwclient.test_connection()
+
+
+@pytest.mark.integration()
+def test_integration_quickbase_api_connection_success():
+    qbclient = QBClient()
+    assert qbclient.test_connection()

tests/test_data_warehouse_client.py

@@ -3,6 +3,7 @@
 import pandas as pd
 import pytest
 from sqlalchemy.engine import Engine
+from sqlalchemy.exc import ArgumentError
 
 from hrqb.utils.data_warehouse import DWClient
 
@@ -72,3 +73,13 @@ def test_dwclient_execute_query_accepts_sql_parameters(sqlite_dwclient):
     )
     assert df.iloc[0].foo == foo_val
     assert df.iloc[0].bar == bar_val
+
+
+def test_dwclient_test_connection_success(sqlite_dwclient):
+    assert sqlite_dwclient.test_connection()
+
+
+def test_dwclient_test_connection_connection_error(sqlite_dwclient):
+    sqlite_dwclient.connection_string = "bad-connection://nothing"
+    with pytest.raises(ArgumentError, match="Could not parse SQLAlchemy URL from string"):
+        sqlite_dwclient.test_connection()

tests/test_qbclient_client.py

@@ -6,6 +6,7 @@
 import pandas as pd
 import pytest
 import requests
+from requests.exceptions import HTTPError
 from requests.models import Response
 
 from hrqb.exceptions import QBFieldNotFoundError
@@ -175,3 +176,22 @@ def test_qbclient_api_non_2xx_response_error(qbclient):
             requests.RequestException, match="Quickbase API error - status 400"
         ):
             assert qbclient.make_request(requests.get, "/always/fail")
+
+
+def test_qbclient_test_connection_success(qbclient):
+    assert qbclient.test_connection()
+
+
+def test_qbclient_test_connection_connection_error(qbclient):
+    error_message = "Invalid API token"
+    with mock.patch.object(type(qbclient), "make_request") as mocked_make_request:
+        mocked_make_request.side_effect = HTTPError(error_message)
+        with pytest.raises(HTTPError, match=error_message):
+            qbclient.test_connection()
+
+
+def test_qbclient_test_connection_response_error(qbclient):
+    with mock.patch.object(type(qbclient), "make_request") as mocked_make_request:
+        mocked_make_request.return_value = {"msg": "this is not expected"}
+        with pytest.raises(ValueError, match="API returned unexpected response"):
+            qbclient.test_connection()