Create is_event_delete function for resources
Not all resources are marked to be deleted by adding a deleted_at attribute, this change will allow for doing this check on a per-resource basis. Change-Id: I2759038233b4a54661cd8a5fe76c8cb4e983d15d
This commit is contained in:
parent
86dea29889
commit
582b9aa4fd
@ -197,7 +197,7 @@ class Resource(db.Model, GetOrCreateMixin):
|
|||||||
raise exceptions.EventTooOld()
|
raise exceptions.EventTooOld()
|
||||||
|
|
||||||
# If we're deleted, then we close the current period.
|
# If we're deleted, then we close the current period.
|
||||||
if 'deleted_at' in event['traits']:
|
if resource.__class__.is_event_delete(event):
|
||||||
period.ended_at = event['traits']['deleted_at']
|
period.ended_at = event['traits']['deleted_at']
|
||||||
elif period.spec != spec:
|
elif period.spec != spec:
|
||||||
period.ended_at = event['generated']
|
period.ended_at = event['generated']
|
||||||
@ -259,6 +259,11 @@ class Instance(Resource):
|
|||||||
|
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def is_event_delete(cls, event):
|
||||||
|
"""is_event_delete"""
|
||||||
|
return 'deleted_at' in event['traits']
|
||||||
|
|
||||||
|
|
||||||
class BigIntegerDateTime(TypeDecorator):
|
class BigIntegerDateTime(TypeDecorator):
|
||||||
"""BigIntegerDateTime"""
|
"""BigIntegerDateTime"""
|
||||||
|
@ -428,6 +428,15 @@ class TestResource(GetOrCreateTestMixin):
|
|||||||
|
|
||||||
@pytest.mark.usefixtures("db_session")
|
@pytest.mark.usefixtures("db_session")
|
||||||
class TestInstance:
|
class TestInstance:
|
||||||
|
def test_is_event_delete(self):
|
||||||
|
event = fake.get_normalized_event()
|
||||||
|
assert models.Instance.is_event_delete(event) == False
|
||||||
|
|
||||||
|
def test_is_event_delete_for_actual_delete(self):
|
||||||
|
event = fake.get_normalized_event()
|
||||||
|
event['traits']['deleted_at'] = event['generated']
|
||||||
|
assert models.Instance.is_event_delete(event) == True
|
||||||
|
|
||||||
def test_is_event_ignored(self):
|
def test_is_event_ignored(self):
|
||||||
event = fake.get_normalized_event()
|
event = fake.get_normalized_event()
|
||||||
assert models.Instance.is_event_ignored(event) == False
|
assert models.Instance.is_event_ignored(event) == False
|
||||||
|
Loading…
x
Reference in New Issue
Block a user