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
74 changes: 19 additions & 55 deletions appium/webdriver/extensions/android/network.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,15 +12,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.

from selenium.common.exceptions import UnknownMethodException
from typing_extensions import Self

from appium.common.helper import extract_const_attributes
from appium.common.logger import logger
from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands
from appium.protocols.webdriver.can_execute_scripts import CanExecuteScripts
from appium.protocols.webdriver.can_remember_extension_presence import CanRememberExtensionPresence
from appium.webdriver.mobilecommand import MobileCommand as Command


class NetSpeed:
Expand All @@ -41,7 +38,7 @@ class NetworkMask:
AIRPLANE_MODE = 0b001


class Network(CanExecuteCommands, CanExecuteScripts, CanRememberExtensionPresence):
class Network(CanExecuteCommands, CanExecuteScripts):
@property
def network_connection(self) -> int:
"""Returns an integer bitmask specifying the network connection type.
Expand All @@ -53,16 +50,12 @@ def network_connection(self) -> int:
since API level 31.
"""
ext_name = 'mobile: getConnectivity'
try:
result_map = self.assert_extension_exists(ext_name).execute_script(ext_name)
return (
(NetworkMask.WIFI if result_map['wifi'] else 0)
| (NetworkMask.DATA if result_map['data'] else 0)
| (NetworkMask.AIRPLANE_MODE if result_map['airplaneMode'] else 0)
)
except UnknownMethodException:
# TODO: Remove the fallback
return self.mark_extension_absence(ext_name).execute(Command.GET_NETWORK_CONNECTION, {})['value']
result_map = self.execute_script(ext_name)
return (
(NetworkMask.WIFI if result_map['wifi'] else 0)
| (NetworkMask.DATA if result_map['data'] else 0)
| (NetworkMask.AIRPLANE_MODE if result_map['airplaneMode'] else 0)
)

def set_network_connection(self, connection_type: int) -> int:
"""Sets the network connection type. Android only.
Expand Down Expand Up @@ -95,20 +88,15 @@ def set_network_connection(self, connection_type: int) -> int:
int: Set network connection type
"""
ext_name = 'mobile: setConnectivity'
try:
return self.assert_extension_exists(ext_name).execute_script(
ext_name,
{
'wifi': bool(connection_type & NetworkMask.WIFI),
'data': bool(connection_type & NetworkMask.DATA),
'airplaneMode': bool(connection_type & NetworkMask.AIRPLANE_MODE),
},
)
except UnknownMethodException:
# TODO: Remove the fallback
return self.mark_extension_absence(ext_name).execute(
Command.SET_NETWORK_CONNECTION, {'parameters': {'type': connection_type}}
)['value']
self.execute_script(
ext_name,
{
'wifi': bool(connection_type & NetworkMask.WIFI),
'data': bool(connection_type & NetworkMask.DATA),
'airplaneMode': bool(connection_type & NetworkMask.AIRPLANE_MODE),
},
)
return self.network_connection

def toggle_wifi(self) -> Self:
"""Toggle the wifi on the device, Android only.
Expand All @@ -119,12 +107,7 @@ def toggle_wifi(self) -> Self:
Union['WebDriver', 'Network']: Self instance
"""
ext_name = 'mobile: setConnectivity'
try:
self.assert_extension_exists(ext_name).execute_script(
ext_name, {'wifi': not (self.network_connection & NetworkMask.WIFI)}
)
except UnknownMethodException:
self.mark_extension_absence(ext_name).execute(Command.TOGGLE_WIFI, {})
self.execute_script(ext_name, {'wifi': not (self.network_connection & NetworkMask.WIFI)})
return self

def set_network_speed(self, speed_type: str) -> Self:
Expand All @@ -149,27 +132,8 @@ def set_network_speed(self, speed_type: str) -> Self:
f'(e.g. {NetSpeed.__name__}.LTE)'
)
ext_name = 'mobile: networkSpeed'
try:
self.assert_extension_exists(ext_name).execute_script(ext_name, {'speed': speed_type})
except UnknownMethodException:
# TODO: Remove the fallback
self.mark_extension_absence(ext_name).execute(Command.SET_NETWORK_SPEED, {'netspeed': speed_type})
self.execute_script(ext_name, {'speed': speed_type})
return self

def _add_commands(self) -> None:
self.command_executor.add_command(Command.TOGGLE_WIFI, 'POST', '/session/$sessionId/appium/device/toggle_wifi')
self.command_executor.add_command(
Command.GET_NETWORK_CONNECTION,
'GET',
'/session/$sessionId/network_connection',
)
self.command_executor.add_command(
Command.SET_NETWORK_CONNECTION,
'POST',
'/session/$sessionId/network_connection',
)
self.command_executor.add_command(
Command.SET_NETWORK_SPEED,
'POST',
'/session/$sessionId/appium/device/network_speed',
)
pass
15 changes: 3 additions & 12 deletions test/unit/webdriver/network_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@ class TestWebDriverNetwork:
@httpretty.activate
def test_network_connection(self):
driver = android_w3c_driver()
httpretty.register_uri(httpretty.GET, appium_command('/session/1234567890/network_connection'), body='{"value": 2}')
httpretty.register_uri(
httpretty.POST,
appium_command('/session/1234567890/execute/sync'),
Expand All @@ -35,24 +34,20 @@ def test_network_connection(self):
@httpretty.activate
def test_set_network_connection(self):
driver = android_w3c_driver()
httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/network_connection'), body='{"value": ""}')
httpretty.register_uri(
httpretty.POST,
appium_command('/session/1234567890/execute/sync'),
body='{"value": {"wifi": true, "data": false, "airplaneMode": false}}',
)
driver.set_network_connection(2)
assert driver.set_network_connection(2) == 2

d = get_httpretty_request_body(httpretty.last_request())
d = get_httpretty_request_body(httpretty.latest_requests()[-4])
assert d['script'] == 'mobile: setConnectivity'
assert d['args'][0]['wifi'] is True

@httpretty.activate
def test_set_network_speed(self):
driver = android_w3c_driver()
httpretty.register_uri(
httpretty.POST,
appium_command('/session/1234567890/appium/device/network_speed'),
)
httpretty.register_uri(
httpretty.POST,
appium_command('/session/1234567890/execute/sync'),
Expand All @@ -65,10 +60,6 @@ def test_set_network_speed(self):
@httpretty.activate
def test_toggle_wifi(self):
driver = android_w3c_driver()
httpretty.register_uri(
httpretty.POST,
appium_command('/session/1234567890/appium/device/toggle_wifi'),
)
httpretty.register_uri(
httpretty.POST,
appium_command('/session/1234567890/execute/sync'),
Expand Down
Loading