KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use test_base method in SeleniumBase
Best Python code snippet using SeleniumBase
test_extensions.py
Source:test_extensions.py 
1#!/usr/bin/env python32# Copyright (c) 2014-present, The osquery authors3#4# This source code is licensed as defined by the LICENSE file found in the5# root directory of this source tree.6#7# SPDX-License-Identifier: (Apache-2.0 OR GPL-2.0-only)8import glob9import os10import psutil11import signal12import subprocess13import sys14import time15import threading16import unittest17# osquery-specific testing utils18import test_base19import utils20EXTENSION_TIMEOUT = 1021class ExtensionTests(test_base.ProcessGenerator, unittest.TestCase):22 def test_1_daemon_without_extensions(self):23 # Start the daemon without thrift, prefer no watchdog because the tests24 # kill the daemon very quickly.25 daemon = self._run_daemon({26 "disable_watchdog": True,27 "disable_extensions": True,28 })29 self.assertTrue(daemon.isAlive())30 # Now try to connect to the disabled API31 client = test_base.EXClient(daemon.options["extensions_socket"])32 self.assertFalse(client.open())33 daemon.kill()34 @test_base.flaky35 def test_2_daemon_api(self):36 daemon = self._run_daemon({"disable_watchdog": True})37 self.assertTrue(daemon.isAlive())38 # Get a python-based thrift client39 client = test_base.EXClient(daemon.options["extensions_socket"])40 self.assertTrue(client.open(timeout=EXTENSION_TIMEOUT))41 em = client.getEM()42 # List the number of extensions43 print(em.ping())44 result = test_base.expect(em.extensions, 0)45 self.assertEqual(len(result), 0)46 # Try the basic ping API47 self.assertEqual(em.ping().code, 0)48 # Try a query49 response = em.query("select * from time")50 self.assertEqual(response.status.code, 0)51 self.assertEqual(len(response.response), 1)52 self.assertTrue("seconds" in response.response[0].keys())53 # Try to get the query columns54 response = em.getQueryColumns("select seconds as s from time")55 self.assertEqual(response.status.code, 0)56 self.assertEqual(len(response.response), 1)57 self.assertTrue("s" in response.response[0])58 client.close()59 daemon.kill()60 @test_base.flaky61 def test_3_example_extension(self):62 daemon = self._run_daemon({"disable_watchdog": True})63 self.assertTrue(daemon.isAlive())64 # Get a python-based thrift client65 client = test_base.EXClient(daemon.options["extensions_socket"])66 self.assertTrue(client.open(timeout=EXTENSION_TIMEOUT))67 em = client.getEM()68 # Make sure there are no extensions registered69 result = test_base.expect(em.extensions, 0)70 self.assertEqual(len(result), 0)71 # Make sure the extension process starts72 extension = self._run_extension(73 path=daemon.options["extensions_socket"],74 timeout=EXTENSION_TIMEOUT,75 )76 self.assertTrue(extension.isAlive())77 # Now that an extension has started, check extension list78 result = test_base.expect(em.extensions, 1)79 self.assertEqual(len(result), 1)80 ex_uuid = result.keys()[0]81 ex_data = result[ex_uuid]82 self.assertEqual(ex_data.name, "example")83 self.assertEqual(ex_data.version, "0.0.1")84 self.assertEqual(ex_data.min_sdk_version, "0.0.0")85 # Get a python-based thrift client to the extension's service86 client2 = test_base.EXClient(daemon.options["extensions_socket"],87 uuid=ex_uuid)88 self.assertTrue(client2.open(timeout=EXTENSION_TIMEOUT))89 ex = client2.getEX()90 self.assertEqual(ex.ping().code, 0)91 # Make sure the extension can receive a call92 em_time = em.call("table", "time", {"action": "columns"})93 ex_time = ex.call("table", "time", {"action": "columns"})94 print(em_time)95 print(ex_time)96 self.assertEqual(ex_time.status.code, 0)97 self.assertTrue(len(ex_time.response) > 0)98 self.assertTrue("name" in ex_time.response[0])99 self.assertEqual(ex_time.status.uuid, ex_uuid)100 # Make sure the extension includes a custom registry plugin101 result = ex.call("table", "example", {"action": "generate"})102 print(result)103 self.assertEqual(result.status.code, 0)104 self.assertEqual(len(result.response), 1)105 self.assertTrue("example_text" in result.response[0])106 self.assertTrue("example_integer" in result.response[0])107 self.assertEqual(result.response[0]["example_text"], "example")108 # Make sure the core can route to the extension109 result = em.call("table", "example", {"action": "generate"})110 print(result)111 client2.close()112 client.close()113 extension.kill()114 daemon.kill()115 @test_base.flaky116 def test_4_extension_dies(self):117 daemon = self._run_daemon({118 "disable_watchdog": True,119 "extensions_interval": "0",120 "verbose": True,121 })122 self.assertTrue(daemon.isAlive())123 # Get a python-based thrift client124 client = test_base.EXClient(daemon.options["extensions_socket"])125 self.assertTrue(client.open(timeout=EXTENSION_TIMEOUT))126 em = client.getEM()127 # Make sure there are no extensions registered128 result = test_base.expect(em.extensions, 0)129 self.assertEqual(len(result), 0)130 # Make sure the extension process starts131 extension = self._run_extension(132 path=daemon.options["extensions_socket"],133 timeout=EXTENSION_TIMEOUT)134 self.assertTrue(extension.isAlive())135 # Now that an extension has started, check extension list136 result = test_base.expect(em.extensions, 1)137 self.assertEqual(len(result), 1)138 # Kill the extension139 extension.kill()140 # Make sure the daemon detects the change141 result = test_base.expect(em.extensions, 0, timeout=EXTENSION_TIMEOUT)142 self.assertEqual(len(result), 0)143 # Make sure the extension restarts144 extension = self._run_extension(145 path=daemon.options["extensions_socket"],146 timeout=EXTENSION_TIMEOUT,147 )148 self.assertTrue(extension.isAlive())149 # With the reset there should be 1 extension again150 result = test_base.expect(em.extensions, 1)151 self.assertEqual(len(result), 1)152 print(em.query("select * from example"))153 # Now tear down the daemon154 client.close()155 daemon.kill()156 # The extension should tear down as well157 self.assertTrue(extension.isDead(extension.pid))158 @test_base.flaky159 def test_5_extension_timeout(self):160 # Start an extension without a daemon, with a timeout.161 extension = self._run_extension(timeout=EXTENSION_TIMEOUT)162 self.assertTrue(extension.isAlive())163 # Now start a daemon164 daemon = self._run_daemon({165 "disable_watchdog": True,166 "extensions_socket": extension.options["extensions_socket"],167 "verbose": True,168 })169 self.assertTrue(daemon.isAlive())170 # Get a python-based thrift client171 client = test_base.EXClient(extension.options["extensions_socket"])172 test_base.expectTrue(client.try_open)173 self.assertTrue(client.open(timeout=EXTENSION_TIMEOUT))174 em = client.getEM()175 # The waiting extension should have connected to the daemon.176 result = test_base.expect(em.extensions, 1)177 self.assertEqual(len(result), 1)178 client.close()179 daemon.kill(True)180 extension.kill()181 @test_base.flaky182 def test_6_extensions_autoload(self):183 loader = test_base.Autoloader(184 [test_base.ARGS.build + "/osquery/example_extension.ext"])185 daemon = self._run_daemon({186 "disable_watchdog": True,187 "extensions_timeout": EXTENSION_TIMEOUT,188 "extensions_autoload": loader.path,189 })190 self.assertTrue(daemon.isAlive())191 # Get a python-based thrift client192 client = test_base.EXClient(daemon.options["extensions_socket"])193 self.assertTrue(client.open(timeout=EXTENSION_TIMEOUT))194 em = client.getEM()195 # The waiting extension should have connected to the daemon.196 result = test_base.expect(em.extensions, 1)197 self.assertEqual(len(result), 1)198 client.close()199 daemon.kill(True)200 @test_base.flaky201 def test_6_extensions_directory_autoload(self):202 utils.copy_file(test_base.ARGS.build + "/osquery/example_extension.ext",203 test_base.CONFIG_DIR)204 loader = test_base.Autoloader([test_base.CONFIG_DIR])205 daemon = self._run_daemon({206 "disable_watchdog": True,207 "extensions_timeout": EXTENSION_TIMEOUT,208 "extensions_autoload": loader.path,209 })210 self.assertTrue(daemon.isAlive())211 # Get a python-based thrift client212 client = test_base.EXClient(daemon.options["extensions_socket"])213 self.assertTrue(client.open(timeout=EXTENSION_TIMEOUT))214 em = client.getEM()215 # The waiting extension should have connected to the daemon.216 result = test_base.expect(em.extensions, 1)217 self.assertEqual(len(result), 1)218 client.close()219 daemon.kill(True)220 @test_base.flaky221 def test_7_extensions_autoload_watchdog(self):222 loader = test_base.Autoloader(223 [test_base.ARGS.build + "/osquery/example_extension.ext"])224 daemon = self._run_daemon({225 "extensions_timeout": EXTENSION_TIMEOUT,226 "extensions_autoload": loader.path,227 })228 self.assertTrue(daemon.isAlive())229 # Get a python-based thrift client230 client = test_base.EXClient(daemon.options["extensions_socket"])231 self.assertTrue(client.open(timeout=EXTENSION_TIMEOUT))232 em = client.getEM()233 # The waiting extension should have connected to the daemon.234 result = test_base.expect(em.extensions, 1)235 self.assertEqual(len(result), 1)236 client.close()237 daemon.kill(True)238 @test_base.flaky239 def test_8_external_config(self):240 loader = test_base.Autoloader(241 [test_base.ARGS.build + "/osquery/example_extension.ext"])242 daemon = self._run_daemon({243 "extensions_autoload": loader.path,244 "extensions_timeout": EXTENSION_TIMEOUT,245 "config_plugin": "example",246 })247 self.assertTrue(daemon.isAlive())248 # Get a python-based thrift client249 client = test_base.EXClient(daemon.options["extensions_socket"])250 self.assertTrue(client.open(timeout=EXTENSION_TIMEOUT))251 em = client.getEM()252 # The waiting extension should have connected to the daemon.253 # If there are no extensions the daemon may have exited (in error).254 result = test_base.expect(em.extensions, 1)255 self.assertEqual(len(result), 1)256 client.close()257 daemon.kill(True)258 @test_base.flaky259 def test_9_external_config_update(self):260 # Start an extension without a daemon, with a timeout.261 extension = self._run_extension(timeout=EXTENSION_TIMEOUT)262 self.assertTrue(extension.isAlive())263 # Now start a daemon264 daemon = self._run_daemon({265 "disable_watchdog": True,266 "extensions_timeout": EXTENSION_TIMEOUT,267 "extensions_socket": extension.options["extensions_socket"],268 })269 self.assertTrue(daemon.isAlive())270 # Get a python-based thrift client to the manager and extension.271 client = test_base.EXClient(extension.options["extensions_socket"])272 test_base.expectTrue(client.try_open)273 self.assertTrue(client.open(timeout=EXTENSION_TIMEOUT))274 em = client.getEM()275 # Need the manager to request the extension's UUID.276 result = test_base.expect(em.extensions, 1)277 self.assertTrue(result is not None)278 ex_uuid = result.keys()[0]279 client2 = test_base.EXClient(extension.options["extensions_socket"],280 uuid=ex_uuid)281 test_base.expectTrue(client2.try_open)282 self.assertTrue(client2.open(timeout=EXTENSION_TIMEOUT))283 ex = client2.getEX()284 # Trigger an async update from the extension.285 request = {286 "action": "update",287 "source": "test",288 "data": "{\"options\": {\"config_plugin\": \"update_test\"}}"}289 ex.call("config", "example", request)290 # The update call in the extension should filter to the core.291 options = em.options()292 self.assertTrue("config_plugin" in options.keys())293 self.assertTrue(options["config_plugin"], "update_test")294 # Cleanup thrift connections and subprocesses.295 client2.close()296 client.close()297 extension.kill()298 daemon.kill()299 @test_base.flaky300 def test_91_extensions_settings(self):301 loader = test_base.Autoloader(302 [test_base.ARGS.build + "/osquery/example_extension.ext"])303 daemon = self._run_daemon({304 "disable_watchdog": True,305 "extensions_timeout": EXTENSION_TIMEOUT,306 "extensions_autoload": loader.path,307 })308 self.assertTrue(daemon.isAlive())309 # Get a python-based thrift client for the manager (core).310 client = test_base.EXClient(daemon.options["extensions_socket"])311 self.assertTrue(client.open(timeout=EXTENSION_TIMEOUT))312 em = client.getEM()313 # The waiting extension should have connected to the daemon.314 # This expect statement will block with a short timeout.315 result = test_base.expect(em.extensions, 1)316 self.assertEqual(len(result), 1)317 # The 'complex_example' table reports several columns.318 # Each is a 'test_type', check each expected value.319 result = em.query("select * from complex_example")320 if len(result.response) == 0:321 # There is a brief race between register and registry broadcast322 # That fast external client fight when querying tables.323 # Other config/logger plugins have wrappers to retry/wait.324 time.sleep(0.5)325 result = em.query("select * from complex_example")326 self.assertEqual(result.response[0]['flag_test'], 'false')327 self.assertEqual(result.response[0]['database_test'], '1')328 client.close()329 daemon.kill(True)330if __name__ == "__main__":331 test_base.assertPermissions()332 module = test_base.Tester()333 # Find and import the thrift-generated python interface334 test_base.loadThriftFromBuild(test_base.ARGS.build)...
test_tokenizer.py
Source:test_tokenizer.py
...91]92def test_normal():93 """ Normal/combined usage.94 """95 test_base(TESTS_NORMAL)96def test_emojis():97 """ Tokenizing emojis/emoticons/decorations.98 """99 test_base(TESTS_EMOJIS)100def test_urls():101 """ Tokenizing URLs.102 """103 test_base(TESTS_URLS)104def test_twitter():105 """ Tokenizing hashtags, mentions and emails.106 """107 test_base(TESTS_TWITTER)108def test_phone_nums():109 """ Tokenizing phone numbers.110 """111 test_base(TESTS_PHONE_NUMS)112def test_datetime():113 """ Tokenizing dates and times.114 """115 test_base(TESTS_DATETIME)116def test_currencies():117 """ Tokenizing currencies.118 """119 test_base(TESTS_CURRENCIES)120def test_num_sym():121 """ Tokenizing combinations of numbers and symbols.122 """123 test_base(TESTS_NUM_SYM)124def test_punctuation():125 """ Tokenizing punctuation and contractions.126 """127 test_base(TESTS_PUNCTUATION)128@nottest129def test_base(tests):130 """ Base function for running tests.131 """132 for (test, expected) in tests:133 actual = tokenize(test)134 assert actual == expected, \135 "Tokenization of \'{}\' failed, expected: {}, actual: {}"\...
app.py
Source:app.py
1import opendp.smartnoise.core as sn2# turn on stack traces from panics3import os4os.environ['RUST_BACKTRACE'] = 'full'5from tests import test_partitioning6from tests import test_components7from tests import test_insertion8from tests import test_mechanisms9from tests import test_base10# establish data information11# data_path = os.path.join('.', 'data', 'PUMS_california_demographics_1000', 'data.csv')12# var_names = ["age", "sex", "educ", "race", "income", "married", "pid"]13# ~~~ SCRAP AREA FOR TESTING ~~~14# print("SimpleGeometric")15# test_base.test_accuracies("SimpleGeometric")16# print("Laplace")17# test_base.test_accuracies("Laplace")18# print("Snapping")19# test_base.test_accuracies("Snapping")20# print("Analytic Gaussian")21# test_base.test_accuracies("AnalyticGaussian")22# print("Gaussian")23# test_base.test_accuracies("Gaussian")24#25# print("Empirical Accuracies")26# print("SimpleGeometric")27# test_base.test_accuracy_empirical("SimpleGeometric")28# print("Laplace")29# test_base.test_accuracy_empirical("Laplace")30# print("Snapping")31# test_base.test_accuracy_empirical("Snapping")32# print("Analytic Gaussian")33# test_base.test_accuracy_empirical("AnalyticGaussian")34# print("Gaussian")...
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 SeleniumBase 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?
