forked from Archives/langchain
docker: attach to container's stdin
- wip image helper for optimized params with common images - gVisor runtime checker - make tests skipped if docker installeddocker-utility-pexpect
parent
17213209e0
commit
7cde1cbfc3
@ -0,0 +1,57 @@
|
||||
"""Optimized parameters for commonly used docker images that can be used by
|
||||
the docker wrapper utility to attach to."""
|
||||
|
||||
from enum import Enum
|
||||
from typing import Optional, List
|
||||
from pydantic import BaseModel, Extra, validator
|
||||
|
||||
|
||||
class BaseImage(BaseModel, extra=Extra.forbid):
|
||||
"""Base docker image class."""
|
||||
tty: bool = False
|
||||
stdin_open: bool = True
|
||||
image: str
|
||||
default_command: Optional[List[str]] = None
|
||||
|
||||
class ShellTypes(str, Enum):
|
||||
"""Enum class for shell types."""
|
||||
bash = '/bin/bash'
|
||||
sh = '/bin/sh'
|
||||
zsh = '/bin/zsh'
|
||||
|
||||
|
||||
class Shell(BaseImage):
|
||||
"""Shell image focused on running shell commands.
|
||||
|
||||
A shell image can be crated by passing a shell alias such as `sh` or `bash`
|
||||
or by passing the full path to the shell binary.
|
||||
"""
|
||||
image: str = 'alpine'
|
||||
shell: str = ShellTypes.bash.value
|
||||
|
||||
@validator('shell')
|
||||
def validate_shell(cls, value: str) -> str:
|
||||
"""Validate shell type."""
|
||||
val = getattr(ShellTypes, value, None)
|
||||
if val:
|
||||
return val.value
|
||||
# elif value in [v.value for v in list(ShellTypes.__members__.values())]:
|
||||
# print(f"docker: overriding shell binary to: {value}")
|
||||
return value
|
||||
|
||||
# example using base image to construct python image
|
||||
class Python(BaseImage):
|
||||
"""Python image class.
|
||||
|
||||
The python image needs to be launced using the `python3 -i` command to keep
|
||||
stdin open.
|
||||
"""
|
||||
image: str = 'python'
|
||||
default_command: List[str] = ['python3', '-i']
|
||||
|
||||
def __setattr__(self, name, value):
|
||||
if name == 'default_command':
|
||||
raise AttributeError(f'running this image with {self.default_command}'
|
||||
' is necessary to keep stdin open.')
|
||||
|
||||
super().__setattr__(name, value)
|
@ -1,25 +1,48 @@
|
||||
"""Test the docker wrapper utility."""
|
||||
|
||||
import pytest
|
||||
from langchain.utilities.docker import DockerWrapper
|
||||
from langchain.utilities.docker import DockerWrapper, gvisor_runtime_available
|
||||
from unittest.mock import MagicMock
|
||||
import subprocess
|
||||
|
||||
|
||||
def test_command_default_image() -> None:
|
||||
def docker_installed() -> bool:
|
||||
"""Checks if docker is installed locally."""
|
||||
try:
|
||||
subprocess.run(['which', 'docker',], check=True)
|
||||
except subprocess.CalledProcessError:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
|
||||
|
||||
@pytest.mark.skipif(not docker_installed(), reason="docker not installed")
|
||||
class TestDockerUtility:
|
||||
|
||||
def test_command_default_image(self) -> None:
|
||||
"""Test running a command with the default alpine image."""
|
||||
docker = DockerWrapper()
|
||||
output = docker.run("cat /etc/os-release")
|
||||
assert output.find(b"alpine")
|
||||
output = docker.run('cat /etc/os-release')
|
||||
assert output.find(b'alpine')
|
||||
|
||||
def test_inner_failing_command() -> None:
|
||||
def test_inner_failing_command(self) -> None:
|
||||
"""Test inner command with non zero exit"""
|
||||
docker = DockerWrapper()
|
||||
output = docker.run("ls /inner-failing-command")
|
||||
output = docker.run('ls /inner-failing-command')
|
||||
assert str(output).startswith("STDERR")
|
||||
|
||||
def test_entrypoint_failure() -> None:
|
||||
def test_entrypoint_failure(self) -> None:
|
||||
"""Test inner command with non zero exit"""
|
||||
docker = DockerWrapper()
|
||||
output = docker.run("todo handle APIError")
|
||||
|
||||
|
||||
|
||||
output = docker.run('todo handle APIError')
|
||||
assert output == 'ERROR'
|
||||
|
||||
def test_check_gvisor_runtime(self) -> None:
|
||||
"""test gVisor runtime verification using a mock docker client"""
|
||||
mock_client = MagicMock()
|
||||
mock_client.info.return_value = {'Runtimes': {'runsc': {'path': 'runsc'}}}
|
||||
assert gvisor_runtime_available(mock_client)
|
||||
mock_client.info.return_value = {'Runtimes': {'runc': {'path': 'runc'}}}
|
||||
assert not gvisor_runtime_available(mock_client)
|
||||
|
Loading…
Reference in New Issue