Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _check_allowed(self, place):
if not place.acquired:
raise UserError("place {} is not acquired".format(place.name))
if gethostname()+'/'+getuser() not in place.allowed:
host, user = place.acquired.split('/')
if user != getuser():
raise UserError("place {} is not acquired by your user, acquired by {}".format(
place.name, user))
if host != gethostname():
raise UserError("place {} is not acquired on this computer, acquired on {}".format(
place.name, host))
"""
result = set()
# reservation token lookup
token = None
if pattern.startswith('+'):
token = pattern[1:]
if not token:
token = os.environ.get('LG_TOKEN', None)
if not token:
return []
for name, place in self.places.items():
if place.reservation == token:
result.add(name)
if not result:
raise UserError("reservation token {} matches nothing".format(token))
return list(result)
# name and alias lookup
for name, place in self.places.items():
if pattern in name:
result.add(name)
for alias in place.aliases:
if ':' in alias:
namespace, alias = alias.split(':', 1)
if namespace != getuser():
continue
elif alias == pattern: # prefer user namespace
return [name]
if pattern in alias:
result.add(name)
return list(result)
args = dict(arg.split('=', 1) for arg in self.args.bootstrap_args)
try:
drv = target.get_driver(OpenOCDDriver)
except NoDriverFoundError:
drv = OpenOCDDriver(target, name=None, **args)
drv.interface.timeout = self.args.wait
break
elif isinstance(resource, NetworkRKUSBLoader):
try:
drv = target.get_driver(RKUSBDriver)
except NoDriverFoundError:
drv = RKUSBDriver(target, name=None)
drv.loader.timeout = self.args.wait
break
if not drv:
raise UserError("target has no compatible resource available")
target.activate(drv)
drv.load(self.args.filename)
def fastboot(self):
place = self.get_acquired_place()
args = self.args.fastboot_args
if not args:
raise UserError("not enough arguments for fastboot")
if args[0] == 'flash':
if len(args) < 3:
raise UserError("not enough arguments for fastboot flash")
args[2] = os.path.abspath(args[2])
elif args[0] == 'boot':
if len(args) < 2:
raise UserError("not enough arguments for fastboot boot")
args[1:] = map(os.path.abspath, args[1:])
target = self._get_target(place)
from ..driver.fastbootdriver import AndroidFastbootDriver
try:
drv = target.get_driver(AndroidFastbootDriver)
except NoDriverFoundError:
drv = AndroidFastbootDriver(target, name=None)
drv.fastboot.timeout = self.args.wait
target.activate(drv)
if args[0] == 'flash':
drv.flash(args[1], args[2])
return
def tmc_query(self):
drv = self._get_tmc()
query = ' '.join(self.args.query)
if not query:
raise UserError("no query given")
result = drv.query(query)
print(result)
def _get_tmc(self):
place = self.get_acquired_place()
target = self._get_target(place)
from ..driver.usbtmcdriver import USBTMCDriver
from ..resource.remote import NetworkUSBTMC
drv = None
for resource in target.resources:
if isinstance(resource, NetworkUSBTMC):
try:
drv = target.get_driver(USBTMCDriver)
except NoDriverFoundError:
drv = USBTMCDriver(target, name=None)
break
if not drv:
raise UserError("target has no compatible resource available")
target.activate(drv)
return drv
def get_place(self, place=None):
pattern = place or self.args.place
if pattern is None:
raise UserError("place pattern not specified")
places = self._match_places(pattern)
if not places:
raise UserError("place pattern {} matches nothing".format(pattern))
if pattern in places:
return self.places[pattern]
if len(places) > 1:
raise UserError(
"pattern {} matches multiple places ({})".
format(pattern, ', '.join(places))
)
return self.places[places[0]]
def get_place(self, place=None):
pattern = place or self.args.place
if pattern is None:
raise UserError("place pattern not specified")
places = self._match_places(pattern)
if not places:
raise UserError("place pattern {} matches nothing".format(pattern))
if pattern in places:
return self.places[pattern]
if len(places) > 1:
raise UserError(
"pattern {} matches multiple places ({})".
format(pattern, ', '.join(places))
)
return self.places[places[0]]
drv = NetworkPowerDriver(target, name=None, delay=delay)
break
elif isinstance(resource, NetworkUSBPowerPort):
try:
drv = target.get_driver(USBPowerDriver)
except NoDriverFoundError:
drv = USBPowerDriver(target, name=None, delay=delay)
break
elif isinstance(resource, PDUDaemonPort):
try:
drv = target.get_driver(PDUDaemonDriver)
except NoDriverFoundError:
drv = PDUDaemonDriver(target, name=None, delay=int(delay))
break
if not drv:
raise UserError("target has no compatible resource available")
target.activate(drv)
res = getattr(drv, action)()
if action == 'get':
print(
"power for place {} is {}".format(
place.name,
'on' if res else 'off',
)