Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_wsman_update_envelope_size_explicit(self):
wsman = WSMan("")
wsman.update_max_payload_size(4096)
assert wsman.max_envelope_size == 4096
# this next value is dependent on a lot of things such as python
# version and rounding differences, we will just assert against a range
assert 1450 <= wsman.max_payload_size <= 1835
def test_jordan():
# executable = "C:\\Windows\\System32\\cmd.exe"
# arguments = ["/c", "timeout", "&&", "30"]
executable = "C:\\Windows\\System32\\WindowsPowershell\\v1.0\\powershell.exe"
# executable = "powershell.exe"
arguments = ["Write-Host", "hi"]
transport = TransportHTTP(server, port, username, password)
wsman = WSMan(transport)
shell = WinRS(wsman)
shell.open()
try:
command_id = shell.run_executable(executable, arguments, no_shell=True)
rc, stdout, stderr = shell.get_output(command_id)
shell.signal(SignalCode.CTRL_C, command_id)
finally:
shell.close()
print()
print("STDOUT:\n%s" % stdout.decode('utf-8'))
print("STDERR:\n%s" % stderr.decode('utf-8'))
print("RC: %d" % rc)
runspace_pool.open()
runspace_pool.disconnect()
assert runspace_pool.state == RunspacePoolState.DISCONNECTED
runspace_pool.connect()
assert runspace_pool.state == RunspacePoolState.OPENED
runspace_pool.disconnect()
actual = []
try:
runspace_pool2 = RunspacePool(wsman_conn)
runspace_pool2.open()
runspace_pool2.disconnect()
# build a new WSMan object with new unique ID and new requests
# session
wsman2 = WSMan(wsman_conn.transport.server)
wsman2.transport = wsman_conn.transport
wsman2.transport.session = None
actual = RunspacePool.get_runspace_pools(wsman2)
assert len(actual) == 2
assert actual[0].id == runspace_pool.id or \
actual[1].id == runspace_pool.id
assert actual[0].id == runspace_pool2.id or \
actual[1].id == runspace_pool2.id
assert actual[0].state == RunspacePoolState.DISCONNECTED
assert actual[1].state == RunspacePoolState.DISCONNECTED
for pool in actual:
pool.connect()
assert pool.state == RunspacePoolState.OPENED
finally:
for pool in actual:
pool.close()
def test_psrp_runspace_host_call_no_host(self):
wsman = WSMan("")
rs = RunspacePool(wsman)
msg = Message(Destination.CLIENT, None, None, RunspacePoolHostCall(),
None)
actual = rs._process_runspacepool_host_call(msg)
assert actual == msg.data
def test_override_default(self):
actual = WSMan("", 8192, 30)
assert actual.max_envelope_size == 8192
assert actual.max_payload_size < actual.max_envelope_size
assert actual.operation_timeout == 30
def test_psrp_open_invalid_state(self):
wsman = WSMan("")
rs = RunspacePool(wsman)
rs.state = RunspacePoolState.DISCONNECTED
with pytest.raises(InvalidRunspacePoolStateError) as err:
rs.open()
assert err.value.action == "open a new Runspace Pool"
assert err.value.current_state == RunspacePoolState.DISCONNECTED
assert err.value.expected_state == RunspacePoolState.BEFORE_OPEN
assert str(err.value) == \
"Cannot 'open a new Runspace Pool' on the current state " \
"'Disconnected', expecting state(s): 'BeforeOpen'"
def test_invoke_get(self, monkeypatch):
def mockuuid():
return uuid.UUID("00000000-0000-0000-0000-000000000000")
monkeypatch.setattr(uuid, 'uuid4', mockuuid)
wsman = WSMan("")
wsman.transport = _TransportTest(WSManAction.GET)
actual = wsman.get("", None)
assert actual.tag == "{http://www.w3.org/2003/05/soap-envelope}Body"
assert actual.text == "body"
def test_wsman_defaults(self):
actual = WSMan("")
assert actual.max_envelope_size == 153600
assert actual.max_payload_size < actual.max_envelope_size
assert actual.operation_timeout == 20
assert isinstance(actual.session_id, str)
# verify we get a unique session id each time this is initialised
new_wsman = WSMan("")
assert actual.session_id != new_wsman.session_id
spawn new cmd command/process
spawn new PowerShell Runspace Pool/Pipeline
copy a file from localhost to the remote Windows host
fetch a file from the remote Windows host to localhost
This is just an easy to use layer on top of the objects WinRS and
RunspacePool/PowerShell. It trades flexibility in favour of simplicity.
If your use case needs some of that flexibility you can use these
functions as a reference implementation for your own functions.
:param server: The server/host to connect to
:param kwargs: The various WSMan args to control the transport
mechanism, see pypsrp.wsman.WSMan for these args
"""
self.wsman = WSMan(server, **kwargs)
def _connect(self):
if not HAS_PYPSRP:
raise AnsibleError("pypsrp or dependencies are not installed: %s"
% to_native(PYPSRP_IMP_ERR))
super(Connection, self)._connect()
self._build_kwargs()
display.vvv("ESTABLISH PSRP CONNECTION FOR USER: %s ON PORT %s TO %s" %
(self._psrp_user, self._psrp_port, self._psrp_host),
host=self._psrp_host)
if not self.runspace:
connection = WSMan(**self._psrp_conn_kwargs)
# create our psuedo host to capture the exit code and host output
host_ui = PSHostUserInterface()
self.host = PSHost(None, None, False, "Ansible PSRP Host", None,
host_ui, None)
self.runspace = RunspacePool(
connection, host=self.host,
configuration_name=self._psrp_configuration_name
)
display.vvvvv(
"PSRP OPEN RUNSPACE: auth=%s configuration=%s endpoint=%s" %
(self._psrp_auth, self._psrp_configuration_name,
connection.transport.endpoint), host=self._psrp_host
)
try: