fluzzi32/connpy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
c4950ed029f8ee1adc05fd158c502e9929a55565
connpy/docs/conn/index.html

1287 lines
59 KiB
HTML
Raw Normal View History

working on doc
2022-04-03 09:38:00 -03:00
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1, minimum-scale=1" />
<meta name="generator" content="pdoc 0.10.0" />
<title>conn API documentation</title>
<meta name="description" content="" />
<link rel="preload stylesheet" as="style" href="https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/11.0.1/sanitize.min.css" integrity="sha256-PK9q560IAAa6WVRRh76LtCaI8pjTJ2z11v0miyNNjrs=" crossorigin>
<link rel="preload stylesheet" as="style" href="https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/11.0.1/typography.min.css" integrity="sha256-7l/o7C8jubJiy74VsKTidCy1yBkRtiUGbVkYBylBqUg=" crossorigin>
<style>:root{--highlight-color:#fe9}.flex{display:flex !important}body{line-height:1.5em}#content{padding:20px}#sidebar{padding:30px;overflow:hidden}#sidebar > *:last-child{margin-bottom:2cm}.http-server-breadcrumbs{font-size:130%;margin:0 0 15px 0}#footer{font-size:.75em;padding:5px 30px;border-top:1px solid #ddd;text-align:right}#footer p{margin:0 0 0 1em;display:inline-block}#footer p:last-child{margin-right:30px}h1,h2,h3,h4,h5{font-weight:300}h1{font-size:2.5em;line-height:1.1em}h2{font-size:1.75em;margin:1em 0 .50em 0}h3{font-size:1.4em;margin:25px 0 10px 0}h4{margin:0;font-size:105%}h1:target,h2:target,h3:target,h4:target,h5:target,h6:target{background:var(--highlight-color);padding:.2em 0}a{color:#058;text-decoration:none;transition:color .3s ease-in-out}a:hover{color:#e82}.title code{font-weight:bold}h2[id^="header-"]{margin-top:2em}.ident{color:#900}pre code{background:#f8f8f8;font-size:.8em;line-height:1.4em}code{background:#f2f2f1;padding:1px 4px;overflow-wrap:break-word}h1 code{background:transparent}pre{background:#f8f8f8;border:0;border-top:1px solid #ccc;border-bottom:1px solid #ccc;margin:1em 0;padding:1ex}#http-server-module-list{display:flex;flex-flow:column}#http-server-module-list div{display:flex}#http-server-module-list dt{min-width:10%}#http-server-module-list p{margin-top:0}.toc ul,#index{list-style-type:none;margin:0;padding:0}#index code{background:transparent}#index h3{border-bottom:1px solid #ddd}#index ul{padding:0}#index h4{margin-top:.6em;font-weight:bold}@media (min-width:200ex){#index .two-column{column-count:2}}@media (min-width:300ex){#index .two-column{column-count:3}}dl{margin-bottom:2em}dl dl:last-child{margin-bottom:4em}dd{margin:0 0 1em 3em}#header-classes + dl > dd{margin-bottom:3em}dd dd{margin-left:2em}dd p{margin:10px 0}.name{background:#eee;font-weight:bold;font-size:.85em;padding:5px 10px;display:inline-block;min-width:40%}.name:hover{background:#e0e0e0}dt:target .name{background:var(--highlight-color)}.name > span:first-child{white-space:nowrap}.name.class > span:nth-child(2){margin-left:.4em}.inherited{color:#999;border-left:5px solid #eee;padding-left:1em}.inheritance em{font-style:normal;font-weight:bold}.desc h2{font-weight:400;font-size:1.25em}.desc h3{font-size:1em}.desc dt code{background:inherit}.source summary,.git-link-div{color:#666;text-align:right;font-weight:400;font-size:.8em;text-transform:uppercase}.source summary > *{white-space:nowrap;cursor:pointer}.git-link{color:inherit;margin-left:1em}.source pre{max-height:500px;overflow:auto;margin:0}.source pre code{font-size:12px;overflow:visible}.hlist{list-style:none}.hlist li{display:inline}.hlist li:after{content:',\2002'}.hlist li:last-child:after{content:none}.hlist .hlist{display:inline;padding-left:1em}img{max-width:100%}td{padding:0 .5em}.admonition{padding:.1em .5em;margin-bottom:1em}.admonition-title{font-weight:bold}.admonition.note,.admonition.info,.admonition.important{background:#aef}.admonition.todo,.admonition.versionadded,.admonition.tip,.admonition.hint{background:#dfd}.admonition.warning,.admonition.versionchanged,.admonition.deprecated{background:#fd4}.admonition.error,.admonition.danger,.admonition.caution{background:lightpink}</style>
<style media="screen and (min-width: 700px)">@media screen and (min-width:700px){#sidebar{width:30%;height:100vh;overflow:auto;position:sticky;top:0}#content{width:70%;max-width:100ch;padding:3em 4em;border-left:1px solid #ddd}pre code{font-size:1em}.item .name{font-size:1em}main{display:flex;flex-direction:row-reverse;justify-content:flex-end}.toc ul ul,#index ul{padding-left:1.5em}.toc > ul > li{margin-top:.5em}}</style>
<style media="print">@media print{#sidebar h1{page-break-before:always}.source{display:none}}@media print{*{background:transparent !important;color:#000 !important;box-shadow:none !important;text-shadow:none !important}a[href]:after{content:" (" attr(href) ")";font-size:90%}a[href][title]:after{content:none}abbr[title]:after{content:" (" attr(title) ")"}.ir a:after,a[href^="javascript:"]:after,a[href^="#"]:after{content:""}pre,blockquote{border:1px solid #999;page-break-inside:avoid}thead{display:table-header-group}tr,img{page-break-inside:avoid}img{max-width:100% !important}@page{margin:0.5cm}p,h2,h3{orphans:3;widows:3}h1,h2,h3,h4,h5,h6{page-break-after:avoid}}</style>
</head>
<body>
<main>
<article id="content">
<header>
<h1 class="title">Package <code>conn</code></h1>
</header>
<section id="section-intro">
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">#!/usr/bin/env python3
&#39;&#39;&#39;
&#39;&#39;&#39;
from .core import node,nodes
from .configfile import configfile
from .connapp import connapp
from pkg_resources import get_distribution
__all__ = [&#34;node&#34;, &#34;nodes&#34;, &#34;configfile&#34;]
__version__ = &#34;2.0.10&#34;
__author__ = &#34;Federico Luzzi&#34;
__pdoc__ = {
&#39;core&#39;: False,
&#39;connapp&#39;: False,
}</code></pre>
</details>
</section>
<section>
</section>
<section>
</section>
<section>
</section>
<section>
<h2 class="section-title" id="header-classes">Classes</h2>
<dl>
<dt id="conn.configfile"><code class="flex name class">
<span>class <span class="ident">configfile</span></span>
<span>(</span><span>conf=None, *, key=None)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">class configfile:
def __init__(self, conf = None, *, key = None):
home = os.path.expanduser(&#34;~&#34;)
self.defaultdir = home + &#39;/.config/conn&#39;
self.defaultfile = self.defaultdir + &#39;/config.json&#39;
self.defaultkey = self.defaultdir + &#39;/.osk&#39;
Path(self.defaultdir).mkdir(parents=True, exist_ok=True)
if conf == None:
self.file = self.defaultfile
else:
self.file = conf
if key == None:
self.key = self.defaultkey
else:
self.key = key
if os.path.exists(self.file):
config = self.loadconfig(self.file)
else:
config = self.createconfig(self.file)
self.config = config[&#34;config&#34;]
self.connections = config[&#34;connections&#34;]
self.profiles = config[&#34;profiles&#34;]
if not os.path.exists(self.key):
self.createkey(self.key)
self.privatekey = RSA.import_key(open(self.key).read())
self.publickey = self.privatekey.publickey()
def loadconfig(self, conf):
jsonconf = open(conf)
return json.load(jsonconf)
def createconfig(self, conf):
defaultconfig = {&#39;config&#39;: {&#39;case&#39;: False, &#39;idletime&#39;: 30}, &#39;connections&#39;: {}, &#39;profiles&#39;: { &#34;default&#34;: { &#34;host&#34;:&#34;&#34;, &#34;protocol&#34;:&#34;ssh&#34;, &#34;port&#34;:&#34;&#34;, &#34;user&#34;:&#34;&#34;, &#34;password&#34;:&#34;&#34;, &#34;options&#34;:&#34;&#34;, &#34;logs&#34;:&#34;&#34; }}}
if not os.path.exists(conf):
with open(conf, &#34;w&#34;) as f:
json.dump(defaultconfig, f, indent = 4)
f.close()
os.chmod(conf, 0o600)
jsonconf = open(conf)
return json.load(jsonconf)
def saveconfig(self, conf):
newconfig = {&#34;config&#34;:{}, &#34;connections&#34;: {}, &#34;profiles&#34;: {}}
newconfig[&#34;config&#34;] = self.config
newconfig[&#34;connections&#34;] = self.connections
newconfig[&#34;profiles&#34;] = self.profiles
with open(conf, &#34;w&#34;) as f:
json.dump(newconfig, f, indent = 4)
f.close()
def createkey(self, keyfile):
key = RSA.generate(2048)
with open(keyfile,&#39;wb&#39;) as f:
f.write(key.export_key(&#39;PEM&#39;))
f.close()
os.chmod(keyfile, 0o600)
def _explode_unique(self, unique):
uniques = unique.split(&#34;@&#34;)
if not unique.startswith(&#34;@&#34;):
result = {&#34;id&#34;: uniques[0]}
else:
result = {}
if len(uniques) == 2:
result[&#34;folder&#34;] = uniques[1]
if result[&#34;folder&#34;] == &#34;&#34;:
return False
elif len(uniques) == 3:
result[&#34;folder&#34;] = uniques[2]
result[&#34;subfolder&#34;] = uniques[1]
if result[&#34;folder&#34;] == &#34;&#34; or result[&#34;subfolder&#34;] == &#34;&#34;:
return False
elif len(uniques) &gt; 3:
return False
return result
def getitem(self, unique, keys = None):
uniques = self._explode_unique(unique)
if unique.startswith(&#34;@&#34;):
if uniques.keys() &gt;= {&#34;folder&#34;, &#34;subfolder&#34;}:
folder = self.connections[uniques[&#34;folder&#34;]][uniques[&#34;subfolder&#34;]]
else:
folder = self.connections[uniques[&#34;folder&#34;]]
newfolder = folder.copy()
newfolder.pop(&#34;type&#34;)
for node in newfolder.keys():
if &#34;type&#34; in newfolder[node].keys():
newfolder[node].pop(&#34;type&#34;)
if keys == None:
return newfolder
else:
f_newfolder = dict((k, newfolder[k]) for k in keys)
return f_newfolder
else:
if uniques.keys() &gt;= {&#34;folder&#34;, &#34;subfolder&#34;}:
node = self.connections[uniques[&#34;folder&#34;]][uniques[&#34;subfolder&#34;]][uniques[&#34;id&#34;]]
elif &#34;folder&#34; in uniques.keys():
node = self.connections[uniques[&#34;folder&#34;]][uniques[&#34;id&#34;]]
else:
node = self.connections[uniques[&#34;id&#34;]]
newnode = node.copy()
newnode.pop(&#34;type&#34;)
return newnode
def _connections_add(self,*, id, host, folder=&#39;&#39;, subfolder=&#39;&#39;, options=&#39;&#39;, logs=&#39;&#39;, password=&#39;&#39;, port=&#39;&#39;, protocol=&#39;&#39;, user=&#39;&#39;, type = &#34;connection&#34; ):
if folder == &#39;&#39;:
self.connections[id] = {&#34;host&#34;: host, &#34;options&#34;: options, &#34;logs&#34;: logs, &#34;password&#34;: password, &#34;port&#34;: port, &#34;protocol&#34;: protocol, &#34;user&#34;: user, &#34;type&#34;: type}
elif folder != &#39;&#39; and subfolder == &#39;&#39;:
self.connections[folder][id] = {&#34;host&#34;: host, &#34;options&#34;: options, &#34;logs&#34;: logs, &#34;password&#34;: password, &#34;port&#34;: port, &#34;protocol&#34;: protocol, &#34;user&#34;: user, &#34;type&#34;: type}
elif folder != &#39;&#39; and subfolder != &#39;&#39;:
self.connections[folder][subfolder][id] = {&#34;host&#34;: host, &#34;options&#34;: options, &#34;logs&#34;: logs, &#34;password&#34;: password, &#34;port&#34;: port, &#34;protocol&#34;: protocol, &#34;user&#34;: user, &#34;type&#34;: type}
def _connections_del(self,*, id, folder=&#39;&#39;, subfolder=&#39;&#39;):
if folder == &#39;&#39;:
del self.connections[id]
elif folder != &#39;&#39; and subfolder == &#39;&#39;:
del self.connections[folder][id]
elif folder != &#39;&#39; and subfolder != &#39;&#39;:
del self.connections[folder][subfolder][id]
def _folder_add(self,*, folder, subfolder = &#39;&#39;):
if subfolder == &#39;&#39;:
if folder not in self.connections:
self.connections[folder] = {&#34;type&#34;: &#34;folder&#34;}
else:
if subfolder not in self.connections[folder]:
self.connections[folder][subfolder] = {&#34;type&#34;: &#34;subfolder&#34;}
def _folder_del(self,*, folder, subfolder=&#39;&#39;):
if subfolder == &#39;&#39;:
del self.connections[folder]
else:
del self.connections[folder][subfolder]
def _profiles_add(self,*, id, host = &#39;&#39;, options=&#39;&#39;, logs=&#39;&#39;, password=&#39;&#39;, port=&#39;&#39;, protocol=&#39;&#39;, user=&#39;&#39; ):
self.profiles[id] = {&#34;host&#34;: host, &#34;options&#34;: options, &#34;logs&#34;: logs, &#34;password&#34;: password, &#34;port&#34;: port, &#34;protocol&#34;: protocol, &#34;user&#34;: user}
def _profiles_del(self,*, id ):
del self.profiles[id]</code></pre>
</details>
<h3>Methods</h3>
<dl>
<dt id="conn.configfile.createconfig"><code class="name flex">
<span>def <span class="ident">createconfig</span></span>(<span>self, conf)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def createconfig(self, conf):
defaultconfig = {&#39;config&#39;: {&#39;case&#39;: False, &#39;idletime&#39;: 30}, &#39;connections&#39;: {}, &#39;profiles&#39;: { &#34;default&#34;: { &#34;host&#34;:&#34;&#34;, &#34;protocol&#34;:&#34;ssh&#34;, &#34;port&#34;:&#34;&#34;, &#34;user&#34;:&#34;&#34;, &#34;password&#34;:&#34;&#34;, &#34;options&#34;:&#34;&#34;, &#34;logs&#34;:&#34;&#34; }}}
if not os.path.exists(conf):
with open(conf, &#34;w&#34;) as f:
json.dump(defaultconfig, f, indent = 4)
f.close()
os.chmod(conf, 0o600)
jsonconf = open(conf)
return json.load(jsonconf)</code></pre>
</details>
</dd>
<dt id="conn.configfile.createkey"><code class="name flex">
<span>def <span class="ident">createkey</span></span>(<span>self, keyfile)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def createkey(self, keyfile):
key = RSA.generate(2048)
with open(keyfile,&#39;wb&#39;) as f:
f.write(key.export_key(&#39;PEM&#39;))
f.close()
os.chmod(keyfile, 0o600)</code></pre>
</details>
</dd>
<dt id="conn.configfile.getitem"><code class="name flex">
<span>def <span class="ident">getitem</span></span>(<span>self, unique, keys=None)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def getitem(self, unique, keys = None):
uniques = self._explode_unique(unique)
if unique.startswith(&#34;@&#34;):
if uniques.keys() &gt;= {&#34;folder&#34;, &#34;subfolder&#34;}:
folder = self.connections[uniques[&#34;folder&#34;]][uniques[&#34;subfolder&#34;]]
else:
folder = self.connections[uniques[&#34;folder&#34;]]
newfolder = folder.copy()
newfolder.pop(&#34;type&#34;)
for node in newfolder.keys():
if &#34;type&#34; in newfolder[node].keys():
newfolder[node].pop(&#34;type&#34;)
if keys == None:
return newfolder
else:
f_newfolder = dict((k, newfolder[k]) for k in keys)
return f_newfolder
else:
if uniques.keys() &gt;= {&#34;folder&#34;, &#34;subfolder&#34;}:
node = self.connections[uniques[&#34;folder&#34;]][uniques[&#34;subfolder&#34;]][uniques[&#34;id&#34;]]
elif &#34;folder&#34; in uniques.keys():
node = self.connections[uniques[&#34;folder&#34;]][uniques[&#34;id&#34;]]
else:
node = self.connections[uniques[&#34;id&#34;]]
newnode = node.copy()
newnode.pop(&#34;type&#34;)
return newnode</code></pre>
</details>
</dd>
<dt id="conn.configfile.loadconfig"><code class="name flex">
<span>def <span class="ident">loadconfig</span></span>(<span>self, conf)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def loadconfig(self, conf):
jsonconf = open(conf)
return json.load(jsonconf)</code></pre>
</details>
</dd>
<dt id="conn.configfile.saveconfig"><code class="name flex">
<span>def <span class="ident">saveconfig</span></span>(<span>self, conf)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def saveconfig(self, conf):
newconfig = {&#34;config&#34;:{}, &#34;connections&#34;: {}, &#34;profiles&#34;: {}}
newconfig[&#34;config&#34;] = self.config
newconfig[&#34;connections&#34;] = self.connections
newconfig[&#34;profiles&#34;] = self.profiles
with open(conf, &#34;w&#34;) as f:
json.dump(newconfig, f, indent = 4)
f.close()</code></pre>
</details>
</dd>
</dl>
</dd>
<dt id="conn.node"><code class="flex name class">
<span>class <span class="ident">node</span></span>
<span>(</span><span>unique, host, options='', logs='', password='', port='', protocol='', user='', config='')</span>
</code></dt>
<dd>
<div class="desc"><p>This class generates a node object. Containts all the information and methods to connect and interact with a device using ssh or telnet.</p>
<p>Attributes:
</p>
<pre><code>- output (str) -- Output of the commands you ran with run or test
-- method.
- result(bool) -- True if expected value is found after running
-- the commands using test method.
</code></pre>
<p>Parameters:
</p>
<pre><code>- unique (str) -- Unique name to assign to the node.
- host (str) -- IP address or hostname of the node.
</code></pre>
<p>Optional Parameters:
</p>
<pre><code>- options (str) -- Additional options to pass the ssh/telnet for
-- connection.
- logs (str) -- Path/file for storing the logs. You can use
-- ${unique},${host}, ${port}, ${user}, ${protocol}
-- as variables.
- password (str) -- Encrypted or plaintext password.
- port (str) -- Port to connect to node, default 22 for ssh and 23
-- for telnet.
- protocol (str) -- Select ssh or telnet. Default is ssh.
- user (str) -- Username to of the node.
- config (obj) -- Pass the object created with class configfile with
-- key for decryption and extra configuration if you
-- are using connection manager.
</code></pre></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">class node:
&#39;&#39;&#39; This class generates a node object. Containts all the information and methods to connect and interact with a device using ssh or telnet.
Attributes:
- output (str) -- Output of the commands you ran with run or test
-- method.
- result(bool) -- True if expected value is found after running
-- the commands using test method.
&#39;&#39;&#39;
def __init__(self, unique, host, options=&#39;&#39;, logs=&#39;&#39;, password=&#39;&#39;, port=&#39;&#39;, protocol=&#39;&#39;, user=&#39;&#39;, config=&#39;&#39;):
&#39;&#39;&#39;
Parameters:
- unique (str) -- Unique name to assign to the node.
- host (str) -- IP address or hostname of the node.
Optional Parameters:
- options (str) -- Additional options to pass the ssh/telnet for
-- connection.
- logs (str) -- Path/file for storing the logs. You can use
-- ${unique},${host}, ${port}, ${user}, ${protocol}
-- as variables.
- password (str) -- Encrypted or plaintext password.
- port (str) -- Port to connect to node, default 22 for ssh and 23
-- for telnet.
- protocol (str) -- Select ssh or telnet. Default is ssh.
- user (str) -- Username to of the node.
- config (obj) -- Pass the object created with class configfile with
-- key for decryption and extra configuration if you
-- are using connection manager.
&#39;&#39;&#39;
if config == &#39;&#39;:
self.idletime = 0
self.key = None
else:
self.idletime = config.config[&#34;idletime&#34;]
self.key = config.key
self.unique = unique
attr = {&#34;host&#34;: host, &#34;logs&#34;: logs, &#34;options&#34;:options, &#34;port&#34;: port, &#34;protocol&#34;: protocol, &#34;user&#34;: user}
for key in attr:
profile = re.search(&#34;^@(.*)&#34;, attr[key])
if profile and config != &#39;&#39;:
setattr(self,key,config.profiles[profile.group(1)][key])
elif attr[key] == &#39;&#39; and key == &#34;protocol&#34;:
try:
setattr(self,key,config.profiles[&#34;default&#34;][key])
except:
setattr(self,key,&#34;ssh&#34;)
else:
setattr(self,key,attr[key])
if isinstance(password,list):
self.password = []
for i, s in enumerate(password):
profile = re.search(&#34;^@(.*)&#34;, password[i])
if profile and config != &#39;&#39;:
self.password.append(config.profiles[profile.group(1)][&#34;password&#34;])
else:
self.password = [password]
def __passtx(self, passwords, *, keyfile=None):
# decrypts passwords, used by other methdos.
dpass = []
if keyfile is None:
keyfile = self.key
if keyfile is not None:
key = RSA.import_key(open(keyfile).read())
decryptor = PKCS1_OAEP.new(key)
for passwd in passwords:
if not re.match(&#39;^b[\&#34;\&#39;].+[\&#34;\&#39;]$&#39;, passwd):
dpass.append(passwd)
else:
try:
decrypted = decryptor.decrypt(ast.literal_eval(passwd)).decode(&#34;utf-8&#34;)
dpass.append(decrypted)
except:
raise ValueError(&#34;Missing or corrupted key&#34;)
return dpass
def _logfile(self, logfile = None):
# translate logs variables and generate logs path.
if logfile == None:
logfile = self.logs
logfile = logfile.replace(&#34;${unique}&#34;, self.unique)
logfile = logfile.replace(&#34;${host}&#34;, self.host)
logfile = logfile.replace(&#34;${port}&#34;, self.port)
logfile = logfile.replace(&#34;${user}&#34;, self.user)
logfile = logfile.replace(&#34;${protocol}&#34;, self.protocol)
now = datetime.datetime.now()
dateconf = re.search(r&#39;\$\{date \&#39;(.*)\&#39;}&#39;, logfile)
if dateconf:
logfile = re.sub(r&#39;\$\{date (.*)}&#39;,now.strftime(dateconf.group(1)), logfile)
return logfile
def _logclean(self, logfile, var = False):
#Remove special ascii characters and other stuff from logfile.
if var == False:
t = open(logfile, &#34;r&#34;).read()
else:
t = logfile
t = t.replace(&#34;\n&#34;,&#34;&#34;,1).replace(&#34;\a&#34;,&#34;&#34;)
t = t.replace(&#39;\n\n&#39;, &#39;\n&#39;)
t = re.sub(r&#39;.\[K&#39;, &#39;&#39;, t)
while True:
tb = re.sub(&#39;.\b&#39;, &#39;&#39;, t, count=1)
if len(t) == len(tb):
break
t = tb
ansi_escape = re.compile(r&#39;\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/ ]*[@-~])&#39;)
t = ansi_escape.sub(&#39;&#39;, t)
if var == False:
d = open(logfile, &#34;w&#34;)
d.write(t)
d.close()
return
else:
return t
def interact(self, debug = False):
&#39;&#39;&#39;
Allow user to interact with the node directly, mostly used by connection manager.
Optional Parameters:
- debug (bool) -- If True, display all the connecting information
-- before interact. Default False.
&#39;&#39;&#39;
connect = self._connect(debug = debug)
if connect == True:
size = re.search(&#39;columns=([0-9]+).*lines=([0-9]+)&#39;,str(os.get_terminal_size()))
self.child.setwinsize(int(size.group(2)),int(size.group(1)))
print(&#34;Connected to &#34; + self.unique + &#34; at &#34; + self.host + (&#34;:&#34; if self.port != &#39;&#39; else &#39;&#39;) + self.port + &#34; via: &#34; + self.protocol)
if &#39;logfile&#39; in dir(self):
self.child.logfile_read = open(self.logfile, &#34;wb&#34;)
elif debug:
self.child.logfile_read = None
if &#39;missingtext&#39; in dir(self):
print(self.child.after.decode(), end=&#39;&#39;)
self.child.interact()
if &#34;logfile&#34; in dir(self) and not debug:
self._logclean(self.logfile)
else:
print(connect)
exit(1)
def run(self, commands,*, folder = &#39;&#39;, prompt = r&#39;&gt;$|#$|\$$|&gt;.$|#.$|\$.$&#39;, stdout = False):
&#39;&#39;&#39;
Run a command or list of commands on the node and return the output.
Parameters:
- commands (str/list) -- Commands to run on the node. Should be
-- str or a list of str.
Optional Named Parameters:
- folder (str) -- Path where output log should be stored, leave
-- empty to disable logging.
- prompt (str) -- Prompt to be expected after a command is finished
-- running. Usually linux uses &#34;&gt;&#34; or EOF while
-- routers use &#34;&gt;&#34; or &#34;#&#34;. The default value should
-- work for most nodes. Change it if your connection
-- need some special symbol.
- stdout (bool) -- Set True to send the command output to stdout.
-- default False.
Returns:
str -&gt; Output of the commands you ran on the node.
&#39;&#39;&#39;
connect = self._connect()
if connect == True:
expects = [prompt, pexpect.EOF]
output = &#39;&#39;
if isinstance(commands, list):
for c in commands:
result = self.child.expect(expects)
self.child.sendline(c)
if result == 0:
output = output + self.child.before.decode() + self.child.after.decode()
if result == 1:
output = output + self.child.before.decode()
else:
result = self.child.expect(expects)
self.child.sendline(commands)
if result == 0:
output = output + self.child.before.decode() + self.child.after.decode()
if result == 1:
output = output + self.child.before.decode()
result = self.child.expect(expects)
if result == 0:
output = output + self.child.before.decode() + self.child.after.decode()
if result == 1:
output = output + self.child.before.decode()
self.child.close()
output = output.lstrip()
if stdout == True:
print(output)
if folder != &#39;&#39;:
with open(folder + &#34;/&#34; + self.unique, &#34;w&#34;) as f:
f.write(output)
f.close()
self._logclean(folder + &#34;/&#34; + self.unique)
self.output = output
return output
else:
self.output = connect
return connect
def test(self, commands, expected, *, prompt = r&#39;&gt;$|#$|\$$|&gt;.$|#.$|\$.$&#39;):
&#39;&#39;&#39;
Run a command or list of commands on the node, then check if expected value appears on the output after the last command.
Parameters:
- commands (str/list) -- Commands to run on the node. Should be
-- str or list of str.
- expected (str) -- Expected text to appear after running
-- all the commands on the node.
Optional Named Parameters:
- prompt (str) -- Prompt to be expected after a command is finished
-- running. Usually linux uses &#34;&gt;&#34; or EOF while
-- routers use &#34;&gt;&#34; or &#34;#&#34;. The default value should
-- work for most nodes. Change it if your connection
-- need some special symbol.
### Returns:
bool -- true if expected value is found after running the commands
false if prompt is found before.
&#39;&#39;&#39;
connect = self._connect()
if connect == True:
expects = [prompt, pexpect.EOF]
output = &#39;&#39;
if isinstance(commands, list):
for c in commands:
result = self.child.expect(expects)
self.child.sendline(c)
if result == 0:
output = output + self.child.before.decode() + self.child.after.decode()
if result == 1:
output = output + self.child.before.decode()
else:
self.child.expect(expects)
self.child.sendline(commands)
output = output + self.child.before.decode() + self.child.after.decode()
expects = [expected, prompt, pexpect.EOF]
results = self.child.expect(expects)
if results == 0:
self.child.close()
self.result = True
output = output + self.child.before.decode() + self.child.after.decode()
output = output.lstrip()
self.output = output
return True
if results in [1, 2]:
self.child.close()
self.result = False
if results == 1:
output = output + self.child.before.decode() + self.child.after.decode()
elif results == 2:
output = output + self.child.before.decode()
output = output.lstrip()
self.output = output
return False
else:
self.result = None
self.output = connect
return connect
def _connect(self, debug = False):
# Method to connect to the node, it parse all the information, create the ssh/telnet command and login to the node.
if self.protocol == &#34;ssh&#34;:
cmd = &#34;ssh&#34;
if self.idletime &gt; 0:
cmd = cmd + &#34; -o ServerAliveInterval=&#34; + str(self.idletime)
if self.user == &#39;&#39;:
cmd = cmd + &#34; -t {}&#34;.format(self.host)
else:
cmd = cmd + &#34; -t {}&#34;.format(&#34;@&#34;.join([self.user,self.host]))
if self.port != &#39;&#39;:
cmd = cmd + &#34; -p &#34; + self.port
if self.options != &#39;&#39;:
cmd = cmd + &#34; &#34; + self.options
if self.logs != &#39;&#39;:
self.logfile = self._logfile()
if self.password[0] != &#39;&#39;:
passwords = self.__passtx(self.password)
else:
passwords = []
expects = [&#39;yes/no&#39;, &#39;refused&#39;, &#39;supported&#39;, &#39;cipher&#39;, &#39;sage&#39;, &#39;timeout&#39;, &#39;unavailable&#39;, &#39;closed&#39;, &#39;[p|P]assword:|[u|U]sername:&#39;, r&#39;&gt;$|#$|\$$|&gt;.$|#.$|\$.$&#39;, &#39;suspend&#39;, pexpect.EOF, &#34;No route to host&#34;, &#34;resolve hostname&#34;, &#34;no matching host key&#34;]
elif self.protocol == &#34;telnet&#34;:
cmd = &#34;telnet &#34; + self.host
if self.port != &#39;&#39;:
cmd = cmd + &#34; &#34; + self.port
if self.options != &#39;&#39;:
cmd = cmd + &#34; &#34; + self.options
if self.logs != &#39;&#39;:
self.logfile = self._logfile()
if self.password[0] != &#39;&#39;:
passwords = self.__passtx(self.password)
else:
passwords = []
expects = [&#39;[u|U]sername:&#39;, &#39;refused&#39;, &#39;supported&#39;, &#39;cipher&#39;, &#39;sage&#39;, &#39;timeout&#39;, &#39;unavailable&#39;, &#39;closed&#39;, &#39;[p|P]assword:&#39;, r&#39;&gt;$|#$|\$$|&gt;.$|#.$|\$.$&#39;, &#39;suspend&#39;, pexpect.EOF, &#34;No route to host&#34;, &#34;resolve hostname&#34;, &#34;no matching host key&#34;]
else:
raise ValueError(&#34;Invalid protocol: &#34; + self.protocol)
child = pexpect.spawn(cmd)
if debug:
child.logfile_read = sys.stdout.buffer
if len(passwords) &gt; 0:
loops = len(passwords)
else:
loops = 1
endloop = False
for i in range(0, loops):
while True:
results = child.expect(expects)
if results == 0:
if self.protocol == &#34;ssh&#34;:
child.sendline(&#39;yes&#39;)
elif self.protocol == &#34;telnet&#34;:
if self.user != &#39;&#39;:
child.sendline(self.user)
else:
self.missingtext = True
break
if results in [1, 2, 3, 4, 5, 6, 7, 12, 13, 14]:
child.close()
return &#34;Connection failed code:&#34; + str(results)
if results == 8:
if len(passwords) &gt; 0:
child.sendline(passwords[i])
else:
self.missingtext = True
break
if results in [9, 11]:
endloop = True
child.sendline()
break
if results == 10:
child.sendline(&#34;\r&#34;)
sleep(2)
if endloop:
break
child.readline(0)
self.child = child
return True</code></pre>
</details>
<h3>Methods</h3>
<dl>
<dt id="conn.node.interact"><code class="name flex">
<span>def <span class="ident">interact</span></span>(<span>self, debug=False)</span>
</code></dt>
<dd>
<div class="desc"><p>Allow user to interact with the node directly, mostly used by connection manager.</p>
<p>Optional Parameters:
</p>
<pre><code>- debug (bool) -- If True, display all the connecting information
-- before interact. Default False.
</code></pre></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def interact(self, debug = False):
&#39;&#39;&#39;
Allow user to interact with the node directly, mostly used by connection manager.
Optional Parameters:
- debug (bool) -- If True, display all the connecting information
-- before interact. Default False.
&#39;&#39;&#39;
connect = self._connect(debug = debug)
if connect == True:
size = re.search(&#39;columns=([0-9]+).*lines=([0-9]+)&#39;,str(os.get_terminal_size()))
self.child.setwinsize(int(size.group(2)),int(size.group(1)))
print(&#34;Connected to &#34; + self.unique + &#34; at &#34; + self.host + (&#34;:&#34; if self.port != &#39;&#39; else &#39;&#39;) + self.port + &#34; via: &#34; + self.protocol)
if &#39;logfile&#39; in dir(self):
self.child.logfile_read = open(self.logfile, &#34;wb&#34;)
elif debug:
self.child.logfile_read = None
if &#39;missingtext&#39; in dir(self):
print(self.child.after.decode(), end=&#39;&#39;)
self.child.interact()
if &#34;logfile&#34; in dir(self) and not debug:
self._logclean(self.logfile)
else:
print(connect)
exit(1)</code></pre>
</details>
</dd>
<dt id="conn.node.run"><code class="name flex">
<span>def <span class="ident">run</span></span>(<span>self, commands, *, folder='', prompt=&#x27;&gt;$|#$|\\$$|&gt;.$|#.$|\\$.$&#x27;, stdout=False)</span>
</code></dt>
<dd>
<div class="desc"><p>Run a command or list of commands on the node and return the output.</p>
<p>Parameters:
</p>
<pre><code>- commands (str/list) -- Commands to run on the node. Should be
-- str or a list of str.
</code></pre>
<p>Optional Named Parameters:
</p>
<pre><code>- folder (str) -- Path where output log should be stored, leave
-- empty to disable logging.
- prompt (str) -- Prompt to be expected after a command is finished
-- running. Usually linux uses "&gt;" or EOF while
-- routers use "&gt;" or "#". The default value should
-- work for most nodes. Change it if your connection
-- need some special symbol.
- stdout (bool) -- Set True to send the command output to stdout.
-- default False.
</code></pre>
<p>Returns:
</p>
<pre><code>str -&gt; Output of the commands you ran on the node.
</code></pre></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def run(self, commands,*, folder = &#39;&#39;, prompt = r&#39;&gt;$|#$|\$$|&gt;.$|#.$|\$.$&#39;, stdout = False):
&#39;&#39;&#39;
Run a command or list of commands on the node and return the output.
Parameters:
- commands (str/list) -- Commands to run on the node. Should be
-- str or a list of str.
Optional Named Parameters:
- folder (str) -- Path where output log should be stored, leave
-- empty to disable logging.
- prompt (str) -- Prompt to be expected after a command is finished
-- running. Usually linux uses &#34;&gt;&#34; or EOF while
-- routers use &#34;&gt;&#34; or &#34;#&#34;. The default value should
-- work for most nodes. Change it if your connection
-- need some special symbol.
- stdout (bool) -- Set True to send the command output to stdout.
-- default False.
Returns:
str -&gt; Output of the commands you ran on the node.
&#39;&#39;&#39;
connect = self._connect()
if connect == True:
expects = [prompt, pexpect.EOF]
output = &#39;&#39;
if isinstance(commands, list):
for c in commands:
result = self.child.expect(expects)
self.child.sendline(c)
if result == 0:
output = output + self.child.before.decode() + self.child.after.decode()
if result == 1:
output = output + self.child.before.decode()
else:
result = self.child.expect(expects)
self.child.sendline(commands)
if result == 0:
output = output + self.child.before.decode() + self.child.after.decode()
if result == 1:
output = output + self.child.before.decode()
result = self.child.expect(expects)
if result == 0:
output = output + self.child.before.decode() + self.child.after.decode()
if result == 1:
output = output + self.child.before.decode()
self.child.close()
output = output.lstrip()
if stdout == True:
print(output)
if folder != &#39;&#39;:
with open(folder + &#34;/&#34; + self.unique, &#34;w&#34;) as f:
f.write(output)
f.close()
self._logclean(folder + &#34;/&#34; + self.unique)
self.output = output
return output
else:
self.output = connect
return connect</code></pre>
</details>
</dd>
<dt id="conn.node.test"><code class="name flex">
<span>def <span class="ident">test</span></span>(<span>self, commands, expected, *, prompt=&#x27;&gt;$|#$|\\$$|&gt;.$|#.$|\\$.$&#x27;)</span>
</code></dt>
<dd>
<div class="desc"><p>Run a command or list of commands on the node, then check if expected value appears on the output after the last command.</p>
<p>Parameters:
</p>
<pre><code>- commands (str/list) -- Commands to run on the node. Should be
-- str or list of str.
- expected (str) -- Expected text to appear after running
-- all the commands on the node.
</code></pre>
<p>Optional Named Parameters: </p>
<pre><code>- prompt (str) -- Prompt to be expected after a command is finished
-- running. Usually linux uses "&gt;" or EOF while
-- routers use "&gt;" or "#". The default value should
-- work for most nodes. Change it if your connection
-- need some special symbol.
</code></pre>
<h3 id="returns">Returns:</h3>
<pre><code>bool -- true if expected value is found after running the commands
false if prompt is found before.
</code></pre></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def test(self, commands, expected, *, prompt = r&#39;&gt;$|#$|\$$|&gt;.$|#.$|\$.$&#39;):
&#39;&#39;&#39;
Run a command or list of commands on the node, then check if expected value appears on the output after the last command.
Parameters:
- commands (str/list) -- Commands to run on the node. Should be
-- str or list of str.
- expected (str) -- Expected text to appear after running
-- all the commands on the node.
Optional Named Parameters:
- prompt (str) -- Prompt to be expected after a command is finished
-- running. Usually linux uses &#34;&gt;&#34; or EOF while
-- routers use &#34;&gt;&#34; or &#34;#&#34;. The default value should
-- work for most nodes. Change it if your connection
-- need some special symbol.
### Returns:
bool -- true if expected value is found after running the commands
false if prompt is found before.
&#39;&#39;&#39;
connect = self._connect()
if connect == True:
expects = [prompt, pexpect.EOF]
output = &#39;&#39;
if isinstance(commands, list):
for c in commands:
result = self.child.expect(expects)
self.child.sendline(c)
if result == 0:
output = output + self.child.before.decode() + self.child.after.decode()
if result == 1:
output = output + self.child.before.decode()
else:
self.child.expect(expects)
self.child.sendline(commands)
output = output + self.child.before.decode() + self.child.after.decode()
expects = [expected, prompt, pexpect.EOF]
results = self.child.expect(expects)
if results == 0:
self.child.close()
self.result = True
output = output + self.child.before.decode() + self.child.after.decode()
output = output.lstrip()
self.output = output
return True
if results in [1, 2]:
self.child.close()
self.result = False
if results == 1:
output = output + self.child.before.decode() + self.child.after.decode()
elif results == 2:
output = output + self.child.before.decode()
output = output.lstrip()
self.output = output
return False
else:
self.result = None
self.output = connect
return connect</code></pre>
</details>
</dd>
</dl>
</dd>
<dt id="conn.nodes"><code class="flex name class">
<span>class <span class="ident">nodes</span></span>
<span>(</span><span>nodes: dict, config='')</span>
</code></dt>
<dd>
<div class="desc"><p>This class generates a nodes object. Contains a list of node class objects and methods to run multiple tasks on nodes simultaneously.</p>
<h3 id="attributes">Attributes:</h3>
<pre><code>- nodelist (list): List of node class objects passed to the init
function.
- output (dict): Dictionary formed by nodes unique as keys,
output of the commands you ran on the node as
value. Created after running methods run or test.
- result (dict): Dictionary formed by nodes unique as keys, value
is True if expected value is found after running
the commands, False if prompt is found before.
Created after running method test.
- &lt;unique&gt; (obj): For each item in nodelist, there is an attribute
generated with the node unique.
</code></pre>
<h3 id="parameters">Parameters:</h3>
<pre><code>- nodes (dict): Dictionary formed by node information:
Keys: Unique name for each node.
Mandatory Subkeys: host(str).
Optional Subkeys: options(str), logs(str), password(str),
port(str), protocol(str), user(str).
For reference on subkeys check node class.
</code></pre>
<p>Optional Parameters:
</p>
<pre><code>- config (obj): Pass the object created with class configfile with key
for decryption and extra configuration if you are using
connection manager.
</code></pre></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">class nodes:
&#39;&#39;&#39; This class generates a nodes object. Contains a list of node class objects and methods to run multiple tasks on nodes simultaneously.
### Attributes:
- nodelist (list): List of node class objects passed to the init
function.
- output (dict): Dictionary formed by nodes unique as keys,
output of the commands you ran on the node as
value. Created after running methods run or test.
- result (dict): Dictionary formed by nodes unique as keys, value
is True if expected value is found after running
the commands, False if prompt is found before.
Created after running method test.
- &lt;unique&gt; (obj): For each item in nodelist, there is an attribute
generated with the node unique.
&#39;&#39;&#39;
def __init__(self, nodes: dict, config = &#39;&#39;):
&#39;&#39;&#39;
### Parameters:
- nodes (dict): Dictionary formed by node information:
Keys: Unique name for each node.
Mandatory Subkeys: host(str).
Optional Subkeys: options(str), logs(str), password(str),
port(str), protocol(str), user(str).
For reference on subkeys check node class.
Optional Parameters:
- config (obj): Pass the object created with class configfile with key
for decryption and extra configuration if you are using
connection manager.
&#39;&#39;&#39;
self.nodelist = []
self.config = config
for n in nodes:
this = node(n, **nodes[n], config = config)
self.nodelist.append(this)
setattr(self,n,this)
def _splitlist(self, lst, n):
#split a list in lists of n members.
for i in range(0, len(lst), n):
yield lst[i:i + n]
def run(self, commands,*, folder = None, prompt = None, stdout = None, parallel = 10):
&#39;&#39;&#39;
Run a command or list of commands on all the nodes in nodelist.
Parameters:
commands (str/list): Commands to run on the node. Should be a str or a list of str.
Optional Named Parameters:
folder (str): Path where output log should be stored, leave empty to disable logging.
prompt (str): Prompt to be expected after a command is finished running. Usually linux uses &#34;&gt;&#34; or EOF while routers use &#34;&gt;&#34; or &#34;#&#34;. The default value should work for most nodes. Change it if your connection need some special symbol.
stdout (bool): Set True to send the command output to stdout. default False.
parallel (int): Number of nodes to run the commands simultaneously. Default is 10, if there are more nodes that this value, nodes are groups in groups with max this number of members.
Returns:
dict: Dictionary formed by nodes unique as keys, Output of the commands you ran on the node as value.
&#39;&#39;&#39;
args = {}
args[&#34;commands&#34;] = commands
if folder != None:
args[&#34;folder&#34;] = folder
if prompt != None:
args[&#34;prompt&#34;] = prompt
if stdout != None:
args[&#34;stdout&#34;] = stdout
output = {}
tasks = []
for n in self.nodelist:
tasks.append(threading.Thread(target=n.run, kwargs=args))
taskslist = list(self._splitlist(tasks, parallel))
for t in taskslist:
for i in t:
i.start()
for i in t:
i.join()
for i in self.nodelist:
output[i.unique] = i.output
self.output = output
return output
def test(self, commands, expected, *, prompt = None, parallel = 10):
&#39;&#39;&#39;
Run a command or list of commands on all the nodes in nodelist, then check if expected value appears on the output after the last command.
Parameters:
commands (str/list): Commands to run on the node. Should be a str or a list of str.
commands (str): Expected text to appear after running all the commands on the node.
Optional Named Parameters:
prompt (str): Prompt to be expected after a command is finished running. Usually linux uses &#34;&gt;&#34; or EOF while routers use &#34;&gt;&#34; or &#34;#&#34;. The default value should work for most nodes. Change it if your connection need some special symbol.
Returns:
dict: Dictionary formed by nodes unique as keys, value is True if expected value is found after running the commands, False if prompt is found before.
&#39;&#39;&#39;
args = {}
args[&#34;commands&#34;] = commands
args[&#34;expected&#34;] = expected
if prompt != None:
args[&#34;prompt&#34;] = prompt
output = {}
result = {}
tasks = []
for n in self.nodelist:
tasks.append(threading.Thread(target=n.test, kwargs=args))
taskslist = list(self._splitlist(tasks, parallel))
for t in taskslist:
for i in t:
i.start()
for i in t:
i.join()
for i in self.nodelist:
result[i.unique] = i.result
output[i.unique] = i.output
self.output = output
self.result = result
return result</code></pre>
</details>
<h3>Methods</h3>
<dl>
<dt id="conn.nodes.run"><code class="name flex">
<span>def <span class="ident">run</span></span>(<span>self, commands, *, folder=None, prompt=None, stdout=None, parallel=10)</span>
</code></dt>
<dd>
<div class="desc"><p>Run a command or list of commands on all the nodes in nodelist.</p>
<p>Parameters:<br>
commands (str/list): Commands to run on the node. Should be a str or a list of str.</p>
<p>Optional Named Parameters:<br>
folder
(str): Path where output log should be stored, leave empty to disable logging.<br>
prompt
(str): Prompt to be expected after a command is finished running. Usually linux uses
"&gt;" or EOF while routers use "&gt;" or "#". The default value should work for most nodes. Change it if your connection need some special symbol.<br>
stdout
(bool): Set True to send the command output to stdout. default False.<br>
parallel (int): Number of nodes to run the commands simultaneously. Default is 10, if there are more nodes that this value, nodes are groups in groups with max this number of members.</p>
<p>Returns:<br>
dict: Dictionary formed by nodes unique as keys, Output of the commands you ran on the node as value.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def run(self, commands,*, folder = None, prompt = None, stdout = None, parallel = 10):
&#39;&#39;&#39;
Run a command or list of commands on all the nodes in nodelist.
Parameters:
commands (str/list): Commands to run on the node. Should be a str or a list of str.
Optional Named Parameters:
folder (str): Path where output log should be stored, leave empty to disable logging.
prompt (str): Prompt to be expected after a command is finished running. Usually linux uses &#34;&gt;&#34; or EOF while routers use &#34;&gt;&#34; or &#34;#&#34;. The default value should work for most nodes. Change it if your connection need some special symbol.
stdout (bool): Set True to send the command output to stdout. default False.
parallel (int): Number of nodes to run the commands simultaneously. Default is 10, if there are more nodes that this value, nodes are groups in groups with max this number of members.
Returns:
dict: Dictionary formed by nodes unique as keys, Output of the commands you ran on the node as value.
&#39;&#39;&#39;
args = {}
args[&#34;commands&#34;] = commands
if folder != None:
args[&#34;folder&#34;] = folder
if prompt != None:
args[&#34;prompt&#34;] = prompt
if stdout != None:
args[&#34;stdout&#34;] = stdout
output = {}
tasks = []
for n in self.nodelist:
tasks.append(threading.Thread(target=n.run, kwargs=args))
taskslist = list(self._splitlist(tasks, parallel))
for t in taskslist:
for i in t:
i.start()
for i in t:
i.join()
for i in self.nodelist:
output[i.unique] = i.output
self.output = output
return output</code></pre>
</details>
</dd>
<dt id="conn.nodes.test"><code class="name flex">
<span>def <span class="ident">test</span></span>(<span>self, commands, expected, *, prompt=None, parallel=10)</span>
</code></dt>
<dd>
<div class="desc"><p>Run a command or list of commands on all the nodes in nodelist, then check if expected value appears on the output after the last command.</p>
<p>Parameters:<br>
commands (str/list): Commands to run on the node. Should be a str or a list of str.<br>
commands (str): Expected text to appear after running all the commands on the node.</p>
<p>Optional Named Parameters:<br>
prompt
(str): Prompt to be expected after a command is finished running. Usually linux uses
"&gt;" or EOF while routers use "&gt;" or "#". The default value should work for most nodes. Change it if your connection need some special symbol.</p>
<p>Returns:<br>
dict: Dictionary formed by nodes unique as keys, value is True if expected value is found after running the commands, False if prompt is found before.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def test(self, commands, expected, *, prompt = None, parallel = 10):
&#39;&#39;&#39;
Run a command or list of commands on all the nodes in nodelist, then check if expected value appears on the output after the last command.
Parameters:
commands (str/list): Commands to run on the node. Should be a str or a list of str.
commands (str): Expected text to appear after running all the commands on the node.
Optional Named Parameters:
prompt (str): Prompt to be expected after a command is finished running. Usually linux uses &#34;&gt;&#34; or EOF while routers use &#34;&gt;&#34; or &#34;#&#34;. The default value should work for most nodes. Change it if your connection need some special symbol.
Returns:
dict: Dictionary formed by nodes unique as keys, value is True if expected value is found after running the commands, False if prompt is found before.
&#39;&#39;&#39;
args = {}
args[&#34;commands&#34;] = commands
args[&#34;expected&#34;] = expected
if prompt != None:
args[&#34;prompt&#34;] = prompt
output = {}
result = {}
tasks = []
for n in self.nodelist:
tasks.append(threading.Thread(target=n.test, kwargs=args))
taskslist = list(self._splitlist(tasks, parallel))
for t in taskslist:
for i in t:
i.start()
for i in t:
i.join()
for i in self.nodelist:
result[i.unique] = i.result
output[i.unique] = i.output
self.output = output
self.result = result
return result</code></pre>
</details>
</dd>
</dl>
</dd>
</dl>
</section>
</article>
<nav id="sidebar">
<h1>Index</h1>
<div class="toc">
<ul></ul>
</div>
<ul id="index">
<li><h3><a href="#header-classes">Classes</a></h3>
<ul>
<li>
<h4><code><a title="conn.configfile" href="#conn.configfile">configfile</a></code></h4>
<ul class="">
<li><code><a title="conn.configfile.createconfig" href="#conn.configfile.createconfig">createconfig</a></code></li>
<li><code><a title="conn.configfile.createkey" href="#conn.configfile.createkey">createkey</a></code></li>
<li><code><a title="conn.configfile.getitem" href="#conn.configfile.getitem">getitem</a></code></li>
<li><code><a title="conn.configfile.loadconfig" href="#conn.configfile.loadconfig">loadconfig</a></code></li>
<li><code><a title="conn.configfile.saveconfig" href="#conn.configfile.saveconfig">saveconfig</a></code></li>
</ul>
</li>
<li>
<h4><code><a title="conn.node" href="#conn.node">node</a></code></h4>
<ul class="">
<li><code><a title="conn.node.interact" href="#conn.node.interact">interact</a></code></li>
<li><code><a title="conn.node.run" href="#conn.node.run">run</a></code></li>
<li><code><a title="conn.node.test" href="#conn.node.test">test</a></code></li>
</ul>
</li>
<li>
<h4><code><a title="conn.nodes" href="#conn.nodes">nodes</a></code></h4>
<ul class="">
<li><code><a title="conn.nodes.run" href="#conn.nodes.run">run</a></code></li>
<li><code><a title="conn.nodes.test" href="#conn.nodes.test">test</a></code></li>
</ul>
</li>
</ul>
</li>
</ul>
</nav>
</main>
<footer id="footer">
<p>Generated by <a href="https://pdoc3.github.io/pdoc" title="pdoc: Python API documentation generator"><cite>pdoc</cite> 0.10.0</a>.</p>
</footer>
</body>
</html>
Reference in New Issue Copy Permalink