Skip to content

Instantly share code, notes, and snippets.

@ggurdanikita

ggurdanikita/Кол-во занятых портов коммутатора

Created March 4, 2020 12:12
Show Gist options
  • Save ggurdanikita/6cb22aa6f8e3207c2fee36dcf3243576 to your computer and use it in GitHub Desktop.
Save ggurdanikita/6cb22aa6f8e3207c2fee36dcf3243576 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
Кол-во занятых портов коммутатора
snmp_switch = <fixtures.switch.Switch object at 0x7fb86c3d1da0>
admin_auth = <misc.service.ServiceManager object at 0x7fb874804358>
task_wait = <function task_wait.<locals>._wait at 0x7fb874b26e18>
server = {'chassis': None,
'configuration': None,
'diag_error': None,
'domain': None,
'hardware_info': None,
'id': 35,
'i...': 'xbepkfbliv', 'size': 10},
'slot': None,
'state': 'ok',
'state_info': None,
'switch_connection': [],
'unit': 2}
@allure.feature('Коммутатор')
@allure.title('Кол-во занятых портов коммутатора')
@allure.description("""
Проверяем, что кол-во занятых портов коммутатора изменяется,
в зависимости от подключения/отключения серверов
""")
def test_switch_using_post(snmp_switch, admin_auth, task_wait, server):
with allure.step('Коммутатор без подключений'):
switch = snmp_switch.info
switch_port_list = snmp_switch.get_port_list()
allure.attach(f'{switch}',
'Информация о коммутаторе')
allure.attach(f'{switch_port_list}',
'Информация о портах')
hc.assert_that(switch,
hc.has_properties(
total_port=hc.equal_to(len(switch_port_list.list)),
using_port=hc.equal_to(0)));
with allure.step('Выберем порт'):
port_id = switch_port_list.list[0].id
allure.attach(f'{port_id}', 'Выбран порт')
with allure.step(f'Подключим к порту #{port_id} сервер'):
connection_info = connection_server_to_switch(admin_auth,
server.id,
port_id)
with allure.step('Проверим кол-во занятых портов после подключения'):
switch_after_connect = SwitchApi(admin_auth.dci_writer)\
.switch_id_get(switch_id=switch.id)
allure.attach(f'{switch_after_connect}', 'Коммутатор после подключения порта')
hc.assert_that(switch_after_connect,
hc.has_properties(
total_port=hc.equal_to(len(switch_port_list.list)),
using_port=hc.equal_to(1)));
with allure.step(f'Удаляем подклоючение #{connection_info.id}'):
connection_del = SwitchConnectionApi(admin_auth.dci_writer)\
> .switch_connection_id_delete(switch_connection_id=connection_info.id)
test/tests/test_switch.py:475:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api/dci/api/switch_connection_api.py:151: in switch_connection_id_delete
(data) = self.switch_connection_id_delete_with_http_info(switch_connection_id, **kwargs) # noqa: E501
api/dci/api/switch_connection_api.py:225: in switch_connection_id_delete_with_http_info
collection_formats=collection_formats)
api/dci/api_client.py:340: in call_api
_preload_content, _request_timeout, _host)
api/dci/api_client.py:171: in __call_api
_request_timeout=_request_timeout)
api/dci/api_client.py:408: in request
body=body)
api/dci/rest.py:266: in DELETE
body=body)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <dci.rest.RESTClientObject object at 0x7fb874804c18>, method = 'DELETE'
url = 'http://10.0.1.8:1500/api/dci/v3/switch_connection/7', query_params = []
headers = {'Accept': 'application/json', 'Content-Type': 'application/json', 'Cookie': 'ses6=1B051C2AFE37E7F6C3DF5E4D', 'Host': 'instance-2', ...}
body = None, post_params = {}, _preload_content = True, _request_timeout = None
def request(self, method, url, query_params=None, headers=None,
body=None, post_params=None, _preload_content=True,
_request_timeout=None):
"""Perform requests.
:param method: http request method
:param url: http request url
:param query_params: query parameters in the url
:param headers: http request headers
:param body: request json body, for `application/json`
:param post_params: request post parameters,
`application/x-www-form-urlencoded`
and `multipart/form-data`
:param _preload_content: if False, the urllib3.HTTPResponse object will
be returned without reading/decoding response
data. Default is True.
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
"""
method = method.upper()
assert method in ['GET', 'HEAD', 'DELETE', 'POST', 'PUT',
'PATCH', 'OPTIONS']
if post_params and body:
raise ApiValueError(
"body parameter cannot be used with post_params parameter."
)
post_params = post_params or {}
headers = headers or {}
timeout = None
if _request_timeout:
if isinstance(_request_timeout, (int, ) if six.PY3 else (int, long)): # noqa: E501,F821
timeout = urllib3.Timeout(total=_request_timeout)
elif (isinstance(_request_timeout, tuple) and
len(_request_timeout) == 2):
timeout = urllib3.Timeout(
connect=_request_timeout[0], read=_request_timeout[1])
if 'Content-Type' not in headers:
headers['Content-Type'] = 'application/json'
try:
# For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:
if query_params:
url += '?' + urlencode(query_params)
if re.search('json', headers['Content-Type'], re.IGNORECASE):
request_body = None
if body is not None:
request_body = json.dumps(body)
r = self.pool_manager.request(
method, url,
body=request_body,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
elif headers['Content-Type'] == 'application/x-www-form-urlencoded': # noqa: E501
r = self.pool_manager.request(
method, url,
fields=post_params,
encode_multipart=False,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
elif headers['Content-Type'] == 'multipart/form-data':
# must del headers['Content-Type'], or the correct
# Content-Type which generated by urllib3 will be
# overwritten.
del headers['Content-Type']
r = self.pool_manager.request(
method, url,
fields=post_params,
encode_multipart=True,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
# Pass a `string` parameter directly in the body to support
# other content types than Json when `body` argument is
# provided in serialized form
elif isinstance(body, str):
request_body = body
r = self.pool_manager.request(
method, url,
body=request_body,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
else:
# Cannot generate the request from given parameters
msg = """Cannot prepare a request message for provided
arguments. Please check that your arguments match
declared content type."""
raise ApiException(status=0, reason=msg)
# For `GET`, `HEAD`
else:
r = self.pool_manager.request(method, url,
fields=query_params,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
except urllib3.exceptions.SSLError as e:
msg = "{0}\n{1}".format(type(e).__name__, str(e))
raise ApiException(status=0, reason=msg)
if _preload_content:
r = RESTResponse(r)
# In the python 3, the response.data is bytes.
# we need to decode it to string.
if six.PY3:
r.data = r.data.decode('utf8')
# log response body
logger.debug("response body: %s", r.data)
if not 200 <= r.status <= 299:
> raise ApiException(http_resp=r)
E dci.exceptions.ApiException: (404)
E Reason: Not Found
E HTTP response headers: HTTPHeaderDict({'Server': 'nginx/1.17.8', 'Date': 'Wed, 04 Mar 2020 09:21:01 GMT', 'Content-Type': 'application/json; charset=utf-8', 'Content-Length': '65', 'Connection': 'keep-alive'})
E HTTP response body: {"error":{"code":8011,"msg":"Switch port not found","value":"7"}}
api/dci/rest.py:229: ApiException
Raw
Подключение сервера к портам коммутатора cisco
snmp_switch = <fixtures.switch.Switch object at 0x7fb86dd6fcc0>
admin_auth = <misc.service.ServiceManager object at 0x7fb874804358>
task_wait = <function task_wait.<locals>._wait at 0x7fb874b26e18>
server = {'chassis': None,
'configuration': None,
'diag_error': None,
'domain': None,
'hardware_info': None,
'id': 29,
'i...': 'jvzdadvelg', 'size': 10},
'slot': None,
'state': 'ok',
'state_info': None,
'switch_connection': [],
'unit': 2}
@allure.feature('Коммутатор')
@allure.title('Подключение сервера к портам коммутатора')
@allure.description("""
Проверим, что сервер можно подключить к нескольким портам
и что подключение можно удалить
""")
def test_server_connection_to_switch(snmp_switch, admin_auth, task_wait, server):
with allure.step('Информация о коммутаторе'):
switch = snmp_switch
allure.attach('{}'.format(switch.info),
'Информация о коммутаторе')
with allure.step('Подключение сервера к портам коммутатора'):
with allure.step('Подключение портов'):
port_list = switch.get_port_list()
#TODO (m.semerikov) Берем первый порт. Рандомность?
first_port_id = port_list.list[0].id
first_connection_info = connection_server_to_switch(admin_auth,
server.id,
first_port_id)
#TODO (m.semerikov) Берем второй порт. Рандомность?
port_id = port_list.list[1].id
connection_info = connection_server_to_switch(admin_auth,
server.id,
port_id)
with allure.step('Опрос коммутатора'):
switch_info = switch_refresh(admin_auth, task_wait, switch.info.id, "ok")
with allure.step('Проверка типа соединения'):
check_connection_type(admin_auth, task_wait, switch.info.id, first_port_id, "connected")
with allure.step('Проверка наличия директивы server'):
check_section_server(admin_auth, task_wait, switch.info.id, first_port_id, server.id)
with allure.step('Проверка удаления подключения'):
with allure.step('Удаление'):
connection_del = SwitchConnectionApi(admin_auth.dci_writer)\
> .switch_connection_id_delete(switch_connection_id=first_connection_info.id)
test/tests/test_switch.py:219:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api/dci/api/switch_connection_api.py:151: in switch_connection_id_delete
(data) = self.switch_connection_id_delete_with_http_info(switch_connection_id, **kwargs) # noqa: E501
api/dci/api/switch_connection_api.py:225: in switch_connection_id_delete_with_http_info
collection_formats=collection_formats)
api/dci/api_client.py:340: in call_api
_preload_content, _request_timeout, _host)
api/dci/api_client.py:171: in __call_api
_request_timeout=_request_timeout)
api/dci/api_client.py:408: in request
body=body)
api/dci/rest.py:266: in DELETE
body=body)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <dci.rest.RESTClientObject object at 0x7fb874804c18>, method = 'DELETE'
url = 'http://10.0.1.8:1500/api/dci/v3/switch_connection/1', query_params = []
headers = {'Accept': 'application/json', 'Content-Type': 'application/json', 'Cookie': 'ses6=1B051C2AFE37E7F6C3DF5E4D', 'Host': 'instance-2', ...}
body = None, post_params = {}, _preload_content = True, _request_timeout = None
def request(self, method, url, query_params=None, headers=None,
body=None, post_params=None, _preload_content=True,
_request_timeout=None):
"""Perform requests.
:param method: http request method
:param url: http request url
:param query_params: query parameters in the url
:param headers: http request headers
:param body: request json body, for `application/json`
:param post_params: request post parameters,
`application/x-www-form-urlencoded`
and `multipart/form-data`
:param _preload_content: if False, the urllib3.HTTPResponse object will
be returned without reading/decoding response
data. Default is True.
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
"""
method = method.upper()
assert method in ['GET', 'HEAD', 'DELETE', 'POST', 'PUT',
'PATCH', 'OPTIONS']
if post_params and body:
raise ApiValueError(
"body parameter cannot be used with post_params parameter."
)
post_params = post_params or {}
headers = headers or {}
timeout = None
if _request_timeout:
if isinstance(_request_timeout, (int, ) if six.PY3 else (int, long)): # noqa: E501,F821
timeout = urllib3.Timeout(total=_request_timeout)
elif (isinstance(_request_timeout, tuple) and
len(_request_timeout) == 2):
timeout = urllib3.Timeout(
connect=_request_timeout[0], read=_request_timeout[1])
if 'Content-Type' not in headers:
headers['Content-Type'] = 'application/json'
try:
# For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:
if query_params:
url += '?' + urlencode(query_params)
if re.search('json', headers['Content-Type'], re.IGNORECASE):
request_body = None
if body is not None:
request_body = json.dumps(body)
r = self.pool_manager.request(
method, url,
body=request_body,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
elif headers['Content-Type'] == 'application/x-www-form-urlencoded': # noqa: E501
r = self.pool_manager.request(
method, url,
fields=post_params,
encode_multipart=False,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
elif headers['Content-Type'] == 'multipart/form-data':
# must del headers['Content-Type'], or the correct
# Content-Type which generated by urllib3 will be
# overwritten.
del headers['Content-Type']
r = self.pool_manager.request(
method, url,
fields=post_params,
encode_multipart=True,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
# Pass a `string` parameter directly in the body to support
# other content types than Json when `body` argument is
# provided in serialized form
elif isinstance(body, str):
request_body = body
r = self.pool_manager.request(
method, url,
body=request_body,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
else:
# Cannot generate the request from given parameters
msg = """Cannot prepare a request message for provided
arguments. Please check that your arguments match
declared content type."""
raise ApiException(status=0, reason=msg)
# For `GET`, `HEAD`
else:
r = self.pool_manager.request(method, url,
fields=query_params,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
except urllib3.exceptions.SSLError as e:
msg = "{0}\n{1}".format(type(e).__name__, str(e))
raise ApiException(status=0, reason=msg)
if _preload_content:
r = RESTResponse(r)
# In the python 3, the response.data is bytes.
# we need to decode it to string.
if six.PY3:
r.data = r.data.decode('utf8')
# log response body
logger.debug("response body: %s", r.data)
if not 200 <= r.status <= 299:
> raise ApiException(http_resp=r)
E dci.exceptions.ApiException: (404)
E Reason: Not Found
E HTTP response headers: HTTPHeaderDict({'Server': 'nginx/1.17.8', 'Date': 'Wed, 04 Mar 2020 09:19:24 GMT', 'Content-Type': 'application/json; charset=utf-8', 'Content-Length': '65', 'Connection': 'keep-alive'})
E HTTP response body: {"error":{"code":8011,"msg":"Switch port not found","value":"1"}}
api/dci/rest.py:229: ApiException
Raw
Подключение сервера к портам коммутатора common
snmp_switch = <fixtures.switch.Switch object at 0x7fb86c4d2be0>
admin_auth = <misc.service.ServiceManager object at 0x7fb874804358>
task_wait = <function task_wait.<locals>._wait at 0x7fb874b26e18>
server = {'chassis': None,
'configuration': None,
'diag_error': None,
'domain': None,
'hardware_info': None,
'id': 30,
'i...': 'fshwjpxyeg', 'size': 10},
'slot': None,
'state': 'ok',
'state_info': None,
'switch_connection': [],
'unit': 2}
@allure.feature('Коммутатор')
@allure.title('Подключение сервера к портам коммутатора')
@allure.description("""
Проверим, что сервер можно подключить к нескольким портам
и что подключение можно удалить
""")
def test_server_connection_to_switch(snmp_switch, admin_auth, task_wait, server):
with allure.step('Информация о коммутаторе'):
switch = snmp_switch
allure.attach('{}'.format(switch.info),
'Информация о коммутаторе')
with allure.step('Подключение сервера к портам коммутатора'):
with allure.step('Подключение портов'):
port_list = switch.get_port_list()
#TODO (m.semerikov) Берем первый порт. Рандомность?
first_port_id = port_list.list[0].id
first_connection_info = connection_server_to_switch(admin_auth,
server.id,
first_port_id)
#TODO (m.semerikov) Берем второй порт. Рандомность?
port_id = port_list.list[1].id
connection_info = connection_server_to_switch(admin_auth,
server.id,
port_id)
with allure.step('Опрос коммутатора'):
switch_info = switch_refresh(admin_auth, task_wait, switch.info.id, "ok")
with allure.step('Проверка типа соединения'):
check_connection_type(admin_auth, task_wait, switch.info.id, first_port_id, "connected")
with allure.step('Проверка наличия директивы server'):
check_section_server(admin_auth, task_wait, switch.info.id, first_port_id, server.id)
with allure.step('Проверка удаления подключения'):
with allure.step('Удаление'):
connection_del = SwitchConnectionApi(admin_auth.dci_writer)\
> .switch_connection_id_delete(switch_connection_id=first_connection_info.id)
test/tests/test_switch.py:219:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api/dci/api/switch_connection_api.py:151: in switch_connection_id_delete
(data) = self.switch_connection_id_delete_with_http_info(switch_connection_id, **kwargs) # noqa: E501
api/dci/api/switch_connection_api.py:225: in switch_connection_id_delete_with_http_info
collection_formats=collection_formats)
api/dci/api_client.py:340: in call_api
_preload_content, _request_timeout, _host)
api/dci/api_client.py:171: in __call_api
_request_timeout=_request_timeout)
api/dci/api_client.py:408: in request
body=body)
api/dci/rest.py:266: in DELETE
body=body)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <dci.rest.RESTClientObject object at 0x7fb874804c18>, method = 'DELETE'
url = 'http://10.0.1.8:1500/api/dci/v3/switch_connection/3', query_params = []
headers = {'Accept': 'application/json', 'Content-Type': 'application/json', 'Cookie': 'ses6=1B051C2AFE37E7F6C3DF5E4D', 'Host': 'instance-2', ...}
body = None, post_params = {}, _preload_content = True, _request_timeout = None
def request(self, method, url, query_params=None, headers=None,
body=None, post_params=None, _preload_content=True,
_request_timeout=None):
"""Perform requests.
:param method: http request method
:param url: http request url
:param query_params: query parameters in the url
:param headers: http request headers
:param body: request json body, for `application/json`
:param post_params: request post parameters,
`application/x-www-form-urlencoded`
and `multipart/form-data`
:param _preload_content: if False, the urllib3.HTTPResponse object will
be returned without reading/decoding response
data. Default is True.
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
"""
method = method.upper()
assert method in ['GET', 'HEAD', 'DELETE', 'POST', 'PUT',
'PATCH', 'OPTIONS']
if post_params and body:
raise ApiValueError(
"body parameter cannot be used with post_params parameter."
)
post_params = post_params or {}
headers = headers or {}
timeout = None
if _request_timeout:
if isinstance(_request_timeout, (int, ) if six.PY3 else (int, long)): # noqa: E501,F821
timeout = urllib3.Timeout(total=_request_timeout)
elif (isinstance(_request_timeout, tuple) and
len(_request_timeout) == 2):
timeout = urllib3.Timeout(
connect=_request_timeout[0], read=_request_timeout[1])
if 'Content-Type' not in headers:
headers['Content-Type'] = 'application/json'
try:
# For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:
if query_params:
url += '?' + urlencode(query_params)
if re.search('json', headers['Content-Type'], re.IGNORECASE):
request_body = None
if body is not None:
request_body = json.dumps(body)
r = self.pool_manager.request(
method, url,
body=request_body,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
elif headers['Content-Type'] == 'application/x-www-form-urlencoded': # noqa: E501
r = self.pool_manager.request(
method, url,
fields=post_params,
encode_multipart=False,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
elif headers['Content-Type'] == 'multipart/form-data':
# must del headers['Content-Type'], or the correct
# Content-Type which generated by urllib3 will be
# overwritten.
del headers['Content-Type']
r = self.pool_manager.request(
method, url,
fields=post_params,
encode_multipart=True,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
# Pass a `string` parameter directly in the body to support
# other content types than Json when `body` argument is
# provided in serialized form
elif isinstance(body, str):
request_body = body
r = self.pool_manager.request(
method, url,
body=request_body,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
else:
# Cannot generate the request from given parameters
msg = """Cannot prepare a request message for provided
arguments. Please check that your arguments match
declared content type."""
raise ApiException(status=0, reason=msg)
# For `GET`, `HEAD`
else:
r = self.pool_manager.request(method, url,
fields=query_params,
preload_content=_preload_content,
timeout=timeout,
headers=headers)
except urllib3.exceptions.SSLError as e:
msg = "{0}\n{1}".format(type(e).__name__, str(e))
raise ApiException(status=0, reason=msg)
if _preload_content:
r = RESTResponse(r)
# In the python 3, the response.data is bytes.
# we need to decode it to string.
if six.PY3:
r.data = r.data.decode('utf8')
# log response body
logger.debug("response body: %s", r.data)
if not 200 <= r.status <= 299:
> raise ApiException(http_resp=r)
E dci.exceptions.ApiException: (404)
E Reason: Not Found
E HTTP response headers: HTTPHeaderDict({'Server': 'nginx/1.17.8', 'Date': 'Wed, 04 Mar 2020 09:19:33 GMT', 'Content-Type': 'application/json; charset=utf-8', 'Content-Length': '65', 'Connection': 'close'})
E HTTP response body: {"error":{"code":8011,"msg":"Switch port not found","value":"3"}}
api/dci/rest.py:229: ApiException
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment