diff --git a/octavia_tempest_plugin/common/decorators.py b/octavia_tempest_plugin/common/decorators.py
index b484497..15c3c17 100644
--- a/octavia_tempest_plugin/common/decorators.py
+++ b/octavia_tempest_plugin/common/decorators.py
@@ -52,3 +52,49 @@
                     message = e.resp_body.get('faultstring', message)
                 raise testtools.TestCase.skipException(message)
     return wrapper
+
+
+def retry_on_port_in_use(start_port, max_retries=3):
+    """Decorator to retry a test function if the specified port is in use.
+
+    This handles cases where a test fails due to a port conflict, typically
+    caused by another service binding the same port on the host. The decorator
+    catches '[Errno 98] Address already in use' errors and retries the test
+    using incrementally higher port numbers.
+
+    The decorated function must accept `port` as its first parameter.
+
+    :param start_port: Initial port to attempt.
+    :param max_retries: Number of retries with incremented port values.
+    """
+    def decorator(func):
+        @wraps(func)
+        def wrapper(self, *args, **kwargs):
+            port = start_port
+            last_exception = None
+
+            for _ in range(max_retries):
+                try:
+                    return func(self, port, *args, **kwargs)
+                except exceptions.NotImplemented as e:
+                    message = (
+                        "The configured provider driver '{driver}' does not "
+                        "support a feature required for this test.".format(
+                            driver=CONF.load_balancer.provider
+                        )
+                    )
+                    if hasattr(e, 'resp_body'):
+                        message = e.resp_body.get('faultstring', message)
+                    raise testtools.TestCase.skipException(message)
+
+                except Exception as e:
+                    if "Address already in use" in str(e):
+                        last_exception = e
+                        port += 1
+                    else:
+                        raise
+
+            raise Exception(f"All port attempts failed after {max_retries} "
+                            f"retries. Last error: {last_exception}")
+        return wrapper
+    return decorator
diff --git a/octavia_tempest_plugin/tests/scenario/v2/test_traffic_ops.py b/octavia_tempest_plugin/tests/scenario/v2/test_traffic_ops.py
index db92352..9cf77ac 100644
--- a/octavia_tempest_plugin/tests/scenario/v2/test_traffic_ops.py
+++ b/octavia_tempest_plugin/tests/scenario/v2/test_traffic_ops.py
@@ -27,6 +27,7 @@
 from tempest.lib import exceptions
 
 from octavia_tempest_plugin.common import constants as const
+from octavia_tempest_plugin.common.decorators import retry_on_port_in_use
 from octavia_tempest_plugin.tests import test_base
 from octavia_tempest_plugin.tests import waiters
 
@@ -912,17 +913,18 @@
     @testtools.skipIf(CONF.load_balancer.test_with_noop,
                       'Traffic tests will not work in noop mode.')
     @decorators.idempotent_id('a446585b-5651-40ce-a4db-cb2ab4d37c03')
-    def test_source_ip_port_http_traffic(self):
+    @retry_on_port_in_use(start_port=60091)
+    def test_source_ip_port_http_traffic(self, port):
         # This is a special case as the reference driver does not support
         # this test. Since it runs with not_implemented_is_error, we must
         # handle this test case special.
         try:
             pool_id = self._listener_pool_create(
-                const.HTTP, 60091,
+                const.HTTP, port,
                 pool_algorithm=const.LB_ALGORITHM_SOURCE_IP_PORT)[1]
             self._test_basic_traffic(
-                const.HTTP, 60091, pool_id,
-                traffic_member_count=1, persistent=False, source_port=60091)
+                const.HTTP, port, pool_id,
+                traffic_member_count=1, persistent=False, source_port=port)
         except exceptions.NotImplemented as e:
             message = ("The configured provider driver '{driver}' "
                        "does not support a feature required for this "
@@ -934,19 +936,20 @@
     @testtools.skipIf(CONF.load_balancer.test_with_noop,
                       'Traffic tests will not work in noop mode.')
     @decorators.idempotent_id('60108f30-d870-487c-ab96-8d8a9b587b94')
-    def test_source_ip_port_tcp_traffic(self):
+    @retry_on_port_in_use(start_port=60092)
+    def test_source_ip_port_tcp_traffic(self, port):
         # This is a special case as the reference driver does not support
         # this test. Since it runs with not_implemented_is_error, we must
         # handle this test case special.
         try:
             listener_id, pool_id = self._listener_pool_create(
-                const.TCP, 60092,
+                const.TCP, port,
                 pool_algorithm=const.LB_ALGORITHM_SOURCE_IP_PORT)
             # Without a delay this can trigger a "Cannot assign requested
             # address" warning setting the source port, leading to failure
             self._test_basic_traffic(
-                const.TCP, 60092, pool_id, traffic_member_count=1,
-                persistent=False, source_port=60092, delay=0.2)
+                const.TCP, port, pool_id, traffic_member_count=1,
+                persistent=False, source_port=port, delay=0.2)
         except exceptions.NotImplemented as e:
             message = ("The configured provider driver '{driver}' "
                        "does not support a feature required for this "
diff --git a/octavia_tempest_plugin/tests/validators.py b/octavia_tempest_plugin/tests/validators.py
index 5ff7bd5..57ba2e2 100644
--- a/octavia_tempest_plugin/tests/validators.py
+++ b/octavia_tempest_plugin/tests/validators.py
@@ -113,6 +113,8 @@
                     session.close()
                 raise
             except Exception as e:
+                if "[Errno 98] Address already in use" in str(e):
+                    raise e
                 LOG.info('Validate URL got exception: %s. '
                          'Retrying.', e)
                 time.sleep(request_interval)
@@ -410,8 +412,10 @@
                               response_counts)
                     time.sleep(1)
                     return
-            except Exception:
+            except Exception as e:
                 LOG.warning('Server is not passing initial traffic. Waiting.')
+                if "[Errno 98] Address already in use" in str(e):
+                    raise e
             time.sleep(request_interval)
 
         LOG.debug('Loadbalancer wait for load balancer response totals: %s',