Skip to content

Commit

Permalink
Wifi 13350 (#893)
Browse files Browse the repository at this point in the history
* rrm command test added to the controller

Signed-off-by: anil-tegala <[email protected]>

* added tx power change test from rrmcmd api

Signed-off-by: anil-tegala <[email protected]>

* modified setting tx power logic as per interface in iwinfo

Signed-off-by: anil-tegala <[email protected]>

* added pass fail check to txpower check

Signed-off-by: anil-tegala <[email protected]>

* set tx-power in config and additional assertions for validation

Signed-off-by: anil-tegala <[email protected]>

---------

Signed-off-by: anil-tegala <[email protected]>
  • Loading branch information
anil-tegala authored Feb 27, 2024
1 parent 407cdae commit 30ea292
Show file tree
Hide file tree
Showing 2 changed files with 140 additions and 22 deletions.
15 changes: 15 additions & 0 deletions libs/tip_2x/controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -860,6 +860,21 @@ def telemetry(self, serial_number, payload):
self.check_response("POST", resp, self.make_headers(), payload, uri)
return resp

def rrm_command(self, serial_number, payload):
uri = self.build_uri("device/" + serial_number + "/rrm")
payld = json.dumps(payload)
logging.info("Sending Command: " + "\n" +
"TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +
"URI: " + str(uri) + "\n" +
"Headers: " + str(self.make_headers()))
allure.attach(name="Sending Command: ", body="Sending Command: " + "\n" +
"TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +
"URI: " + str(uri) + "\n" +
"Headers: " + str(self.make_headers()))
resp = requests.post(uri, data=payld, headers=self.make_headers(), verify=False, timeout=120)
self.check_response("POST", resp, self.make_headers(), payld, uri)
return resp

def get_rtty_params(self, serial_number):
uri = self.build_uri("device/" + serial_number + "/rtty")
logging.info("Sending Command: " + "\n" +
Expand Down
147 changes: 125 additions & 22 deletions tests/controller_tests/ucentral_gateway/test_gatewayservice.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,10 +7,14 @@
import json
import logging
import random
import datetime
import re
import time
from time import sleep

import allure
import pytest
import requests


@pytest.mark.uc_sanity
Expand All @@ -23,13 +27,21 @@ class TestUcentralGatewayService(object):
"""
"""
configuration = {
"uuid": 1,
"uuid": 2,
"radios": [
{
"band": "5G",
"country": "CA",
"channel": 52,
"channel-mode": "HE",
"channel-width": 80
"channel-width": 80,
"country": "CA"
},
{
"band": "2G",
"channel": 11,
"channel-mode": "HE",
"channel-width": 20,
"country": "CA"
}
],

Expand All @@ -52,7 +64,7 @@ class TestUcentralGatewayService(object):
{
"name": "OpenWifi",
"wifi-bands": [
"5G"
"2G", "5G"
],
"bss-mode": "ap",
"encryption": {
Expand Down Expand Up @@ -82,22 +94,7 @@ class TestUcentralGatewayService(object):
"lease-count": 100,
"lease-time": "6h"
}
},
"ssids": [
{
"name": "OpenWifi",
"wifi-bands": [
"5G"
],
"bss-mode": "ap",
"encryption": {
"proto": "psk2",
"key": "OpenWifi",
"ieee80211w": "optional"
}
}
]

}
}
],
"metrics": {
Expand All @@ -107,11 +104,14 @@ class TestUcentralGatewayService(object):
},
"health": {
"interval": 120
},
"wifi-frames": {
"filters": ["probe", "auth"]
}
},
"services": {
"lldp": {
"describe": "2.x",
"describe": "uCentral",
"location": "universe"
},
"ssh": {
Expand Down Expand Up @@ -765,7 +765,6 @@ def test_gw_service_delete_update_read_radius_proxy_configuration(self, get_targ
resp = get_target_object.controller_library_object.get_radius_proxy_configuration()
assert resp.status_code == 200


@pytest.mark.gw_country_code_for_ip_address
@allure.title("Get the country code for an IP address")
@allure.testcase(name="WIFI-12558",
Expand Down Expand Up @@ -1050,3 +1049,107 @@ def test_gw_service_get_radius_sessions(self, get_target_object, get_testbed_det
device_name = get_testbed_details['device_under_tests'][0]['identifier']
resp = get_target_object.controller_library_object.get_radius_sessions(device_name)
assert resp.status_code == 200

@pytest.mark.rrmcmd_tx_power
@allure.title("Verify Dynamic change of Tx Power using RRM action command")
@allure.testcase(name="WIFI-13350",
url="https://telecominfraproject.atlassian.net/browse/WIFI-13350")
def test_rrmcmd_tx_power(self, get_target_object, get_testbed_details):
"""
Test to SEND RRM commands from device present in Gateway UI
Unique marker:pytest -m "rrmcmd_tx_power"
"""
action_body = {
"actions": [
{
"action": "tx_power",
"bssid": "",
"level": 20
}
]
}
serial_number = get_target_object.device_under_tests_info[0]["identifier"]
for i in range(len(self.configuration['radios'])):
self.configuration['radios'][i]["tx-power"] = 23
payload = {"configuration": self.configuration, "serialNumber": serial_number, "UUID": 1}
uri = get_target_object.controller_library_object.build_uri("device/" + serial_number + "/configure")
basic_cfg_str = json.dumps(payload)
logging.info("Sending Command: Configure " + "\n" +
"TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +
"URI: " + str(uri) + "\n" +
"Data: " + str(json.dumps(payload, indent=2)) + "\n" +
"Headers: " + str(get_target_object.controller_library_object.make_headers()))
allure.attach(name="Sending Command: Configure", body="Sending Command: " + "\n" +
"TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +
"URI: " + str(uri) + "\n" +
"Data: " + str(payload) + "\n" +
"Headers: " + str(
get_target_object.controller_library_object.make_headers()))
logging.info("wait until the configuration push get's applied...")
resp = requests.post(uri, data=basic_cfg_str, verify=False, timeout=240,
headers=get_target_object.controller_library_object.make_headers())
if resp and resp.status_code == 200:
logging.info(f"Status:{resp.status_code} - Configuration push successful")
logging.info(resp.json())
else:
logging.error("Failed to push the config")
pytest.exit(f"Reason:{resp.reason} - Error while pushing the configuration")
logging.info("iwinfo before applying Tx Power using RRM action command: \n")
cmd_response = get_target_object.get_dut_library_object().get_iwinfo(attach_allure=False)
allure.attach(body=cmd_response, name="iwinfo before applying Tx Power using RRM action command:")
if str(cmd_response) != "pop from empty list":
interfaces = {}
interface_matches = re.finditer(
r'wlan\d\s+ESSID:\s+".*?"\s+Access Point:\s+([0-9A-Fa-f:]+).*?Tx-Power:\s+([\d\s]+)', cmd_response,
re.DOTALL)
if interface_matches:
for match in interface_matches:
interface_name = f'wlan{match.group(0)[4]}'
access_point = match.group(1)
tx_power = match.group(2).strip()
interfaces[interface_name] = {'Access Point': access_point, 'Tx-Power': tx_power}
logging.info(interfaces)
else:
logging.error("Failed to get iwinfo")
pytest.exit("Failed to get iwinfo")
for i in interfaces:
action_body["actions"][0]["bssid"] = interfaces[i]['Access Point']
action_body["actions"][0]["level"] = 20
response = get_target_object.controller_library_object.rrm_command(payload=action_body,
serial_number=serial_number)
logging.info(response.json())
time.sleep(2)
allure.attach(name=f"Response: {response.status_code} - {response.reason}", body=str(response.json()),
attachment_type=allure.attachment_type.JSON)
time.sleep(3)
logging.info("iwinfo After applying Tx Power using RRM action command: \n")
cmd_response1 = get_target_object.get_dut_library_object().get_iwinfo(attach_allure=False)
allure.attach(body=cmd_response1, name="iwinfo before applying Tx Power using RRM action command:")
if cmd_response1 == {}:
assert False, "Empty iwinfo reponse from AP through minicom"
interfaces1 = {}
interface_matches1 = re.finditer(
r'wlan\d\s+ESSID:\s+".*?"\s+Access Point:\s+([0-9A-Fa-f:]+).*?Tx-Power:\s+([\d\s]+)', cmd_response1,
re.DOTALL)
if interface_matches1:
for match1 in interface_matches1:
interface_name1 = f'wlan{match1.group(0)[4]}'
access_point1 = match1.group(1)
tx_power1 = match1.group(2).strip()
interfaces1[interface_name1] = {'Access Point': access_point1, 'Tx-Power': tx_power1}
logging.info(interfaces1)
else:
logging.error("Failed to get iwinfo")
pytest.exit("Failed to get iwinfo")
key_to_check = ('Tx-Power', '20')
logging.info(interfaces1.items())
for key, value in interfaces1.items():
logging.info(key, value)
if key_to_check in value:
assert True
else:
assert False, "failed to set tx power using RRM CMD"
elif cmd_response == {}:
assert False, "Empty iwinfo reponse from AP through minicom"
else:
assert False, "Failed to get iwinfo from minicom"

0 comments on commit 30ea292

Please sign in to comment.