Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions winrm/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,12 @@

import collections.abc
import re
import reprlib
import typing as t
import warnings
import xml.etree.ElementTree as ET
from base64 import b64encode
from dataclasses import dataclass

from winrm.protocol import Protocol

Expand All @@ -21,15 +23,16 @@
FEATURE_PROXY_SUPPORT = True


class Response(object):
@dataclass
class Response:
"""Response from a remote command execution"""

def __init__(self, args: tuple[bytes, bytes, int]) -> None:
self.std_out, self.std_err, self.status_code = args
std_out: bytes
std_err: bytes
status_code: int = 0

def __repr__(self) -> str:
# TODO put tree dots at the end if out/err was truncated
return '<Response code {0}, out "{1!r}", err "{2!r}">'.format(self.status_code, self.std_out[:20], self.std_err[:20])
return f'<Response code {self.status_code}, out "{reprlib.repr(self.std_out)}", err "{reprlib.repr(self.std_err)}">'


class Session(object):
Expand All @@ -43,7 +46,7 @@ def run_cmd(self, command: str, args: collections.abc.Iterable[str | bytes] = ()
# TODO optimize perf. Do not call open/close shell every time
shell_id = self.protocol.open_shell()
command_id = self.protocol.run_command(shell_id, command, args)
rs = Response(self.protocol.get_command_output(shell_id, command_id))
rs = Response(*self.protocol.get_command_output(shell_id, command_id))
self.protocol.cleanup_command(shell_id, command_id)
self.protocol.close_shell(shell_id)
return rs
Expand Down
17 changes: 16 additions & 1 deletion winrm/tests/test_session.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import pytest

from winrm import Session
from winrm import Response, Session


def test_run_cmd(protocol_fake):
Expand Down Expand Up @@ -89,3 +89,18 @@ def test_decode_clixml_invalid_xml():
actual = s._clean_error_msg(msg)

assert actual == msg


def test_response_repr_short():
r = Response(std_out=b"short std out", std_err=b"short std err")

assert repr(r) == "<Response code 0, out \"b'short std out'\", err \"b'short std err'\">"


def test_response_repr_long():
r = Response(
std_out=b"some very long std out that take more than 20 chars",
std_err=b"some very long std err that take more than 20 chars",
)

assert repr(r) == "<Response code 0, out \"b'some very l...than 20 chars'\", err \"b'some very l...than 20 chars'\">"