KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use publish_type method in localstack
Best Python code snippet using localstack_python
publish.py
Source:publish.py 
1import re2import time3from PySide2 import QtCore, QtWidgets4from akapipe.core import database as akadatabase5from akapipe.core import db as akadb6from akapipe.core import templates as akatemplates7from . import base8from .. import actions, log9from .. import common10from .. import database11from .. import images12from .. import ui13from ..external import ffmpeg14AnimPublish = 015CompPublish = 116PRESETS = {17 AnimPublish: {18 'name': 'Work in Progress',19 'token': akatemplates.AV_AnimationPublish,20 'formats': (21 'png',22 'jpg',23 ),24 },25 CompPublish: {26 'name': 'Comp & Final',27 'token': akatemplates.AV_CompPublish,28 'formats': (29 'png',30 'jpg',31 'exr',32 'tif',33 ),34 },35}36def get_stamp_text(publish_type, source=None):37 v = 'Published on {time} by {user} to {publish_type}'.format(38 time=time.strftime('%d/%m/%Y %H:%M:%S'),39 publish_type=publish_type,40 user=common.get_username(),41 )42 if source:43 v += '\n Source:\n{}'.format(source)44 return v45@common.error46@common.debug47def stamp_publish(server, job, root, stamp_path, source, publish_type):48 # Write stamp file to publish dir49 with open(stamp_path, 'w') as f:50 v = get_stamp_text(publish_type, source=source)51 f.write(v)52 # Save the stamp info in the bookmark database53 db = database.get_db(server, job, root)54 common.proxy_path(source)55 with db.connection():56 db.setValue(57 common.proxy_path(source),58 'description',59 get_stamp_text(publish_type),60 table=database.AssetTable61 )62def get_progress_bar(frames):63 pbar = QtWidgets.QProgressDialog()64 common.set_stylesheet(pbar)65 pbar.setFixedWidth(common.size(common.DefaultWidth))66 pbar.setLabelText('Publishing files...')67 pbar.setMinimum(1)68 pbar.setMaximum(len(frames))69 pbar.setRange(1, len(frames))70 return pbar71def publish_footage(72 preset,73 source,74 frames,75 seq=None,76 shot=None,77 server=None,78 job=None,79 root=None,80 asset=None,81 task=None,82 make_movie=True,83 add_timecode=None,84 video_size=None,85 ffmpeg_preset=None,86 copy_path=True,87 reveal_publish=True,88):89 """Our footage publish script.90 Args:91 ref (weakref): A reference to an item's data.92 publish_type (unicode): The publish type.93 reveal (bool): Show the publish folder in the file explorer.94 """95 is_collapsed = common.is_collapsed(source)96 if not is_collapsed:97 raise RuntimeError('Item is not a sequence.')98 if not frames:99 raise RuntimeError('Sequence seems to be empty.')100 ext = source.split('.')[-1]101 if make_movie and ext.lower() not in images.oiio_image_extensions:102 raise RuntimeError(f'{ext} is not a accepted image format.')103 # Let's check if the template file exists104 if not all((seq, shot)):105 seq, shot = base.get_seq_shot(source)106 if not all((seq, shot)):107 raise RuntimeError(108 'The item cannot be published because it is not in a shot folder.'109 )110 if ext.lower() not in PRESETS[preset]['formats']:111 raise RuntimeError('Not a valid publish format.')112 client, project = base.get_client_project()113 if not all((client, project)):114 raise RuntimeError(115 'Could not get the client and project names from the the current job.'116 )117 destination = akatemplates.expand_token(118 PRESETS[preset]['token'],119 PR_NetworkPath=server,120 client=client,121 project=project,122 sequence=seq,123 shot=shot,124 )125 if not QtCore.QFileInfo(destination).exists():126 if not QtCore.QDir(destination).mkpath('.'):127 raise RuntimeError(f'Could not create {destination}.')128 # Get the publish type from the path129 regex = r'|'.join(akatemplates.tokens[f] for f in akatemplates.publish_types)130 publish_type = re.search(regex, destination).group(0)131 publish_type = re.sub(r'[0-9_-]+', '', publish_type)132 # Initialize the aka database and get the client and project prefixes133 akadatabase.init_table_data(akadb.CL)134 akadatabase.init_table_data(akadb.PR)135 cl_abbrev = akadatabase.get_value(akadb.CL, client, akadb.CL_Abbreviation)136 pr_abbrev = akadatabase.get_value(akadb.PR, project, akadb.PR_Abbreviation)137 padding = akatemplates.tokens[akatemplates.AV_PublishSequenceFile].count('#')138 frames = sorted(frames, key=lambda x: int(x))139 pbar = get_progress_bar(frames)140 pbar.open()141 for idx, frame in enumerate(frames, 1):142 if pbar.wasCanceled():143 break144 pbar.setValue(idx)145 QtWidgets.QApplication.instance().processEvents()146 source_frame = f'{is_collapsed.group(1)}{frame}{is_collapsed.group(3)}'147 _framename = akatemplates.expand_token(148 akatemplates.AV_PublishSequenceFile,149 CL_Abbreviation=cl_abbrev,150 PR_Abbreviation=pr_abbrev,151 publish_type=publish_type,152 sequence=seq,153 shot=shot,154 ext=ext,155 )156 _framename = re.sub(r'[#]+', str(int(idx)).zfill(padding), _framename)157 destination_frame = f'{destination}/{_framename}'158 # Make thumbnail159 try:160 if idx == 1:161 # Now let's create a thumbnail for this publish162 images.ImageCache.oiio_make_thumbnail(163 source_frame,164 f'{destination}/thumbnail.png',165 common.thumbnail_size166 )167 except:168 log.error('Could not make thumbnail.')169 # Create a movie170 if make_movie and idx == 1:171 try:172 movie_path = ffmpeg.convert(173 source_frame,174 ffmpeg_preset,175 server=server,176 job=job,177 root=root,178 asset=asset,179 task=task,180 size=video_size,181 timecode=add_timecode182 )183 _moviename = akatemplates.expand_token(184 akatemplates.AV_PublishFile,185 CL_Abbreviation=cl_abbrev,186 PR_Abbreviation=pr_abbrev,187 publish_type=publish_type,188 sequence=seq,189 shot=shot,190 ext=next(191 v['output_extension'] for v in ffmpeg.PRESETS.values()192 if v['preset'] == ffmpeg_preset193 ),194 )195 destination_movie = f'{destination}/{_moviename}'196 destination_movie_file = QtCore.QFile(destination_movie)197 if destination_movie_file.exists():198 if not destination_movie_file.remove():199 raise RuntimeError('Could not remove movie file')200 if movie_path and not QtCore.QFile.copy(movie_path, destination_movie):201 log.error(f'Could copy {movie_path}')202 if copy_path:203 actions.copy_path(204 destination_movie,205 mode=common.WindowsPath206 )207 except Exception as e:208 log.error('Failed to make movie.')209 destination_file = QtCore.QFile(destination_frame)210 # The actual copy operation211 if destination_file.exists():212 if not destination_file.remove():213 log.error(f'Could not remove {destination_frame}')214 QtCore.QFile.copy(source_frame, destination_frame)215 # Stamp the publish so we can backtrace if needed216 try:217 stampname = akatemplates.expand_token(218 akatemplates.AV_PublishFile,219 CL_Abbreviation=cl_abbrev,220 PR_Abbreviation=pr_abbrev,221 publish_type=publish_type,222 sequence=seq,223 shot=shot,224 ext='txt',225 )226 stamp_publish(227 server, job, root,228 f'{destination}/{stampname}',229 source,230 publish_type231 )232 except Exception as e:233 log.error(e)234 try:235 from ..teams import message236 db = database.get_db(server, job, root)237 webhook = db.value(db.source(), 'teamstoken', database.BookmarkTable)238 if webhook:239 payload = message.get_payload(240 message.PUBLISH_MESSAGE,241 thumbnail=f'{destination}/thumbnail.png',242 seq=seq,243 shot=shot,244 path=destination,245 date=time.strftime('%d/%m/%Y %H:%M:%S'),246 user=common.get_username(),247 publish_type=publish_type,248 )249 message.send(webhook, payload)250 except Exception as e:251 log.error(e)252 if copy_path:253 ui.OkBox(254 'Item published successfully.',255 'The path to the published movie file has been copied to the clipboard.'256 ).open()257 else:258 ui.OkBox(259 'Item published successfully.'260 ).open()261 if reveal_publish:...
test_redunlive_models.py
Source:test_redunlive_models.py
1# -*- coding: utf-8 -*-2"""Tests for `models` in redunlive webapp."""3import httpretty4import pytest5from epipearl import Epipearl6from cadash.redunlive.models import CaptureAgent7from cadash.redunlive.models import CaLocation8epiphan_url = 'http://fake.example.edu'9class TestCaptureAgentModel(object):10 def setup(self):11 p = CaptureAgent('ABCD1111', 'fake1.example.edu')12 p.channels['live']['channel'] = '1'13 p.channels['live']['publish_type'] = '0'14 p.channels['lowBR']['channel'] = '2'15 p.channels['lowBR']['publish_type'] = '0'16 p.client = Epipearl(epiphan_url, 'user', 'passwd')17 self.ca = p18 @httpretty.activate19 def test_get_converging_live_status(self):20 live = self.ca.channels['live']21 lowBR = self.ca.channels['lowBR']22 assert live['publish_type'] == '0'23 httpretty.register_uri(24 httpretty.GET, '%s/admin/channel%s/get_params.cgi' %25 (epiphan_url, live['channel']), body='publish_type = 6')26 httpretty.register_uri(27 httpretty.GET, '%s/admin/channel%s/get_params.cgi' %28 (epiphan_url, lowBR['channel']), body='publish_type = 6')29 self.ca.sync_live_status()30 assert live['publish_type'] == lowBR['publish_type']31 assert live['publish_type'] == '6'32 @httpretty.activate33 def test_get_diverging_live_status(self):34 live = self.ca.channels['live']35 lowBR = self.ca.channels['lowBR']36 httpretty.register_uri(37 httpretty.GET, '%s/admin/channel%s/get_params.cgi' %38 (epiphan_url, live['channel']), body='publish_type = 0')39 httpretty.register_uri(40 httpretty.GET, '%s/admin/channel%s/get_params.cgi' %41 (epiphan_url, lowBR['channel']), body='publish_type = 6')42 httpretty.register_uri(43 httpretty.GET, '%s/admin/channel%s/set_params.cgi' %44 (epiphan_url, lowBR['channel']), body='', status=201)45 self.ca.sync_live_status()46 assert live['publish_type'] == lowBR['publish_type']47 assert live['publish_type'] == '0'48 @httpretty.activate49 def test_set_live_status(self):50 live = self.ca.channels['live']51 lowBR = self.ca.channels['lowBR']52 httpretty.register_uri(53 httpretty.GET, '%s/admin/channel%s/set_params.cgi' %54 (epiphan_url, live['channel']), body='', status=201)55 httpretty.register_uri(56 httpretty.GET, '%s/admin/channel%s/set_params.cgi' %57 (epiphan_url, lowBR['channel']), body='', status=201)58 new_publish_status = '6'59 self.ca.write_live_status(new_publish_status)60 assert live['publish_type'] == lowBR['publish_type']61 assert live['publish_type'] == new_publish_status62class TestCaLocationModel(object):63 def setup(self):64 p = CaptureAgent('ABCD1111', 'fake1.example.edu')65 p.channels['live']['channel'] = '1'66 p.channels['live']['publish_type'] = '0'67 p.channels['lowBR']['channel'] = '2'68 p.channels['lowBR']['publish_type'] = '0'69 self.primary = p70 s = CaptureAgent('ABCD2222', 'fake2.example.edu')71 s.channels['live']['channel'] = '7'72 s.channels['live']['publish_type'] = '0'73 s.channels['lowBR']['channel'] = '8'74 s.channels['lowBR']['publish_type'] = '0'75 self.secondary = s76 e1 = CaptureAgent('ABCD3333', 'fake3.example.edu')77 e1.channels['live']['channel'] = '3'78 e1.channels['live']['publish_type'] = '0'79 e1.channels['lowBR']['channel'] = '4'80 e1.channels['lowBR']['publish_type'] = '0'81 self.experimental1 = e182 e2 = CaptureAgent('ABCD4444', 'fake4.example.edu')83 e2.channels['live']['channel'] = '3'84 e2.channels['live']['publish_type'] = '0'85 e2.channels['lowBR']['channel'] = '4'86 e2.channels['lowBR']['publish_type'] = '0'87 self.experimental2 = e288 l = CaLocation('loc1')89 l.primary_ca = self.primary90 l.secondary_ca = self.secondary91 l.experimental_cas = [self.experimental1, self.experimental2]92 self.loc = l93 def test_simple_location(self):94 assert self.loc.primary_ca.name == 'fake1'95 assert self.loc.secondary_ca.name == 'fake2'96 assert len(self.loc.experimental_cas) == 297 assert self.loc.active_livestream is None98 def test_same_ca_for_secondary(self):99 with pytest.raises(ValueError):100 self.loc.secondary_ca = self.primary101 def test_same_ca_for_primary(self):102 with pytest.raises(ValueError):103 self.loc.primary_ca = self.secondary104 def test_toggle_active_livestream(self):105 assert self.loc.active_livestream is None106 self.primary.channels['live']['publish_type'] = '6'107 assert self.loc.active_livestream == 'primary'108 self.primary.channels['live']['publish_type'] = '0'109 self.secondary.channels['live']['publish_type'] = '6'...
scraper.py
Source:scraper.py
1import requests2from pprint import pprint3import json4from multiprocessing.dummy import Pool as ThreadPool5import time6def read_config(filename):7 data = None8 with open(filename, 'r') as f:9 data = json.loads(f.read())10 return data11config = read_config('config.json')12puzzle_ids = []13game_url_template = "https://nyt-games-prd.appspot.com/svc/crosswords/v6/game/"14security_header = config['s-header']15user_id = config['user-id']16game_data = []17# for multithreading18urls = [19 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-01-01&date_end=2019-02-28',20 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-03-01&date_end=2019-04-30',21 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-05-01&date_end=2019-06-30',22 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-07-01&date_end=2019-07-31',23 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-08-01&date_end=2019-10-31',24 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-11-01&date_end=2020-01-31',25 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2020-02-01&date_end=2020-03-31',26 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2020-04-01&date_end=2020-06-31',27 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2020-07-01&date_end=2020-08-31',28 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2020-09-01&date_end=2020-11-31',29 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2020-12-01&date_end=2021-02-28',30 'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2021-03-01&date_end=2021-03-08'31]32def get_cword_data(cword):33 data_url = game_url_template + str(cword['puzzle_id']) + ".json"34 cword_response = requests.get(data_url, headers=security_header)35 cword_response_data = json.loads(cword_response.content)36 cword["board"] = cword_response_data["board"]37 return cword38def main():39 start = time.time()40 # scrape crossword lists for publish dates and game ids41 for url in urls:42 response = requests.get(url, headers=security_header)43 data = json.loads(response.content)['results']44 solved_cwords = [cword for cword in data if cword['solved']]45 with ThreadPool(20) as pool:46 results = pool.map(get_cword_data, solved_cwords)47 game_data.extend(results)48 # output the list of boards somehow (for now, writing to a json file)49 with open("data.json", "w") as f:50 f.write(json.dumps(game_data))51 end = time.time()52 print("script finished in", end - start, "s")53if __name__ == "__main__":...
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 localstack 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?
