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.