KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _auth method in tempest

Best Python code snippet using tempest_python

emotiv_api.py

Source:emotiv_api.py

1""" emotiv_api.py: All the api json implemented by the emotiv instruction2website: https://emotiv.github.io/cortex-docs/#data-types3"""4__author__ = "Isaac Sim"5__copyright__ = "Copyright 2019, The Realtime EEG Analysis Project"6__credits__ = ["Isaac Sim"]7__license__ = ""8__version__ = "1.0.0"9__maintainer__ = ["Isaac Sim", "Dongjoon Jeon"]10__email__ = "gilgarad@igsinc.co.kr"11__status__ = "Development"12# Mandatory Methods by Cortex API13def logout(user_id: str):14    """ Logout15    :param user_id:16    :return:17    """18    return {19        "jsonrpc": "2.0",20        "method": "logout",21        "params": {22            "username": user_id23        },24        "id": 125    }26def login(user_id: str, password: str, client_id: str, client_secret: str):27    """ Login28    :param user_id:29    :param password:30    :param client_id:31    :param client_secret:32    :return:33    """34    return {35        "jsonrpc": "2.0",36        "method": "login",37        "params": {38            "username": user_id,39            "password": password,40            "client_id": client_id,41            "client_secret": client_secret42        },43        "id": 144    }45def getUserLogin():46    """ Get all users logged in to Cortex47    :return:48    """49    return {50        "jsonrpc": "2.0",51        "method": "getUserLogin",52        "id": 153    }54def authorize(client_id: str, client_secret: str):55    """ Authenticate a user56    :param client_id:57    :param client_secret:58    :return:59    """60    return {61        "jsonrpc": "2.0",62        "method": "authorize",63        "params": {64            "client_id": client_id,65            "client_secret": client_secret,66            # "license": license_id,67            # "debit": 068        },69        "id": 170    }71def acceptLicense(_auth: str):72    """ Accept license in order to use Cortex73    :param _auth:74    :return:75    """76    return {77        "jsonrpc": "2.0",78        "method": "acceptLicense",79        "params": {80            "_auth": _auth81        },82        "id": 183    }84def getLicenseInfo(_auth: str):85    """ Get license information of a user86    :param _auth:87    :return:88    """89    return {90        "jsonrpc": "2.0",91        "method": "getLicenseInfo",92        "params": {93            "_auth": _auth94        },95        "id": 196    }97def queryHeadsets():98    """ Shows the detailed list of headsets connected99    :return:100    """101    return {102        "jsonrpc": "2.0",103        "method": "queryHeadsets",104        "params": {},105        "id": 1106    }107def createSession(_auth: str):108    """ Creates a new session109    :param _auth:110    :return:111    """112    # active, close, startRecord, stopRecord, addTags, removeTags113    return {114        "jsonrpc": "2.0",115        "method": "createSession",116        "params": {117            "_auth": _auth,118            # "status": "open"119            # "status": "close"120            "status": "active"121        },122        "id": 1123    }124def updateSession(_auth: str, sess_id: str):125    """ Update the session with new information126    :param _auth:127    :param sess_id:128    :return:129    """130    # status: active, close, startRecord, stopRecord, addTags, removeTags131    return {132        "jsonrpc": "2.0",133        "method": "updateSession",134        "params": {135            "_auth": _auth,136            "session": sess_id,137            "status": "close"138        },139        "id": 1140    }141def querySessions(_auth: str):142    """ Query the list of all sessions143    :param _auth:144    :return:145    """146    return {147        "jsonrpc": "2.0",148        "method": "querySessions",149        "params": {150            "_auth": _auth151        },152        "id": 1153    }154def subscribe(_auth: str):155    """ Subscribe to the streams of dev, eeg, pow, met156    :param _auth:157    :return:158    """159    return {160        "jsonrpc": "2.0",161        "method": "subscribe",162        "params": {163            "_auth": _auth,164            "streams": [165                "dev", # connection status to the cortex166                "eeg", # eeg rawdata167                "pow", # theta alpha betaL betaH gamma168                "met" # emotiv's analyzed result of169                        # interest, stress, relaxation, excitement, engagement, long term excitement, focus170            ]171        },172        "id": 1173    }174def updateNote(_auth: str):175    """ Edit note when recording is on a session176    :param _auth:177    :return:178    """179    return {180        "jsonrpc": "2.0",181        "method": "updateNote",182        "params": {183            "_auth": _auth,184            "session": "abcd",185            "note": "new note",186            "record": "1234"187        },188        "id": 1189    }190def unsubscribe(_auth: str):191    """ Stop subscribe to the streams of dev, eeg, pow, met192    :param _auth:193    :return:194    """195    return {196        "jsonrpc": "2.0",197        "method": "unsubscribe",198        "params": {199            "_auth": _auth,200            "streams": [201                "eeg",202                "dev",203                "pow",204                "met"205            ]206        },207        "id": 1208    }209def injectMarker(_auth: str, sess: str):210    """ Injects a marker into the data stream for a headset211    :param _auth:212    :param sess:213    :return:214    """215    return {216        "jsonrpc": "2.0",217        "method": "injectMarker",218        "params": {219            "_auth": _auth,220            "session": sess,221            "label": "test1",222            "value": "record-1",223            "port": "USB",224            "time": 123456789225        },226        "id": 1227    }228def getDetectionInfo():229    """ This request return all useful informations for set up training Mental Command and Facial Expression230    :return:231    """232    return {233        "jsonrpc": "2.0",234        "method": "getDetectionInfo",235        "params": {236            "detection": "mentalCommand"237        },238        "id": 1...

test__auth.py

Source:test__auth.py

1# Copyright 2016 Google Inc. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#      http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14import mock15import google.auth.credentials16import google_auth_httplib217import httplib218import oauth2client.client19import unittest220from googleapiclient import _auth21class TestAuthWithGoogleAuth(unittest2.TestCase):22    def setUp(self):23        _auth.HAS_GOOGLE_AUTH = True24        _auth.HAS_OAUTH2CLIENT = False25    def tearDown(self):26        _auth.HAS_GOOGLE_AUTH = True27        _auth.HAS_OAUTH2CLIENT = True28    def test_default_credentials(self):29        with mock.patch('google.auth.default', autospec=True) as default:30            default.return_value = (31                mock.sentinel.credentials, mock.sentinel.project)32            credentials = _auth.default_credentials()33            self.assertEqual(credentials, mock.sentinel.credentials)34    def test_with_scopes_non_scoped(self):35        credentials = mock.Mock(spec=google.auth.credentials.Credentials)36        returned = _auth.with_scopes(credentials, mock.sentinel.scopes)37        self.assertEqual(credentials, returned)38    def test_with_scopes_scoped(self):39        class CredentialsWithScopes(40                google.auth.credentials.Credentials,41                google.auth.credentials.Scoped):42            pass43        credentials = mock.Mock(spec=CredentialsWithScopes)44        credentials.requires_scopes = True45        returned = _auth.with_scopes(credentials, mock.sentinel.scopes)46        self.assertNotEqual(credentials, returned)47        self.assertEqual(returned, credentials.with_scopes.return_value)48        credentials.with_scopes.assert_called_once_with(mock.sentinel.scopes)49    def test_authorized_http(self):50        credentials = mock.Mock(spec=google.auth.credentials.Credentials)51        authorized_http = _auth.authorized_http(credentials)52        self.assertIsInstance(53            authorized_http,54            google_auth_httplib2.AuthorizedHttp)55        self.assertEqual(authorized_http.credentials, credentials)56        self.assertIsInstance(authorized_http.http, httplib2.Http)57        self.assertIsInstance(authorized_http.http.timeout, int)58        self.assertGreater(authorized_http.http.timeout, 0)59class TestAuthWithOAuth2Client(unittest2.TestCase):60    def setUp(self):61        _auth.HAS_GOOGLE_AUTH = False62        _auth.HAS_OAUTH2CLIENT = True63    def tearDown(self):64        _auth.HAS_GOOGLE_AUTH = True65        _auth.HAS_OAUTH2CLIENT = True66    def test_default_credentials(self):67        default_patch = mock.patch(68            'oauth2client.client.GoogleCredentials.get_application_default')69        with default_patch as default:70            default.return_value = mock.sentinel.credentials71            credentials = _auth.default_credentials()72            self.assertEqual(credentials, mock.sentinel.credentials)73    def test_with_scopes_non_scoped(self):74        credentials = mock.Mock(spec=oauth2client.client.Credentials)75        returned = _auth.with_scopes(credentials, mock.sentinel.scopes)76        self.assertEqual(credentials, returned)77    def test_with_scopes_scoped(self):78        credentials = mock.Mock(spec=oauth2client.client.GoogleCredentials)79        credentials.create_scoped_required.return_value = True80        returned = _auth.with_scopes(credentials, mock.sentinel.scopes)81        self.assertNotEqual(credentials, returned)82        self.assertEqual(returned, credentials.create_scoped.return_value)83        credentials.create_scoped.assert_called_once_with(mock.sentinel.scopes)84    def test_authorized_http(self):85        credentials = mock.Mock(spec=oauth2client.client.Credentials)86        authorized_http = _auth.authorized_http(credentials)87        http = credentials.authorize.call_args[0][0]88        self.assertEqual(authorized_http, credentials.authorize.return_value)89        self.assertIsInstance(http, httplib2.Http)90        self.assertIsInstance(http.timeout, int)91        self.assertGreater(http.timeout, 0)92class TestAuthWithoutAuth(unittest2.TestCase):93    def setUp(self):94        _auth.HAS_GOOGLE_AUTH = False95        _auth.HAS_OAUTH2CLIENT = False96    def tearDown(self):97        _auth.HAS_GOOGLE_AUTH = True98        _auth.HAS_OAUTH2CLIENT = True99    def test_default_credentials(self):100        with self.assertRaises(EnvironmentError):101            print(_auth.default_credentials())102class TestGoogleAuthWithoutHttplib2(unittest2.TestCase):103    def setUp(self):104        _auth.google_auth_httplib2 = None105    def tearDown(self):106        _auth.google_auth_httplib2 = google_auth_httplib2107    def test_default_credentials(self):108        credentials = mock.Mock(spec=google.auth.credentials.Credentials)109        with self.assertRaises(ValueError):...

eeg_control.py

Source:eeg_control.py

1# Importing the packages and the libraries2from credentials import *3import json4from websocket import create_connection5import ssl6import time7import requests8# Create object of class create_connection9ws = create_connection("wss://emotivcortex.com:54321", sslopt={"cert_reqs": ssl.CERT_NONE})10# Create a session with the Emotiv Insight11ws.send(json.dumps({12 "jsonrpc": "2.0",13 "method": "createSession",14 "params": {15     "_auth": _auth,16     "headset":"INSIGHT-5A688F16",17     "status": "open"18 },19 "id": 120 }))21print(ws.recv())22# Subscribe to sys stream23ws.send(json.dumps({24 "jsonrpc": "2.0",25 "method": "subscribe", 26 "params": {27     "_auth":_auth,28     "streams":[29     "sys"30     ]31 },32 "id": 133 }))34print(ws.recv())35# Begin training mental commands36# Training neutral37ws.send(json.dumps( {38 "jsonrpc": "2.0", 39 "method": "training", 40 "params": {41   "_auth":_auth,42   "detection":"mentalCommand",43   "action":"neutral",44   "status":"start"45 }, 46 "id": 147 }))48print(ws.recv())49time.sleep(5)50print(ws.recv())51time.sleep(10)52print(ws.recv())53ws.send(json.dumps( {54 "jsonrpc": "2.0", 55 "method": "training", 56 "params": {57     "_auth":_auth,58     "detection":"mentalCommand",59     "action":"neutral",60     "status":"accept"61 }, 62 "id": 163 }64))65print(ws.recv())66time.sleep(2)67print(ws.recv())68# Training push (forward)69ws.send(json.dumps( {70 "jsonrpc": "2.0", 71 "method": "training", 72 "params": {73   "_auth":_auth,74   "detection":"mentalCommand",75   "action":"push",76   "status":"start"77 }, 78 "id": 179 }))80print(ws.recv())81time.sleep(5)82print(ws.recv())83time.sleep(10)84print(ws.recv())85ws.send(json.dumps( {86 "jsonrpc": "2.0", 87 "method": "training", 88 "params": {89     "_auth":_auth,90     "detection":"mentalCommand",91     "action":"push",92     "status":"accept"93 }, 94 "id": 195 }96))97print(ws.recv())98time.sleep(2)99print(ws.recv())100# Training left (pivot left)101ws.send(json.dumps( {102 "jsonrpc": "2.0", 103 "method": "training", 104 "params": {105   "_auth":_auth,106   "detection":"mentalCommand",107   "action":"left",108   "status":"start"109 }, 110 "id": 1111 }))112print(ws.recv())113time.sleep(5)114print(ws.recv())115time.sleep(10)116print(ws.recv())117ws.send(json.dumps( {118 "jsonrpc": "2.0", 119 "method": "training", 120 "params": {121     "_auth":_auth,122     "detection":"mentalCommand",123     "action":"left",124     "status":"accept"125 }, 126 "id": 1127 }128))129print(ws.recv())130time.sleep(2)131print(ws.recv())132# Training right (pivot right)133ws.send(json.dumps( {134 "jsonrpc": "2.0", 135 "method": "training", 136 "params": {137   "_auth":_auth,138   "detection":"mentalCommand",139   "action":"right",140   "status":"start"141 }, 142 "id": 1143 }))144print(ws.recv())145time.sleep(5)146print(ws.recv())147time.sleep(10)148print(ws.recv())149ws.send(json.dumps( {150 "jsonrpc": "2.0", 151 "method": "training", 152 "params": {153     "_auth":_auth,154     "detection":"mentalCommand",155     "action":"right",156     "status":"accept"157 }, 158 "id": 1159 }160))161print(ws.recv())162time.sleep(2)163print(ws.recv())164# Obtain stream of mental commands165# Subscribe to com stream166ws.send(json.dumps({167 "jsonrpc": "2.0",168 "method": "subscribe", 169 "params": {170     "_auth":_auth,171     "streams":[172     "com"173 ]174 },175 "id": 1176 }))177print(ws.recv())178# loop for rover control179while True:180 thought = json.loads(ws.recv())["com"][0]181 print(thought)182 183 if(thought == "push"):184     url_get = "http://192.168.0.8:5000/forward"185     res = requests.get(url_get)186 elif(thought == "left"):187     url_get = "http://192.168.0.8:5000/pivot_left"188     res = requests.get(url_get)189 elif(thought == "right"):190     url_get = "http://192.168.0.8:5000/pivot_right"...

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.