Fix intermittent send-to integration failures (#254)

* move DataServers into session-scoped fixtures with different ports to prevent intermittent connection problems

* style

Author: timabrmsn

tests/integration/conftest.py

@@ -2,6 +2,7 @@
 from shlex import split as split_command
 
 import pytest
+from tests.integration.util import DataServer
 
 from code42cli.errors import Code42CLIError
 from code42cli.main import cli
@@ -46,3 +47,15 @@ def _encode_response(line, encoding_type=_ENCODING_TYPE):
 
 def append_profile(command):
     return "{} --profile {}".format(command, TEST_PROFILE_NAME)
+
+
+@pytest.yield_fixture(scope="session")
+def udp_dataserver():
+    with DataServer(protocol="UDP"):
+        yield
+
+
+@pytest.yield_fixture(scope="session")
+def tcp_dataserver():
+    with DataServer(protocol="TCP"):
+        yield

tests/integration/test_alerts.py

@@ -5,7 +5,6 @@
 import pytest
 from tests.integration.conftest import append_profile
 from tests.integration.util import assert_test_is_successful
-from tests.integration.util import DataServer
 
 from code42cli.main import cli
 
@@ -24,17 +23,24 @@ def test_alerts_search_command_returns_success_return_code(
 
 
 @pytest.mark.integration
-@pytest.mark.parametrize(
-    "protocol", ["TCP", "UDP"],
-)
-def test_alerts_send_to_returns_success_return_code(
-    runner, integration_test_profile, protocol
+def test_alerts_send_to_tcp_returns_success_return_code(
+    runner, integration_test_profile, tcp_dataserver
 ):
-    command = "alerts send-to localhost:5140 -p {} -b {}".format(
-        protocol, begin_date_str
+    command = append_profile(
+        f"alerts send-to localhost:5140 -p TCP -b '{begin_date_str}'"
     )
-    with DataServer(protocol=protocol):
-        result = runner.invoke(cli, split_command(append_profile(command)))
+    result = runner.invoke(cli, split_command(command))
+    assert result.exit_code == 0
+
+
+@pytest.mark.integration
+def test_alerts_send_to_udp_returns_success_return_code(
+    runner, integration_test_profile, udp_dataserver
+):
+    command = append_profile(
+        f"alerts send-to localhost:5141 -p UDP -b '{begin_date_str}'"
+    )
+    result = runner.invoke(cli, split_command(command))
     assert result.exit_code == 0
 
 

tests/integration/test_auditlogs.py

@@ -5,7 +5,6 @@
 import pytest
 from tests.integration.conftest import append_profile
 from tests.integration.util import assert_test_is_successful
-from tests.integration.util import DataServer
 
 from code42cli.main import cli
 
@@ -17,17 +16,24 @@
 
 
 @pytest.mark.integration
-@pytest.mark.parametrize(
-    "protocol", ["TCP", "UDP"],
-)
-def test_auditlogs_send_to_command_returns_success_return_code(
-    runner, integration_test_profile, protocol
+def test_auditlogs_send_to_tcp_command_returns_success_return_code(
+    runner, integration_test_profile, tcp_dataserver
 ):
-    command = "audit-logs send-to localhost:5140 -p {} -b '{}'".format(
-        protocol, begin_date_str
+    command = append_profile(
+        f"audit-logs send-to localhost:5140 -p TCP -b '{begin_date_str}'"
     )
-    with DataServer(protocol=protocol):
-        result = runner.invoke(cli, split_command(append_profile(command)))
+    result = runner.invoke(cli, split_command(command))
+    assert result.exit_code == 0
+
+
+@pytest.mark.integration
+def test_auditlogs_send_to_udp_command_returns_success_return_code(
+    runner, integration_test_profile, udp_dataserver
+):
+    command = append_profile(
+        f"audit-logs send-to localhost:5141 -p UDP -b '{begin_date_str}'"
+    )
+    result = runner.invoke(cli, split_command(command))
     assert result.exit_code == 0
 
 

tests/integration/test_securitydata.py

@@ -4,23 +4,31 @@
 
 import pytest
 from tests.integration.conftest import append_profile
-from tests.integration.util import DataServer
 
 from code42cli.main import cli
 
 
 @pytest.mark.integration
-@pytest.mark.parametrize(
-    "protocol", ["TCP", "UDP"],
-)
-def test_security_data_send_to_return_success_return_code(
-    runner, integration_test_profile, protocol
+def test_security_data_send_to_tcp_return_success_return_code(
+    runner, integration_test_profile, tcp_dataserver
 ):
     begin_date = datetime.utcnow() - timedelta(days=20)
     begin_date_str = begin_date.strftime("%Y-%m-%d")
-    command = "security-data send-to localhost:5140 -p {} -b {}".format(
-        protocol, begin_date_str
+    command = append_profile(
+        f"security-data send-to localhost:5140 -p TCP -b '{begin_date_str}'"
     )
-    with DataServer(protocol=protocol):
-        result = runner.invoke(cli, split_command(append_profile(command)))
+    result = runner.invoke(cli, split_command(command))
+    assert result.exit_code == 0
+
+
+@pytest.mark.integration
+def test_security_data_send_to_udp_return_success_return_code(
+    runner, integration_test_profile, udp_dataserver
+):
+    begin_date = datetime.utcnow() - timedelta(days=20)
+    begin_date_str = begin_date.strftime("%Y-%m-%d")
+    command = append_profile(
+        f"security-data send-to localhost:5141 -p UDP -b '{begin_date_str}'"
+    )
+    result = runner.invoke(cli, split_command(command))
     assert result.exit_code == 0

tests/integration/util.py

@@ -38,7 +38,7 @@ def wrapper():
 
 class DataServer:
     TCP_SERVER_COMMAND = "ncat -l 5140"
-    UDP_SERVER_COMMAND = "ncat -ul 5140"
+    UDP_SERVER_COMMAND = "ncat -ul 5141"
 
     def __init__(self, protocol="TCP"):
         if protocol.upper() == "UDP":