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

How to use connect_remote method in tempest

Best Python code snippet using tempest_python

transmogrifier.py

Source:transmogrifier.py

1import os2import sys3import urllib24import xbmc5import xbmcgui6import xbmcaddon7import hashlib8import base649from urllib2 import URLError, HTTPError10WINDOW_PREFIX = 'transmogrifier'11try: 12	import simplejson as json13except ImportError: 14	import json15def set_property(k, v):16	k = "%s.%s" % (WINDOW_PREFIX, k)17	xbmcgui.Window(10000).setProperty(k, str(v))18	19def get_property(k):20	k = "%s.%s" % (WINDOW_PREFIX, k)21	p = xbmcgui.Window(10000).getProperty(k)22	if p == 'false': return False23	if p == 'true': return True24	return p25def clear_property(k):26	k = "%s.%s" % (WINDOW_PREFIX, k)27	xbmcgui.Window(10000).clearProperty(k)28	29class TransmogrifierAPI:30	def __init__(self):31		self.use_remote_host = xbmcaddon.Addon(id='service.transmogrifier').getSetting('connect_remote') == 'true'32		if self.use_remote_host:33			self.host = xbmcaddon.Addon(id='service.transmogrifier').getSetting('remote_host')34			self.pin = xbmcaddon.Addon(id='service.transmogrifier').getSetting('remote_auth_pin')35			self.token = xbmcaddon.Addon(id='service.transmogrifier').getSetting('remote_auth_token')36			self.port = xbmcaddon.Addon(id='service.transmogrifier').getSetting('remote_control_port')37		else:38			self.host = 'localhost'39			self.pin = xbmcaddon.Addon(id='service.transmogrifier').getSetting('auth_pin')40			self.token = xbmcaddon.Addon(id='service.transmogrifier').getSetting('auth_token')41			self.port = xbmcaddon.Addon(id='service.transmogrifier').getSetting('control_port')42		self.save_directory = xbmcaddon.Addon(id='service.transmogrifier').getSetting('save_directory')43		self._authorize()44		45	def get_progress(self):46		return self._call('progress')47	48	def notify(self, message):49		image = os.path.join(xbmc.translatePath( xbmcaddon.Addon(id='service.transmogrifier').getAddonInfo('path') ), 'icon.png')50		cmd = "XBMC.Notification(TransmogrifierAPI, %s, 1500, %s)" % (message, image)51		xbmc.executebuiltin(cmd)52	53	def _authorize(self):54		if self.token == '':55			response = self._call("authorize")56			self.token = response['token']57			if xbmcaddon.Addon(id='service.transmogrifier').getSetting('connect_remote') == 'true':58				xbmcaddon.Addon(id='service.transmogrifier').setSetting('remote_auth_token', self.token)59			else:60				xbmcaddon.Addon(id='service.transmogrifier').setSetting('auth_token', self.token)61		else:62			response = self._call("validate_token")63			if 'success' not in response.keys():64				if xbmcaddon.Addon(id='service.transmogrifier').getSetting('connect_remote') == 'true':65					xbmcaddon.Addon(id='service.transmogrifier').setSetting('remote_auth_token', '')66				else:67					xbmcaddon.Addon(id='service.transmogrifier').setSetting('auth_token', '')68	69	def get_cached_file(self, title, season=None, episode=None, year=None):70		from dudehere.routines.vfs import VFSClass71		vfs = VFSClass()72		if season is not None:73			for extension in ['avi', 'mkv', 'mov', 'mp4', 'flv']:74				path = vfs.join(self.save_directory, "TV Shows/%s %sx%s.%s" % (title, season, episode, extension))75				if vfs.exists(path):76					return path77			return False78		else:79			for extension in ['avi', 'mkv', 'mov', 'mp4', 'flv']:80				path = vfs.join(self.save_directory,"Movies/%s (%s).%s" % (title, year, extension))81				if vfs.exists(path):82					return path83			return False84	85	def get_streaming_url(self, url, host=None):86		if not url.lower().startswith(("http://", "https://")): return url87		88		try:89			response = self._call('status')90			if response['status'] != 200: 91				print 'TransmogrifierAPI Error: Proxy Host down, restart service?'92				return url93		except:94			print 'TransmogrifierAPI Error: Proxy Host down, restart service?'95			return url96		if host is not None:97			response = self._call('blacklist', {"host": host})98			if response['status'] == 406:99				print 'TransmogrifierAPI Notice: Host [ %s ] is blacklist' % host100				return url101		file_id = hashlib.md5(url).hexdigest()102		set_property('streaming', "true")103		set_property('file_id', file_id)104		clear_property('abort_id')105		hash_url = base64.b64encode(url)106		stream_url = "http://%s:%s/stream/%s/%s" % (self.host, self.port, hash_url, file_id)107		return stream_url108	109	def start_buffering(self, url):110		url = url + '?start_buffering=true'111		request = urllib2.Request(url)112		f = urllib2.urlopen(request)113		f.read()114		f.close()115	116	def clean_queue(self):117		from dudehere.routines.plugin import Plugin118		c = Plugin().confirm("Clean Queue", "Remove complete and failed?")119		if c: self._call("clear_queue")120	121	def enqueue(self, videos):122		if type(videos) is dict: videos = [videos]123		data = {"videos": videos}124		result = self._call('enqueue', data)125		try:126			if result['status'] == 200:127				self.notify('Successful Enqueue')128			else:129				self.notify('Enqueue Failed, review log for details')130		except:131			self.notify('Enqueue Failed, review log for details')132		return result133	134	def abort(self, file_id):135		return self._call('abort', {"file_id": file_id})136	137	def restart(self, ids):138		if type(ids) is int: ids = [ids]139		data = {"videos": []}140		for id in ids:141			data['videos'].append({"id": id})142		return self._call('restart', data)143	144	def delete(self, ids):145		if type(ids) is int: ids = [ids]146		data = {"videos": []}147		for id in ids:148			data['videos'].append({"id": id})149		return self._call('delete', data)150	151	152	153	def change_priority(self, id, priority):154		data = {"videos": [{"id": id, "priority": priority}]}155		return self._call('change_priority', data)156		157	def get_videos(self, media):158		from dudehere.routines.vfs import VFSClass159		vfs = VFSClass()160		if media == 'tv':161			path = vfs.join(self.save_directory, "TV Shows")162		else:163			path = vfs.join(self.save_directory, "Movies")164		videos = vfs.ls(path, pattern="(avi|mp4|mkv|mov|flv)$")[1]165		return path, videos166	167	def get_queue(self):168		return self._call('queue')169	170	def get_poll(self):171		return self._call('poll')172	173	def get_progress(self):174		return self._call('progress')175	176	def _build_url(self):177		url = "http://%s:%s/api.json" % (self.host, self.port)178		return url179	180	def _build_request(self, method):181		if method=='authorize':182			request = {"method": method, "pin": self.pin}183		else:184			request = {"method": method, "token": self.token}185		return request186	187	def _call(self, method, data=None):188		189		url = self._build_url()190		request = self._build_request(method)191		if data:192			for key in data.keys():193				request[key] = data[key]194		json_data = json.dumps(request)195		headers = {'Content-Type': 'application/json'}196		try:197			request = urllib2.Request(url, data=json_data, headers=headers)198			f = urllib2.urlopen(request)199			response = f.read()200			return json.loads(response)201		except HTTPError as e:...

repo_kwlink.py

Source:repo_kwlink.py

1#!/usr/bin/env python32''' GitHub Python scraper3Linking github repositories to their keywords/topics.4This code hits the abuse detection mechanism, even with the pausing.5'''6from github import Github7from github import RateLimitExceededException8from github.GithubException import UnknownObjectException9from py2neo import Graph10from json import load11import random12from time import sleep13with open('.gitignore') as gi:14    good = False15    # This simply checks to see if you have a connection string in your repo.16    # I use `strip` to remove whitespace/newlines.17    for line in gi:18        if line.strip() == "connect_remote.json":19            good = True20            break21if good is False:22    print("The connect_remote.json file is not in your .gitignore file. \23           Please add it!")24with open('./connect_remote.json') as f:25    data = load(f)26with open('./gh.token') as f:27    gh_token = f.read().splitlines()28g = Github(gh_token[2])29graph = Graph(**data[1])30tx = graph.begin()31cypher = """MATCH (:TYPE {type:"schema:CodeRepository"})-[:isType]-(cr:OBJECT)32            WHERE NOT EXISTS(cr.errorType)33            OPTIONAL MATCH (cr)-[]-(:ANNOTATION)-[]-(k:KEYWORD)34            WITH DISTINCT cr.name AS repos, k.keyword AS kw35            WHERE kw IS NULL36            RETURN repos;"""37print("Matching existing repositories")38dbs = graph.run(cypher).data()39random.shuffle(dbs)40short_db = dbs41for db in dbs:42    print(db['repos'])43    while True:44        sleep(2)45        try:46            repo = g.get_repo(db['repos'])47            repo_top = repo.get_topics()48            if len(repo_top) > 0:49                print("Hitting a new repository with topics! -", repo.name)50                print(repo_top)51                addFiles = {'id': repo.id, 'keywords': repo_top}52                with open('cql/github_topics.cql', mode="r") as gitadd:53                    silent = graph.run(gitadd.read(), addFiles)54            silent = short_db.pop(short_db.index(db))55            break56        except UnknownObjectException as inst:57            print("Missing repo:", str(inst))58            with open('cql/repo404.cql', mode="r") as gitadd:59                silent = graph.run(gitadd.read(), db)60            print("Marked node.")61            break62        except RateLimitExceededException as inst:63            print("Rate exceeded:", str(inst))64            print('Oops, broke for ' + db['repos']65                  + ' with repo call.')66            sleep(300)67            continue68        except Exception as inst:69            print(inst)70            sleep(120)...

testing_return.py

Source:testing_return.py

...38else:39    writefunc()404142def connect_remote():43    # repo_name = readfunc()44    print(emoji.emojize("Connecting to your remote directory...:lying_face:"))45    # os.system("{} {}{}.git".format(remote, user_repo_link, repo_name))46    print("{} {}{}.git".format(remote, user_repo_link, repo_name))47    # print(repo_name)4849
...

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.