|
45 | 45 | @pytest.fixture(autouse=True)
|
46 | 46 | def cleanup_endpoints(mb_sagemaker_session) -> Generator[None, None, None]:
|
47 | 47 | """Clean up any existing endpoints before and after tests."""
|
| 48 | + sagemaker_client = mb_sagemaker_session.sagemaker_client |
| 49 | + |
48 | 50 | # Pre-test cleanup
|
49 |
| - endpoints = mb_sagemaker_session.list_endpoints() |
50 |
| - for endpoint in endpoints["Endpoints"]: |
51 |
| - try: |
52 |
| - mb_sagemaker_session.delete_endpoint(endpoint_name=endpoint["EndpointName"]) |
53 |
| - mb_sagemaker_session.delete_endpoint_config( |
54 |
| - endpoint_config_name=endpoint["EndpointConfigName"] |
55 |
| - ) |
56 |
| - except Exception as e: |
57 |
| - print(f"Error cleaning up endpoint {endpoint['EndpointName']}: {e}") |
| 51 | + try: |
| 52 | + endpoints = sagemaker_client.list_endpoints() |
| 53 | + for endpoint in endpoints["Endpoints"]: |
| 54 | + try: |
| 55 | + sagemaker_client.delete_endpoint(EndpointName=endpoint["EndpointName"]) |
| 56 | + sagemaker_client.delete_endpoint_config( |
| 57 | + EndpointConfigName=endpoint["EndpointConfigName"] |
| 58 | + ) |
| 59 | + except Exception as e: |
| 60 | + print(f"Error cleaning up endpoint {endpoint['EndpointName']}: {e}") |
| 61 | + except Exception as e: |
| 62 | + print(f"Error listing endpoints: {e}") |
58 | 63 |
|
59 | 64 | yield
|
60 | 65 |
|
61 |
| - # Post-test cleanup - this is technically redundant with the existing cleanup_model_resources |
62 |
| - # but serves as a safety net |
63 |
| - endpoints = mb_sagemaker_session.list_endpoints() |
64 |
| - for endpoint in endpoints["Endpoints"]: |
65 |
| - try: |
66 |
| - mb_sagemaker_session.delete_endpoint(endpoint_name=endpoint["EndpointName"]) |
67 |
| - mb_sagemaker_session.delete_endpoint_config( |
68 |
| - endpoint_config_name=endpoint["EndpointConfigName"] |
69 |
| - ) |
70 |
| - except Exception as e: |
71 |
| - print(f"Error cleaning up endpoint {endpoint['EndpointName']}: {e}") |
| 66 | + # Post-test cleanup |
| 67 | + try: |
| 68 | + endpoints = sagemaker_client.list_endpoints() |
| 69 | + for endpoint in endpoints["Endpoints"]: |
| 70 | + try: |
| 71 | + sagemaker_client.delete_endpoint(EndpointName=endpoint["EndpointName"]) |
| 72 | + sagemaker_client.delete_endpoint_config( |
| 73 | + EndpointConfigName=endpoint["EndpointConfigName"] |
| 74 | + ) |
| 75 | + except Exception as e: |
| 76 | + print(f"Error cleaning up endpoint {endpoint['EndpointName']}: {e}") |
| 77 | + except Exception as e: |
| 78 | + print(f"Error listing endpoints: {e}") |
72 | 79 |
|
73 | 80 |
|
74 | 81 | @pytest.fixture(scope="module")
|
|
0 commit comments