Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use check_dependencies method in Sure
Best Python code snippet using sure_python
doctor.py
Source:doctor.py 
1import click2import errno3import os4import platform5import re6import requests7import shutil8import socket9import sys10from enum import Enum, unique11from json import JSONEncoder12from two1.commands.config import \13 TWO1_LOGGER_SERVER, TWO1_HOST, TWO1_PROVIDER_HOST, \14 TWO1_PYPI_HOST, TWO1_POOL_URL, TWO1_MERCHANT_HOST, \15 TWO1_VERSION, TWO1_USER_FOLDER, TWO1_CONFIG_FILE16from two1.commands.status import has_bitcoinkit17from two1.lib.util.exceptions import TwoOneError18from two1.lib.server.analytics import capture_usage19from two1.lib.util.decorators import json_output20from two1.lib.util.uxstring import UxString21# Doctor Constants22MIN_VERSION_21 = (0, 3, 0)23MIN_VERSION_OS = (1, 0, 0)24if platform.system() == "Darwin":25 MIN_VERSION_OS = (14, 0, 0)26elif platform.system() == "Linux":27 MIN_VERSION_OS = (4, 0, 0)28MIN_VERSION_PYTHON = (3, 3, 0)29SOCKET_TIMEOUT = 1030DEMO_ENDPOINTS = [31 {"url": "/bitcoin_auth/token", "method": "get"},32 {"url": "/phone/send-sms", "method": "post"},33 {"url": "/search/bing", "method": "post"},34]35@unique36class DoctorStatus(Enum):37 OK, Fail, Warning = range(3)38 def prettyprint(self):39 color = "white"40 if self.name == "OK":41 color = "green"42 elif self.name == "Fail":43 color = "red"44 elif self.name == "Warning":45 color = "orange"46 return click.style(self.name, fg=color)47class DoctorCheck:48 def __init__(self, config):49 self.config = config;50 self.checks = []51 self.summary = {}52 self.passed = 053 self.warnings = 054 def addCheck(self, name, value, status, error=None):55 self.checks.append({56 "name": name,57 "value": value,58 "status": status.name,59 "error": error60 })61 self.config.log(" {: <30} -> {:<35} [{}]".format(name, value, status.prettyprint()))62 if status is DoctorStatus.OK:63 self.passed += 164 elif status is DoctorStatus.Warning:65 self.passed += 166 self.warnings += 167 elif status is DoctorStatus.Fail:68 if error:69 self.config.log(UxString.doctor_error + error)70 def generateSummary(self):71 final_status = DoctorStatus.Fail72 if self.passed == len(self.checks):73 final_status = DoctorStatus.OK74 if self.warnings > 0:75 final_status = DoctorStatus.Warning76 self.summary = {77 "passed": self.passed,78 "warnings": self.warnings,79 "total": len(self.checks),80 "status": final_status.name81 }82 DoctorCheck.printSummary(self.config, self.summary)83 # static method84 def printSummary(config, *summary):85 total = {86 "passed": 0,87 "total": 0,88 "warnings": 089 }90 for s in summary:91 total["passed"] += s["passed"]92 total["total"] += s["total"]93 total["warnings"] += s["warnings"]94 final_status = DoctorStatus.Fail95 if total["passed"] == total["total"]:96 final_status = DoctorStatus.OK97 if total["warnings"] > 0:98 final_status = DoctorStatus.Warning99 config.log("\n{} / {} tests passed, {} warnings. [{}]\n".format(\100 total["passed"],101 total["total"],102 total["warnings"],103 final_status.prettyprint()))104 def json(self):105 return {106 "checks": self.checks,107 "summary": self.summary108 }109@click.command()110@json_output111def doctor(config):112 """Checks on the health of the tool.113 """114 return _doctor(config)115@capture_usage116def _doctor(config):117 def assertEqual(checker, name, val, expected_val):118 if val == expected_val:119 checker.addCheck(name, val, DoctorStatus.OK)120 else:121 checker.addCheck(name, val, DoctorStatus.Fail, "{} != {}.".format(val, expected_val))122 def assertTrue(checker, name, val, success_msg="Exists", error_msg = None):123 if val == True or val == "Yes":124 checker.addCheck(name, success_msg, DoctorStatus.OK)125 else:126 if error_msg == None:127 error_msg = "{} is not True.".format(val)128 checker.addCheck(name, "No", DoctorStatus.Fail, error_msg)129 def assertAny(checker, name, val):130 if val != None and val != "":131 checker.addCheck(name, val, DoctorStatus.OK)132 else:133 checker.addCheck(name, val, DoctorStatus.Fail, "{} cannot be None.".format(name))134 def assertIn(checker, name, val, expected_val_array):135 if val in expected_val_array:136 checker.addCheck(name, val, DoctorStatus.OK)137 else:138 checker.addCheck(name, val, DoctorStatus.Fail, "{} must be one of {}".format(val, expected_val_array)) 139 def assertGte(checker, name, val, min_val):140 if val >= min_val:141 checker.addCheck(name, val, DoctorStatus.OK)142 else:143 checker.addCheck(name, val, DoctorStatus.Fail, "{} must be >= {}".format(val, min_val))144 def assertVersionGte(checker, name, version, min_version):145 ok = True146 version_str = "{}.{}.{}".format(version[0],version[1],version[2])147 # compare major148 if int(version[0]) > min_version[0]:149 ok = True150 elif int(version[0]) < min_version[0]:151 ok = False152 else:153 # compare minor154 if int(version[1]) > min_version[1]:155 ok = True156 elif int(version[1]) < min_version[1]:157 ok = False158 else:159 # compare revision160 rev = int(version[2].split('-')[0])161 if rev >= min_version[2]:162 ok = True163 elif rev < min_version[2]:164 ok = False165 if ok:166 checker.addCheck(name, version_str, DoctorStatus.OK)167 else:168 checker.addCheck(name, version_str, DoctorStatus.Fail, "Version must be >= {}.{}.{}. Your version is {}.{}.{}.".format(min_version[0], min_version[1], min_version[2], version[0],version[1],version[2]))169 def assertHTTPStatusCode(checker, name, url, method="get", expected_status_code=402):170 try:171 request_method = getattr(requests, method)172 r = request_method(url, timeout=SOCKET_TIMEOUT)173 if r.status_code == expected_status_code:174 checker.addCheck("{} {}".format(method.upper(), name), r.status_code, DoctorStatus.OK)175 else:176 checker.addCheck("{} {}".format(method.upper(), name), r.status_code, DoctorStatus.Fail, "Expected status code '{}'".format(expected_status_code))177 except requests.exceptions.ConnectionError:178 checker.addCheck("{} {}".format(method.upper(), name), "Failed", DoctorStatus.Fail, "Could not connect to '{}'".format(url))179 except Exception as e:180 checker.addCheck("{} {}".format(method.upper(), name), "Failed", DoctorStatus.Fail, str(e))181 182 def assertSocket(checker, name, url):183 protocol = "http"184 port = 80185 hostname = ""186 url_components = url.split("://")187 if len(url_components) >= 2:188 protocol = url_components[0]189 hostname = url_components[1]190 host_components = hostname.split(":")191 if len(host_components) == 2:192 hostname = host_components[0]193 port = int(host_components[1])194 else:195 checker.addCheck(name, url, DoctorStatus.Fail, "Invalid url " + url)196 return197 if protocol == "https":198 port = 443199 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)200 s.settimeout(SOCKET_TIMEOUT)201 try:202 result = s.connect_ex((hostname, port))203 s.close()204 if result > 0:205 checker.addCheck(name, url, DoctorStatus.Fail, errno.errorcode[result])206 else:207 checker.addCheck(name, "{}:{}".format(hostname, port), DoctorStatus.OK)208 except Exception as e:209 s.close()210 checker.addCheck(name, url, DoctorStatus.Fail, str(e))211 def assertCommandExists(checker, name, cmd, error_msg=None):212 cmd_path = None213 try:214 cmd_path = shutil.which(cmd)215 except Exception as e:216 pass217 if error_msg == None:218 error_msg = "'{}' does not exist.".format(cmd)219 assertTrue(check_dependencies, name, cmd_path != None, error_msg=error_msg)220 def assertPathExists(checker, name, path):221 return assertTrue(checker, name, os.path.exists(path), error_msg="'{}' does not exist.".format(path))222 # doctor code start223 config.log(UxString.doctor_start)224 # Checking OS225 config.log(UxString.doctor_general)226 check_general = DoctorCheck(config)227 assertVersionGte(check_general, "21 Tool version", TWO1_VERSION.split('.'), MIN_VERSION_21)228 assertIn(check_general, "OS System", platform.system(), ["Windows", "Linux", "Darwin"])229 assertVersionGte(check_general, "OS Release", platform.release().split('.'), MIN_VERSION_OS)230 assertVersionGte(check_general, "Python version", platform.python_version_tuple(), MIN_VERSION_PYTHON)231 assertTrue(check_general, "Has Bitcoin kit", has_bitcoinkit(), success_msg="Yes", error_msg="Bitcoin kit not detected.")232 try:233 s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)234 s.connect(("8.8.8.8", 80))235 assertAny(check_general, "IP Address", s.getsockname()[0])236 except Exception as e: 237 assertTrue(check_general, "IP Address", False, error_msg="No Internet Connectivity Found.")238 check_general.generateSummary()239 # Checking dependencies240 config.log(UxString.doctor_dependencies)241 check_dependencies = DoctorCheck(config)242 assertTrue(check_dependencies, "two1 python library", 'two1' in sys.modules, error_msg="two1 lib does not exist.")243 cli_path = None244 try:245 cli_path = shutil.which('21')246 except Exception as e:247 pass248 assertCommandExists(check_dependencies, "21 CLI", cmd="21", error_msg="21 CLI not bound to '21'.")249 assertCommandExists(check_dependencies, "Zerotier CLI", cmd="zerotier-cli", error_msg="Zerotier CLI not bound.")250 assertCommandExists(check_dependencies, "apt-get", cmd="apt-get")251 assertCommandExists(check_dependencies, "minerd", cmd="minerd")252 assertCommandExists(check_dependencies, "wallet", cmd="wallet")253 assertPathExists(check_dependencies, ".two1 folder", TWO1_USER_FOLDER)254 assertPathExists(check_dependencies, ".two1 config file", TWO1_CONFIG_FILE)255 check_dependencies.generateSummary()256 # Checking demo endpoints257 config.log(UxString.doctor_demo_endpoints)258 check_demos = DoctorCheck(config)259 for demo_endpoint in DEMO_ENDPOINTS:260 assertHTTPStatusCode(check_demos, name=demo_endpoint["url"], url=(TWO1_MERCHANT_HOST + demo_endpoint["url"]), method=demo_endpoint["method"], expected_status_code=402)261 check_demos.generateSummary()262 # Check servers263 config.log(UxString.doctor_servers)264 check_servers = DoctorCheck(config)265 assertSocket(check_servers, "Pool2 Api", TWO1_HOST)266 assertSocket(check_servers, "Pool2 Tcp", TWO1_POOL_URL)267 assertSocket(check_servers, "Log server", TWO1_LOGGER_SERVER)268 assertSocket(check_servers, "Merchant", TWO1_MERCHANT_HOST)269 assertSocket(check_servers, "Blockchain", TWO1_PROVIDER_HOST)270 assertSocket(check_servers, "PyPi Host", TWO1_PYPI_HOST)271 assertHTTPStatusCode(check_servers, name="21co slack", url="https://slack.21.co", expected_status_code=200)272 assertHTTPStatusCode(check_servers, name="Raspbian package repo", url="http://mirrordirector.raspbian.org/raspbian", expected_status_code=200)273 assertHTTPStatusCode(check_servers, name="Chain.com API", url="https://api.chain.com", expected_status_code=401)274 275 check_servers.generateSummary()276 config.log(UxString.doctor_total)277 DoctorCheck.printSummary(config, \278 check_general.summary, \279 check_dependencies.summary, \280 check_demos.summary, \281 check_servers.summary)282 result = {283 "general": check_general.json(),284 "dependencies": check_dependencies.json(),285 "demo": check_demos.json(),286 "servers": check_servers.json()287 }288 if all([doctor_check['summary']['total'] == doctor_check['summary']['passed'] for doctor_check in result.values()]):289 return result290 else:...
test_check_dependencies.py
Source:test_check_dependencies.py
...9 monkeypatch.setitem(sys.modules, "uvicorn", module_mock)10 # mock sys.exit11 exit_mock = MagicMock()12 monkeypatch.setattr(sys, "exit", exit_mock)13 check_dependencies()14 assert not exit_mock.called15@pytest.mark.parametrize("mode", ["all", "", "frontend", "backend", None])16def test_no_streamlit(mode, monkeypatch):17 # mock missing streamlit module18 uvicorn_mock = MagicMock()19 monkeypatch.setitem(sys.modules, "uvicorn", uvicorn_mock)20 monkeypatch.setitem(sys.modules, "streamlit", None)21 # mock sys.exit22 exit_mock = MagicMock()23 monkeypatch.setattr(sys, "exit", exit_mock)24 if mode is None:25 check_dependencies()26 else:27 check_dependencies(mode=mode)28 if mode == "backend":29 assert not exit_mock.called30 return31 assert exit_mock.called32 assert exit_mock.call_args[0][0] == (33 "You need to install streamlit to start the frontend. "34 "To install streamlit, run: pip install streamlit"35 )36@pytest.mark.parametrize("mode", ["all", "", "backend", "frontend", None])37def test_no_uvicorn(mode, monkeypatch):38 # mock missing uvicorn module39 st_mock = MagicMock()40 monkeypatch.setitem(sys.modules, "streamlit", st_mock)41 monkeypatch.setitem(sys.modules, "uvicorn", None)42 # mock sys.exit43 exit_mock = MagicMock()44 monkeypatch.setattr(sys, "exit", exit_mock)45 if mode is None:46 check_dependencies()47 else:48 check_dependencies(mode=mode)49 if mode == "frontend":50 assert not exit_mock.called51 return52 assert exit_mock.called53 assert exit_mock.call_args[0][0] == (54 "You need to install uvicorn to start the backend. "55 "To install uvicorn, run: pip install uvicorn"...
test_player.py
Source:test_player.py
1import os2from unittest import mock3import pytest4import castero.config5from castero.config import Config6from castero.episode import Episode7from castero.feed import Feed8from castero.player import Player, PlayerDependencyError9my_dir = os.path.dirname(os.path.realpath(__file__))10feed = Feed(file=my_dir + "/feeds/valid_basic.xml")11episode = Episode(12 feed,13 title="episode title",14 description="episode description",15 link="episode link",16 pubdate="episode pubdate",17 copyright="episode copyright",18 enclosure="episode enclosure",19)20SomePlayer = mock.MagicMock()21available_players = {"someplayer": SomePlayer}22def test_player_create_instance_success_direct():23 Config.data = {"player": "someplayer"}24 Player.create_instance(available_players, "t", "p", episode)25 assert SomePlayer.check_dependencies.call_count == 126 SomePlayer.assert_called_with("t", "p", episode)27def test_player_create_instance_success_indirect():28 Config.data = {"player": ""}29 Player.create_instance(available_players, "t", "p", episode)30 SomePlayer.check_dependencies.assert_called = 231 SomePlayer.assert_called_with("t", "p", episode)32def test_player_create_instance_dep_error_direct():33 Config.data = {"player": "someplayer"}34 SomePlayer.check_dependencies.side_effect = PlayerDependencyError()35 with pytest.raises(PlayerDependencyError):36 Player.create_instance(available_players, "t", "p", episode)37 assert SomePlayer.check_dependencies.call_count == 138def test_player_create_instance_dep_error_indirect():39 Config.data = {"player": ""}40 SomePlayer.check_dependencies.side_effect = PlayerDependencyError()41 with pytest.raises(PlayerDependencyError):42 Player.create_instance(available_players, "t", "p", episode)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Sure automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
