Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
total_size, total_count = assetsize(folder)
o, r, w = recprocess(gee_type, gee_id)
# print(gee_id,gee_type,total_size,total_count,o,r,w)
try:
with open(output, "a") as csvfile:
writer = csv.writer(csvfile, delimiter=",", lineterminator="\n")
writer.writerow(
[gee_type, gee_id, total_count, total_size, o, r, w]
)
csvfile.close()
except Exception as e:
print(e)
for collection in collection_list:
gee_id = collection
gee_type = "collection"
logger.info("Processing Collection {}".format(gee_id))
total_size, total_count = assetsize(collection)
o, r, w = recprocess(gee_type, gee_id)
# print(gee_id,gee_type,total_size,total_count,o,r,w)
with open(output, "a") as csvfile:
writer = csv.writer(csvfile, delimiter=",", lineterminator="\n")
writer.writerow([gee_type, gee_id, total_count, total_size, o, r, w])
csvfile.close()
for table in table_list:
gee_id = table
gee_type = "table"
logger.info("Processing table {}".format(gee_id))
total_size, total_count = assetsize(table)
o, r, w = recprocess(gee_type, gee_id)
# print(gee_id,gee_type,total_size,total_count,o,r,w)
with open(output, "a") as csvfile:
writer = csv.writer(csvfile, delimiter=",", lineterminator="\n")
d = uiautomator2.connect(device_url)
devinfo = d.device_info
serial = devinfo['serial']
logger.info("serial %s, udid %s", serial, devinfo['udid'])
aserver_url = devinfo.get(
"serverUrl",
"http://wifiphone.nie.netease.com") # TODO(atx-agent should udpate)
logger.info("atx-server url %s", aserver_url)
r = requests.get(
aserver_url + "/devices/" + devinfo['udid'] + "/info").json()
pvd = r.get('provider')
if not pvd:
logger.info("u2init not connected")
return "http://" + d.wlan_ip + ":7912/packages"
pkg_url = 'http://%s:%d/devices/%s/pkgs' % (pvd['ip'], pvd['port'], serial)
logger.info("package url %s", pkg_url)
return pkg_url
confirm = (
yes
or input(f"\nDisplay {len(search_results)} results? (y/n) ") in ("y", "Y")
)
if confirm:
for result in search_results:
result_num += 1
title = result.get('title', "")
artist = result.get('artist', "")
album = result.get('album', "")
song_id = result['id']
logger.info(
f"{result_num:>{pad}}/{total} {title} -- {artist} -- {album} ({song_id})"
)
else:
logger.info("No songs found matching query")
mc.logout()
logger.info("All done!")
def read_response(self) -> tuple:
"""
Returns:
tuple: (output, finished: bool)
"""
line = self._p.stdout.readline()
logger.info("READ: %r", line)
if self.is_closed():
self._p = None
return line, True
if line.rstrip().endswith(self._magic):
return line.rstrip()[:-len(self._magic)], True
return line, False
def unblock_services(self, services=None):
"""
Unblock this application from accessing its services on all its known hosts.
:param services: List[String]; List of service names to unblock, will target all if unset.
"""
cfg = Config()
for app_instance in self.instances:
cmds = []
for service in self.services:
if service['type'] in cfg['service-whitelist']:
continue
if services and service['name'] not in services:
continue
logger.info("Unblocking %s for %s:%s", service['name'], app_instance['diego_id'], app_instance['cont_ip'])
for (sip, protocol, port) in service['hosts']:
cmd = ['sudo', 'iptables', '-D', 'FORWARD', '-s', app_instance['cont_ip'],
'-d', sip, '-p', protocol]
if port != 'all':
cmd.extend(['--dport', port])
cmd.extend(['-j', 'DROP'])
for _ in range(TIMES_TO_REMOVE):
cmds.append(' '.join(map(str, cmd)))
cmd = ['sudo', 'iptables', '-D', 'FORWARD', '-d', app_instance['cont_ip'],
'-s', sip, '-p', protocol]
if port != 'all':
cmd.extend(['--sport', port])
cmd.extend(['-j', 'DROP'])
for _ in range(TIMES_TO_REMOVE):
cmds.append(' '.join(map(str, cmd)))
def sc_notify(event):
logger.info("SmartContract Runtime.Notify event: %s", event)
# Make sure that the event payload list has at least one element.
if not isinstance(event.event_payload, ContractParameter) or event.event_payload.Type != ContractParameterType.Array or not len(event.event_payload.Value):
return
# The event payload list has at least one element. As developer of the smart contract
# you should know what data-type is in the bytes, and how to decode it. In this example,
# it's just a string, so we decode it with utf-8:
logger.info("- payload part 1: %s", event.event_payload.Value[0].Value.decode("utf-8"))
# test/integration/targets/filter_random_mac
integration_tests_files.extend(glob.glob(os.path.join(targets_dir, f'{plugin_type}_{plugin_name}')))
# aliased integration tests
# https://github.com/ansible-community/collection_migration/issues/326
integration_tests_files.extend(get_processed_aliases(checkout_dir).get(plugin_name, []))
# (filename, marked_for_removal)
# we do not mark integration tests dependencies (meta/main.yml) for removal yet as at this point
# we do not know if they are used by some other target, we do the cleanup at the end of the migration
# see process_integration_tests_deps and actually_remove functions
files = [(filename, True) for filename in integration_tests_files]
deps = []
for fname, dummy_to_remove in files:
logger.info('Found integration tests for %s %s in %s', plugin_type, plugin_name, fname)
deps.extend(process_integration_tests_deps(checkout_dir, fname))
return files + deps
def update_preset(self, preset, sumoregion, accesskeyid, accesskey):
logger.info('Updating preset in credential store.')
if preset:
if self.credentialstore.name_exists(preset):
try:
result = self.credentialstore.update_creds(preset, sumoregion, accesskeyid, accesskey)
except Exception as e:
logger.exception(e)
self.errorbox('Something went wrong\n\n' + str(e))
else:
self.errorbox('Something went wrong. That preset does not exist in the database.')
return
if monarch.pcf.util.cf_target(org, space):
logger.error("Failed to target org %s and space %s!", org, space)
return None
app = App(org, space, appname)
app.add_services_from_cfg()
if not app.find_guid():
logger.error("App discovery failed because GUID could not be found!")
return None
if not app.find_instances():
logger.error("App discovery failed because no application instances could be found!")
return None
if app.find_services() is None:
logger.error("App discovery failed because there was an error when finding services!")
return None
logger.info("Successfully discovered %s in %s %s.", appname, org, space)
logger.debug(json.dumps(app.serialize(), indent=2))
return app