Skip to content

Instantly share code, notes, and snippets.

@topiaruss
Last active February 25, 2016 16:02
Show Gist options
  • Save topiaruss/01100f5e9da9da85b796 to your computer and use it in GitHub Desktop.
Save topiaruss/01100f5e9da9da85b796 to your computer and use it in GitHub Desktop.
> assert orig == ret
E assert TimedTask(tid..., 13, 665000)) == TimedTask(tid=..., 13, 664999))
E Full diff:
E - TimedTask(tid=UUID('6...27'), created=datetime.datetime(2016, 2, 25, 14, 32, 13, 665000))
E ? ^^^^
E + TimedTask(tid=UUID('6...27'), created=datetime.datetime(2016, 2, 25, 14, 32, 13, 664999))
E ? ^^^^
Note, the time saved was already rounded to milliseconds. The time returned is one ms less than the saved time.
# -*- coding: utf-8 -*-
import uuid
from datetime import datetime
from cassandra.cqlengine import columns
from cassandra.cqlengine import connection
from cassandra.cqlengine.management import (
create_keyspace_simple, drop_keyspace, sync_table
)
from cassandra.cqlengine.models import Model
from sirmodels.settings import CASSANDRA_HOSTS
from sirmodels.settings import CASSANDRA_KEYSPACE
class TimedTask(Model):
tid = columns.UUID(primary_key=True, required=True)
created = columns.DateTime()
def reset_keyspace():
connection.setup(CASSANDRA_HOSTS, CASSANDRA_KEYSPACE,
protocol_version=3)
drop_keyspace(CASSANDRA_KEYSPACE)
create_keyspace_simple(CASSANDRA_KEYSPACE, 1)
def round_now():
"""
return a now with ms granularity
"""
now = datetime.now()
microsecond = now.microsecond
microsecond = microsecond / 1000 * 1000
return datetime.now().replace(microsecond=microsecond)
def test_timed_task():
reset_keyspace()
model = TimedTask
sync_table(model)
for i in range(1000):
print i
now = round_now()
tid = uuid.uuid4()
orig = model.create(tid=tid, created=now)
ret = model.objects.get(tid=tid)
assert orig == ret
print '1000 ok'
assert 0
@topiaruss
Copy link
Author

Caused by commit: datastax/python-driver@a98078b ??

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment