Delete tars after --non-blocking completes

conda/dev.yml

@@ -8,8 +8,7 @@ dependencies:
   - pip=22.2.2
   - python=3.9.13
   - six=1.16.0
-  - globus-sdk=3.2.1
-  - fair-research-login=0.2.6
+  - globus-sdk=3.15.0
   # Developer Tools
   # =================
   # If versions are updated, also update 'rev' in `.pre-commit.config.yaml`

tests/scripts/globus_auth.bash

@@ -0,0 +1,125 @@
+setup()
+{
+  echo "##########################################################################################################"
+  local case_name="${1}"
+  local src_dir="${2}"
+  echo "Testing: ${case_name}"
+  full_dir="${src_dir}/${case_name}"
+  rm -rf ${full_dir}
+  mkdir -p ${full_dir}
+  cd ${full_dir}
+
+  mkdir zstash_demo
+  mkdir zstash_demo/empty_dir
+  mkdir zstash_demo/dir
+  echo -n '' > zstash_demo/file_empty.txt
+  echo 'file0 stuff' > zstash_demo/dir/file0.txt
+}
+
+check_log_has()
+{
+    local expected_grep="${1}"
+    local log_file="${2}"
+    grep "${expected_grep}" ${log_file}
+    if [ $? != 0 ]; then
+        echo "Expected grep '${expected_grep}' not found in ${log_file}. Test failed."
+        exit 2
+    fi
+}
+
+check_log_does_not_have()
+{
+    local not_expected_grep="${1}"
+    local log_file="${2}"
+    grep "${not_expected_grep}" ${log_file}
+    if [ $? == 0 ]; then
+        echo "Not-expected grep '${expected_grep}' was found in ${log_file}. Test failed."
+        exit 2
+    fi
+}
+
+run_test_cases()
+{
+    # This script requires user input and thus cannot be run automatically as part of a test suite.
+
+    # To start fresh with Globus:
+    # 1. Log into endpoints (LCRC Improv DTN, NERSC Perlmutter) at globus.org: File Manager > Add the endpoints in the "Collection" fields
+    # 2. To start fresh, with no consents: https://auth.globus.org/v2/web/consents > Manage Your Consents > Globus Endpoint Performance Monitoring > rescind all"
+
+    # Before each run:
+    # Perlmutter:
+    # cd /global/homes/f/forsyth/zstash/tests/
+    # rm -rf test_globus_auth_try1 # Or just change $DST_DIR to a new directory
+    #
+    # Chrysalis:
+    # cd ~/ez/zstash/
+    # conda activate <env-name>
+    # pre-commit run --all-files
+    # python -m pip install .
+    # cd tests/scripts
+    # ./globus_auth.bash
+
+    PERLMUTTER_ENDPOINT=6bdc7956-fc0f-4ad2-989c-7aa5ee643a79
+
+    SRC_DIR=/lcrc/group/e3sm/ac.forsyth2/zstash_testing/test_globus_auth # Chrysalis
+    DST_DIR=globus://${PERLMUTTER_ENDPOINT}/global/homes/f/forsyth/zstash/tests/test_globus_auth_try5
+
+    GLOBUS_CFG=/home/ac.forsyth2/.globus-native-apps.cfg
+    INI_PATH=/home/ac.forsyth2/.zstash.ini
+    TOKEN_FILE=/home/ac.forsyth2/.zstash_globus_tokens.json
+
+    # Start fresh
+    rm -rf ${GLOBUS_CFG}
+    rm -rf ${INI_PATH}
+    rm -rf ${TOKEN_FILE}
+
+    echo "Running globus_auth test"
+    echo "Exit codes: 0 -- success, 1 -- zstash failed, 2 -- grep check failed"
+
+    case_name="run1"
+    setup ${case_name} "${SRC_DIR}"
+    # Expecting to see exactly 1 authentication prompt
+    zstash create --hpss=${DST_DIR}/${case_name} zstash_demo 2>&1 | tee ${case_name}.log
+    if [ $? != 0 ]; then
+        echo "${case_name} failed. Check ${case_name}_create.log for details. Cannot continue."
+        exit 1
+    fi
+    echo "${case_name} completed successfully. Checking ${case_name}.log now."
+    # From check_state_files
+    check_log_does_not_have "WARNING: Globus CFG ${GLOBUS_CFG} exists. This may be left over from earlier versions of zstash, and may cause issues. Consider deleting." ${case_name}.log
+    check_log_has "INFO: ${INI_PATH} does NOT exist. This means we won't be able to read the local endpoint ID from it." ${case_name}.log
+    check_log_has "INFO: Token file ${TOKEN_FILE} does NOT exist. This means we won't be able to load tokens from it." ${case_name}.log
+    # From get_local_endpoint_id
+    check_log_has "INFO: Writing to empty ${INI_PATH}" ${case_name}.log
+    check_log_has "INFO: Setting local_endpoint_id based on FQDN chrlogin2.lcrc.anl.gov:" ${case_name}.log
+    # From get_transfer_client_with_auth
+    check_log_has "INFO: No stored tokens found - starting authentication" ${case_name}.log
+    check_log_has "Please go to this URL and login:" ${case_name}.log # Our one expected authentication prompt
+    # From save_tokens
+    check_log_has "INFO: Tokens saved successfully" ${case_name}.log
+
+
+    case_name="run2"
+    setup ${case_name} "${SRC_DIR}"
+    # Expecting to see exactly 0 authentication prompts
+    zstash create --hpss=${DST_DIR}/${case_name} zstash_demo 2>&1 | tee ${case_name}.log
+    if [ $? != 0 ]; then
+        echo "${case_name} failed. Check ${case_name}_create.log for details. Cannot continue."
+        exit 1
+    fi
+    echo "${case_name} completed successfully. Checking ${case_name}.log now."
+    # From check_state_files
+    check_log_does_not_have "WARNING: Globus CFG ${GLOBUS_CFG} exists. This may be left over from earlier versions of zstash, and may cause issues. Consider deleting." ${case_name}.log
+    check_log_has "INFO: ${INI_PATH} exists. We can try to read the local endpoint ID from it." ${case_name}.log # Differs from run1
+    check_log_has "INFO: Token file ${TOKEN_FILE} exists. We can try to load tokens from it." ${case_name}.log # Differs from run1
+    # From get_local_endpoint_id
+    check_log_has "INFO: Setting local_endpoint_id based on ${INI_PATH}" ${case_name}.log # Differs from run1
+    check_log_has "INFO: Setting local_endpoint_id based on FQDN chrlogin2.lcrc.anl.gov:" ${case_name}.log
+    # From get_transfer_client_with_auth
+    check_log_has "INFO: Found stored refresh token - using it" ${case_name}.log # Differs from run1
+    check_log_does_not_have "Please go to this URL and login:" ${case_name}.log # There should be no login prompts for run2!
+    # From save_tokens
+    check_log_does_not_have "INFO: Tokens saved successfully" ${case_name}.log # Differs from run1
+}
+
+run_test_cases

zstash/create.py

@@ -6,14 +6,15 @@
 import os.path
 import sqlite3
 import sys
-from typing import Any, List, Tuple
+from typing import Any, List, Optional, Tuple
 
 from six.moves.urllib.parse import urlparse
 
 from .globus import globus_activate, globus_finalize
 from .hpss import hpss_put
 from .hpss_utils import add_files
 from .settings import DEFAULT_CACHE, config, get_db_filename, logger
+from .transfer_tracking import GlobusTransferCollection, HPSSTransferCollection
 from .utils import (
     create_tars_table,
     get_files_to_archive,
@@ -52,12 +53,13 @@ def create():
         logger.error(input_path_error_str)
         raise NotADirectoryError(input_path_error_str)
 
+    gtc: Optional[GlobusTransferCollection] = None
     if hpss != "none":
         url = urlparse(hpss)
         if url.scheme == "globus":
             # identify globus endpoints
-            logger.debug(f"{ts_utc()}:Calling globus_activate(hpss)")
-            globus_activate(hpss)
+            logger.debug(f"{ts_utc()}:Calling globus_activate()")
+            gtc = globus_activate(hpss)
         else:
             # config.hpss is not "none", so we need to
             # create target HPSS directory
@@ -88,14 +90,23 @@ def create():
 
     # Create and set up the database
     logger.debug(f"{ts_utc()}: Calling create_database()")
-    failures: List[str] = create_database(cache, args)
+    htc: HPSSTransferCollection = HPSSTransferCollection()
+    failures: List[str] = create_database(cache, args, gtc=gtc, htc=htc)
 
     # Transfer to HPSS. Always keep a local copy.
     logger.debug(f"{ts_utc()}: calling hpss_put() for {get_db_filename(cache)}")
-    hpss_put(hpss, get_db_filename(cache), cache, keep=args.keep, is_index=True)
+    hpss_put(
+        hpss,
+        get_db_filename(cache),
+        cache,
+        keep=args.keep,
+        is_index=True,
+        gtc=gtc,
+        # htc=htc,
+    )
 
     logger.debug(f"{ts_utc()}: calling globus_finalize()")
-    globus_finalize(non_blocking=args.non_blocking)
+    globus_finalize(gtc, htc, non_blocking=args.non_blocking)
 
     if len(failures) > 0:
         # List the failures
@@ -204,7 +215,12 @@ def setup_create() -> Tuple[str, argparse.Namespace]:
     return cache, args
 
 
-def create_database(cache: str, args: argparse.Namespace) -> List[str]:
+def create_database(
+    cache: str,
+    args: argparse.Namespace,
+    gtc: Optional[GlobusTransferCollection],
+    htc: Optional[HPSSTransferCollection],
+) -> List[str]:
     # Create new database
     logger.debug(f"{ts_utc()}:Creating index database")
     if os.path.exists(get_db_filename(cache)):
@@ -263,26 +279,7 @@ def create_database(cache: str, args: argparse.Namespace) -> List[str]:
     files: List[str] = get_files_to_archive(cache, args.include, args.exclude)
 
     failures: List[str]
-    if args.follow_symlinks:
-        try:
-            # Add files to archive
-            failures = add_files(
-                cur,
-                con,
-                -1,
-                files,
-                cache,
-                args.keep,
-                args.follow_symlinks,
-                skip_tars_md5=args.no_tars_md5,
-                non_blocking=args.non_blocking,
-                error_on_duplicate_tar=args.error_on_duplicate_tar,
-                overwrite_duplicate_tars=args.overwrite_duplicate_tars,
-                force_database_corruption=args.for_developers_force_database_corruption,
-            )
-        except FileNotFoundError:
-            raise Exception("Archive creation failed due to broken symlink.")
-    else:
+    try:
         # Add files to archive
         failures = add_files(
             cur,
@@ -297,7 +294,14 @@ def create_database(cache: str, args: argparse.Namespace) -> List[str]:
             error_on_duplicate_tar=args.error_on_duplicate_tar,
             overwrite_duplicate_tars=args.overwrite_duplicate_tars,
             force_database_corruption=args.for_developers_force_database_corruption,
+            gtc=gtc,
+            htc=htc,
         )
+    except FileNotFoundError as e:
+        if args.follow_symlinks:
+            raise Exception("Archive creation failed due to broken symlink.")
+        else:
+            raise e
 
     # Close database
     con.commit()