Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _update_location method in localstack

Best Python code snippet using localstack_python

hardlink_quarantine.py

Source:hardlink_quarantine.py

1# Copyright 2018 Gentoo Foundation2# Distributed under the terms of the GNU General Public License v23from portage import os4from portage.repository.storage.interface import (5	RepoStorageException,6	RepoStorageInterface,7)8from portage.util.futures import asyncio9from portage.util.futures.compat_coroutine import (10	coroutine,11	coroutine_return,12)13from _emerge.SpawnProcess import SpawnProcess14class HardlinkQuarantineRepoStorage(RepoStorageInterface):15	"""16	This is the default storage module, since its quite compatible with17	most configurations.18	It's desirable to be able to create shared hardlinks between the19	download directory and the normal repository, and this is facilitated20	by making the download directory be a subdirectory of the normal21	repository location (ensuring that no mountpoints are crossed).22	Shared hardlinks are created by using the rsync --link-dest option.23	Since the download is initially unverified, it is safest to save24	it in a quarantine directory. The quarantine directory is also25	useful for making the repository update more atomic, so that it26	less likely that normal repository location will be observed in27	a partially synced state.28	"""29	def __init__(self, repo, spawn_kwargs):30		self._user_location = repo.location31		self._update_location = None32		self._spawn_kwargs = spawn_kwargs33		self._current_update = None34	@coroutine35	def _check_call(self, cmd):36		"""37		Run cmd and raise RepoStorageException on failure.38		@param cmd: command to executre39		@type cmd: list40		"""41		p = SpawnProcess(args=cmd, scheduler=asyncio._wrap_loop(), **self._spawn_kwargs)42		p.start()43		if (yield p.async_wait()) != os.EX_OK:44			raise RepoStorageException('command exited with status {}: {}'.\45				format(p.returncode, ' '.join(cmd)))46	@coroutine47	def init_update(self):48		update_location = os.path.join(self._user_location, '.tmp-unverified-download-quarantine')49		yield self._check_call(['rm', '-rf', update_location])50		# Use  rsync --link-dest to hardlink a files into self._update_location,51		# since cp -l is not portable.52		yield self._check_call(['rsync', '-a', '--link-dest', self._user_location,53			'--exclude=/distfiles', '--exclude=/local', '--exclude=/lost+found', '--exclude=/packages',54			'--exclude', '/{}'.format(os.path.basename(update_location)),55			self._user_location + '/', update_location + '/'])56		self._update_location = update_location57		coroutine_return(self._update_location)58	@property59	def current_update(self):60		if self._update_location is None:61			raise RepoStorageException('current update does not exist')62		return self._update_location63	@coroutine64	def commit_update(self):65		update_location = self.current_update66		self._update_location = None67		yield self._check_call(['rsync', '-a', '--delete',68			'--exclude=/distfiles', '--exclude=/local', '--exclude=/lost+found', '--exclude=/packages',69			'--exclude', '/{}'.format(os.path.basename(update_location)),70			update_location + '/', self._user_location + '/'])71		yield self._check_call(['rm', '-rf', update_location])72	@coroutine73	def abort_update(self):74		if self._update_location is not None:75			update_location = self._update_location76			self._update_location = None77			yield self._check_call(['rm', '-rf', update_location])78	@coroutine79	def garbage_collection(self):...

mover.py

Source:mover.py

...34        btn_down.grid(row=1, column=1)35        btn_right.grid(row=1, column=2)36        self.lbl_info.grid(row=2, column=0)37        self.btn_info_to_clipboard.grid(row=2, column=1)38        self._update_location()39    def _update_location(self):40        """Updates internal values for widget location & updates display accordingly"""41        self.target_widget.update()42        self.target_parent.update()43        self.widget_x = self.target_widget.winfo_x()44        self.widget_y = self.target_widget.winfo_y()45        self.lbl_info.config(46            text="(x=" + str(self.widget_x) + ", y=" + str(self.widget_y) + ")"47        )48    def _move_target(self, x: int, y: int):49        """Checks that move is within bounds and if so, re-places widget at x,y"""50        if x >= 0 and x <= self.target_parent.winfo_width():51            if y >= 0 and y <= self.target_parent.winfo_height():52                self.target_widget.place(x=x, y=y)53    def copy_to_clipboard(self):54        pyperclip.copy(self.lbl_info["text"])55    def left(self):56        x = self.widget_x - self.increment57        y = self.widget_y58        self._move_target(x, y)59        self._update_location()60    def up(self):61        x = self.widget_x62        y = self.widget_y - self.increment63        self._move_target(x, y)64        self._update_location()65    def down(self):66        x = self.widget_x67        y = self.widget_y + self.increment68        self._move_target(x, y)69        self._update_location()70    def right(self):71        x = self.widget_x + self.increment72        y = self.widget_y73        self._move_target(x, y)...

inplace.py

Source:inplace.py

1# Copyright 2018 Gentoo Foundation2# Distributed under the terms of the GNU General Public License v23from portage.repository.storage.interface import (4	RepoStorageException,5	RepoStorageInterface,6)7from portage.util.futures.compat_coroutine import coroutine, coroutine_return8class InplaceRepoStorage(RepoStorageInterface):9	"""10	Legacy repo storage behavior, where updates are applied in-place.11	This module is not recommended, since the repository is left in an12	unspecified (possibly malicious) state if the update fails.13	"""14	def __init__(self, repo, spawn_kwargs):15		self._user_location = repo.location16		self._update_location = None17	@coroutine18	def init_update(self):19		self._update_location = self._user_location20		coroutine_return(self._update_location)21		yield None22	@property23	def current_update(self):24		if self._update_location is None:25			raise RepoStorageException('current update does not exist')26		return self._update_location27	@coroutine28	def commit_update(self):29		self.current_update30		self._update_location = None31		coroutine_return()32		yield None33	@coroutine34	def abort_update(self):35		self._update_location = None36		coroutine_return()37		yield None38	@coroutine39	def garbage_collection(self):40		coroutine_return()...

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.