Fix timeout handling, add timeout testing

anaconda_anon_usage/heartbeat.py

@@ -27,6 +27,11 @@
 COM_REPO = "https://repo.anaconda.com/pkgs/"
 REPOS = (CLD_REPO, COM_REPO, ORG_REPO)
 HEARTBEAT_PATH = "noarch/activate-0.0.0-0.conda"
+# How long to attempt the connection. Shorter timeouts reduces
+# the likelihood that a slow connection will yield a successful
+# heartbeat, but in that situation, a longer timeout would delay
+# the activation process and compromise user experience
+TIMEOUT = 0.25
 
 
 def _print(msg, *args, standalone=False, error=False):
@@ -45,11 +50,15 @@ def _print(msg, *args, standalone=False, error=False):
 
 def _ping(session, url, wait):
     try:
-        response = session.head(url, proxies=session.proxies)
+        # A short timeout is necessary here so that the activation
+        # is not unduly delayed by a blocked internet connection
+        response = session.head(url, proxies=session.proxies, timeout=TIMEOUT)
         _print("Status code (expect 404): %s", response.status_code)
     except Exception as exc:
         if type(exc).__name__ != "ConnectionError":
-            _print("Heartbeat error: %s", exc, error=True)
+            _print("Unexpected heartbeat error: %s", exc, error=True)
+        elif "timeout=" in str(exc):
+            _print("Timeout exceeded; heartbeat likely not sent.")
 
 
 def attempt_heartbeat(channel=None, path=None, wait=False):
@@ -87,11 +96,13 @@ def attempt_heartbeat(channel=None, path=None, wait=False):
     if DRY_RUN:
         _print("Dry run selected, not sending heartbeat.")
     else:
+        context.remote_max_retries = 1
         session = get_session(url)
-        t = Thread(target=_ping, args=(session, url, wait), daemon=True)
+        # Run in the background so we can proceed with the rest of the
+        # activation tasks while the request fires. The process will wait
+        # to terminate until the thread is complete.
+        t = Thread(target=_ping, args=(session, url, wait), daemon=False)
         t.start()
-        _print("%saiting for response", "W" if wait else "Not w")
-        t.join(timeout=None if wait else 0.1)
     _print(line, standalone=True)
 
 

tests/integration/proxy_tester.py

@@ -274,6 +274,9 @@ def do_CONNECT(self):
             with self.server.lock:
                 cert_file, key_file = read_or_create_cert(host)
 
+            if self.server.delay:
+                time.sleep(self.server.delay)
+
             # Establish tunnel
             self.send_response(200, "Connection Established")
             self._multiline_log(
@@ -315,7 +318,7 @@ def do_CONNECT(self):
                 self._forward_data(client, remote)
 
         except Exception as exc:
-            self._log("CONNECT error: %s", exc, level="exception")
+            self._log("CONNECT error: %s", exc, level="error")
             try:
                 self.send_error(502)
             except Exception:
@@ -394,6 +397,13 @@ def main():
         default=8080,
         help="Port for the proxy server (default: 8080)",
     )
+    parser.add_argument(
+        "--delay",
+        type=float,
+        action="store",
+        default=0,
+        help="Add a delay, in seconds, to each connection request, to test connection issues.",
+    )
     parser.add_argument(
         "--keep-certs",
         action="store_true",
@@ -442,6 +452,7 @@ def cleanup():
 
     # Start and configure server
     server = ThreadingHTTPServer(("0.0.0.0", args.port), ProxyHandler)
+    server.delay = max(0, args.delay)
 
     # Enable interception if any response-related args are provided
     if (

tests/integration/test_heartbeats.py

@@ -20,7 +20,6 @@
 if os.path.isfile("/etc/conda/machine_token"):
     expected += ("m",)
 
-os.environ["ANACONDA_ANON_USAGE_DEBUG"] = "1"
 
 ALL_FIELDS = {"aau", "aid", "c", "s", "e", "u", "h", "n", "m", "o", "U", "H", "N"}
 
@@ -48,11 +47,9 @@ def get_test_envs():
 all_environments = set()
 
 
-def verify_user_agent(output, expected, envname=None, marker=None):
+def verify_user_agent(user_agent, expected, envname=None, marker=None):
     other_tokens["n"] = envname if envname else "base"
 
-    match = re.search(r"^.*User-Agent: (.+)$", output, re.MULTILINE)
-    user_agent = match.groups()[0] if match else ""
     new_values = [t.split("/", 1) for t in user_agent.split(" ") if "/" in t]
     new_values = {k: v for k, v in new_values if k in ALL_FIELDS}
     header = " ".join(f"{k}/{v}" for k, v in new_values.items())
@@ -97,20 +94,20 @@ def verify_user_agent(output, expected, envname=None, marker=None):
 urls = [u for c in context.channels for u in Channel(c).urls()]
 urls.extend(u.rstrip("/") for u in context.channel_alias.urls())
 if any(".anaconda.cloud" in u for u in urls):
-    hb_url = "https://repo.anaconda.cloud/"
+    exp_host = "repo.anaconda.cloud:443"
 elif any(".anaconda.com" in u for u in urls):
-    hb_url = "https://repo.anaconda.com/"
+    exp_host = "repo.anaconda.com:443"
 elif any(".anaconda.org" in u for u in urls):
-    hb_url = "https://conda.anaconda.org/"
+    exp_host = "conda.anaconda.org:443"
 else:
-    hb_url = None
-if hb_url:
-    hb_url += "pkgs/main/noarch/activate-0.0.0-0.conda"
-print("Expected heartbeat url:", hb_url)
-print("Expected user agent tokens:", ",".join(expected))
+    raise RuntimeError("No heartbeat URL available.")
+exp_path = "/pkgs/main/noarch/activate-0.0.0-0.conda"
+print("Expected host:", exp_host)
+print("Expected path:", exp_path)
+print("Expected tokens:", ",".join(expected))
 need_header = True
-for hval in ("true", "false"):
-    os.environ["CONDA_ANACONDA_HEARTBEAT"] = hval
+for hval in ("true", "false", "delay"):
+    os.environ["CONDA_ANACONDA_HEARTBEAT"] = str(hval != "false").lower()
     for envname in envs:
         # Do each one twice to make sure the user agent string
         # remains correct on repeated attempts
@@ -120,8 +117,10 @@ def verify_user_agent(output, expected, envname=None, marker=None):
             # also has the advantage of making sure our code respects proxies properly
             pscript = join(dirname(__file__), "proxy_tester.py")
             # fmt: off
-            cmd = ["python", pscript, "--return-code", "404", "--",
-                   "python", "-m", "conda", "shell." + stype, "activate", envname]
+            cmd = ["python", pscript, "--return-code", "404"]
+            if hval == "delay":
+                cmd.extend(["--delay", "1.0"])
+            cmd.extend(["--", "python", "-m", "conda", "shell." + stype, "activate", envname])
             # fmt: on
             proc = subprocess.run(
                 cmd,
@@ -130,26 +129,22 @@ def verify_user_agent(output, expected, envname=None, marker=None):
                 text=True,
             )
             header = status = ""
-            no_hb_url = "No valid heartbeat channel" in proc.stderr
-            hb_urls = {
-                line.rsplit(" ", 1)[-1]
-                for line in proc.stderr.splitlines()
-                if "Heartbeat url:" in line
-            }
-            status = ""
-            if hval == "true":
-                if not (no_hb_url or hb_urls):
-                    status = "NOT ENABLED"
-                elif hb_url and not hb_urls:
-                    status = "NO HEARTBEAT URL"
-                elif not hb_url and hb_urls:
-                    status = "UNEXPECTED URLS: " + ",".join(hb_urls)
-                elif hb_url and any(hb_url not in u for u in hb_urls):
-                    status = "INCORRECT URLS: " + ",".join(hb_urls)
-            elif hval == "false" and (no_hb_url or hb_urls):
+            t_host = re.search(r"^.* CONNECT (.*) HTTP/1.1$", proc.stdout, re.MULTILINE)
+            t_host = t_host.groups()[0] if t_host else ""
+            t_path = re.search(r"^.* HEAD (.*) HTTP/1.1$", proc.stdout, re.MULTILINE)
+            t_path = t_path.groups()[0] if t_path else ""
+            t_uagent = re.search(r"^  . User-Agent: (.*)", proc.stdout, re.MULTILINE)
+            t_uagent = t_uagent.groups()[0] if t_uagent else ""
+            if hval != "false" and not t_host:
+                status = "NOT ENABLED"
+            elif hval == "false" and t_host:
                 status = "NOT DISABLED"
-            if hb_urls and not status:
-                status, header = verify_user_agent(proc.stdout, expected, envname)
+            elif hval == "delay" and t_path:
+                status = "TIMEOUT FAILED"
+            elif t_host and t_path and (t_host != exp_host or t_path != exp_path):
+                status = f"INCORRECT URL: {t_host}{t_path}"
+            if not status and hval == "true":
+                status, header = verify_user_agent(t_uagent, expected, envname)
             if need_header:
                 if header:
                     print("|", header)