write integration tests for pty-server in a container

Dockerfile

@@ -7,11 +7,7 @@ WORKDIR /usr/src
 # Install pexpect and clone the repository
 RUN python -m venv .venv \
     && . .venv/bin/activate \
-    && pip install interactive-process \
-    && git clone https://github.com/breba-apps/breba-docs.git
-
-# Add breba-docs to the Python path
-ENV PYTHONPATH="/usr/src/breba-docs:$PYTHONPATH"
+    && pip install pty-server
 
 # Run the listener script in detached mode
-CMD ["/bin/bash", "-c", ". .venv/bin/activate && python breba-docs/breba_docs/socket_server/listener.py"]
+CMD ["/bin/bash", "-c", "VIRTUAL_ENV_DISABLE_PROMPT=1 . .venv/bin/activate && pty-server"]

pyproject.toml

@@ -33,6 +33,7 @@ langchain-openai = "^0.2.10"
 langgraph = "^0.2.53"
 pytest-asyncio = "^0.25.0"
 interactive-process = "^0.2.0"
+pty-server = "^0.1.6"
 
 
 [tool.poetry.group.test.dependencies]
@@ -53,6 +54,7 @@ markers = [
 # log_cli = "True"
 # log_cli_level = "INFO"
 
+
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.3.2"
 pytest-mock = "^3.14.0"

tests/integration/test_container_async.py

@@ -0,0 +1,82 @@
+import json
+import time
+
+import pytest
+
+from breba_docs.container import container_setup
+
+from pty_server import AsyncPtyClient
+from breba_docs.socket_server.listener import PORT
+from pty_server.async_client import STATUS_TIMEOUT, STATUS_COMPLETED, PtyServerResponse
+
+
+@pytest.fixture(scope="module")
+def container():
+    # To Run the container from terminal from breba_docs package dir
+    # docker run -d -it \
+    #   -v $(pwd)/breba_docs/socket_server:/usr/src/socket_server \
+    #   -w /usr/src \
+    #   -p 44440:44440 \
+    #   python:3 \
+    #   /bin/bash
+    started_container = container_setup(dev=True)
+    time.sleep(2)
+    yield started_container
+    started_container.stop()
+    started_container.remove()
+
+async def response_accumulator(response: PtyServerResponse, timeout):
+    data = ""
+    async for chunk in response.stream(timeout):
+        data += chunk
+    return data
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_execute_command(container):
+    async with AsyncPtyClient() as client:
+        response_install = await client.send_command('pip install pexpect')
+        reponse_text = await response_accumulator(response_install, 2)
+        assert response_install.status == STATUS_COMPLETED
+        assert "Successfully installed pexpect" in reponse_text
+
+        response_uninstall = await client.send_command('pip uninstall pexpect')
+        reponse_text = await response_accumulator(response_uninstall, 2)
+        assert response_uninstall.status == STATUS_TIMEOUT
+        assert "Proceed (Y/n)" in reponse_text
+
+        command = {"input": 'Y'}
+        await client.send_message(json.dumps(command))
+        response_text = await response_accumulator(response_uninstall, 2)
+        assert response_uninstall.status == STATUS_COMPLETED
+        assert "Successfully uninstalled" in response_text
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_execute_ampersand_command(container):
+    async with AsyncPtyClient() as client:
+        command = 'mkdir test && cd test && pwd && echo "more testing is needed"'
+        response = await client.send_command(command)
+        response_text = await response_accumulator(response, 0.01)
+    assert "/usr/src/test" in response_text
+    assert "No such file or directory" not in response_text
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_multiple_connections(container):
+    async with AsyncPtyClient() as client:
+        command = 'mkdir test2 && cd test2 && pwd && echo "more testing is needed"'
+        response = await client.send_command(command)
+        response_text = await response_accumulator(response, 0.01)
+
+    assert "/usr/src/test" in response_text
+    assert "No such file or directory" not in response_text
+
+    async with AsyncPtyClient() as client:
+        command = 'mkdir test3 && cd test3 && pwd && echo "more testing is needed"'
+        response = await client.send_command(command)
+        response_text = await response_accumulator(response, 0.01)
+    assert "/usr/src/test3" in response_text
+    assert "No such file or directory" not in response_text