Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_expired():
# valid
expiration = dt.utcnow() + timedelta(seconds=10)
expiration = expiration.replace(tzinfo=isodate.UTC)
expiration = isodate.strftime(expiration, isodate.DT_EXT_COMPLETE)
agent = Agent(expiration=expiration)
admin = agent.Admin()
print admin.echo('FUTURE')
# invalid
agent = Agent(expiration='12345')
admin = agent.Admin()
print admin.echo('INVALID')
# expired
expiration = dt.utcnow() - timedelta(seconds=10)
expiration = expiration.replace(tzinfo=isodate.UTC)
expiration = isodate.strftime(expiration, isodate.DT_EXT_COMPLETE)
agent = Agent(wait=0, expiration=expiration)
admin = agent.Admin()
print admin.echo('EXPIRED')
sys.exit(0)
def executePost(self, db=True):
if self.procedurename is None:
raise Exception(
"POST action without procedure name not allowed")
from datetime import datetime
now = datetime.now(iso.UTC)
non_blocking_exceptions = []
# path to file for logging requests
dir_name = os.path.dirname(__file__)
file_name = os.path.join(dir_name, 'logs', 'fast_insert_log.csv')
# Create data array
data = self.waEnviron['wsgi_input'].decode().split(";")
data_log = data[:]
# Assigned id always in the first position
assignedid = data[0]
if len(data) == 4: # regular time series
mode = self.MODE_REGULAR
def utc_tz():
"""
Get the UTC timezone.
@rtype: datetime.tzinfo instance
@return: a tzinfo instance representing the utc timezone
"""
return isodate.UTC
def utc_tz():
"""
Get the UTC timezone.
@rtype: datetime.tzinfo instance
@return: a tzinfo instance representing the utc timezone
"""
return isodate.UTC
def utc_tz():
"""
Get the UTC timezone.
@rtype: datetime.tzinfo instance
@return: a tzinfo instance representing the utc timezone
"""
return isodate.UTC
self.last_updated = last_updated or time.time()
self.name = id
self.options = options or {}
self.principal = principal
self.resource = resource
self.schedule = schedule
self.task = task
self.total_run_count = total_run_count
self.kwargs['scheduled_call_id'] = self.id
if first_run is None:
# get the date and time from the iso_schedule value, and if it does not have a date and
# time, use the current date and time
self.first_run = dateutils.format_iso8601_datetime(
dateutils.parse_iso8601_interval(iso_schedule)[1] or
datetime.utcnow().replace(tzinfo=isodate.UTC))
elif isinstance(first_run, datetime):
self.first_run = dateutils.format_iso8601_datetime(first_run)
else:
self.first_run = first_run
if remaining_runs is None:
self.remaining_runs = dateutils.parse_iso8601_interval(iso_schedule)[2]
else:
self.remaining_runs = remaining_runs
self.next_run = self.calculate_next_run()
for importer in collection.find():
modified = False
if config_key not in importer:
importer[config_key] = {}
modified = True
# If the key doesn't exist, or does exist but has no value, set it based on the
# last sync time, if possible. Otherwise, set it to now.
if not importer.get(updated_key, None):
try:
importer[updated_key] = isodate.parse_datetime(importer['last_sync'])
# The attribute doesn't exist, or parsing failed. It's safe to set a newer timestamp.
except: # noqa: 722
importer[updated_key] = datetime.datetime.now(tz=isodate.UTC)
modified = True
if modified:
collection.save(importer)