Python datetime 模块，timedelta() 实例源码
我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用datetime.timedelta()。
def get_jwt(self):
exp = datetime.datetime.utcnow() + datetime.timedelta(minutes=10)
exp = calendar.timegm(exp.timetuple())
# Generate the JWT
payload = {
# issued at time
'iat': int(time.time()),
# JWT expiration time (10 minute maximum)
'exp': exp,
# Integration's GitHub identifier
'iss': options.get('github.integration-app-id'),
}
return jwt.encode(
payload, options.get('github.integration-private-key'), algorithm='RS256'
)
def razzia_wizard(request):
if request.method == 'POST':
return redirect(
reverse("razzia_view") + "?start={0}-{1}-{2}&end={3}-{4}-{5}&products={6}&username=&razzia_title={7}"
.format(int(request.POST['start_year']),
int(request.POST['start_month']),
int(request.POST['start_day']),
int(request.POST['end_year']), int(request.POST['end_month']),
int(request.POST['end_day']),
request.POST.get('products'),
request.POST.get('razzia_title')))
suggested_start_date = timezone.now() - datetime.timedelta(days=-180)
suggested_end_date = timezone.now()
start_date_picker = fields.DateField(
widget=extras.SelectDateWidget(years=[x for x in range(2000, timezone.now().year + 1)]))
end_date_picker = fields.DateField(
widget=extras.SelectDateWidget(years=[x for x in range(2000, timezone.now().year + 1)]))
return render(request, 'admin/stregsystem/razzia/wizard.html',
{
'start_date_picker': start_date_picker.widget.render("start", suggested_start_date),
'end_date_picker': end_date_picker.widget.render("end", suggested_end_date)},
)
def nodeinfo_view(request):
"""Generate a NodeInfo document."""
site = Site.objects.get_current()
usage = {"users": {}}
if settings.SOCIALHOME_STATISTICS:
usage = {
"users": {
"total": User.objects.count(),
"activeHalfyear": User.objects.filter(last_login__gte=now() - datetime.timedelta(days=180)).count(),
"activeMonth": User.objects.filter(last_login__gte=now() - datetime.timedelta(days=30)).count(),
},
"localPosts": Content.objects.filter(author__user__isnull=False, content_type=ContentType.CONTENT).count(),
"localComments": Content.objects.filter(author__user__isnull=False, content_type=ContentType.REPLY).count(),
}
nodeinfo = NodeInfo(
software={"name": "socialhome", "version": version},
protocols={"inbound": ["diaspora"], "outbound": ["diaspora"]},
services={"inbound": [], "outbound": []},
open_registrations=settings.ACCOUNT_ALLOW_REGISTRATION,
usage=usage,
metadata={"nodeName": site.name}
)
return JsonResponse(nodeinfo.doc)
def test_view_responds_stats_on(self):
self.get(NODEINFO_DOCUMENT_PATH)
self.response_200()
self.assertEqual(
json.loads(decode_if_bytes(self.last_response.content))["usage"],
{
"users": {
"total": User.objects.count(),
"activeHalfyear": User.objects.filter(last_login__gte=now() - datetime.timedelta(days=180)).count(),
"activeMonth": User.objects.filter(last_login__gte=now() - datetime.timedelta(days=30)).count(),
},
"localPosts": Content.objects.filter(
author__user__isnull=False, content_type=ContentType.CONTENT).count(),
"localComments": Content.objects.filter(
author__user__isnull=False, content_type=ContentType.REPLY).count(),
}
)
def test_has_support_for_elapsed_time_in_template_render(settings):
settings.PERFORMANCE_LIMITS = {
'Template.render': {
'time': {
'total': 0
}
}
}
template = loader.get_template('all-group-names.markdown')
with freeze_time('2016-09-29 15:52:01') as frozen_time:
class SlowIterable(object):
def __iter__(self):
yield 'foo'
frozen_time.tick(timedelta(seconds=5))
yield 'bar'
with pytest.raises(LimitViolationError) as excinfo:
template.render(context={'groups': SlowIterable()})
assert excinfo.value.context == {'template': ['all-group-names.markdown']}
def test(self):
if self.user_dict['password_enabled'] == 'true' :
last_changed = dateutil.parser.parse(self.user_dict['password_last_changed'])
now = datetime.datetime.utcnow().replace(tzinfo=last_changed.tzinfo)
diff = now - last_changed
delta = datetime.timedelta(
days=config.config['PASSWORD_ROTATION_DAYS']
)
if diff > delta:
self.reason = 'Password has not been changed in {0} days'.format(
delta.days
)
self.status = common.CheckState.FAIL
else:
self.status = common.CheckState.PASS
elif self.user_dict['password_last_changed'] == 'not_supported':
self.reason = 'password_last_changed field is not supported'
self.status = common.CheckState.ERROR
else:
self.reason = 'Password is not enabled'
self.status = common.CheckState.PASS
def key_rotated(self, key_id):
active_key = 'access_key_{0}_active'.format(key_id)
if self.user_dict[active_key] != 'true':
return True #since the key is not active, call it rotated
last_rotated_key = 'access_key_{0}_last_rotated'.format(key_id)
last_rotated = self.user_dict[last_rotated_key]
try:
last_rotated_date = dateutil.parser.parse(last_rotated)
except ValueError as e:
return False #The key has not been rotated so the value is N/A
delta = datetime.timedelta(days=config.config['ACCESS_KEY_ROTATION_DAYS'])
now = datetime.datetime.now().replace(tzinfo=last_rotated_date.tzinfo)
diff = now-last_rotated_date
if diff > delta:
return False
return True
def half_life(issues):
"""Calculate the half life of the service's issues.
Args:
issues (:py:class:`list`): The service's issue data.
Returns:
:py:class:`datetime.timedelta`: The half life of the issues.
"""
lives = []
for issue in issues:
start = safe_parse(issue.get('created_at'))
end = safe_parse(issue.get('closed_at'))
if start and end:
lives.append(end - start)
if lives:
lives.sort()
size = len(lives)
return lives[((size + (size % 2)) // 2) - 1]
def health_summary(self, half_life):
"""Calculate the health of the service.
Args:
half_life (:py:class:`datetime.timedelta`): The half life of
the service's issues.
Returns:
:py:class:`str`: The health of the service, either ``'ok'``,
``'neutral'`` or ``'error'``.
"""
if half_life is None:
return 'neutral'
if half_life <= timedelta(days=self.ok_threshold):
return 'ok'
elif half_life <= timedelta(days=self.neutral_threshold):
return 'neutral'
return 'error'
def test_am_pm_behaviour(self):
check_time = datetime.datetime(
year=2016, month=11, day=7, hour=22,
minute=10, second=0, microsecond=1)
PreHourlyProcessorUtil.get_data_provider().set_last_processed(
date_time=(check_time + datetime.timedelta(hours=-12)))
self.assertTrue(PreHourlyProcessorUtil.is_time_to_run(check_time))
def _time_frame(self, tf):
_fmt = '%Y%m%d%H%M'
delta = {
"last_day": datetime.timedelta(days=1),
"last_3_days": datetime.timedelta(days=3),
"last_week": datetime.timedelta(days=7),
"all": None,
"custom": None,
}
if delta[tf]:
now = datetime.datetime.now()
to_time = now.strftime(_fmt)
from_time = now - delta[tf]
from_time = from_time.strftime(_fmt)
else:
from_time = None
to_time = None
return from_time, to_time
def check_if_media_sync_offset_satisfied(logger, settings, audit):
"""
Check if the media sync offset is satisfied. The media sync offset is a duration in seconds specified in the
configuration file. This duration is the amount of time audit media is given to sync up with SafetyCulture servers
before this tool exports the audit data.
:param logger: The logger
:param settings: Settings from command line and configuration file
:param audit: Audit JSON
:return: Boolean - True if the media sync offset is satisfied, otherwise, returns false.
"""
modified_at = dateutil.parser.parse(audit['modified_at'])
now = datetime.utcnow()
elapsed_time_difference = (pytz.utc.localize(now) - modified_at)
# if the media_sync_offset has been satisfied
if not elapsed_time_difference > timedelta(seconds=settings[MEDIA_SYNC_OFFSET_IN_SECONDS]):
logger.info('Audit {0} modified too recently, some media may not have completed syncing. Skipping export until next sync cycle'.format(
audit['audit_id']))
return False
return True
def uptime(self):
with open('/proc/uptime', 'r') as f:
uptime, idletime = f.readline().split()
up_seconds = int(float(uptime))
idle_seconds = int(float(idletime))
# in some machine like Linode VPS, idle time may bigger than up time
if idle_seconds > up_seconds:
cpu_count = multiprocessing.cpu_count()
idle_seconds = idle_seconds/cpu_count
# in some VPS, this value may still bigger than up time
# may be the domain 0 machine has more cores
# we calclate approximately for it
if idle_seconds > up_seconds:
for n in range(2,10):
if idle_seconds/n < up_seconds:
idle_seconds = idle_seconds/n
break
fmt = '{days} ? {hours} ?? {minutes} ? {seconds} ?'
uptime_string = strfdelta(datetime.timedelta(seconds = up_seconds), fmt)
idletime_string = strfdelta(datetime.timedelta(seconds = idle_seconds), fmt)
return {
'up': uptime_string,
'idle': idletime_string,
'idle_rate': div_percent(idle_seconds, up_seconds),
}
def computeTrends(self, trends):
time = datetime.datetime.now()
t1 = time - datetime.timedelta(minutes=3)
t2 = time - datetime.timedelta(minutes=10)
for value in self.values:
key = "{0}.{1}".format(self.deviceId, value.id)
if key in trends and len(trends[key])>2:
previous = filter(lambda x: x.time > t2 and x.time <= t1, trends[key])
current = filter(lambda x: x.time > t1 and x.time <= time, trends[key])
if len(previous) >= 5 and len(current) >= 2:
previous_values = [x.value for x in previous]
previous_avg = sum(previous_values)/len(previous_values)
current_values = [x.value for x in current]
current_avg = sum(current_values)/len(current_values)
if current_avg > previous_avg + 0.02:
value.trend = 1
if current_avg < previous_avg - 0.02:
value.trend = -1
return self
def calculate_etl(self, downloaded, total, start=None):
"""Calculates the estimated time left, based on how long it took us
to reach "downloaded" and how many messages we have left.
If no start time is given, the time will simply by estimated by how
many chunks are left, which will NOT work if what is being downloaded is media"""
left = total - downloaded
if not start:
# We add chunk size - 1 because division will truncate the decimal places,
# so for example, if we had a chunk size of 8:
# 7 messages + 7 = 14 -> 14 // 8 = 1 chunk download required
# 8 messages + 7 = 15 -> 15 // 8 = 1 chunk download required
# 9 messages + 7 = 16 -> 16 // 8 = 2 chunks download required
#
# Clearly, both 7 and 8 fit in one chunk, but 9 doesn't.
chunks_left = (left + self.download_chunk_size - 1) // self.download_chunk_size
etl = chunks_left * self.download_delay
else:
if downloaded:
delta_time = (datetime.now() - start).total_seconds() / downloaded
etl = left * delta_time
else:
etl = 0
return timedelta(seconds=round(etl, 1))
def tearDown(self):
"""clean up the test
"""
import datetime
import transaction
from stalker import defaults
from stalker.db.session import DBSession
from stalker.db.declarative import Base
from pyramid import testing
testing.tearDown()
# clean up test database
connection = DBSession.connection()
engine = connection.engine
connection.close()
Base.metadata.drop_all(engine)
transaction.commit()
DBSession.remove()
defaults.timing_resolution = datetime.timedelta(hours=1)
def tearDown(self):
"""clean up the test
"""
import datetime
import transaction
from stalker import defaults
from stalker.db.declarative import Base
from stalker.db.session import DBSession
from pyramid import testing
testing.tearDown()
# clean up test database
connection = DBSession.connection()
engine = connection.engine
connection.close()
Base.metadata.drop_all(engine)
transaction.commit()
DBSession.remove()
defaults.timing_resolution = datetime.timedelta(hours=1)
def get_time(cls, request, time_attr):
"""Extracts a time object from the given request
:param request: the request object
:param time_attr: the attribute name
:return: datetime.timedelta
"""
time_part = datetime.datetime.strptime(
request.params[time_attr][:-4],
'%a, %d %b %Y %H:%M:%S'
)
return datetime.timedelta(
hours=time_part.hour,
minutes=time_part.minute
)
def test_create_oauth2_token(self):
admin_user = mixer.blend('auth.User', is_staff=True, is_superuser=True)
app = Application.objects.create(
name='SuperAPI OAUTH2 APP',
user=admin_user,
client_type=Application.CLIENT_PUBLIC,
authorization_grant_type=Application.GRANT_PASSWORD,
)
assert Application.objects.count() == 1, "Should be equal"
random = get_random_string(length=16)
admin_token = AccessToken.objects.create(
user=admin_user,
scope='read write',
# ?? ????? . . .
expires=timezone.now() + timedelta(minutes=5),
token=f'{random}---{admin_user.username}',
application=app
)
assert admin_token is not None, "??? ???"
def next_market_minute(self, start):
"""
Get the next market minute after @start. This is either the immediate
next minute, the open of the same day if @start is before the market
open on a trading day, or the open of the next market day after @start.
"""
if self.is_trading_day(start):
market_open, market_close = self.get_open_and_close(start)
# If start before market open on a trading day, return market open.
if start < market_open:
return market_open
# If start is during trading hours, then get the next minute.
elif start < market_close:
return start + datetime.timedelta(minutes=1)
# If start is not in a trading day, or is after the market close
# then return the open of the *next* trading day.
return self.next_open_and_close(start)[0]
def previous_market_minute(self, start):
"""
Get the next market minute before @start. This is either the immediate
previous minute, the close of the same day if @start is after the close
on a trading day, or the close of the market day before @start.
"""
if self.is_trading_day(start):
market_open, market_close = self.get_open_and_close(start)
# If start after the market close, return market close.
if start > market_close:
return market_close
# If start is during trading hours, then get previous minute.
if start > market_open:
return start - datetime.timedelta(minutes=1)
# If start is not a trading day, or is before the market open
# then return the close of the *previous* trading day.
return self.previous_open_and_close(start)[1]
def create_trade_history(sid, prices, amounts, interval, sim_params, env,
source_id="test_factory"):
trades = []
current = sim_params.first_open
oneday = timedelta(days=1)
use_midnight = interval >= oneday
for price, amount in zip(prices, amounts):
if use_midnight:
trade_dt = current.replace(hour=0, minute=0)
else:
trade_dt = current
trade = create_trade(sid, price, amount, trade_dt, source_id)
trades.append(trade)
current = get_next_trading_dt(current, interval, env)
assert len(trades) == len(prices)
return trades
def setUp(self):
setup_logger(self)
self.sim_params = factory.create_simulation_parameters(num_days=4,
env=self.env)
trade_history = factory.create_trade_history(
133,
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
self.sim_params,
self.env
)
self.source = SpecificEquityTrades(
event_list=trade_history,
env=self.env,
)
self.df_source, self.df = \
factory.create_test_df_source(self.sim_params, self.env)
self.panel_source, self.panel = \
factory.create_test_panel_source(self.sim_params, self.env)
def setUp(self):
setup_logger(self)
self.env = TradingEnvironment()
self.sim_params = factory.create_simulation_parameters(num_days=4,
env=self.env)
self.env.write_data(equities_identifiers=[1, 133])
trade_history = factory.create_trade_history(
1,
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
self.sim_params,
self.env
)
self.source = SpecificEquityTrades(
event_list=trade_history,
env=self.env,
)
self.df_source, self.df = \
factory.create_test_df_source(self.sim_params, self.env)
def setUp(self):
self.sim_params = factory.create_simulation_parameters(num_days=4,
env=self.env)
self.trade_history = factory.create_trade_history(
self.sid,
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
self.sim_params,
self.env
)
self.source = SpecificEquityTrades(
event_list=self.trade_history,
env=self.env,
)
def setUp(self):
self.sim_params = factory.create_simulation_parameters(
num_days=4, env=self.env
)
self.trade_history = factory.create_trade_history(
self.sidint,
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
self.sim_params,
self.env,
)
self.source = SpecificEquityTrades(
event_list=self.trade_history,
env=self.env,
)
def test_iterate_over_rl(self):
sim_params = factory.create_simulation_parameters(
start=list(LEVERAGED_ETFS.keys())[0], num_days=4, env=self.env)
trade_history = factory.create_trade_history(
'BZQ',
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
sim_params,
env=self.env
)
self.source = SpecificEquityTrades(event_list=trade_history,
env=self.env)
algo = IterateRLAlgo(symbol='BZQ', sim_params=sim_params, env=self.env)
algo.run(self.source)
self.assertTrue(algo.found)
def test_algo_without_rl_violation(self):
sim_params = factory.create_simulation_parameters(
start=list(LEVERAGED_ETFS.keys())[0], num_days=4,
env=self.env)
trade_history = factory.create_trade_history(
'AAPL',
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
sim_params,
env=self.env
)
self.source = SpecificEquityTrades(event_list=trade_history,
env=self.env)
algo = RestrictedAlgoWithoutCheck(symbol='AAPL',
sim_params=sim_params,
env=self.env)
algo.run(self.source)
def test_algo_with_rl_violation_after_knowledge_date(self):
sim_params = factory.create_simulation_parameters(
start=list(
LEVERAGED_ETFS.keys())[0] + timedelta(days=7), num_days=5,
env=self.env)
trade_history = factory.create_trade_history(
'BZQ',
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
sim_params,
env=self.env
)
self.source = SpecificEquityTrades(event_list=trade_history,
env=self.env)
algo = RestrictedAlgoWithoutCheck(symbol='BZQ',
sim_params=sim_params,
env=self.env)
with self.assertRaises(TradingControlViolation) as ctx:
algo.run(self.source)
self.check_algo_exception(algo, ctx, 0)
def test_algo_without_rl_violation_after_delete(self):
with security_list_copy():
# add a delete statement removing bzq
# write a new delete statement file to disk
add_security_data([], ['BZQ'])
sim_params = factory.create_simulation_parameters(
start=self.extra_knowledge_date, num_days=3)
trade_history = factory.create_trade_history(
'BZQ',
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
sim_params,
env=self.env,
)
self.source = SpecificEquityTrades(event_list=trade_history,
env=self.env)
algo = RestrictedAlgoWithoutCheck(
symbol='BZQ', sim_params=sim_params, env=self.env
)
algo.run(self.source)
def test_partial_month(self):
start = datetime.datetime(
year=1991,
month=1,
day=1,
hour=0,
minute=0,
tzinfo=pytz.utc)
# 1992 and 1996 were leap years
total_days = 365 * 5 + 2
end = start + datetime.timedelta(days=total_days)
sim_params90s = SimulationParameters(
period_start=start,
period_end=end,
env=self.env,
)
returns = factory.create_returns_from_range(sim_params90s)
returns = returns[:-10] # truncate the returns series to end mid-month
metrics = risk.RiskReport(returns, sim_params90s, env=self.env)
total_months = 60
self.check_metrics(metrics, total_months, start)
def sell_positions(self):
q = Query()
test_func = lambda closed: not closed
docs = self.position_db.search(q.closed.test(test_func))
# Sell and remove position if >1hr old
for doc in docs:
if arrow.get(doc["at"]) < (arrow.now() - datetime.timedelta(hours=1)):
self.logger.log("Trader/Seller", "informative", "Selling position for contract " + doc["contract_id"] + "!")
if self.web_interface.have_position_in_market(doc["contract_id"]):
self.web_interface.sell(doc["contract_id"], doc["side"], doc["amount"])
self.position_db.update({ "closed": True }, eids=[doc.eid])
# Make a trade based on the result
def __init__(
self,
interval_in_seconds,
service_name,
result_dict,
max_delay_seconds,
disable=False
):
super(SensuAlertManager, self).__init__(interval_in_seconds)
self._service_name = service_name
self._setup_ok_result_dict(result_dict)
self._setup_delayed_result_dict()
self._setup_disabled_alert_dict()
self._log = logging.getLogger('{}.util.sensu_alert_manager'.format(service_name))
self._disable = disable
self._should_send_sensu_disabled_message = False
self._max_delay = timedelta(seconds=max_delay_seconds)
def __init__(self, session, api_id, api_hash,
proxy=None, timeout=timedelta(seconds=5)):
"""Initializes the Telegram client with the specified API ID and Hash.
Session must always be a Session instance, and an optional proxy
can also be specified to be used on the connection.
"""
self.session = session
self.api_id = int(api_id)
self.api_hash = api_hash
self.proxy = proxy
self._timeout = timeout
self._logger = logging.getLogger(__name__)
# Cache "exported" senders 'dc_id: TelegramBareClient' and
# their corresponding sessions not to recreate them all
# the time since it's a (somewhat expensive) process.
self._cached_clients = {}
# These will be set later
self.dc_options = None
self._sender = None
# endregion
# region Connecting
def login():
db = UserDb(app.config['LOCAL_DB'])
form = request.form
user = form.get('user')
pwd = form.get('pwd')
password = db.login(user)
del db
if pwd == password:
# ??????
session.permanent = True
# session????
app.permanent_session_lifetime = timedelta(minutes=30)
session.update(dict(user=user))
return render_template('index.html')
elif password is None:
return render_template('login.html', info="??????!")
else:
return render_template('login.html', info="?????!")
def itermonthdates(self, year, month):
"""
Return an iterator for one month. The iterator will yield datetime.date
values and will always iterate through complete weeks, so it will yield
dates outside the specified month.
"""
date = datetime.date(year, month, 1)
# Go back to the beginning of the week
days = (date.weekday() - self.firstweekday) % 7
date -= datetime.timedelta(days=days)
oneday = datetime.timedelta(days=1)
while True:
yield date
date += oneday
if date.month != month and date.weekday() == self.firstweekday:
break
def use_testing_credentials(args, credentials):
print("Skipping AWS API calls because AWSMFA_TESTING_MODE is set.",
file=sys.stderr)
# AWS returns offset-aware UTC times, so we fake that in order to
# verify consistent code paths between py2 and py3 datetime.
fake_expiration = (datetime.datetime.now(tz=pytz.utc) +
datetime.timedelta(minutes=5))
fake_credentials = {
'AccessKeyId': credentials.get(args.identity_profile,
'aws_access_key_id'),
'SecretAccessKey': credentials.get(args.identity_profile,
'aws_secret_access_key'),
'SessionToken': "420",
'Expiration': fake_expiration,
}
print_expiration_time(fake_expiration)
update_credentials_file(args.aws_credentials,
args.target_profile,
args.identity_profile,
credentials,
fake_credentials)
def is_dst(zonename):
"""Check if current time in a time zone is in dst.
From: http://stackoverflow.com/a/19778845/1489738
"""
tz = pytz.timezone(zonename)
now = pytz.utc.localize(datetime.datetime.utcnow())
return now.astimezone(tz).dst() != datetime.timedelta(0)
def fetch_og_preview(content, urls):
"""Fetch first opengraph entry for a list of urls."""
for url in urls:
# See first if recently cached already
if OpenGraphCache.objects.filter(url=url, modified__gte=now() - datetime.timedelta(days=7)).exists():
opengraph = OpenGraphCache.objects.get(url=url)
Content.objects.filter(id=content.id).update(opengraph=opengraph)
return opengraph
try:
og = OpenGraph(url=url, parser="lxml")
except AttributeError:
continue
if not og or ("title" not in og and "site_name" not in og and "description" not in og and "image" not in og):
continue
try:
title = og.title if "title" in og else og.site_name if "site_name" in og else ""
description = og.description if "description" in og else ""
image = og.image if "image" in og and not content.is_nsfw else ""
try:
with transaction.atomic():
opengraph = OpenGraphCache.objects.create(
url=url,
title=truncate_letters(safe_text(title), 250),
description=safe_text(description),
image=safe_text(image),
)
except DataError:
continue
except IntegrityError:
# Some other process got ahead of us
opengraph = OpenGraphCache.objects.get(url=url)
Content.objects.filter(id=content.id).update(opengraph=opengraph)
return opengraph
Content.objects.filter(id=content.id).update(opengraph=opengraph)
return opengraph
return False
def fetch_oembed_preview(content, urls):
"""Fetch first oembed content for a list of urls."""
for url in urls:
# See first if recently cached already
if OEmbedCache.objects.filter(url=url, modified__gte=now()-datetime.timedelta(days=7)).exists():
oembed = OEmbedCache.objects.get(url=url)
Content.objects.filter(id=content.id).update(oembed=oembed)
return oembed
# Fetch oembed
options = {}
if url.startswith("https://twitter.com/"):
# This probably has little effect since we fetch these on the backend...
# But, DNT is always good to communicate if possible :)
options = {"dnt": "true"}
try:
oembed = PyEmbed(discoverer=OEmbedDiscoverer()).embed(url, **options)
except (PyEmbedError, PyEmbedDiscoveryError, PyEmbedConsumerError, ValueError):
continue
if not oembed:
continue
# Ensure width is 100% not fixed
oembed = re.sub(r'width="[0-9]*"', 'width="100%"', oembed)
oembed = re.sub(r'height="[0-9]*"', "", oembed)
try:
with transaction.atomic():
oembed = OEmbedCache.objects.create(url=url, oembed=oembed)
except IntegrityError:
# Some other process got ahead of us
oembed = OEmbedCache.objects.get(url=url)
Content.objects.filter(id=content.id).update(oembed=oembed)
return oembed
Content.objects.filter(id=content.id).update(oembed=oembed)
return oembed
return False
def test_edited_is_true_for_newly_created_content_after_15_minutes_grace_period(self):
with freeze_time(self.public_content.created + datetime.timedelta(minutes=16)):
self.public_content.save()
self.assertTrue(self.public_content.edited)
def test_dict_for_view_edited_post(self):
with freeze_time(self.public_content.created + datetime.timedelta(minutes=16)):
self.public_content.save()
self.assertEqual(self.public_content.dict_for_view(self.user), {
"author": self.public_content.author_id,
"author_guid": self.public_content.author.guid,
"author_handle": self.public_content.author.handle,
"author_home_url": self.public_content.author.home_url,
"author_image": self.public_content.author.safer_image_url_small,
"author_is_local": bool(self.public_content.author.user),
"author_name": self.public_content.author.handle,
"author_profile_url": self.public_content.author.get_absolute_url(),
"content_type": self.public_content.content_type.string_value,
"delete_url": reverse("content:delete", kwargs={"pk": self.public_content.id}),
"detail_url": self.public_content.get_absolute_url(),
"formatted_timestamp": self.public_content.timestamp,
"guid": self.public_content.guid,
"has_shared": False,
"humanized_timestamp": "%s (edited)" % self.public_content.humanized_timestamp,
"id": self.public_content.id,
"is_authenticated": True,
"is_author": True,
"is_following_author": False,
"parent": "",
"profile_id": self.public_content.author.id,
"rendered": self.public_content.rendered,
"reply_count": 0,
"reply_url": reverse("content:reply", kwargs={"pk": self.public_content.id}),
"shares_count": 0,
"slug": self.public_content.slug,
"through": self.public_content.id,
"update_url": reverse("content:update", kwargs={"pk": self.public_content.id}),
})
def test_if_cached_already_but_older_than_7_days_then_fetch(self, og):
with freeze_time(datetime.date.today() - datetime.timedelta(days=8)):
OpenGraphCacheFactory(url=self.urls[0])
fetch_og_preview(self.content, self.urls)
og.assert_called_once_with(url=self.urls[0], parser="lxml")
def test_cache_updated_if_previous_found_older_than_7_days(self, embed):
with freeze_time(datetime.date.today() - datetime.timedelta(days=8)):
OEmbedCacheFactory(url=self.urls[0])
fetch_oembed_preview(self.content, self.urls)
embed.assert_called_once_with(self.urls[0])
def edited(self):
"""Determine whether Content has been edited.
Because we do multiple saves in some cases on creation, for example for oEmbed or OpenGraph,
and a remote content could be delivered multiple times within a short time period, for example via
relay and original node, we allow 15 minutes before deciding that the content has been edited.
TODO: it would make sense to store an "edited" flag on the model itself.
"""
return self.modified > self.created + datetime.timedelta(minutes=15)
def test_captures_and_measures_elapsed_time(seconds):
with capture_result_collected() as captured:
with freeze_time('2016-09-22 15:57:01') as frozen_time:
with TimeCollector():
frozen_time.tick(timedelta(seconds=seconds))
assert len(captured.calls) == 1
assert pytest.approx(seconds) == captured.calls[0]['results'][0].value
def test_can_limit_elapsed_seconds(seconds):
with freeze_time('2016-09-22 15:57:01') as frozen_time:
with pytest.raises(LimitViolationError) as excinfo:
with TimeLimit(total=0):
frozen_time.tick(timedelta(seconds=seconds))
assert excinfo.value.base_error_msg == \
'Too many ({}) total elapsed seconds (limit: 0)'.format(seconds)
def code_that_fails(self):
self.frozen_time.tick(timedelta(seconds=5))
def __str__(self):
self.frozen_time.tick(timedelta(seconds=self.render_in_seconds))
return 'rendered slowly in {} seconds'.format(
self.render_in_seconds)
def test(self):
last_used_times = []
if self.user_dict['access_key_1_active'] == 'true':
last_used_times.append(
dateutil.parser.parse(
self.user_dict['access_key_1_last_used_date']
)
)
if self.user_dict['access_key_2_active'] == 'true':
last_used_times.append(
dateutil.parser.parse(
self.user_dict['access_key_2_last_used_date']
)
)
if self.user_dict['password_enabled'] in ['true', 'not_supported'] and \
self.user_dict['password_last_used'] != 'no_information':
last_used_times.append(
dateutil.parser.parse(
self.user_dict['password_last_used']
)
)
if len(last_used_times) == 0:
self.reason = 'Account has never been used'
self.status = common.CheckState.FAIL
return
last_used = max(last_used_times)
now = datetime.datetime.utcnow()
now = now.replace(tzinfo=last_used.tzinfo)
delta = datetime.timedelta(days=config.config['ACCOUNT_INACTIVE_DAYS'])
difference = now - last_used
if delta < difference:
self.reason = 'Account last used {0} days ago.'.format(difference.days)
self.status = common.CheckState.FAIL
else:
self.status = common.CheckState.PASS