Coverage for agentos/marketplace/skills/docker/docker.py: 23%

1""" 

2docker — Docker container/image operations via CLI. 

3 

4Actions: ps, images, logs, inspect, stats 

5""" 

6 

7import subprocess 

8from typing import Any 

9 

10 

11def run(action: str = "ps", container: str = "", image: str = "", **kwargs: Any) -> str: 

12 """Docker operations. 

13 

14 Args: 

15 action: ps | images | logs | inspect | stats | version 

16 container: container name/id (for logs, inspect, stats) 

17 image: image name/id (for inspect on images) 

18 """ 

19 cmds = { 

20 "ps": ["docker", "ps", "--format", "table {{.Names}}\t{{.Status}}\t{{.Ports}}"], 

21 "ps_all": ["docker", "ps", "-a", "--format", "table {{.Names}}\t{{.Status}}\t{{.Ports}}"], 

22 "images": ["docker", "images", "--format", "table {{.Repository}}\t{{.Tag}}\t{{.Size}}"], 

23 "version": ["docker", "version", "--format", "{{.Server.Version}}"], 

24 } 

25 

26 if action == "logs" and container: 

27 return _run(["docker", "logs", "--tail", "50", container]) 

28 if action == "inspect" and container: 

29 return _run(["docker", "inspect", container]) 

30 if action == "stats": 

31 return _run(["docker", "stats", "--no-stream", "--all"]) 

32 if action in cmds: 

33 return _run(cmds[action]) 

34 

35 return f"[docker] Unknown action: {action}. Available: ps, ps_all, images, logs, inspect, stats, version" 

36 

37 

38def _run(cmd: list[str]) -> str: 

39 try: 

40 r = subprocess.run(cmd, capture_output=True, text=True, timeout=15) 

41 return r.stdout or r.stderr or "[docker] No output." 

42 except FileNotFoundError: 

43 return "[docker] Docker not installed." 

44 except Exception as e: 

45 return f"[docker] Error: {e}" 

46 

47 

48__all__ = ["run"]