update test case

Author: lusu-msft

sdk/ai/azure-ai-agentserver-core/tests/conftest.py

@@ -8,6 +8,7 @@
 import json
 import logging
 import os
+import socket
 import subprocess
 import sys
 import time
@@ -50,21 +51,32 @@ def __init__(
         sample_name: str,
         script_name: str,
         endpoint: str = "/responses",  # Default endpoint
-        base_url: str = "http://localhost:8088",
+        base_url: Optional[str] = None,
         env_vars: Optional[Dict[str, str]] = None,
         timeout: int = 120,
+        port: Optional[int] = None,
     ):
         self.sample_name = sample_name
         self.script_name = script_name
-        self.base_url = base_url
         self.endpoint = endpoint
         self.timeout = timeout
 
         # Setup paths
         self.project_root = project_root  # Use already defined project_root
-        self.sample_dir = self.project_root / "samples" / "python" / sample_name
+        self.sample_dir = self.project_root / "samples" / sample_name
         self.original_dir = os.getcwd()
 
+        # Determine port assignment priority: explicit param > env override > random
+        if env_vars and env_vars.get("DEFAULT_AD_PORT"):
+            self.port = int(env_vars["DEFAULT_AD_PORT"])
+        elif port is not None:
+            self.port = port
+        else:
+            self.port = self._find_free_port()
+
+        # Configure base URL for client requests
+        self.base_url = (base_url or f"http://127.0.0.1:{self.port}").rstrip("/")
+
         # Setup environment
         # Get Agent Framework configuration (new format)
         azure_ai_project_endpoint = os.getenv("AZURE_AI_PROJECT_ENDPOINT", "")
@@ -118,13 +130,29 @@ def __init__(
         if env_vars:
             self.env_vars.update(env_vars)
 
+        # Ensure server picks the dynamically assigned port and clients know how to reach it
+        self.env_vars.setdefault("DEFAULT_AD_PORT", str(self.port))
+        self.env_vars.setdefault("AGENT_BASE_URL", self.base_url)
+
         self.process = None
         self.session = requests.Session()
 
+    @staticmethod
+    def _find_free_port() -> int:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            sock.bind(("127.0.0.1", 0))
+            return sock.getsockname()[1]
+
     def setup(self):
         """Setup test environment."""
         os.chdir(self.sample_dir)
 
+        logger.info(
+            "Configured %s to listen on %s",
+            self.sample_name,
+            f"{self.base_url}{self.endpoint}",
+        )
+
         # Validate critical environment variables
         # For Agent Framework samples, check new env vars first
         required_vars = []
@@ -175,10 +203,17 @@ def setup(self):
 
     def start_server(self):
         """Start the agent server."""
-        logger.info(f"Starting {self.sample_name} server in {self.sample_dir}")
+        logger.info(
+            "Starting %s server in %s on port %s",
+            self.sample_name,
+            self.sample_dir,
+            self.port,
+        )
 
         env = os.environ.copy()
         env.update(self.env_vars)
+        env["DEFAULT_AD_PORT"] = str(self.port)
+        env.setdefault("AGENT_BASE_URL", self.base_url)
 
         # Use unbuffered output to capture logs in real-time
         self.process = subprocess.Popen(
@@ -195,7 +230,11 @@ def start_server(self):
 
     def wait_for_ready(self, max_attempts: int = 30, delay: float = 1.0) -> bool:
         """Wait for server to be ready."""
-        logger.info(f"Waiting for server to be ready (max {max_attempts} attempts)")
+        logger.info(
+            "Waiting for server to be ready at %s (max %s attempts)",
+            f"{self.base_url}{self.endpoint}",
+            max_attempts,
+        )
 
         for i in range(max_attempts):
             # Check process status first

sdk/ai/azure-ai-agentserver-core/tests/test_custom.py

@@ -7,6 +7,7 @@
 """
 
 import os
+import socket
 import subprocess
 import sys
 import time
@@ -35,7 +36,8 @@ def __init__(self, sample_name: str, script_name: str):
         self.sample_name = sample_name
         self.script_name = script_name
         self.sample_dir = project_root / "samples" / sample_name
-        self.base_url = "http://localhost:8088"
+        self.port = self._find_free_port()
+        self.base_url = f"http://127.0.0.1:{self.port}"
         self.responses_endpoint = f"{self.base_url}/responses"
         self.process = None
         self.original_dir = os.getcwd()
@@ -49,6 +51,8 @@ def start_server(self):
         # Prepare environment with UTF-8 encoding to handle emoji in agent output
         env = os.environ.copy()
         env["PYTHONIOENCODING"] = "utf-8"
+        env["DEFAULT_AD_PORT"] = str(self.port)
+        env.setdefault("AGENT_BASE_URL", self.base_url)
 
         # Use subprocess.DEVNULL to avoid buffering issues
         self.process = subprocess.Popen(
@@ -127,6 +131,12 @@ def cleanup(self):
 
         os.chdir(self.original_dir)
 
+    @staticmethod
+    def _find_free_port() -> int:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            sock.bind(("127.0.0.1", 0))
+            return sock.getsockname()[1]
+
 
 class TestSimpleMockAgent:
     """Test suite for Simple Mock Agent - uses shared server."""