1# Copyright 2023 The Chromium Authors 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4 5from __future__ import annotations 6 7import abc 8import contextlib 9from typing import TYPE_CHECKING, Iterator, Optional 10 11from crossbench import plt 12from crossbench.network.traffic_shaping.live import NoTrafficShaper 13 14if TYPE_CHECKING: 15 from crossbench.browsers.attributes import BrowserAttributes 16 from crossbench.flags.base import Flags 17 from crossbench.network.traffic_shaping.base import TrafficShaper 18 from crossbench.runner.groups.session import BrowserSessionRunGroup 19 20 21class Network(abc.ABC): 22 23 def __init__(self, 24 traffic_shaper: Optional[TrafficShaper] = None, 25 browser_platform: plt.Platform = plt.PLATFORM) -> None: 26 self._traffic_shaper = traffic_shaper or NoTrafficShaper(browser_platform) 27 self._browser_platform = browser_platform 28 self._host_platform = browser_platform.host_platform 29 self._is_running: bool = False 30 31 @property 32 def traffic_shaper(self) -> TrafficShaper: 33 return self._traffic_shaper 34 35 @property 36 def browser_platform(self) -> plt.Platform: 37 return self._browser_platform 38 39 @property 40 def host_platform(self) -> plt.Platform: 41 return self._host_platform 42 43 @property 44 def is_running(self) -> bool: 45 return self._is_running 46 47 @property 48 def is_live(self) -> bool: 49 """Return True if the network is the default live/direct connection, as 50 opposed to a replay network or local file server.""" 51 return False 52 53 @property 54 def is_wpr(self) -> bool: 55 """Return True if the network is the replay network.""" 56 return False 57 58 @property 59 def is_local_file_server(self) -> bool: 60 """Return True if the network is the local file server network.""" 61 return False 62 63 @property 64 def http_port(self) -> Optional[int]: 65 """HTTP port for non-live server-based networks.""" 66 return None 67 68 @property 69 def https_port(self) -> Optional[int]: 70 """HTTPS port for non-live server-based networks.""" 71 return None 72 73 @property 74 def host(self) -> Optional[str]: 75 """Host for non-live server-based networks.""" 76 return None 77 78 def extra_flags(self, browser_attributes: BrowserAttributes) -> Flags: 79 assert self.is_running, "Network is not running." 80 return self.traffic_shaper.extra_flags(browser_attributes) 81 82 @contextlib.contextmanager 83 def open(self, session: BrowserSessionRunGroup) -> Iterator[Network]: 84 del session 85 assert not self._is_running, "Cannot start network more than once." 86 self._is_running = True 87 try: 88 yield self 89 finally: 90 self._is_running = False 91